Support timestamp in Kibana being int (not DateTime) column in ClickHouse (#1381)

Basically support for #1380

After: we can add a data view with a timestamp and all seems to be
looking good while `ts` being an `UInt64` 😉
![Screenshot 2025-03-26 at 00 29
30](https://github.com/user-attachments/assets/f8292b79-4b82-47f5-a021-e385904c0cc2)

---------

Signed-off-by: Przemyslaw Delewski <[email protected]>
Co-authored-by: Przemyslaw Delewski <[email protected]>

Author: trzysiek

platform/clickhouse/table.go

@@ -191,3 +191,8 @@ func (t *Table) GetFieldInfo(ctx context.Context, fieldName string) FieldInfo {
 	}
 	return ExistsAndIsBaseType
 }
+
+func (t *Table) IsInt(fieldName string) bool {
+	col, ok := t.Cols[fieldName]
+	return ok && col.Type != nil && strings.Contains(col.Type.String(), "Int")
+}

platform/config/config.go

@@ -59,6 +59,10 @@ type QuesmaConfiguration struct {
 	MapFieldsDiscoveringEnabled bool
 }
 
+func NewQuesmaConfigurationIndexConfigOnly(indexConfig map[string]IndexConfiguration) QuesmaConfiguration {
+	return QuesmaConfiguration{IndexConfig: indexConfig}
+}
+
 func (c *QuesmaConfiguration) AliasFields(indexName string) map[string]string {
 	aliases := make(map[string]string)
 	if indexConfig, found := c.IndexConfig[indexName]; found {

platform/frontend_connectors/schema_transformer.go

@@ -3,13 +3,15 @@
 package frontend_connectors
 
 import (
+	"context"
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/clickhouse"
 	"github.com/QuesmaOrg/quesma/platform/common_table"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/QuesmaOrg/quesma/platform/logger"
 	"github.com/QuesmaOrg/quesma/platform/model"
 	"github.com/QuesmaOrg/quesma/platform/model/typical_queries"
+	"github.com/QuesmaOrg/quesma/platform/parsers/elastic_query_dsl"
 	"github.com/QuesmaOrg/quesma/platform/schema"
 	"github.com/QuesmaOrg/quesma/platform/transformations"
 	"sort"
@@ -58,7 +60,8 @@ func (s *SchemaCheckPass) applyBooleanLiteralLowering(index schema.Schema, query
 			if strings.Contains(boolLiteral, "true") || strings.Contains(boolLiteral, "false") {
 				boolLiteral = strings.TrimLeft(boolLiteral, "'")
 				boolLiteral = strings.TrimRight(boolLiteral, "'")
-				return model.NewLiteralWithEscapeType(boolLiteral, e.EscapeType)
+				var asAny any = boolLiteral
+				return e.CloneAndOverride(&asAny, nil, nil)
 			}
 		}
 		return e.Clone()
@@ -235,7 +238,7 @@ func (s *SchemaCheckPass) applyGeoTransformations(schemaInstance schema.Schema,
 			}
 		}
 
-		return model.NewFunction(e.Name, b.VisitChildren(e.Args)...)
+		return visitFunction(b, e)
 	}
 
 	visitor.OverrideVisitSelectCommand = func(v *model.BaseExprVisitor, query model.SelectCommand) interface{} {
@@ -534,7 +537,7 @@ func (s *SchemaCheckPass) applyWildcardExpansion(indexSchema schema.Schema, quer
 
 	for _, selectColumn := range query.SelectCommand.Columns {
 
-		if selectColumn == model.NewWildcardExpr {
+		if model.IsWildcard(selectColumn) {
 			hasWildcard = true
 		} else {
 			newColumns = append(newColumns, selectColumn)
@@ -615,9 +618,9 @@ func (s *SchemaCheckPass) applyFullTextField(indexSchema schema.Schema, query *m
 
 				if len(fullTextFields) == 0 {
 					if (strings.ToUpper(e.Op) == "LIKE" || strings.ToUpper(e.Op) == "ILIKE") && model.AsString(e.Right) == "'%'" {
-						return model.NewLiteral(true)
+						return model.TrueExpr
 					}
-					return model.NewLiteral(false)
+					return model.FalseExpr
 				}
 
 				var expressions []model.Expr
@@ -632,7 +635,7 @@ func (s *SchemaCheckPass) applyFullTextField(indexSchema schema.Schema, query *m
 			}
 		}
 
-		return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), e.Op, e.Right.Accept(b).(model.Expr))
+		return visitInfix(b, e)
 	}
 
 	expr := query.SelectCommand.Accept(visitor)
@@ -750,7 +753,7 @@ func (s *SchemaCheckPass) applyFieldEncoding(indexSchema schema.Schema, query *m
 			if hasAttributesValuesColumn {
 				return model.NewArrayAccess(model.NewColumnRef(clickhouse.AttributesValuesColumn), model.NewLiteral(fmt.Sprintf("'%s'", e.ColumnName)))
 			} else {
-				return model.NewLiteral("NULL")
+				return model.NullExpr
 			}
 		}
 	}
@@ -870,7 +873,7 @@ func (s *SchemaCheckPass) convertQueryDateTimeFunctionToClickhouse(indexSchema s
 			// add more
 
 		default:
-			return model.NewFunction(e.Name, b.VisitChildren(e.Args)...)
+			return visitFunction(b, e)
 		}
 	}
 
@@ -904,7 +907,7 @@ func (s *SchemaCheckPass) checkAggOverUnsupportedType(indexSchema schema.Schema,
 								if strings.HasPrefix(col.InternalPropertyType, dbTypePrefix) {
 									logger.Warn().Msgf("Aggregation '%s' over unsupported type '%s' in column '%s'", e.Name, dbTypePrefix, col.InternalPropertyName.AsString())
 									args := b.VisitChildren(e.Args)
-									args[0] = model.NewLiteral("NULL")
+									args[0] = model.NullExpr
 									return model.NewFunction(e.Name, args...)
 								}
 							}
@@ -915,15 +918,15 @@ func (s *SchemaCheckPass) checkAggOverUnsupportedType(indexSchema schema.Schema,
 						if access.ColumnRef.ColumnName == clickhouse.AttributesValuesColumn {
 							logger.Warn().Msgf("Unsupported case. Aggregation '%s' over attribute named: '%s'", e.Name, access.Index)
 							args := b.VisitChildren(e.Args)
-							args[0] = model.NewLiteral("NULL")
+							args[0] = model.NullExpr
 							return model.NewFunction(e.Name, args...)
 						}
 					}
 				}
 			}
 		}
 
-		return model.NewFunction(e.Name, b.VisitChildren(e.Args)...)
+		return visitFunction(b, e)
 	}
 
 	expr := query.SelectCommand.Accept(visitor)
@@ -974,6 +977,76 @@ func (s *SchemaCheckPass) applyAliasColumns(indexSchema schema.Schema, query *mo
 	return query, nil
 }
 
+func visitFunction(b *model.BaseExprVisitor, f model.FunctionExpr) interface{} {
+	return model.NewFunction(f.Name, b.VisitChildren(f.Args)...)
+}
+
+func visitInfix(b *model.BaseExprVisitor, e model.InfixExpr) interface{} {
+	return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), e.Op, e.Right.Accept(b).(model.Expr))
+}
+
+func (s *SchemaCheckPass) acceptIntsAsTimestamps(indexSchema schema.Schema, query *model.Query) (*model.Query, error) {
+	table, exists := s.tableDiscovery.TableDefinitions().Load(query.TableName)
+	if !exists {
+		return nil, fmt.Errorf("table %s not found", query.TableName)
+	}
+
+	dateManager := elastic_query_dsl.NewDateManager(context.Background())
+	visitor := model.NewBaseVisitor()
+
+	visitor.OverrideVisitInfix = func(b *model.BaseExprVisitor, e model.InfixExpr) interface{} {
+		col, okLeft := model.ExtractColRef(e.Left)
+		lit, _ := model.ToLiteral(e.Right)
+		ts, okRight := model.ToLiteralsValue(e.Right)
+		if okLeft && okRight && table.IsInt(col.ColumnName) {
+			format := ""
+			if f, ok := lit.Format(); ok {
+				format = f
+			}
+			expr, ok := dateManager.ParseDateUsualFormat(ts, clickhouse.DateTime64, format)
+			if !ok {
+				// FIXME hacky but seems working
+				if tsStr, ok_ := ts.(string); ok_ && len(tsStr) > 2 {
+					expr, ok = dateManager.ParseDateUsualFormat(tsStr[1:len(tsStr)-1], clickhouse.DateTime64, format)
+				}
+			}
+			if ok {
+				if f, okF := model.ToFunction(expr); okF && f.Name == "fromUnixTimestamp64Milli" && 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])
+						}
+					}
+				}
+			}
+		}
+		return visitInfix(b, e)
+	}
+
+	visitor.OverrideVisitFunction = func(b *model.BaseExprVisitor, f model.FunctionExpr) interface{} {
+		if f.Name == "toUnixTimestamp64Milli" && 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 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 visitFunction(b, f)
+	}
+
+	expr := query.SelectCommand.Accept(visitor)
+	if _, ok := expr.(*model.SelectCommand); ok {
+		query.SelectCommand = *expr.(*model.SelectCommand)
+	}
+
+	return query, nil
+}
+
 func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.ExecutionPlan, error) {
 
 	transformationChain := []struct {
@@ -988,6 +1061,7 @@ func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.Execution
 			return transformations.ApplyAllNecessaryCommonTransformations(query, schema, s.cfg.MapFieldsDiscoveringEnabled)
 		}},
 		{TransformationName: "AliasColumnsTransformation", Transformation: s.applyAliasColumns},
+		{TransformationName: "AcceptIntsAsTimestamps", Transformation: s.acceptIntsAsTimestamps},
 
 		// Section 2: generic schema based transformations
 		//
@@ -1082,7 +1156,7 @@ func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *m
 			okLeft = true
 			lhsCol = lhsT.ColumnRef
 		default:
-			return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), e.Op, e.Right.Accept(b).(model.Expr))
+			return visitInfix(b, e)
 		}
 
 		rhsValue, ok := rhs.Value.(string)
@@ -1091,7 +1165,7 @@ func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *m
 				// only strings can be ILIKEd, everything else is a simple =
 				return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), "=", e.Right.Accept(b).(model.Expr))
 			} else {
-				return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), e.Op, e.Right.Accept(b).(model.Expr))
+				return visitInfix(b, e)
 			}
 		}
 
@@ -1131,7 +1205,7 @@ func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *m
 			switch field.Type.String() {
 			case schema.QuesmaTypeInteger.Name, schema.QuesmaTypeLong.Name, schema.QuesmaTypeUnsignedLong.Name, schema.QuesmaTypeFloat.Name, schema.QuesmaTypeBoolean.Name:
 				rhs.Value = strings.Trim(rhsValue, "%")
-				rhs.EscapeType = model.NormalNotEscaped
+				rhs.Attrs[model.EscapeKey] = model.NormalNotEscaped
 				return equal()
 			case schema.QuesmaTypeKeyword.Name:
 				return equal()
@@ -1168,14 +1242,17 @@ func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *m
 
 								// sanity check for map type with two elements
 								if len(kvTypes) == 2 {
-									rhsValue := rhs.Value.(string)
+									rhsValue = rhs.Value.(string)
 									rhsValue = strings.TrimPrefix(rhsValue, "'")
 									rhsValue = strings.TrimSuffix(rhsValue, "'")
 
 									// here we check if the value of the map is string or not
 
 									if strings.Contains(kvTypes[1], "String") {
-										return model.NewInfixExpr(arrayElementFn.Accept(b).(model.Expr), "iLIKE", model.NewLiteralWithEscapeType(rhsValue, model.NotEscapedLikeFull))
+										newRhs := rhs.Clone()
+										newRhs.Value = rhsValue
+										newRhs.Attrs[model.EscapeKey] = model.NotEscapedLikeFull
+										return model.NewInfixExpr(arrayElementFn.Accept(b).(model.Expr), "iLIKE", newRhs)
 									} else {
 										return model.NewInfixExpr(arrayElementFn.Accept(b).(model.Expr), "=", e.Right.Accept(b).(model.Expr))
 									}
@@ -1190,7 +1267,7 @@ func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *m
 		if e.Op == model.MatchOperator {
 			logger.Error().Msgf("Match operator is not supported for column %v (expr: %v)", lhsCol, e)
 		}
-		return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), e.Op, e.Right.Accept(b).(model.Expr))
+		return visitInfix(b, e)
 	}
 
 	expr := query.SelectCommand.Accept(visitor)