Implement hydrolix ddl

platform/backend_connectors/hydrolix_backend_connector.go

@@ -4,16 +4,23 @@
 package backend_connectors
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
+	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/config"
+	"github.com/QuesmaOrg/quesma/platform/logger"
+	"io"
+	"net/http"
 
 	quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
 )
 
 type HydrolixBackendConnector struct {
 	BasicSqlBackendConnector
-	cfg *config.RelationalDbConfiguration
+	cfg         *config.RelationalDbConfiguration
+	IngestURL   string
+	AccessToken string
 }
 
 func (p *HydrolixBackendConnector) GetId() quesma_api.BackendConnectorType {
@@ -48,5 +55,34 @@ func (p *HydrolixBackendConnector) InstanceName() string {
 }
 
 func (p *HydrolixBackendConnector) Exec(ctx context.Context, query string, args ...interface{}) error {
+	if p.IngestURL == "" || p.AccessToken == "" {
+		logger.Info().Msg("missing ingest URL or access token")
+		// TODO for fallback, execute the query directly on the database connection
+		_, err := p.connection.ExecContext(ctx, query)
+		return err
+	}
+
+	// Create HTTP request using the JSON payload from query
+	req, err := http.NewRequestWithContext(ctx, "POST", p.IngestURL, bytes.NewBufferString(query))
+	if err != nil {
+		return fmt.Errorf("failed to create request: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+p.AccessToken)
+	req.Header.Set("Content-Type", "application/json")
+
+	// Execute HTTP request
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	// Handle error response
+	if resp.StatusCode >= 400 {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("ingest failed: %s — %s", resp.Status, string(body))
+	}
+
 	return nil
 }

platform/ingest/hydrolixlowerer.go

@@ -3,14 +3,19 @@
 package ingest
 
 import (
+	"encoding/json"
+	"fmt"
 	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"
+	"strings"
+	"sync/atomic"
 )
 
 type HydrolixLowerer struct {
 	virtualTableStorage persistence.JSONDatabase
+	ingestCounter       int64
 }
 
 func NewHydrolixLowerer(virtualTableStorage persistence.JSONDatabase) *HydrolixLowerer {
@@ -19,37 +24,137 @@ func NewHydrolixLowerer(virtualTableStorage persistence.JSONDatabase) *HydrolixL
 	}
 }
 
-func (l *HydrolixLowerer) LowerToDDL(validatedJsons []types.JSON,
+func (ip *HydrolixLowerer) GenerateIngestContent(table *chLib.Table,
+	data types.JSON,
+	inValidJson types.JSON,
+	encodings map[schema.FieldEncodingKey]schema.EncodedFieldName) ([]AlterStatement, types.JSON, []NonSchemaField, error) {
+
+	if len(table.Config.Attributes) == 0 {
+		return nil, data, nil, nil
+	}
+
+	mDiff := DifferenceMap(data, table) // TODO change to DifferenceMap(m, t)
+
+	if len(mDiff) == 0 && len(inValidJson) == 0 { // no need to modify, just insert 'js'
+		return nil, data, nil, nil
+	}
+
+	// check attributes precondition
+	if len(table.Config.Attributes) <= 0 {
+		return nil, nil, nil, fmt.Errorf("no attributes config, but received non-schema fields: %s", mDiff)
+	}
+	attrsMap, _ := BuildAttrsMap(mDiff, table.Config)
+
+	// generateNewColumns is called on original attributes map
+	// before adding invalid fields to it
+	// otherwise it would contain invalid fields e.g. with wrong types
+	// we only want to add fields that are not part of the schema e.g we don't
+	// have columns for them
+	var alterStatements []AlterStatement
+	atomic.AddInt64(&ip.ingestCounter, 1)
+	//if ok, alteredAttributesIndexes := ip.shouldAlterColumns(table, attrsMap); ok {
+	//	alterStatements = ip.generateNewColumns(attrsMap, table, alteredAttributesIndexes, encodings)
+	//}
+	// If there are some invalid fields, we need to add them to the attributes map
+	// to not lose them and be able to store them later by
+	// generating correct update query
+	// addInvalidJsonFieldsToAttributes returns a new map with invalid fields added
+	// this map is then used to generate non-schema fields string
+	attrsMapWithInvalidFields := addInvalidJsonFieldsToAttributes(attrsMap, inValidJson)
+	nonSchemaFields, err := generateNonSchemaFields(attrsMapWithInvalidFields)
+
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	onlySchemaFields := RemoveNonSchemaFields(data, table)
+
+	return alterStatements, onlySchemaFields, nonSchemaFields, nil
+}
+
+func (l *HydrolixLowerer) LowerToDDL(
+	validatedJsons []types.JSON,
 	table *chLib.Table,
 	invalidJsons []types.JSON,
 	encodings map[schema.FieldEncodingKey]schema.EncodedFieldName,
-	createTableCmd CreateTableStatement) ([]string, error) {
+	createTableCmd CreateTableStatement,
+) ([]string, error) {
+	// Construct columns array
+	var columnsJSON strings.Builder
+	columnsJSON.WriteString("[\n")
+
+	for i, col := range createTableCmd.Columns {
+		if i > 0 {
+			columnsJSON.WriteString(",\n")
+		}
+		columnsJSON.WriteString(fmt.Sprintf(`  { "name": "%s", "type": "%s"`, col.ColumnName, col.ColumnType))
+		if col.Comment != "" {
+			columnsJSON.WriteString(fmt.Sprintf(`, "comment": "%s"`, col.Comment))
+		}
+		if col.AdditionalMetadata != "" {
+			columnsJSON.WriteString(fmt.Sprintf(`, "metadata": "%s"`, col.AdditionalMetadata))
+		}
+		columnsJSON.WriteString(" }")
+	}
+
+	columnsJSON.WriteString("\n]")
+
+	const timeColumnName = "ingest_time"
+
+	const (
+		partitioningStrategy    = "strategy"
+		partitioningField       = "field"
+		partitioningGranularity = "granularity"
+
+		defaultStrategy    = "time"
+		defaultField       = "ingest_time"
+		defaultGranularity = "day"
+	)
+	partitioningJSON := fmt.Sprintf(`"partitioning": {
+  "%s": "%s",
+  "%s": "%s",
+  "%s": "%s"
+}`,
+		partitioningStrategy, defaultStrategy,
+		partitioningField, defaultField,
+		partitioningGranularity, defaultGranularity)
+	events := make(map[string]any)
 	for i, preprocessedJson := range validatedJsons {
-		_ = i
-		_ = preprocessedJson
+		_, onlySchemaFields, nonSchemaFields, err := l.GenerateIngestContent(table, preprocessedJson,
+			invalidJsons[i], encodings)
+		if err != nil {
+			return nil, fmt.Errorf("error BuildInsertJson, tablename: '%s' : %v", table.Name, err)
+		}
+		if err != nil {
+			return nil, fmt.Errorf("error BuildInsertJson, tablename: '%s' : %v", table.Name, err)
+		}
+		content := convertNonSchemaFieldsToMap(nonSchemaFields)
+
+		for k, v := range onlySchemaFields {
+			content[k] = v
+		}
+
+		for k, v := range content {
+			events[k] = v
+		}
+	}
+
+	eventList := []map[string]any{events}
+	eventBytes, err := json.MarshalIndent(eventList, "    ", "  ")
+	if err != nil {
+		return nil, err
 	}
+	eventJSON := string(eventBytes)
 
-	result := []string{`{
+	result := fmt.Sprintf(`{
   "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"
-    }
+    "project": "%s",
+    "name": "%s",
+    "time_column": "%s",
+    "columns": %s,
+    %s,
   },
-  "events": [
-    {
-      "new_field": "bar"
-    }
-  ]
-}`}
-
-	return result, nil
+  "events": %s
+}`, table.DatabaseName, table.Name, timeColumnName, columnsJSON.String(), partitioningJSON, eventJSON)
+	return []string{result}, nil
 }

platform/ingest/insert_test.go

@@ -480,25 +480,27 @@ func TestHydrolixIngest(t *testing.T) {
 
 			expectedStatements: []string{
 				fmt.Sprintf(`{
-  "schema": {
+"schema": {
     "project": "%s",
     "name": "test_index",
     "time_column": "ingest_time",
     "columns": [
-      { "name": "new_field", "type": "string" },
-      { "name": "ingest_time", "type": "datetime", "default": "NOW" }
-    ],
+  { "name": "@timestamp", "type": "DateTime64(3)", "metadata": "DEFAULT now64()" },
+  { "name": "attributes_values", "type": "Map(String,String)" },
+  { "name": "attributes_metadata", "type": "Map(String,String)" },
+  { "name": "new_field", "type": "Nullable(String)", "comment": "quesmaMetadataV1:fieldName=new_field" }
+],
     "partitioning": {
-      "strategy": "time",
-      "field": "ingest_time",
-      "granularity": "day"
-    }
+  "strategy": "time",
+  "field": "ingest_time",
+  "granularity": "day"
+},
   },
   "events": [
-    {
-      "new_field": "bar"
-    }
-  ]
+      {
+        "new_field": "bar"
+      }
+    ]
 }`, projectName),
 			},
 		},

platform/ingest/processor.go

@@ -627,54 +627,6 @@ func (ip *SqlLowerer) shouldAlterColumns(table *chLib.Table, attrsMap map[string
 	return false, nil
 }
 
-func (ip *SqlLowerer) GenerateIngestContent(table *chLib.Table,
-	data types.JSON,
-	inValidJson types.JSON,
-	encodings map[schema.FieldEncodingKey]schema.EncodedFieldName) ([]AlterStatement, types.JSON, []NonSchemaField, error) {
-
-	if len(table.Config.Attributes) == 0 {
-		return nil, data, nil, nil
-	}
-
-	mDiff := DifferenceMap(data, table) // TODO change to DifferenceMap(m, t)
-
-	if len(mDiff) == 0 && len(inValidJson) == 0 { // no need to modify, just insert 'js'
-		return nil, data, nil, nil
-	}
-
-	// check attributes precondition
-	if len(table.Config.Attributes) <= 0 {
-		return nil, nil, nil, fmt.Errorf("no attributes config, but received non-schema fields: %s", mDiff)
-	}
-	attrsMap, _ := BuildAttrsMap(mDiff, table.Config)
-
-	// generateNewColumns is called on original attributes map
-	// before adding invalid fields to it
-	// otherwise it would contain invalid fields e.g. with wrong types
-	// we only want to add fields that are not part of the schema e.g we don't
-	// have columns for them
-	var alterStatements []AlterStatement
-	atomic.AddInt64(&ip.ingestCounter, 1)
-	if ok, alteredAttributesIndexes := ip.shouldAlterColumns(table, attrsMap); ok {
-		alterStatements = ip.generateNewColumns(attrsMap, table, alteredAttributesIndexes, encodings)
-	}
-	// If there are some invalid fields, we need to add them to the attributes map
-	// to not lose them and be able to store them later by
-	// generating correct update query
-	// addInvalidJsonFieldsToAttributes returns a new map with invalid fields added
-	// this map is then used to generate non-schema fields string
-	attrsMapWithInvalidFields := addInvalidJsonFieldsToAttributes(attrsMap, inValidJson)
-	nonSchemaFields, err := generateNonSchemaFields(attrsMapWithInvalidFields)
-
-	if err != nil {
-		return nil, nil, nil, err
-	}
-
-	onlySchemaFields := RemoveNonSchemaFields(data, table)
-
-	return alterStatements, onlySchemaFields, nonSchemaFields, nil
-}
-
 func generateInsertJson(nonSchemaFields []NonSchemaField, onlySchemaFields types.JSON) (string, error) {
 	result := convertNonSchemaFieldsToMap(nonSchemaFields)