UseCommonTable global per processor (#962)

This PR moves global `useCommonTable` to processor level (under config)
instead of fully global :
```
processors:
  - name: my-query-processor
    type: quesma-v1-processor-query
    config:
      useCommonTable: true
```

Author: pdelewski

ci/it/configs/quesma-with-dual-writes-and-common-table.yml.template

@@ -33,11 +33,13 @@ processors:
         logs-dual-query:
           target: [ c, e ]
         logs-4:
-          useCommonTable: true
-          target: [ c ]
+          target:
+            - c:
+                useCommonTable: true
         logs-5:
-          useCommonTable: true
-          target: [ c ]
+          target:
+            - c:
+                useCommonTable: true
         "*":
           target: [ e ]
   - name: IP
@@ -53,13 +55,13 @@ processors:
         logs-dual-query:
           target: [ c, e ]
         logs-4:
-          useCommonTable: true
-          target: [ c ]
+          target:
+            - c:
+                useCommonTable: true
+        logs-5:
+          target:
         "*":
           target: [ e ]
-        logs-5:
-          useCommonTable: true
-          target: [  ]
 
 pipelines:
   - name: my-elasticsearch-proxy-read

quesma/quesma/config/config_v2.go

@@ -112,7 +112,8 @@ const DefaultWildcardIndexName = "*"
 
 // Configuration of QuesmaV1ProcessorQuery and QuesmaV1ProcessorIngest
 type QuesmaProcessorConfig struct {
-	IndexConfig map[string]IndexConfiguration `koanf:"indexes"`
+	UseCommonTable bool                          `koanf:"useCommonTable"`
+	IndexConfig    map[string]IndexConfiguration `koanf:"indexes"`
 }
 
 func LoadV2Config() QuesmaNewConfiguration {
@@ -296,6 +297,9 @@ func (c *QuesmaNewConfiguration) validatePipelines() error {
 			queryProcessor.Type != QuesmaV1ProcessorNoOp {
 			return fmt.Errorf("query pipeline must have query or noop processor")
 		}
+		if queryProcessor.Config.UseCommonTable != ingestProcessor.Config.UseCommonTable {
+			return fmt.Errorf("query and ingest processors must have the same configuration of 'useCommonTable'")
+		}
 		if !(queryProcessor.Type == QuesmaV1ProcessorNoOp) {
 			if _, found := queryProcessor.Config.IndexConfig[DefaultWildcardIndexName]; !found {
 				return fmt.Errorf("the default index configuration (under the name '%s') must be defined in the query processor", DefaultWildcardIndexName)
@@ -565,6 +569,10 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 				if val, exists := target.properties["useCommonTable"]; exists {
 					conf.CreateCommonTable = val == "true"
 					conf.UseCommonTableForWildcard = val == "true"
+				} else {
+					// inherit setting from the processor level
+					conf.CreateCommonTable = queryProcessor.Config.UseCommonTable
+					conf.UseCommonTableForWildcard = queryProcessor.Config.UseCommonTable
 				}
 			}
 
@@ -600,6 +608,9 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 					}
 					if val, exists := target.properties["useCommonTable"]; exists {
 						processedConfig.UseCommonTable = val == "true"
+					} else {
+						// inherit setting from the processor level
+						processedConfig.UseCommonTable = queryProcessor.Config.UseCommonTable
 					}
 					if val, exists := target.properties["tableName"]; exists {
 						processedConfig.Override = val.(string)
@@ -660,6 +671,10 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 			if val, exists := target.properties["useCommonTable"]; exists {
 				conf.CreateCommonTable = val == "true"
 				conf.UseCommonTableForWildcard = val == "true"
+			} else {
+				// inherit setting from the processor level
+				conf.CreateCommonTable = queryProcessor.Config.UseCommonTable
+				conf.UseCommonTableForWildcard = queryProcessor.Config.UseCommonTable
 			}
 		}
 		if defaultConfig.SchemaOverrides != nil {
@@ -686,6 +701,10 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 			if val, exists := target.properties["useCommonTable"]; exists {
 				conf.CreateCommonTable = val == "true"
 				conf.UseCommonTableForWildcard = val == "true"
+			} else {
+				// inherit setting from the processor level
+				conf.CreateCommonTable = ingestProcessor.Config.UseCommonTable
+				conf.UseCommonTableForWildcard = ingestProcessor.Config.UseCommonTable
 			}
 		}
 		if ingestProcessorDefaultIndexConfig.SchemaOverrides != nil {
@@ -730,6 +749,9 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 				}
 				if val, exists := target.properties["useCommonTable"]; exists {
 					processedConfig.UseCommonTable = val == true
+				} else {
+					// inherit setting from the processor level
+					processedConfig.UseCommonTable = queryProcessor.Config.UseCommonTable
 				}
 				if val, exists := target.properties["tableName"]; exists {
 					processedConfig.Override = val.(string)
@@ -779,6 +801,9 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
 				}
 				if val, exists := target.properties["useCommonTable"]; exists {
 					processedConfig.UseCommonTable = val == true
+				} else {
+					// inherit setting from the processor level
+					processedConfig.UseCommonTable = ingestProcessor.Config.UseCommonTable
 				}
 				if val, exists := target.properties["tableName"]; exists {
 					processedConfig.Override = val.(string)

quesma/quesma/config/config_v2_test.go

@@ -269,7 +269,6 @@ func TestTargetNewVariant(t *testing.T) {
 }
 
 func TestUseCommonTableGlobalProperty(t *testing.T) {
-	t.Skip()
 	os.Setenv(configFileLocationEnvVar, "./test_configs/use_common_table_global_property.yaml")
 	cfg := LoadV2Config()
 	if err := cfg.Validate(); err != nil {

quesma/quesma/config/test_configs/use_common_table_global_property.yaml

@@ -1,6 +1,5 @@
 # TEST CONFIGURATION
 licenseKey: "cdd749a3-e777-11ee-bcf8-0242ac150004"
-useCommonTable: true
 frontendConnectors:
   - name: elastic-ingest
     type: elasticsearch-fe-ingest
@@ -29,6 +28,7 @@ processors:
   - name: my-query-processor
     type: quesma-v1-processor-query
     config:
+      useCommonTable: true
       indexes:
         kibana_sample_data_ecommerce:
           target:
@@ -43,6 +43,7 @@ processors:
   - name: my-ingest-processor
     type: quesma-v1-processor-ingest
     config:
+      useCommonTable: true
       indexes:
         kibana_sample_data_ecommerce:
           target: