Add support for common table // kick out TableResolver (#1089)

In this next-gen processor, we don't need an instance of table resolver.

Each decision can be derived straight from the processor configuration.

Author: mieciu

quesma/ingest/processor2.go

@@ -23,7 +23,6 @@ import (
 	"quesma/quesma/types"
 	"quesma/schema"
 	"quesma/stats"
-	"quesma/table_resolver"
 	"quesma/telemetry"
 	"quesma/util"
 	quesma_api "quesma_v2/core"
@@ -50,7 +49,6 @@ type (
 		ingestFieldStatistics     IngestFieldStatistics
 		ingestFieldStatisticsLock sync.Mutex
 		virtualTableStorage       persistence.JSONDatabase
-		tableResolver             table_resolver.TableResolver
 	}
 )
 
@@ -442,63 +440,46 @@ func (lm *IngestProcessor2) Ingest(ctx context.Context, tableName string, jsonDa
 	return lm.ProcessInsertQuery(ctx, tableName, jsonData, transformer, nameFormatter)
 }
 
+func (lm *IngestProcessor2) useCommonTable(tableName string) bool {
+	if tableConfig, ok := lm.cfg.IndexConfig[tableName]; ok {
+		return tableConfig.UseCommonTable
+	}
+	return false
+}
+
 func (lm *IngestProcessor2) ProcessInsertQuery(ctx context.Context, tableName string,
 	jsonData []types.JSON, transformer jsonprocessor.IngestTransformer,
 	tableFormatter TableColumNameFormatter) error {
 
-	decision := lm.tableResolver.Resolve(quesma_api.IngestPipeline, tableName)
-
-	if decision.Err != nil {
-		return decision.Err
-	}
-
-	if decision.IsEmpty { // TODO
-		return fmt.Errorf("table %s not found", tableName)
-	}
-
-	if decision.IsClosed { // TODO
-		return fmt.Errorf("table %s is closed", tableName)
-	}
-
-	for _, connectorDecision := range decision.UseConnectors {
-
-		var clickhouseDecision *quesma_api.ConnectorDecisionClickhouse
-
-		var ok bool
-		if clickhouseDecision, ok = connectorDecision.(*quesma_api.ConnectorDecisionClickhouse); !ok {
-			continue
+	if lm.useCommonTable(tableName) {
+		// we have clone the data, because we want to process it twice
+		var clonedJsonData []types.JSON
+		for _, jsonValue := range jsonData {
+			clonedJsonData = append(clonedJsonData, jsonValue.Clone())
 		}
 
-		if clickhouseDecision.IsCommonTable { // TODO: TABLE_RESOLVER DECIDES WHETHER WE'RE DEALING WITH COMMON TABLE
-			// we have clone the data, because we want to process it twice
-			var clonedJsonData []types.JSON
-			for _, jsonValue := range jsonData {
-				clonedJsonData = append(clonedJsonData, jsonValue.Clone())
-			}
-
-			err := lm.processInsertQueryInternal(ctx, tableName, clonedJsonData, transformer, tableFormatter, true)
-			if err != nil {
-				// we ignore an error here, because we want to process the data and don't lose it
-				logger.ErrorWithCtx(ctx).Msgf("error processing insert query - virtual table schema update: %v", err)
-			}
-
-			pipeline := jsonprocessor.IngestTransformerPipeline{}
-			pipeline = append(pipeline, &common_table.IngestAddIndexNameTransformer{IndexName: tableName})
-			pipeline = append(pipeline, transformer)
+		err := lm.processInsertQueryInternal(ctx, tableName, clonedJsonData, transformer, tableFormatter, true)
+		if err != nil {
+			// we ignore an error here, because we want to process the data and don't lose it
+			logger.ErrorWithCtx(ctx).Msgf("error processing insert query - virtual table schema update: %v", err)
+		}
 
-			err = lm.processInsertQueryInternal(ctx, common_table.TableName, jsonData, pipeline, tableFormatter, false)
-			if err != nil {
-				return fmt.Errorf("error processing insert query to a common table: %w", err)
-			}
+		pipeline := jsonprocessor.IngestTransformerPipeline{}
+		pipeline = append(pipeline, &common_table.IngestAddIndexNameTransformer{IndexName: tableName})
+		pipeline = append(pipeline, transformer)
 
-		} else {
-			err := lm.processInsertQueryInternal(ctx, clickhouseDecision.ClickhouseTableName, jsonData, transformer, tableFormatter, false)
-			if err != nil {
-				return fmt.Errorf("error processing insert query: %w", err)
-			}
+		err = lm.processInsertQueryInternal(ctx, common_table.TableName, jsonData, pipeline, tableFormatter, false)
+		if err != nil {
+			return fmt.Errorf("error processing insert query to a common table: %w", err)
 		}
 
+	} else {
+		err := lm.processInsertQueryInternal(ctx, tableName, jsonData, transformer, tableFormatter, false)
+		if err != nil {
+			return fmt.Errorf("error processing insert query: %w", err)
+		}
 	}
+
 	return nil
 }
 
@@ -667,9 +648,9 @@ func (ip *IngestProcessor2) Ping() error {
 	return ip.chDb.Open()
 }
 
-func NewIngestProcessor2(cfg *config.QuesmaConfiguration, chDb quesma_api.BackendConnector, phoneHomeAgent telemetry.PhoneHomeAgent, loader chLib.TableDiscovery, schemaRegistry schema.Registry, virtualTableStorage persistence.JSONDatabase, tableResolver table_resolver.TableResolver, esBackendConn backend_connectors.ElasticsearchBackendConnector) *IngestProcessor2 {
+func NewIngestProcessor2(cfg *config.QuesmaConfiguration, chDb quesma_api.BackendConnector, phoneHomeAgent telemetry.PhoneHomeAgent, loader chLib.TableDiscovery, schemaRegistry schema.Registry, virtualTableStorage persistence.JSONDatabase, esBackendConn backend_connectors.ElasticsearchBackendConnector) *IngestProcessor2 {
 	ctx, cancel := context.WithCancel(context.Background())
-	return &IngestProcessor2{ctx: ctx, cancel: cancel, chDb: chDb, tableDiscovery: loader, cfg: cfg, phoneHomeAgent: phoneHomeAgent, schemaRegistry: schemaRegistry, virtualTableStorage: virtualTableStorage, tableResolver: tableResolver, es: esBackendConn}
+	return &IngestProcessor2{ctx: ctx, cancel: cancel, chDb: chDb, tableDiscovery: loader, cfg: cfg, phoneHomeAgent: phoneHomeAgent, schemaRegistry: schemaRegistry, virtualTableStorage: virtualTableStorage, es: esBackendConn}
 }
 
 // validateIngest validates the document against the table schema

quesma/main.go

@@ -66,6 +66,14 @@ func buildIngestOnlyQuesma() quesma_api.QuesmaBuilder {
 				"test_index_2": {
 					Name: "test_index_2",
 				},
+				"tab1": {
+					Name:           "tab1",
+					UseCommonTable: true,
+				},
+				"tab2": {
+					Name:           "tab2",
+					UseCommonTable: true,
+				},
 				"*": {
 					IngestTarget: []string{config.ElasticsearchTarget},
 				},

quesma/processors/es_to_ch_ingest/elasticsearch_to_clickhouse_ingest_processor.go

@@ -78,9 +78,7 @@ func (p *ElasticsearchToClickHouseIngestProcessor) prepareTemporaryIngestProcess
 	tableDisco := clickhouse.NewTableDiscovery2(oldQuesmaConfig, chBackendConn, virtualTableStorage)
 	schemaRegistry := schema.NewSchemaRegistry(clickhouse.TableDiscoveryTableProviderAdapter{TableDiscovery: tableDisco}, oldQuesmaConfig, clickhouse.SchemaTypeAdapter{})
 
-	v2TableResolver := NewNextGenTableResolver()
-
-	ip := ingest.NewIngestProcessor2(oldQuesmaConfig, chBackendConn, nil, tableDisco, schemaRegistry, virtualTableStorage, v2TableResolver, esBackendConn)
+	ip := ingest.NewIngestProcessor2(oldQuesmaConfig, chBackendConn, nil, tableDisco, schemaRegistry, virtualTableStorage, esBackendConn)
 
 	ip.Start()
 	return ip
@@ -103,7 +101,8 @@ func (p *ElasticsearchToClickHouseIngestProcessor) Handle(metadata map[string]in
 		messageAsHttpReq := mCasted.OriginalRequest
 
 		if _, present := p.config.IndexConfig[indexNameFromIncomingReq]; !present && metadata[IngestAction] == DocIndexAction {
-			// route to Elasticsearch, `bulk` request might be sent to ClickHouse depending on the request payload
+			// `_doc` at this point can go directly to Elasticsearch,
+			// `_bulk` request might be still sent to ClickHouse as the req payload may contain documents targeting CH tables
 			resp := p.legacyIngestProcessor.SendToElasticsearch(messageAsHttpReq)
 			respBody, err := ReadResponseBody(resp)
 			if err != nil {