Extending map keys types discovering (#1378)

This PR:
- handles `Map(LowCardinality(String),...)`
- fixes `extractMapValueType`

<!-- A note on testing your PR -->
<!-- Basic unit test run is executed against each commit in the PR.
If you want to run a full integration test suite, you can trigger it by
commenting
     with '/run-integration-tests' or '/run-it' -->

Author: pdelewski

platform/clickhouse/schema.go

@@ -240,7 +240,7 @@ func ResolveType(clickHouseTypeName string) reflect.Type {
 		return reflect.TypeOf(true)
 	case "JSON":
 		return reflect.TypeOf(map[string]interface{}{})
-	case "Map(String, Nullable(String))", "Map(String, String)", "Map(LowCardinality(String), String)":
+	case "Map(String, Nullable(String))", "Map(String, String)", "Map(LowCardinality(String), String)", "Map(LowCardinality(String), Nullable(String))":
 		return reflect.TypeOf(map[string]string{})
 	case "Unknown":
 		return reflect.TypeOf(UnknownType{})

platform/clickhouse/table_discovery.go

@@ -614,23 +614,27 @@ func removePrecision(str string) string {
 }
 
 // extractMapValueType extracts the value type from a ClickHouse Map definition
+// extractMapValueType extracts the value type from a ClickHouse Map definition.
 func extractMapValueType(mapType string) (string, error) {
-	// Regular expression to match Map(String, valueType)
-	re := regexp.MustCompile(`Map\(String,\s*([^)]+)\)`)
-	matches := re.FindStringSubmatch(mapType)
+	// Regular expression to match "Map(String, <valueType>)"
+	re := regexp.MustCompile(`Map\((?:LowCardinality\()?String\)?,\s*(.+)\)$`)
 
+	matches := re.FindStringSubmatch(mapType)
 	if len(matches) < 2 {
-		return "", fmt.Errorf("invalid map type format: %s", mapType)
+		return "", errors.New("invalid map type format: " + mapType)
 	}
 
+	// Trim spaces and return the full value type
 	return strings.TrimSpace(matches[1]), nil
 }
 
 func (td *tableDiscovery) enrichTableWithMapFields(inputTable map[string]map[string]columnMetadata) map[string]map[string]columnMetadata {
 	outputTable := make(map[string]map[string]columnMetadata)
 	for table, columns := range inputTable {
 		for colName, columnMeta := range columns {
-			if strings.HasPrefix(columnMeta.colType, "Map(String") {
+			columnType := strings.TrimSpace(columnMeta.colType)
+			if strings.HasPrefix(columnType, "Map(String") ||
+				strings.HasPrefix(columnType, "Map(LowCardinality(String") {
 				logger.Debug().Msgf("Discovered map column: %s.%s", table, colName)
 				// Ensure the table exists in outputTable
 				if _, ok := outputTable[table]; !ok {
@@ -661,7 +665,7 @@ func (td *tableDiscovery) enrichTableWithMapFields(inputTable map[string]map[str
 					// with origin set to mapping
 					mapKeyCol := colName + "." + key
 					var valueType string
-					valueType, err = extractMapValueType(columnMeta.colType)
+					valueType, err = extractMapValueType(columnType)
 					if err != nil {
 						logger.Error().Msgf("Error extracting value type for table, column: %s, %s, %v", table, colName, err)
 						continue

platform/clickhouse/table_discovery_test.go

@@ -303,3 +303,35 @@ func Test_resolveColumn_Nullable(t *testing.T) {
 		})
 	}
 }
+
+func TestExtractMapValueType(t *testing.T) {
+	tests := []struct {
+		input    string
+		expected string
+		hasError bool
+	}{
+		{"Map(String, Int32)", "Int32", false},
+		{"Map(String, Nullable(String))", "Nullable(String)", false},
+		{"Map(String, Array(UInt8))", "Array(UInt8)", false},
+		{"Map(LowCardinality(String), UInt64)", "UInt64", false},
+		{"Map(LowCardinality(String), Decimal(10,2))", "Decimal(10,2)", false},
+		{"Map(LowCardinality(String), FixedString(16))", "FixedString(16)", false},
+		{"Map(String,)", "", true},                                       // Invalid format
+		{"Map(LowCardinality(String),)", "", true},                       // Invalid format
+		{"Map(String)", "", true},                                        // Missing value type
+		{"RandomType(String, Int32)", "", true},                          // Not a Map type
+		{"Map(String, Map(String, Int32))", "Map(String, Int32)", false}, // Nested map
+	}
+
+	for _, test := range tests {
+		t.Run(test.input, func(t *testing.T) {
+			result, err := extractMapValueType(test.input)
+			if (err != nil) != test.hasError {
+				t.Errorf("unexpected error state for input %q: got error %v", test.input, err)
+			}
+			if result != test.expected {
+				t.Errorf("expected %q, got %q for input %q", test.expected, result, test.input)
+			}
+		})
+	}
+}