Introducing map keys discovering (#1364)

This PR adds an automatic map key discovery feature needed for field
caps and dashboard creation.

```
CREATE TABLE IF NOT EXISTS "foo"
(
    "@timestamp" DateTime64 DEFAULT now64(),
    `bar` Map(String, Nullable(String))
)
ENGINE = MergeTree
ORDER BY ("@timestamp");    

insert into foo values ('2020-01-01 00:00:00', {'a': 'b'});
insert into foo values ('2020-01-01 00:00:00', {'c': 'b'});
insert into foo values ('2020-01-01 00:00:00', {'e': 'b'});
```

Feature is hidden behind the global flag:

```
mapFieldsDiscoveringEnabled: true
```

<img width="1688" alt="image"
src="https://github.com/user-attachments/assets/0ca89753-907d-4bb0-b536-39a735b06742"
/>


<!-- 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/schema.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/QuesmaOrg/quesma/platform/logger"
+	"github.com/QuesmaOrg/quesma/platform/schema"
 	"github.com/QuesmaOrg/quesma/platform/util"
 	"math"
 	"reflect"
@@ -60,6 +61,7 @@ type (
 		Modifiers string
 		Codec     Codec // TODO currently not used, it's part of Modifiers
 		Comment   string
+		Origin    schema.FieldSource // TODO this field is just added to have way to forward information to the schema registry and should be considered as a technical debt
 	}
 	DateTimeType int
 )

platform/clickhouse/table_discovery.go

@@ -16,6 +16,7 @@ import (
 	"github.com/QuesmaOrg/quesma/platform/util"
 	quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
 	"github.com/goccy/go-json"
+	"regexp"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -67,6 +68,7 @@ type columnMetadata struct {
 	// we use it as persistent storage and load it
 	// in the case when we don't control ingest
 	comment string
+	origin  schema.FieldSource // TODO this field is just added to have way to forward information to the schema registry and should be considered as a technical debt
 }
 
 func NewTableDiscovery(cfg *config.QuesmaConfiguration, dbConnPool quesma_api.BackendConnector, virtualTablesDB persistence.JSONDatabase) TableDiscovery {
@@ -115,6 +117,7 @@ func (t TableDiscoveryTableProviderAdapter) TableDefinitions() map[string]schema
 				Name:    column.Name,
 				Type:    column.Type.String(),
 				Comment: column.Comment,
+				Origin:  column.Origin,
 			}
 		}
 		table.DatabaseName = value.DatabaseName
@@ -205,6 +208,9 @@ func (td *tableDiscovery) ReloadTableDefinitions() {
 		td.tableDefinitionsLastReloadUnixSec.Store(time.Now().Unix())
 		return
 	} else {
+		if td.cfg.MapFieldsDiscoveringEnabled {
+			tables = td.enrichTableWithMapFields(tables)
+		}
 		if td.AutodiscoveryEnabled() {
 			configuredTables = td.autoConfigureTables(tables, databaseName)
 		} else {
@@ -379,6 +385,7 @@ func (td *tableDiscovery) populateTableDefinitions(configuredTables map[string]d
 			column := resolveColumn(col, columnMeta.colType)
 			if column != nil {
 				column.Comment = columnMeta.comment
+				column.Origin = columnMeta.origin
 				columnsMap[col] = column
 			} else {
 				logger.Warn().Msgf("column '%s.%s' type: '%s' not resolved. table will be skipped", tableName, col, columnMeta.colType)
@@ -606,6 +613,70 @@ func removePrecision(str string) string {
 	}
 }
 
+// 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)
+
+	if len(matches) < 2 {
+		return "", fmt.Errorf("invalid map type format: %s", mapType)
+	}
+
+	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") {
+				// 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
+				}
+
+				// Ensure the table exists in outputTable
+				if _, ok := outputTable[table]; !ok {
+					outputTable[table] = make(map[string]columnMetadata)
+				}
+
+				// Process returned keys and add them as virtual columns
+				for rows.Next() {
+					var key string
+					if err := rows.Scan(&key); err != nil {
+						fmt.Println("Error scanning key:", 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{
+							colType: valueType,
+							origin:  schema.FieldSourceMapping,
+						}
+					}
+				}
+				rows.Close() // Close after processing
+			} else {
+				// Copy other columns as-is
+				if _, ok := outputTable[table]; !ok {
+					outputTable[table] = make(map[string]columnMetadata)
+				}
+				outputTable[table][colName] = columnMeta
+			}
+		}
+	}
+	return outputTable
+}
+
 func (td *tableDiscovery) readTables(database string) (map[string]map[string]columnMetadata, error) {
 	logger.Debug().Msgf("describing tables: %s", database)
 

platform/config/config.go

@@ -56,6 +56,7 @@ type QuesmaConfiguration struct {
 	DefaultQueryTarget          []string
 	DefaultIngestOptimizers     map[string]OptimizerConfiguration
 	DefaultQueryOptimizers      map[string]OptimizerConfiguration
+	MapFieldsDiscoveringEnabled bool
 }
 
 func (c *QuesmaConfiguration) AliasFields(indexName string) map[string]string {
@@ -264,6 +265,7 @@ Quesma Configuration:
 	UseCommonTableForWildcard: %t,
 	DefaultIngestTarget: %v,
 	DefaultQueryTarget: %v,
+	MapFieldsDiscoveringEnabled: %t
 `,
 		c.TransparentProxy,
 		elasticUrl,
@@ -285,6 +287,7 @@ Quesma Configuration:
 		c.UseCommonTableForWildcard,
 		c.DefaultIngestTarget,
 		c.DefaultQueryTarget,
+		c.MapFieldsDiscoveringEnabled,
 	)
 }
 

platform/config/config_v2.go

@@ -39,15 +39,16 @@ const (
 )
 
 type QuesmaNewConfiguration struct {
-	BackendConnectors  []BackendConnector   `koanf:"backendConnectors"`
-	FrontendConnectors []FrontendConnector  `koanf:"frontendConnectors"`
-	InstallationId     string               `koanf:"installationId"`
-	LicenseKey         string               `koanf:"licenseKey"`
-	Logging            LoggingConfiguration `koanf:"logging"`
-	IngestStatistics   bool                 `koanf:"ingestStatistics"`
-	Processors         []Processor          `koanf:"processors"`
-	Pipelines          []Pipeline           `koanf:"pipelines"`
-	DisableTelemetry   bool                 `koanf:"disableTelemetry"`
+	BackendConnectors           []BackendConnector   `koanf:"backendConnectors"`
+	FrontendConnectors          []FrontendConnector  `koanf:"frontendConnectors"`
+	InstallationId              string               `koanf:"installationId"`
+	LicenseKey                  string               `koanf:"licenseKey"`
+	Logging                     LoggingConfiguration `koanf:"logging"`
+	IngestStatistics            bool                 `koanf:"ingestStatistics"`
+	Processors                  []Processor          `koanf:"processors"`
+	Pipelines                   []Pipeline           `koanf:"pipelines"`
+	DisableTelemetry            bool                 `koanf:"disableTelemetry"`
+	MapFieldsDiscoveringEnabled bool                 `koanf:"mapFieldsDiscoveringEnabled"`
 }
 
 type LoggingConfiguration struct {
@@ -576,6 +577,8 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 	conf.InstallationId = c.InstallationId
 	conf.LicenseKey = c.LicenseKey
 
+	conf.MapFieldsDiscoveringEnabled = c.MapFieldsDiscoveringEnabled
+
 	conf.AutodiscoveryEnabled = false
 	conf.Connectors = make(map[string]RelationalDbConfiguration)
 	relDBConn, connType, relationalDBErr := c.getRelationalDBConf()

platform/schema/registry.go

@@ -65,6 +65,7 @@ type (
 		Name    string
 		Type    string // FIXME: change to schema.Type
 		Comment string
+		Origin  FieldSource // TODO this field is just added to have way to forward information to the schema registry and should be considered as a technical debt
 	}
 )
 
@@ -327,7 +328,7 @@ func (s *schemaRegistry) populateSchemaFromTableDefinition(definitions map[strin
 			}
 			if existing, exists := fields[propertyName]; !exists {
 				if quesmaType, resolved := s.dataSourceTypeAdapter.Convert(column.Type); resolved {
-					fields[propertyName] = Field{PropertyName: propertyName, InternalPropertyName: FieldName(column.Name), InternalPropertyType: column.Type, Type: quesmaType}
+					fields[propertyName] = Field{PropertyName: propertyName, InternalPropertyName: FieldName(column.Name), InternalPropertyType: column.Type, Type: quesmaType, Origin: column.Origin}
 				} else {
 					logger.Debug().Msgf("type %s not supported, falling back to keyword", column.Type)
 					fields[propertyName] = Field{PropertyName: propertyName, InternalPropertyName: FieldName(column.Name), InternalPropertyType: column.Type, Type: QuesmaTypeKeyword}