Fix broken _ids (#1409)

<img width="1717" alt="image"
src="https://github.com/user-attachments/assets/67850d0c-d4f6-47ce-bb91-01cd018767e2"
/>

Author: mieciu

platform/frontend_connectors/schema_transformer.go

@@ -656,6 +656,13 @@ func (s *SchemaCheckPass) applyTimestampField(indexSchema schema.Schema, query *
 	if column, ok := indexSchema.Fields[model.TimestampFieldName]; ok {
 		timestampColumnName = column.InternalPropertyName.AsString()
 	}
+	table, ok := s.tableDiscovery.TableDefinitions().Load(query.TableName)
+	if !ok {
+		return nil, fmt.Errorf("table %s not found", query.TableName)
+	}
+	if table.DiscoveredTimestampFieldName != nil {
+		timestampColumnName = *table.DiscoveredTimestampFieldName
+	}
 
 	// if not found, check if the table has a timestamp field discovered somehow
 	// This is commented out for now.

platform/frontend_connectors/schema_transformer_test.go

@@ -25,6 +25,92 @@ func (f fixedTableProvider) TableDefinitions() map[string]schema.Table {
 func (f fixedTableProvider) AutodiscoveryEnabled() bool                              { return false }
 func (f fixedTableProvider) RegisterTablesReloadListener(chan<- types.ReloadMessage) {}
 
+func TestApplyTimestampField(t *testing.T) {
+	indexConfig := map[string]config.IndexConfiguration{
+		"test": {},
+	}
+
+	fields := map[schema.FieldName]schema.Field{
+		"@timestamp":  {PropertyName: "@timestamp", InternalPropertyName: "@timestamp", InternalPropertyType: "DateTime64", Type: schema.QuesmaTypeDate},
+		"other_field": {PropertyName: "other_field", InternalPropertyName: "other_field", InternalPropertyType: "String", Type: schema.QuesmaTypeText},
+	}
+
+	indexSchema := schema.Schema{
+		Fields: fields,
+	}
+
+	tableMap := clickhouse.NewTableMap()
+	tableDiscovery := clickhouse.NewEmptyTableDiscovery()
+	tableDiscovery.TableMap = tableMap
+
+	tableMap.Store("test", &clickhouse.Table{
+		Name: "test",
+		DiscoveredTimestampFieldName: func() *string {
+			field := "discovered_timestamp"
+			return &field
+		}(),
+	})
+
+	transform := NewSchemaCheckPass(&config.QuesmaConfiguration{IndexConfig: indexConfig}, tableDiscovery, defaultSearchAfterStrategy)
+
+	tests := []struct {
+		name     string
+		query    *model.Query
+		expected *model.Query
+	}{
+		{
+			name: "replace @timestamp with discovered timestamp",
+			query: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					Columns: []model.Expr{
+						model.NewColumnRef("@timestamp"),
+					},
+				},
+			},
+			expected: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					Columns: []model.Expr{
+						model.NewColumnRef("discovered_timestamp"),
+					},
+				},
+			},
+		},
+		{
+			name: "no replacement needed",
+			query: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					Columns: []model.Expr{
+						model.NewColumnRef("other_field"),
+					},
+				},
+			},
+			expected: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					Columns: []model.Expr{
+						model.NewColumnRef("other_field"),
+					},
+				},
+			},
+		},
+	}
+
+	for i, tt := range tests {
+		t.Run(util.PrettyTestName(tt.name, i), func(t *testing.T) {
+			tt.query.Schema = indexSchema
+			tt.query.Indexes = []string{tt.query.TableName}
+
+			actual, err := transform.applyTimestampField(indexSchema, tt.query)
+			assert.NoError(t, err)
+
+			assert.Equal(t, model.AsString(tt.expected.SelectCommand), model.AsString(actual.SelectCommand))
+		})
+	}
+}
+
 func Test_ipRangeTransform(t *testing.T) {
 	const isIPAddressInRangePrimitive = "isIPAddressInRange"
 	const CASTPrimitive = "CAST"

platform/parsers/elastic_query_dsl/query_parser.go

@@ -347,9 +347,14 @@ func (cw *ClickhouseQueryTranslator) parseIds(queryMap QueryMap) model.SimpleQue
 		}
 	}
 
-	// TODO replace with cw.Schema
 	var idToSql func(string) (model.Expr, error)
-	timestampColumnName := model.TimestampFieldName
+	var timestampColumnName string
+	if cw.Table.DiscoveredTimestampFieldName != nil {
+		timestampColumnName = *cw.Table.DiscoveredTimestampFieldName
+	} else {
+		timestampColumnName = model.TimestampFieldName
+	}
+
 	if column, ok := cw.Table.Cols[timestampColumnName]; ok {
 		switch column.Type.String() {
 		case clickhouse.DateTime64.String():