Fix metadata tuple parsing (#1190)

* Fix metadata tuple parsing

* adding failing tests for tuple and map in getCassandraType

helpers.go

@@ -132,14 +132,20 @@ func getCassandraType(name string) TypeInfo {
 			Elem:       getCassandraType(strings.TrimPrefix(name[:len(name)-1], "list<")),
 		}
 	} else if strings.HasPrefix(name, "map<") {
-		names := strings.SplitN(strings.TrimPrefix(name[:len(name)-1], "map<"), ", ", 2)
+		names := splitCompositeTypes(strings.TrimPrefix(name[:len(name)-1], "map<"))
+		if len(names) != 2 {
+			Logger.Printf("Error parsing map type, it has %d subelements, expecting 2\n", len(names))
+			return NativeType{
+				typ: TypeCustom,
+			}
+		}
 		return CollectionType{
 			NativeType: NativeType{typ: TypeMap},
 			Key:        getCassandraType(names[0]),
 			Elem:       getCassandraType(names[1]),
 		}
 	} else if strings.HasPrefix(name, "tuple<") {
-		names := strings.Split(strings.TrimPrefix(name[:len(name)-1], "tuple<"), ", ")
+		names := splitCompositeTypes(strings.TrimPrefix(name[:len(name)-1], "tuple<"))
 		types := make([]TypeInfo, len(names))
 
 		for i, name := range names {
@@ -157,6 +163,34 @@ func getCassandraType(name string) TypeInfo {
 	}
 }
 
+func splitCompositeTypes(name string) []string {
+	if !strings.Contains(name, "<") {
+		return strings.Split(name, ", ")
+	}
+	var parts []string
+	lessCount := 0
+	segment := ""
+	for _, char := range name {
+		if char == ',' && lessCount == 0 {
+			if segment != "" {
+				parts = append(parts, strings.TrimSpace(segment))
+			}
+			segment = ""
+			continue
+		}
+		segment += string(char)
+		if char == '<' {
+			lessCount++
+		} else if char == '>' {
+			lessCount--
+		}
+	}
+	if segment != "" {
+		parts = append(parts, strings.TrimSpace(segment))
+	}
+	return parts
+}
+
 func getApacheCassandraType(class string) Type {
 	switch strings.TrimPrefix(class, apacheCassandraTypePrefix) {
 	case "AsciiType":

helpers_test.go

@@ -77,6 +77,100 @@ func TestGetCassandraType(t *testing.T) {
 				},
 			},
 		},
+		{
+			"frozen<tuple<frozen<tuple<text, frozen<list<frozen<tuple<int, int>>>>>>, frozen<tuple<text, frozen<list<frozen<tuple<int, int>>>>>>,  frozen<map<text, frozen<list<frozen<tuple<int, int>>>>>>>>",
+			TupleTypeInfo{
+				NativeType: NativeType{typ: TypeTuple},
+				Elems: []TypeInfo{
+					TupleTypeInfo{
+						NativeType: NativeType{typ: TypeTuple},
+						Elems: []TypeInfo{
+							NativeType{typ: TypeText},
+							CollectionType{
+								NativeType: NativeType{typ: TypeList},
+								Elem: TupleTypeInfo{
+									NativeType: NativeType{typ: TypeTuple},
+									Elems: []TypeInfo{
+										NativeType{typ: TypeInt},
+										NativeType{typ: TypeInt},
+									},
+								},
+							},
+						},
+					},
+					TupleTypeInfo{
+						NativeType: NativeType{typ: TypeTuple},
+						Elems: []TypeInfo{
+							NativeType{typ: TypeText},
+							CollectionType{
+								NativeType: NativeType{typ: TypeList},
+								Elem: TupleTypeInfo{
+									NativeType: NativeType{typ: TypeTuple},
+									Elems: []TypeInfo{
+										NativeType{typ: TypeInt},
+										NativeType{typ: TypeInt},
+									},
+								},
+							},
+						},
+					},
+					CollectionType{
+						NativeType: NativeType{typ: TypeMap},
+						Key:        NativeType{typ: TypeText},
+						Elem: CollectionType{
+							NativeType: NativeType{typ: TypeList},
+							Elem: TupleTypeInfo{
+								NativeType: NativeType{typ: TypeTuple},
+								Elems: []TypeInfo{
+									NativeType{typ: TypeInt},
+									NativeType{typ: TypeInt},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			"frozen<tuple<frozen<tuple<int, int>>, int, frozen<tuple<int, int>>>>", TupleTypeInfo{
+				NativeType: NativeType{typ: TypeTuple},
+
+				Elems: []TypeInfo{
+					TupleTypeInfo{
+						NativeType: NativeType{typ: TypeTuple},
+
+						Elems: []TypeInfo{
+							NativeType{typ: TypeInt},
+							NativeType{typ: TypeInt},
+						},
+					},
+					NativeType{typ: TypeInt},
+					TupleTypeInfo{
+						NativeType: NativeType{typ: TypeTuple},
+
+						Elems: []TypeInfo{
+							NativeType{typ: TypeInt},
+							NativeType{typ: TypeInt},
+						},
+					},
+				},
+			},
+		},
+		{
+			"frozen<map<frozen<tuple<int, int>>, int>>", CollectionType{
+				NativeType: NativeType{typ: TypeMap},
+
+				Key: TupleTypeInfo{
+					NativeType: NativeType{typ: TypeTuple},
+
+					Elems: []TypeInfo{
+						NativeType{typ: TypeInt},
+						NativeType{typ: TypeInt},
+					},
+				},
+				Elem: NativeType{typ: TypeInt},
+			},
+		},
 	}
 
 	for _, test := range tests {