String column as keyword field

cmd/experimental/main.go

@@ -165,7 +165,7 @@ func main() {
 
 	virtualTableStorage := persistence.NewElasticJSONDatabase(cfg.Elasticsearch, common_table.VirtualTableElasticIndexName)
 	tableDisco := clickhouse.NewTableDiscovery(&cfg, connectionPool, virtualTableStorage)
-	schemaRegistry := schema.NewSchemaRegistry(clickhouse.TableDiscoveryTableProviderAdapter{TableDiscovery: tableDisco}, &cfg, clickhouse.SchemaTypeAdapter{})
+	schemaRegistry := schema.NewSchemaRegistry(clickhouse.TableDiscoveryTableProviderAdapter{TableDiscovery: tableDisco}, &cfg, clickhouse.NewSchemaTypeAdapter(cfg.StringColumnIsKeywordField))
 	schemaRegistry.Start()
 
 	im := elasticsearch.NewIndexManagement(cfg.Elasticsearch)

cmd/main.go

@@ -88,7 +88,7 @@ func main() {
 
 	virtualTableStorage := persistence.NewElasticJSONDatabase(cfg.Elasticsearch, common_table.VirtualTableElasticIndexName)
 	tableDisco := clickhouse.NewTableDiscovery(&cfg, connectionPool, virtualTableStorage)
-	schemaRegistry := schema.NewSchemaRegistry(clickhouse.TableDiscoveryTableProviderAdapter{TableDiscovery: tableDisco}, &cfg, clickhouse.SchemaTypeAdapter{})
+	schemaRegistry := schema.NewSchemaRegistry(clickhouse.TableDiscoveryTableProviderAdapter{TableDiscovery: tableDisco}, &cfg, clickhouse.NewSchemaTypeAdapter(cfg.StringColumnIsKeywordField))
 	schemaRegistry.Start()
 
 	im := elasticsearch.NewIndexManagement(cfg.Elasticsearch)

platform/clickhouse/type_adapter.go

@@ -8,6 +8,14 @@ import (
 )
 
 type SchemaTypeAdapter struct {
+	stringColumnIsKeywordField bool
+}
+
+func NewSchemaTypeAdapter(stringColumnIsKeywordField bool) SchemaTypeAdapter {
+
+	return SchemaTypeAdapter{
+		stringColumnIsKeywordField: stringColumnIsKeywordField,
+	}
 }
 
 func (c SchemaTypeAdapter) Convert(s string) (schema.QuesmaType, bool) {
@@ -23,7 +31,12 @@ func (c SchemaTypeAdapter) Convert(s string) (schema.QuesmaType, bool) {
 
 	switch s {
 	case "String":
-		return schema.QuesmaTypeText, true
+		if c.stringColumnIsKeywordField {
+			return schema.QuesmaTypeKeyword, true
+		} else {
+			return schema.QuesmaTypeText, true
+		}
+
 	case "LowCardinality(String)", "UUID", "FixedString":
 		return schema.QuesmaTypeKeyword, true
 	case "Int", "Int8", "Int16", "Int32", "Int64":

platform/config/config.go

@@ -57,6 +57,7 @@ type QuesmaConfiguration struct {
 	DefaultIngestOptimizers     map[string]OptimizerConfiguration
 	DefaultQueryOptimizers      map[string]OptimizerConfiguration
 	MapFieldsDiscoveringEnabled bool
+	StringColumnIsKeywordField  bool
 }
 
 func NewQuesmaConfigurationIndexConfigOnly(indexConfig map[string]IndexConfiguration) QuesmaConfiguration {
@@ -269,7 +270,8 @@ Quesma Configuration:
 	UseCommonTableForWildcard: %t,
 	DefaultIngestTarget: %v,
 	DefaultQueryTarget: %v,
-	MapFieldsDiscoveringEnabled: %t
+	MapFieldsDiscoveringEnabled: %t,
+	StringColumnIsKeywordField: %t
 `,
 		c.TransparentProxy,
 		elasticUrl,
@@ -292,6 +294,7 @@ Quesma Configuration:
 		c.DefaultIngestTarget,
 		c.DefaultQueryTarget,
 		c.MapFieldsDiscoveringEnabled,
+		c.StringColumnIsKeywordField,
 	)
 }
 

platform/config/config_v2.go

@@ -49,6 +49,7 @@ type QuesmaNewConfiguration struct {
 	Pipelines                   []Pipeline           `koanf:"pipelines"`
 	DisableTelemetry            bool                 `koanf:"disableTelemetry"`
 	MapFieldsDiscoveringEnabled bool                 `koanf:"mapFieldsDiscoveringEnabled"`
+	StringColumnIsKeywordField  bool                 `koanf:"stringColumnIsKeywordField"`
 }
 
 type LoggingConfiguration struct {
@@ -591,6 +592,7 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 	conf.LicenseKey = c.LicenseKey
 
 	conf.MapFieldsDiscoveringEnabled = c.MapFieldsDiscoveringEnabled
+	conf.StringColumnIsKeywordField = c.StringColumnIsKeywordField
 
 	conf.AutodiscoveryEnabled = false
 	conf.Connectors = make(map[string]RelationalDbConfiguration)

platform/config/config_v2_test.go

@@ -359,3 +359,27 @@ func TestPartitionBy(t *testing.T) {
 
 	assert.Equal(t, Hourly, legacyConf.DefaultPartitioningStrategy)
 }
+
+func TestStringColumnIsTextDefaultBehavior(t *testing.T) {
+	os.Setenv(configFileLocationEnvVar, "./test_configs/partition_by.yaml")
+	cfg := LoadV2Config()
+	if err := cfg.Validate(); err != nil {
+		t.Fatalf("error validating config: %v", err)
+	}
+	legacyConf := cfg.TranslateToLegacyConfig()
+
+	assert.Equal(t, false, legacyConf.StringColumnIsKeywordField)
+
+}
+
+func TestStringColumnIsKeyword(t *testing.T) {
+	os.Setenv(configFileLocationEnvVar, "./test_configs/string_column_is_keyword_field.yaml")
+	cfg := LoadV2Config()
+	if err := cfg.Validate(); err != nil {
+		t.Fatalf("error validating config: %v", err)
+	}
+	legacyConf := cfg.TranslateToLegacyConfig()
+
+	assert.Equal(t, true, legacyConf.StringColumnIsKeywordField)
+
+}

platform/config/test_configs/string_column_is_keyword_field.yaml

@@ -0,0 +1,68 @@
+# TEST CONFIGURATION
+licenseKey: "cdd749a3-e777-11ee-bcf8-0242ac150004"
+stringColumnIsKeywordField: true
+frontendConnectors:
+  - name: elastic-ingest
+    type: elasticsearch-fe-ingest
+    config:
+      listenPort: 8080
+  - name: elastic-query
+    type: elasticsearch-fe-query
+    config:
+      listenPort: 8080
+backendConnectors:
+  - name: my-minimal-elasticsearch
+    type: elasticsearch
+    config:
+      url: "http://localhost:9200"
+  - name: my-clickhouse-data-source
+    type: clickhouse-os
+    config:
+      url: "clickhouse://localhost:9000"
+ingestStatistics: true
+internalTelemetryUrl: "https://api.quesma.com/phone-home"
+logging:
+  remoteUrl: "https://api.quesma.com/phone-home"
+  path: "logs"
+  level: "info"
+processors:
+  - name: my-query-processor
+    type: quesma-v1-processor-query
+    config:
+      indexes:
+        example-index:
+          target:
+            - my-clickhouse-data-source
+        kibana_sample_data_ecommerce:
+          target:
+            - my-clickhouse-data-source
+          partitioningStrategy: daily
+        "*":
+          target:
+            - my-minimal-elasticsearch
+          partitioningStrategy: hourly
+  - name: my-ingest-processor
+    type: quesma-v1-processor-ingest
+    config:
+      indexes:
+        example-index:
+          target:
+            - my-clickhouse-data-source
+        kibana_sample_data_ecommerce:
+          target:
+            - my-clickhouse-data-source
+          partitioningStrategy: daily
+        "*":
+          target:
+            - my-minimal-elasticsearch
+          partitioningStrategy: hourly
+pipelines:
+  - name: my-pipeline-elasticsearch-query-clickhouse
+    frontendConnectors: [ elastic-query ]
+    processors: [ my-query-processor ]
+    backendConnectors: [ my-minimal-elasticsearch, my-clickhouse-data-source ]
+  - name: my-pipeline-elasticsearch-ingest-to-clickhouse
+    frontendConnectors: [ elastic-ingest ]
+    processors: [ my-ingest-processor ]
+    backendConnectors: [ my-minimal-elasticsearch, my-clickhouse-data-source ]
+