Introducing datetime doris transformation (#1503)

This PR introduces an abstract logical __quesma_date_hour function to be
used at the parsing level, and adds a new DateTimeFunction
transformation that lowers it to a backend-specific implementation.

Author: pdelewski

platform/database_common/log_manager.go

@@ -77,6 +77,7 @@ type LogManagerIFace interface {
 	CountMultiple(ctx context.Context, tables ...*Table) (int64, error)
 	Count(ctx context.Context, table *Table) (int64, error)
 	GetTableDefinitions() (TableMap, error)
+	GetBackendConnector() quesma_api.BackendConnector
 }
 
 func NewTableMap() *TableMap {
@@ -427,3 +428,7 @@ func (c *ChTableConfig) GetAttributes() []Attribute {
 func (l *LogManager) IsInTransparentProxyMode() bool {
 	return l.cfg.TransparentProxy
 }
+
+func (l *LogManager) GetBackendConnector() quesma_api.BackendConnector {
+	return l.chDb
+}

platform/frontend_connectors/schema_transformer.go

@@ -14,6 +14,7 @@ import (
 	"github.com/QuesmaOrg/quesma/platform/parsers/elastic_query_dsl"
 	"github.com/QuesmaOrg/quesma/platform/schema"
 	"github.com/QuesmaOrg/quesma/platform/transformations"
+	quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
 	"sort"
 	"strconv"
 	"strings"
@@ -901,7 +902,6 @@ func (s *SchemaCheckPass) applyRuntimeMappings(indexSchema schema.Schema, query
 func (s *SchemaCheckPass) convertQueryDateTimeFunctionToClickhouse(indexSchema schema.Schema, query *model.Query) (*model.Query, error) {
 
 	visitor := model.NewBaseVisitor()
-
 	visitor.OverrideVisitFunction = func(b *model.BaseExprVisitor, e model.FunctionExpr) interface{} {
 
 		switch e.Name {
@@ -926,7 +926,36 @@ func (s *SchemaCheckPass) convertQueryDateTimeFunctionToClickhouse(indexSchema s
 		query.SelectCommand = *expr.(*model.SelectCommand)
 	}
 	return query, nil
+}
+
+// it convers out internal date time related fuction to clickhouse functions
+func (s *SchemaCheckPass) convertQueryDateTimeFunctionToDoris(indexSchema schema.Schema, query *model.Query) (*model.Query, error) {
+
+	visitor := model.NewBaseVisitor()
+	visitor.OverrideVisitFunction = func(b *model.BaseExprVisitor, e model.FunctionExpr) interface{} {
+
+		switch e.Name {
+
+		case model.DateHourFunction:
+			if len(e.Args) != 1 {
+				return e
+			}
+			return model.NewFunction("HOUR", e.Args[0].Accept(b).(model.Expr))
+
+			// TODO this is a place for over date/time related functions
+			// add more
 
+		default:
+			return visitFunction(b, e)
+		}
+	}
+
+	expr := query.SelectCommand.Accept(visitor)
+
+	if _, ok := expr.(*model.SelectCommand); ok {
+		query.SelectCommand = *expr.(*model.SelectCommand)
+	}
+	return query, nil
 }
 
 func (s *SchemaCheckPass) checkAggOverUnsupportedType(indexSchema schema.Schema, query *model.Query) (*model.Query, error) {
@@ -1116,21 +1145,46 @@ func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.Execution
 		{TransformationName: "FullTextFieldTransformation", Transformation: s.applyFullTextField},
 		{TransformationName: "TimestampFieldTransformation", Transformation: s.applyTimestampField},
 		{TransformationName: "ApplySearchAfterParameter", Transformation: s.applySearchAfterParameter},
-
-		// Section 3: clickhouse specific transformations
-		{TransformationName: "QuesmaDateFunctions", Transformation: s.convertQueryDateTimeFunctionToClickhouse},
-		{TransformationName: "IpTransformation", Transformation: s.applyIpTransformations},
-		{TransformationName: "GeoTransformation", Transformation: s.applyGeoTransformations},
-		{TransformationName: "ArrayTransformation", Transformation: s.applyArrayTransformations},
-		{TransformationName: "MapTransformation", Transformation: s.applyMapTransformations},
-		{TransformationName: "MatchOperatorTransformation", Transformation: s.applyMatchOperator},
-		{TransformationName: "AggOverUnsupportedType", Transformation: s.checkAggOverUnsupportedType},
-		{TransformationName: "ApplySelectFromCluster", Transformation: s.ApplySelectFromCluster},
-
-		// Section 4: compensations and checks
-		{TransformationName: "BooleanLiteralTransformation", Transformation: s.applyBooleanLiteralLowering},
 	}
 
+	// Section 3: backend specific transformations
+	// fallback to clickhouse date functions if no backend connector is set
+	if plan.BackendConnector == nil {
+		transformationChain = append(transformationChain, struct {
+			TransformationName string
+			Transformation     func(schema.Schema, *model.Query) (*model.Query, error)
+		}{TransformationName: "QuesmaDateFunctions", Transformation: s.convertQueryDateTimeFunctionToClickhouse})
+	} else {
+		if plan.BackendConnector.GetId() == quesma_api.ClickHouseSQLBackend {
+			transformationChain = append(transformationChain, struct {
+				TransformationName string
+				Transformation     func(schema.Schema, *model.Query) (*model.Query, error)
+			}{TransformationName: "QuesmaDateFunctions", Transformation: s.convertQueryDateTimeFunctionToClickhouse})
+		}
+
+		if plan.BackendConnector.GetId() == quesma_api.DorisSQLBackend {
+			transformationChain = append(transformationChain, struct {
+				TransformationName string
+				Transformation     func(schema.Schema, *model.Query) (*model.Query, error)
+			}{TransformationName: "QuesmaDateFunctions", Transformation: s.convertQueryDateTimeFunctionToDoris})
+		}
+	}
+	transformationChain = append(transformationChain,
+		[]struct {
+			TransformationName string
+			Transformation     func(schema.Schema, *model.Query) (*model.Query, error)
+		}{
+			{TransformationName: "IpTransformation", Transformation: s.applyIpTransformations},
+			{TransformationName: "GeoTransformation", Transformation: s.applyGeoTransformations},
+			{TransformationName: "ArrayTransformation", Transformation: s.applyArrayTransformations},
+			{TransformationName: "MapTransformation", Transformation: s.applyMapTransformations},
+			{TransformationName: "MatchOperatorTransformation", Transformation: s.applyMatchOperator},
+			{TransformationName: "AggOverUnsupportedType", Transformation: s.checkAggOverUnsupportedType},
+			{TransformationName: "ApplySelectFromCluster", Transformation: s.ApplySelectFromCluster},
+			{TransformationName: "BooleanLiteralTransformation", Transformation: s.applyBooleanLiteralLowering},
+		}...,
+	)
+
 	for k, query := range plan.Queries {
 		var err error
 

platform/frontend_connectors/search.go

@@ -521,6 +521,7 @@ func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern strin
 			logger.InfoWithCtx(ctx).Msgf("Input SQL query %d: %s", i, query.SelectCommand.String())
 		}
 	}
+	plan.BackendConnector = q.logManager.GetBackendConnector()
 	err = q.transformQueries(plan)
 
 	if err != nil {

platform/model/query.go

@@ -5,6 +5,7 @@ package model
 import (
 	"github.com/QuesmaOrg/quesma/platform/parsers/painful"
 	"github.com/QuesmaOrg/quesma/platform/schema"
+	quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
 	"time"
 )
 
@@ -120,6 +121,8 @@ type ExecutionPlan struct {
 	MergeSiblingResults func(plan *ExecutionPlan, results [][]QueryResultRow) (*ExecutionPlan, [][]QueryResultRow)
 
 	SiblingQueries map[int][]int // Map of query IDs to their sibling query IDs
+
+	BackendConnector quesma_api.BackendConnector // Backend connector used for executing the queries
 }
 
 // NewExecutionPlan creates a new instance of model.ExecutionPlan

platform/parsers/elastic_query_dsl/aggregation_parser.go

@@ -406,7 +406,7 @@ func (cw *ClickhouseQueryTranslator) parseFieldFromScriptField(queryMap QueryMap
 	wantedRegex := regexp.MustCompile(`^doc\['(\w+)']\.value\.(?:getHour\(\)|hourOfDay)$`)
 	matches := wantedRegex.FindStringSubmatch(source)
 	if len(matches) == 2 {
-		return model.NewFunction("toHour", model.NewColumnRef(matches[1])), true
+		return model.NewFunction(model.DateHourFunction, model.NewColumnRef(matches[1])), true
 	}
 
 	// b) source: "if (doc['field_name_1'].value == doc['field_name_2'].value") { return 1; } else { return 0; }"

platform/parsers/elastic_query_dsl/aggregation_parser_test.go

@@ -704,9 +704,9 @@ func Test_parseFieldFromScriptField(t *testing.T) {
 		expectedMatch   model.Expr
 		expectedSuccess bool
 	}{
-		{goodQueryMap("doc['field1'].value.getHour()"), model.NewFunction("toHour", model.NewColumnRef("field1")), true},
+		{goodQueryMap("doc['field1'].value.getHour()"), model.NewFunction(model.DateHourFunction, model.NewColumnRef("field1")), true},
 		{goodQueryMap("doc['field1'].value.getHour() + doc['field2'].value.getHour()"), nil, false},
-		{goodQueryMap("doc['field1'].value.hourOfDay"), model.NewFunction("toHour", model.NewColumnRef("field1")), true},
+		{goodQueryMap("doc['field1'].value.hourOfDay"), model.NewFunction(model.DateHourFunction, model.NewColumnRef("field1")), true},
 		{goodQueryMap("doc['field1'].value"), nil, false},
 		{goodQueryMap("value.getHour() + doc['field2'].value.getHour()"), nil, false},
 		{QueryMap{}, nil, false},

platform/testdata/opensearch-visualize/aggregation_requests.go

@@ -1342,8 +1342,8 @@ var AggregationTests = []testdata.AggregationTestCase{
 			}},
 		},
 		ExpectedPancakeSQL: `
-			SELECT maxOrNull(toHour("timestamp")) AS "metric__maxAgg_col_0",
-			  minOrNull(toHour("timestamp")) AS "metric__minAgg_col_0"
+			SELECT maxOrNull(__quesma_date_hour("timestamp")) AS "metric__maxAgg_col_0",
+			  minOrNull(__quesma_date_hour("timestamp")) AS "metric__minAgg_col_0"
 			FROM ` + TableName + ``,
 	},
 	{ // [9]
@@ -1457,9 +1457,9 @@ var AggregationTests = []testdata.AggregationTestCase{
 			}},
 		},
 		ExpectedPancakeSQL: `
-			SELECT toHour("timestamp") AS "aggr__2__key_0", count(*) AS "aggr__2__count"
+			SELECT __quesma_date_hour("timestamp") AS "aggr__2__key_0", count(*) AS "aggr__2__count"
 			FROM ` + TableName + `
-			GROUP BY toHour("timestamp") AS "aggr__2__key_0"
+			GROUP BY __quesma_date_hour("timestamp") AS "aggr__2__key_0"
 			ORDER BY "aggr__2__key_0" ASC`,
 	},
 }

platform/testdata/opensearch-visualize/pipeline_aggregation_requests.go

@@ -850,7 +850,7 @@ var PipelineAggregationTests = []testdata.AggregationTestCase{
 		ExpectedPancakeSQL: `
 			SELECT toInt64(toUnixTimestamp64Milli("timestamp") / 600000) AS "aggr__2__key_0",
 			  count(*) AS "aggr__2__count",
-			  sumOrNull(toHour("timestamp")) AS "metric__2__1-metric_col_0"
+			  sumOrNull(__quesma_date_hour("timestamp")) AS "metric__2__1-metric_col_0"
 			FROM __quesma_table_name
 			GROUP BY toInt64(toUnixTimestamp64Milli("timestamp") / 600000) AS "aggr__2__key_0"
 			ORDER BY "aggr__2__key_0" ASC`,
@@ -1726,7 +1726,7 @@ var PipelineAggregationTests = []testdata.AggregationTestCase{
 		ExpectedPancakeSQL: `
 			SELECT toInt64(toUnixTimestamp64Milli("timestamp") / 600000) AS "aggr__2__key_0",
 			  count(*) AS "aggr__2__count",
-			  sumOrNull(toHour("timestamp")) AS "metric__2__1-metric_col_0"
+			  sumOrNull(__quesma_date_hour("timestamp")) AS "metric__2__1-metric_col_0"
 			FROM __quesma_table_name
 			GROUP BY toInt64(toUnixTimestamp64Milli("timestamp") / 600000) AS "aggr__2__key_0"
 			ORDER BY "aggr__2__key_0" ASC`,