Map fields discovery fix, additional logging (#1368)

This PR fixes the addition of a map field to the schema in case of
subsequent query failures and adds more logging.


<!-- 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' -->

Author: pdelewski

platform/clickhouse/table_discovery.go

@@ -631,40 +631,58 @@ func (td *tableDiscovery) enrichTableWithMapFields(inputTable map[string]map[str
 	for table, columns := range inputTable {
 		for colName, columnMeta := range columns {
 			if strings.HasPrefix(columnMeta.colType, "Map(String") {
-				// Query ClickHouse for map keys in the given column
-				rows, err := td.dbConnPool.Query(context.Background(), fmt.Sprintf("SELECT arrayJoin(mapKeys(%s)) FROM %s", colName, table))
-				if err != nil {
-					fmt.Println("Error querying map keys:", err)
-					continue
-				}
-
+				logger.Debug().Msgf("Discovered map column: %s.%s", table, colName)
 				// Ensure the table exists in outputTable
 				if _, ok := outputTable[table]; !ok {
 					outputTable[table] = make(map[string]columnMetadata)
 				}
+				if _, ok := outputTable[table][colName]; !ok {
+					// Update origin for incoming map column
+					columnMeta.origin = schema.FieldSourceIngest
+					outputTable[table][colName] = columnMeta
+					logger.Debug().Msgf("Added column: %s.%s", table, colName)
+				}
 
-				// Process returned keys and add them as virtual columns
+				// Query ClickHouse for map keys in the given column
+				rows, err := td.dbConnPool.Query(context.Background(), fmt.Sprintf("SELECT arrayJoin(mapKeys(%s)) FROM %s", colName, table))
+				if err != nil {
+					logger.Error().Msgf("Error querying map keys for table, column: %s, %s, %v", table, colName, err)
+					continue
+				}
+				foundKeys := false
 				for rows.Next() {
+					foundKeys = true
 					var key string
 					if err := rows.Scan(&key); err != nil {
-						fmt.Println("Error scanning key:", err)
+						logger.Error().Msgf("Error scanning key for table, column: %s, %s, %v", table, colName, err)
 						continue
 					}
-					// Update origin for incoming map column
-					columnMeta.origin = schema.FieldSourceIngest
-					outputTable[table][colName] = columnMeta
 					// Add virtual column for each key in the map
 					// with origin set to mapping
-					virtualColName := colName + "." + key
-					valueType, err := extractMapValueType(columnMeta.colType)
-					if err == nil {
-						outputTable[table][virtualColName] = columnMetadata{
+					mapKeyCol := colName + "." + key
+					var valueType string
+					valueType, err = extractMapValueType(columnMeta.colType)
+					if err != nil {
+						logger.Error().Msgf("Error extracting value type for table, column: %s, %s, %v", table, colName, err)
+						continue
+					} else {
+						outputTable[table][mapKeyCol] = columnMetadata{
 							colType: valueType,
 							origin:  schema.FieldSourceMapping,
 						}
+						logger.Debug().Msgf("Added map key column: %s.%s", table, mapKeyCol)
 					}
 				}
-				rows.Close() // Close after processing
+				if !foundKeys {
+					logger.Debug().Msgf("No map keys found for table, column: %s, %s", table, colName)
+				}
+				if err := rows.Err(); err != nil {
+					logger.Error().Msgf("Error iterating map keys for %s.%s: %v", table, colName, err)
+				}
+				err = rows.Close() // Close after processing
+				if err != nil {
+					logger.Error().Msgf("Error closing rows for table, column: %s, %s, %v", table, colName, err)
+				}
 			} else {
 				// Copy other columns as-is
 				if _, ok := outputTable[table]; !ok {

platform/config/index_config.go

@@ -75,6 +75,7 @@ func (c IndexConfiguration) String(indexName string) string {
 	if c.UseCommonTable {
 		builder.WriteString(", useSingleTable: true")
 	}
+	builder.WriteString(fmt.Sprintf(", enableFieldMapSyntax: %v", c.EnableFieldMapSyntax))
 
 	return builder.String()
 }