Introducing logical fromUnixTime func (#1504)

This PR introduces an abstract logical `FromUnixTimeFunction` and
`FromUnixTimeFunction64mili` functions to be used at the parsing level,
and adds a extend DateTimeFunction transformation that lowers it to a
backend-specific implementation.

---------

Co-authored-by: przemyslaw <[email protected]>

Author: pdelewski

platform/frontend_connectors/physical_functions.go

@@ -0,0 +1,14 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package frontend_connectors
+
+const (
+	CLickhouseDateHourFunction           = "toHour"
+	ClickhouseFromUnixTimeFunction64mili = "fromUnixTimestamp64Milli"
+	ClickhouseFromUnixTimeFunction       = "fromUnixTimestamp"
+	ClickhouseToTimezone                 = "toTimezone"
+	ClickhousetoUnixTimestamp64Milli     = "toUnixTimestamp64Milli"
+	DorisDateHourFunction                = "HOUR"
+	DorisFromUnixTimeFunction            = "FROM_UNIXTIME"
+	DorisFromUnixTimeFunction64mili      = "FROM_MILLISECOND"
+)

platform/frontend_connectors/schema_transformer.go

@@ -20,6 +20,11 @@ import (
 	"strings"
 )
 
+type TransformationsChain struct {
+	TransformationName string
+	Transformation     func(schema.Schema, *model.Query) (*model.Query, error)
+}
+
 type SchemaCheckPass struct {
 	cfg                 *config.QuesmaConfiguration
 	tableDiscovery      database_common.TableDiscovery
@@ -910,10 +915,15 @@ func (s *SchemaCheckPass) convertQueryDateTimeFunctionToClickhouse(indexSchema s
 			if len(e.Args) != 1 {
 				return e
 			}
-			return model.NewFunction("toHour", e.Args[0].Accept(b).(model.Expr))
+			return model.NewFunction(CLickhouseDateHourFunction, e.Args[0].Accept(b).(model.Expr))
+
+		case model.FromUnixTimeFunction64mili:
+			args := b.VisitChildren(e.Args)
+			return model.NewFunction(ClickhouseFromUnixTimeFunction64mili, args...)
 
-			// TODO this is a place for over date/time related functions
-			// add more
+		case model.FromUnixTimeFunction:
+			args := b.VisitChildren(e.Args)
+			return model.NewFunction(ClickhouseFromUnixTimeFunction, args...)
 
 		default:
 			return visitFunction(b, e)
@@ -940,11 +950,14 @@ func (s *SchemaCheckPass) convertQueryDateTimeFunctionToDoris(indexSchema schema
 			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
-
+			return model.NewFunction(DorisDateHourFunction, e.Args[0].Accept(b).(model.Expr))
+		case model.FromUnixTimeFunction:
+			args := b.VisitChildren(e.Args)
+			return model.NewFunction(DorisFromUnixTimeFunction, args...)
+
+		case model.FromUnixTimeFunction64mili:
+			args := b.VisitChildren(e.Args)
+			return model.NewFunction(DorisFromUnixTimeFunction64mili, args...)
 		default:
 			return visitFunction(b, e)
 		}
@@ -1083,7 +1096,7 @@ func (s *SchemaCheckPass) acceptIntsAsTimestamps(indexSchema schema.Schema, quer
 				}
 			}
 			if ok {
-				if f, okF := model.ToFunction(expr); okF && f.Name == "fromUnixTimestamp64Milli" && len(f.Args) == 1 {
+				if f, okF := model.ToFunction(expr); okF && f.Name == model.FromUnixTimeFunction64mili && len(f.Args) == 1 {
 					if l, okL := model.ToLiteral(f.Args[0]); okL {
 						if _, exists := l.Format(); exists { // heuristics: it's a date <=> it has a format
 							return model.NewInfixExpr(col, e.Op, f.Args[0])
@@ -1096,16 +1109,16 @@ func (s *SchemaCheckPass) acceptIntsAsTimestamps(indexSchema schema.Schema, quer
 	}
 
 	visitor.OverrideVisitFunction = func(b *model.BaseExprVisitor, f model.FunctionExpr) interface{} {
-		if f.Name == "toUnixTimestamp64Milli" && len(f.Args) == 1 {
+		if f.Name == ClickhousetoUnixTimestamp64Milli && len(f.Args) == 1 {
 			if col, ok := model.ExtractColRef(f.Args[0]); ok && table.IsInt(col.ColumnName) {
 				// erases toUnixTimestamp64Milli
 				return f.Args[0]
 			}
 		}
-		if f.Name == "toTimezone" && len(f.Args) == 2 {
+		if f.Name == ClickhouseToTimezone && len(f.Args) == 2 {
 			if col, ok := model.ExtractColRef(f.Args[0]); ok && table.IsInt(col.ColumnName) {
 				// adds fromUnixTimestamp64Milli
-				return model.NewFunction("toTimezone", model.NewFunction("fromUnixTimestamp64Milli", f.Args[0]), f.Args[1])
+				return model.NewFunction(ClickhouseToTimezone, model.NewFunction(model.FromUnixTimeFunction64mili, f.Args[0]), f.Args[1])
 			}
 		}
 		return visitFunction(b, f)
@@ -1119,12 +1132,8 @@ func (s *SchemaCheckPass) acceptIntsAsTimestamps(indexSchema schema.Schema, quer
 	return query, nil
 }
 
-func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.ExecutionPlan, error) {
-
-	transformationChain := []struct {
-		TransformationName string
-		Transformation     func(schema.Schema, *model.Query) (*model.Query, error)
-	}{
+func (s *SchemaCheckPass) makeTransformations(backendConnectorType quesma_api.BackendConnectorType) []TransformationsChain {
+	transformationChain := []TransformationsChain{
 		// Section 1: from logical to physical
 		{TransformationName: "PhysicalFromExpressionTransformation", Transformation: s.applyPhysicalFromExpression},
 		{TransformationName: "WildcardExpansion", Transformation: s.applyWildcardExpansion},
@@ -1149,31 +1158,23 @@ func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.Execution
 
 	// Section 3: backend specific transformations
 	// fallback to clickhouse date functions if no backend connector is set
-	if plan.BackendConnector == nil {
+
+	if backendConnectorType == quesma_api.ClickHouseSQLBackend {
 		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 {
+
+	if backendConnectorType == 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,
+		[]TransformationsChain{
 			{TransformationName: "IpTransformation", Transformation: s.applyIpTransformations},
 			{TransformationName: "GeoTransformation", Transformation: s.applyGeoTransformations},
 			{TransformationName: "ArrayTransformation", Transformation: s.applyArrayTransformations},
@@ -1184,7 +1185,20 @@ func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.Execution
 			{TransformationName: "BooleanLiteralTransformation", Transformation: s.applyBooleanLiteralLowering},
 		}...,
 	)
+	return transformationChain
+}
+
+func (s *SchemaCheckPass) determineBackendConnectorType(plan *model.ExecutionPlan) quesma_api.BackendConnectorType {
+	if plan != nil && plan.BackendConnector != nil {
+		return plan.BackendConnector.GetId()
+	}
+	return quesma_api.ClickHouseSQLBackend
+}
+
+func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.ExecutionPlan, error) {
 
+	backendConnectorType := s.determineBackendConnectorType(plan)
+	transformationChain := s.makeTransformations(backendConnectorType)
 	for k, query := range plan.Queries {
 		var err error
 

platform/frontend_connectors/schema_transformer_test.go

@@ -2039,7 +2039,7 @@ func Test_acceptIntsAsTimestamps(t *testing.T) {
 										model.NewInfixExpr(
 											model.NewFunction("timeZoneOffset", model.NewFunction(
 												"toTimezone",
-												model.NewFunction("fromUnixTimestamp64Milli", model.NewColumnRef("timestampInt")),
+												model.NewFunction(model.FromUnixTimeFunction64mili, model.NewColumnRef("timestampInt")),
 												model.NewLiteral("'Europe/Warsaw'")),
 											),
 											"*",

platform/frontend_connectors/search_opensearch_test.go

@@ -55,6 +55,7 @@ func TestSearchOpensearch(t *testing.T) {
 			assert.NoError(t, err, "no ParseQuery error")
 			assert.True(t, len(queries) > 0, "len queries > 0")
 			whereClause := model.AsString(queries[0].SelectCommand.WhereClause)
+			// This checks where clause after parsing and before transformations
 			assert.Contains(t, tt.WantedSql, whereClause, "contains wanted sql")
 
 			for _, wantedQuery := range tt.WantedQueries {

platform/frontend_connectors/search_test.go

@@ -83,7 +83,7 @@ func TestAsyncSearchHandler(t *testing.T) {
 		},
 	}
 
-	for i, tt := range testdata.TestsAsyncSearch {
+	for i, tt := range testdata.TestsAsyncSearchAfterTransformations {
 		t.Run(util.PrettyTestName(tt.Name, i), func(t *testing.T) {
 			conn, mock := util.InitSqlMockWithPrettySqlAndPrint(t, false)
 			db := backend_connectors.NewClickHouseBackendConnectorWithConnection("", conn)
@@ -302,7 +302,7 @@ func TestSearchHandler(t *testing.T) {
 		},
 	}
 
-	for i, tt := range testdata.TestsSearch {
+	for i, tt := range testdata.TestsSearchAfterTransformations {
 		t.Run(util.PrettyTestName(tt.Name, i), func(t *testing.T) {
 			var conn *sql.DB
 			var mock sqlmock.Sqlmock
@@ -433,7 +433,7 @@ func TestSearchHandlerNoAttrsConfig(t *testing.T) {
 		},
 	}
 
-	for i, tt := range testdata.TestsSearchNoAttrs {
+	for i, tt := range testdata.TestsSearchNoAttrsAfterTransformations {
 		t.Run(util.PrettyTestName(tt.Name, i), func(t *testing.T) {
 			conn, mock := util.InitSqlMockWithPrettyPrint(t, false)
 			defer conn.Close()
@@ -482,7 +482,7 @@ func TestAsyncSearchFilter(t *testing.T) {
 			},
 		},
 	}
-	for i, tt := range testdata.TestSearchFilter {
+	for i, tt := range testdata.TestSearchFilterAfterTransformations {
 		t.Run(util.PrettyTestName(tt.Name, i), func(t *testing.T) {
 			var conn *sql.DB
 			var mock sqlmock.Sqlmock

platform/frontend_connectors/terms_enum_test.go

@@ -128,8 +128,8 @@ func testHandleTermsEnumRequest(t *testing.T, requestBody []byte, fieldName stri
 	ctx = context.WithValue(context.Background(), tracing.RequestIdCtxKey, "test")
 	qt := &elastic_query_dsl.ClickhouseQueryTranslator{Table: table, Ctx: ctx, Schema: s.Tables[schema.IndexName(testTableName)]}
 	// Here we additionally verify that terms for `_tier` are **NOT** included in the SQL query
-	expectedQuery1 := fmt.Sprintf(`SELECT DISTINCT %s FROM %s WHERE (("epoch_time">=fromUnixTimestamp(1709036700) AND "epoch_time"<=fromUnixTimestamp(1709037659)) AND ("epoch_time_datetime64">=fromUnixTimestamp64Milli(1709036700000) AND "epoch_time_datetime64"<=fromUnixTimestamp64Milli(1709037659999))) LIMIT 13`, fieldName, testTableName)
-	expectedQuery2 := fmt.Sprintf(`SELECT DISTINCT %s FROM %s WHERE (("epoch_time">=fromUnixTimestamp(1709036700) AND "epoch_time"<=fromUnixTimestamp(1709037659)) AND ("epoch_time_datetime64">=fromUnixTimestamp64Milli(1709036700000) AND "epoch_time_datetime64"<=fromUnixTimestamp64Milli(1709037659999))) LIMIT 13`, fieldName, testTableName)
+	expectedQuery1 := fmt.Sprintf(`SELECT DISTINCT %s FROM %s WHERE (("epoch_time">=__quesma_from_unixtime(1709036700) AND "epoch_time"<=__quesma_from_unixtime(1709037659)) AND ("epoch_time_datetime64">=__quesma_from_unixtime64mili(1709036700000) AND "epoch_time_datetime64"<=__quesma_from_unixtime64mili(1709037659999))) LIMIT 13`, fieldName, testTableName)
+	expectedQuery2 := fmt.Sprintf(`SELECT DISTINCT %s FROM %s WHERE (("epoch_time">=__quesma_from_unixtime(1709036700) AND "epoch_time"<=__quesma_from_unixtime(1709037659)) AND ("epoch_time_datetime64">=__quesma_from_unixtime64mili(1709036700000) AND "epoch_time_datetime64"<=__quesma_from_unixtime64mili(1709037659999))) LIMIT 13`, fieldName, testTableName)
 
 	// Once in a while `AND` conditions could be swapped, so we match both cases
 	mock.ExpectQuery(fmt.Sprintf("%s|%s", regexp.QuoteMeta(expectedQuery1), regexp.QuoteMeta(expectedQuery2))).

platform/model/constants.go

@@ -7,6 +7,8 @@ const (
 	FullTextFieldNamePlaceHolder = "__quesma_fulltext_field_name"
 	TimestampFieldName           = "@timestamp"
 
-	DateHourFunction = "__quesma_date_hour"
-	MatchOperator    = "__quesma_match"
+	DateHourFunction           = "__quesma_date_hour"
+	MatchOperator              = "__quesma_match"
+	FromUnixTimeFunction       = "__quesma_from_unixtime"
+	FromUnixTimeFunction64mili = "__quesma_from_unixtime64mili"
 )

platform/model/where_visitor.go

@@ -19,14 +19,14 @@ func FindTimestampLowerBound(field ColumnRef, whereClause Expr) (timestampInMill
 	visitor := NewBaseVisitor()
 	visitor.OverrideVisitInfix = func(visitor *BaseExprVisitor, e InfixExpr) interface{} {
 		if columnRef, ok := e.Left.(ColumnRef); ok && columnRef == field && e.Op == ">=" || e.Op == ">" {
-			if fun, ok := e.Right.(FunctionExpr); ok && fun.Name == "fromUnixTimestamp64Milli" && len(fun.Args) == 1 {
+			if fun, ok := e.Right.(FunctionExpr); ok && fun.Name == FromUnixTimeFunction64mili && len(fun.Args) == 1 {
 				if rhs, ok := fun.Args[0].(LiteralExpr); ok {
 					if rhsInt64, ok := util.ExtractInt64Maybe(rhs.Value); ok {
 						timestampInMillis = min(timestampInMillis, rhsInt64)
 						found = true
 					}
 				}
-			} else if fun, ok := e.Right.(FunctionExpr); ok && fun.Name == "fromUnixTimestamp" && len(fun.Args) == 1 {
+			} else if fun, ok := e.Right.(FunctionExpr); ok && fun.Name == FromUnixTimeFunction64mili && len(fun.Args) == 1 {
 				if rhs, ok := fun.Args[0].(LiteralExpr); ok {
 					if rhsInt64, ok := util.ExtractInt64Maybe(rhs.Value); ok {
 						timestampInMillis = min(timestampInMillis, rhsInt64*1000) // seconds -> milliseconds

platform/parsers/elastic_query_dsl/dates.go

@@ -85,7 +85,7 @@ func (dm DateManager) ParseDateUsualFormat(exprFromRequest any, datetimeType dat
 		case database_common.DateTime64:
 			threeDigitsOfPrecisionSuffice := utcTs.UnixNano()%1_000_000 == 0
 			if threeDigitsOfPrecisionSuffice {
-				return model.NewFunction("fromUnixTimestamp64Milli", addFormat(utcTs.UnixMilli())), true
+				return model.NewFunction(model.FromUnixTimeFunction64mili, addFormat(utcTs.UnixMilli())), true
 			} else {
 				return model.NewFunction(
 					"toDateTime64",
@@ -98,7 +98,7 @@ func (dm DateManager) ParseDateUsualFormat(exprFromRequest any, datetimeType dat
 				), true
 			}
 		case database_common.DateTime:
-			return model.NewFunction("fromUnixTimestamp", addFormat(utcTs.Unix())), true
+			return model.NewFunction(model.FromUnixTimeFunction, addFormat(utcTs.Unix())), true
 		default:
 			logger.WarnWithCtx(dm.ctx).Msgf("Unknown datetimeType: %v", datetimeType)
 		}

platform/parsers/elastic_query_dsl/pancake_sql_query_generation_test.go

@@ -107,8 +107,7 @@ func TestPancakeQueryGeneration(t *testing.T) {
 				}
 				prettyExpectedSql := util.SqlPrettyPrint([]byte(strings.TrimSpace(expectedSql)))
 
-				util.AssertSqlEqual(t, prettyExpectedSql, prettyPancakeSql)
-
+				assert.Equal(t, util.RemoveAllWhitespaces(prettyPancakeSql), util.RemoveAllWhitespaces(prettyExpectedSql))
 				_, ok := pancakeSql.Type.(PancakeQueryType)
 				if !ok {
 					assert.Fail(t, "Expected pancake query type")