Exctracting ingest ddl transformation (#1472)

This PR extracts and introduces a new transformation interface:

```
type Lowerer interface {
    LowerToDDL(validatedJsons []types.JSON,
        table *chLib.Table,
        invalidJsons []types.JSON,
        encodings map[schema.FieldEncodingKey]schema.EncodedFieldName,
        createTableCmd CreateTableStatement) ([]string, error)
}
```
The intention is to transpile input into other DDL representations, with
the primary goal of supporting Hydrolix ingestion.

It also introduces basic abstractions for building DDL using an object
model instead of raw strings.

This interface may evolve over time to adapt to our needs.

---------

Signed-off-by: Przemyslaw Delewski <[email protected]>
Co-authored-by: Przemysław Hejman <[email protected]>

Author: pdelewski

cmd/main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/QuesmaOrg/quesma/platform/table_resolver"
 	"github.com/QuesmaOrg/quesma/platform/telemetry"
 	"github.com/QuesmaOrg/quesma/platform/ui"
+	quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
 	"log"
 	"os"
 	"os/signal"
@@ -111,8 +112,11 @@ func main() {
 			// Ensure common table exists. This table have to be created before ingest processor starts
 			common_table.EnsureCommonTableExists(connectionPool, cfg.ClusterName)
 		}
-
-		ingestProcessor = ingest.NewIngestProcessor(&cfg, connectionPool, phoneHomeAgent, tableDisco, schemaRegistry, virtualTableStorage, tableResolver)
+		sqlLowerer := ingest.NewSqlLowerer(virtualTableStorage)
+		hydrolixLowerer := ingest.NewHydrolixLowerer(virtualTableStorage)
+		ingestProcessor = ingest.NewIngestProcessor(&cfg, connectionPool, phoneHomeAgent, tableDisco, schemaRegistry, sqlLowerer, tableResolver)
+		ingestProcessor.RegisterLowerer(sqlLowerer, quesma_api.ClickHouseSQLBackend)
+		ingestProcessor.RegisterLowerer(hydrolixLowerer, quesma_api.HydrolixSQLBackend)
 	} else {
 		logger.Info().Msg("Ingest processor is disabled.")
 	}

platform/backend_connectors/hydrolix_backend_connector.go

@@ -0,0 +1,52 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+
+package backend_connectors
+
+import (
+	"context"
+	"database/sql"
+	"github.com/QuesmaOrg/quesma/platform/config"
+
+	quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
+)
+
+type HydrolixBackendConnector struct {
+	BasicSqlBackendConnector
+	cfg *config.RelationalDbConfiguration
+}
+
+func (p *HydrolixBackendConnector) GetId() quesma_api.BackendConnectorType {
+	return quesma_api.HydrolixSQLBackend
+}
+
+func (p *HydrolixBackendConnector) Open() error {
+	conn, err := initDBConnection(p.cfg)
+	if err != nil {
+		return err
+	}
+	p.connection = conn
+	return nil
+}
+
+func NewHydrolixBackendConnector(configuration *config.RelationalDbConfiguration) *HydrolixBackendConnector {
+	return &HydrolixBackendConnector{
+		cfg: configuration,
+	}
+}
+
+func NewHydrolixBackendConnectorWithConnection(_ string, conn *sql.DB) *HydrolixBackendConnector {
+	return &HydrolixBackendConnector{
+		BasicSqlBackendConnector: BasicSqlBackendConnector{
+			connection: conn,
+		},
+	}
+}
+
+func (p *HydrolixBackendConnector) InstanceName() string {
+	return "hydrolix" // TODO add name taken from config
+}
+
+func (p *HydrolixBackendConnector) Exec(ctx context.Context, query string, args ...interface{}) error {
+	return nil
+}

platform/ingest/alter_table_test.go

@@ -51,11 +51,11 @@ func TestAlterTable(t *testing.T) {
 
 	ip := newIngestProcessorWithEmptyTableMap(fieldsMap, &config.QuesmaConfiguration{})
 	for i := range rowsToInsert {
-		alter, onlySchemaFields, nonSchemaFields, err := ip.GenerateIngestContent(table, types.MustJSON(rowsToInsert[i]), nil, encodings)
+		alter, onlySchemaFields, nonSchemaFields, err := ip.lowerer.GenerateIngestContent(table, types.MustJSON(rowsToInsert[i]), nil, encodings)
 		assert.NoError(t, err)
 		insert, err := generateInsertJson(nonSchemaFields, onlySchemaFields)
 		assert.Equal(t, expectedInsert[i], insert)
-		assert.Equal(t, alters[i], alter[0])
+		assert.Equal(t, alters[i], alter[0].ToSql())
 		// Table will grow with each iteration
 		assert.Equal(t, i+1, len(table.Cols))
 		for _, col := range columns[:i+1] {
@@ -128,9 +128,9 @@ func TestAlterTableHeuristic(t *testing.T) {
 			previousRow = currentRow
 		}
 
-		assert.Equal(t, int64(0), ip.ingestCounter)
+		assert.Equal(t, int64(0), ip.lowerer.ingestCounter)
 		for i := range rowsToInsert {
-			_, _, _, err := ip.GenerateIngestContent(table, types.MustJSON(rowsToInsert[i]), nil, encodings)
+			_, _, _, err := ip.lowerer.GenerateIngestContent(table, types.MustJSON(rowsToInsert[i]), nil, encodings)
 			assert.NoError(t, err)
 		}
 		assert.Equal(t, tc.expected, len(table.Cols))

platform/ingest/ast.go

@@ -0,0 +1,124 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package ingest
+
+import (
+	"fmt"
+	"github.com/QuesmaOrg/quesma/platform/util"
+	"strings"
+)
+
+type ColumnStatement struct {
+	ColumnName         string
+	ColumnType         string
+	Comment            string
+	PropertyName       string
+	AdditionalMetadata string
+}
+
+type CreateTableStatement struct {
+	Name       string
+	Cluster    string // Optional: ON CLUSTER
+	Columns    []ColumnStatement
+	Indexes    string // Optional: INDEXES
+	Comment    string
+	PostClause string // e.g. ENGINE, ORDER BY, etc.
+}
+
+type AlterStatementType int
+
+const (
+	AddColumn AlterStatementType = iota
+	CommentColumn
+)
+
+type AlterStatement struct {
+	Type       AlterStatementType
+	TableName  string
+	OnCluster  string
+	ColumnName string
+	ColumnType string // used only for AddColumn
+	Comment    string // used only for CommentColumn
+}
+
+type InsertStatement struct {
+	TableName    string
+	InsertValues string // expected to be JSONEachRow-compatible content
+}
+
+func (ct CreateTableStatement) ToSQL() string {
+	if ct.Name == "" {
+		return ""
+	}
+	var b strings.Builder
+
+	if ct.Cluster != "" {
+		b.WriteString(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" ON CLUSTER "%s"`+" \n(\n\n", ct.Name, ct.Cluster))
+	} else {
+		b.WriteString(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s"`, ct.Name))
+	}
+
+	first := true
+
+	if len(ct.Columns) > 0 {
+		b.WriteString(" \n(\n\n")
+
+	}
+
+	for _, column := range ct.Columns {
+		if first {
+			first = false
+		} else {
+			b.WriteString(",\n")
+		}
+		b.WriteString(util.Indent(1))
+		b.WriteString(fmt.Sprintf("\"%s\" %s", column.ColumnName, column.ColumnType))
+		if column.Comment != "" {
+			b.WriteString(fmt.Sprintf(" COMMENT '%s'", column.Comment))
+		}
+		if column.AdditionalMetadata != "" {
+			b.WriteString(fmt.Sprintf(" %s", column.AdditionalMetadata))
+		}
+	}
+
+	b.WriteString(ct.Indexes)
+
+	if len(ct.Columns) > 0 {
+		b.WriteString("\n)\n")
+	}
+
+	if ct.PostClause != "" {
+		b.WriteString(ct.PostClause + "\n")
+	}
+	if ct.Comment != "" {
+		b.WriteString(fmt.Sprintf("COMMENT '%s'", ct.Comment))
+	}
+
+	return b.String()
+}
+
+func (stmt AlterStatement) ToSql() string {
+	var onCluster string
+	if stmt.OnCluster != "" {
+		onCluster = fmt.Sprintf(` ON CLUSTER "%s"`, stmt.OnCluster)
+	}
+
+	switch stmt.Type {
+	case AddColumn:
+		return fmt.Sprintf(
+			`ALTER TABLE "%s"%s ADD COLUMN IF NOT EXISTS "%s" %s`,
+			stmt.TableName, onCluster, stmt.ColumnName, stmt.ColumnType,
+		)
+	case CommentColumn:
+		return fmt.Sprintf(
+			`ALTER TABLE "%s"%s COMMENT COLUMN "%s" '%s'`,
+			stmt.TableName, onCluster, stmt.ColumnName, stmt.Comment,
+		)
+	default:
+		panic(fmt.Sprintf("unsupported AlterStatementType: %v", stmt.Type))
+	}
+}
+
+func (s InsertStatement) ToSQL() string {
+	return fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow %s`, s.TableName, s.InsertValues)
+}

platform/ingest/common_table_test.go

@@ -209,7 +209,7 @@ func TestIngestToCommonTable(t *testing.T) {
 
 			ingest := newIngestProcessorWithEmptyTableMap(tables, quesmaConfig)
 			ingest.chDb = db
-			ingest.virtualTableStorage = virtualTableStorage
+			ingest.lowerer.virtualTableStorage = virtualTableStorage
 			ingest.schemaRegistry = schemaRegistry
 			ingest.tableResolver = resolver
 
@@ -227,7 +227,7 @@ func TestIngestToCommonTable(t *testing.T) {
 				}
 
 				tables.Store(indexName, testTable)
-				err = ingest.storeVirtualTable(testTable)
+				err = storeVirtualTable(testTable, ingest.lowerer.virtualTableStorage)
 				if err != nil {
 					t.Fatalf("error storing virtual table: %v", err)
 				}

platform/ingest/ddllowerer.go

@@ -0,0 +1,20 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package ingest
+
+import (
+	chLib "github.com/QuesmaOrg/quesma/platform/clickhouse"
+	"github.com/QuesmaOrg/quesma/platform/schema"
+	"github.com/QuesmaOrg/quesma/platform/types"
+)
+
+// The main purpose of Lowerer interface is to infers a schema from input JSON
+// data and then generates a backend-specific DDL (Data Definition Language) representation, such as a CREATE TABLE statement.
+// or other DDL commands that are needed to create or modify a table in the database.
+type Lowerer interface {
+	LowerToDDL(validatedJsons []types.JSON,
+		table *chLib.Table,
+		invalidJsons []types.JSON,
+		encodings map[schema.FieldEncodingKey]schema.EncodedFieldName,
+		createTableCmd CreateTableStatement) ([]string, error)
+}

platform/ingest/hydrolixlowerer.go

@@ -0,0 +1,55 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package ingest
+
+import (
+	chLib "github.com/QuesmaOrg/quesma/platform/clickhouse"
+	"github.com/QuesmaOrg/quesma/platform/persistence"
+	"github.com/QuesmaOrg/quesma/platform/schema"
+	"github.com/QuesmaOrg/quesma/platform/types"
+)
+
+type HydrolixLowerer struct {
+	virtualTableStorage persistence.JSONDatabase
+}
+
+func NewHydrolixLowerer(virtualTableStorage persistence.JSONDatabase) *HydrolixLowerer {
+	return &HydrolixLowerer{
+		virtualTableStorage: virtualTableStorage,
+	}
+}
+
+func (l *HydrolixLowerer) LowerToDDL(validatedJsons []types.JSON,
+	table *chLib.Table,
+	invalidJsons []types.JSON,
+	encodings map[schema.FieldEncodingKey]schema.EncodedFieldName,
+	createTableCmd CreateTableStatement) ([]string, error) {
+	for i, preprocessedJson := range validatedJsons {
+		_ = i
+		_ = preprocessedJson
+	}
+
+	result := []string{`{
+  "schema": {
+    "project": "",
+    "name": "test_index",
+    "time_column": "ingest_time",
+    "columns": [
+      { "name": "new_field", "type": "string" },
+      { "name": "ingest_time", "type": "datetime", "default": "NOW" }
+    ],
+    "partitioning": {
+      "strategy": "time",
+      "field": "ingest_time",
+      "granularity": "day"
+    }
+  },
+  "events": [
+    {
+      "new_field": "bar"
+    }
+  ]
+}`}
+
+	return result, nil
+}