Experimental - Field map syntax (#1344)

This is experimental support for accessing Map columns (String,...). 


For example:
```
CREATE TABLE IF NOT EXISTS "foo"
(
    "@timestamp" DateTime64 DEFAULT now64(),
    `bar` Map(String, Nullable(String))
)
ENGINE = MergeTree
ORDER BY ("@timestamp");        

insert into foo values ('2020-01-01 00:00:00', {'a': 'b'});
insert into foo values ('2020-01-01 00:00:00', {'c': 'd', 'e': 'f'});
insert into foo values ('2020-01-01 00:00:00', {'g': 'h', 'i': 'j', 'k': 'l'});
```

We can access the value of the map by using `fieldname.key` syntax
```
{
    "query": {
        "bool": {
            "must": [
                {
                    "match": {
                        "bar.a": "b"
                    }
                }
            ]
        },
        "match_all": {}
    }
}
```

Feature is disabled by default, it  can be enabled by:
```
        foo:
            enableFieldMapSyntax: true
            target: [ my-clickhouse-data-source ]
```

---------

Co-authored-by: Przemysław Hejman <[email protected]>

Author: nablaone

platform/clickhouse/schema.go

@@ -233,7 +233,7 @@ func ResolveType(clickHouseTypeName string) reflect.Type {
 		return reflect.TypeOf(true)
 	case "JSON":
 		return reflect.TypeOf(map[string]interface{}{})
-	case "Map(String, Nullable(String))", "Map(String, String)":
+	case "Map(String, Nullable(String))", "Map(String, String)", "Map(LowCardinality(String), String)":
 		return reflect.TypeOf(map[string]string{})
 	case "Unknown":
 		return reflect.TypeOf(UnknownType{})

platform/clickhouse/type_adapter.go

@@ -38,7 +38,7 @@ func (c SchemaTypeAdapter) Convert(s string) (schema.QuesmaType, bool) {
 		return schema.QuesmaTypeDate, true
 	case "Point":
 		return schema.QuesmaTypePoint, true
-	case "Map(String, Nullable(String))", "Map(String, String)":
+	case "Map(String, Nullable(String))", "Map(String, String)", "Map(LowCardinality(String), Nullable(String))", "Map(LowCardinality(String), String)":
 		return schema.QuesmaTypeMap, true
 	default:
 		return schema.QuesmaTypeUnknown, false

platform/config/index_config.go

@@ -20,6 +20,8 @@ type IndexConfiguration struct {
 	UseCommonTable  bool                              `koanf:"useCommonTable"`
 	Target          any                               `koanf:"target"`
 
+	EnableFieldMapSyntax bool `koanf:"enableFieldMapSyntax"`
+
 	// Computed based on the overall configuration
 	QueryTarget  []string
 	IngestTarget []string

platform/frontend_connectors/schema_transformer.go

@@ -29,6 +29,19 @@ func NewSchemaCheckPass(cfg *config.QuesmaConfiguration, tableDiscovery clickhou
 	}
 }
 
+func (s *SchemaCheckPass) isFieldMapSyntaxEnabled(query *model.Query) bool {
+
+	var enabled bool
+
+	if len(query.Indexes) == 1 {
+		if indexConf, ok := s.cfg.IndexConfig[query.Indexes[0]]; ok {
+			enabled = indexConf.EnableFieldMapSyntax
+		}
+	}
+
+	return enabled
+}
+
 func (s *SchemaCheckPass) applyBooleanLiteralLowering(index schema.Schema, query *model.Query) (*model.Query, error) {
 
 	visitor := model.NewBaseVisitor()
@@ -716,6 +729,26 @@ func (s *SchemaCheckPass) applyFieldEncoding(indexSchema schema.Schema, query *m
 		if resolvedField, ok := indexSchema.ResolveField(e.ColumnName); ok {
 			return model.NewColumnRefWithTable(resolvedField.InternalPropertyName.AsString(), e.TableAlias)
 		} else {
+			// here we didn't find a column by field name,
+			// we try some other options
+
+			// 1. we check if the field name point to the map
+			if s.isFieldMapSyntaxEnabled(query) {
+				elements := strings.Split(e.ColumnName, ".")
+				if len(elements) > 1 {
+					if mapField, ok := indexSchema.ResolveField(elements[0]); ok {
+						// check if we have map type, especially  Map(String, any) here
+						if mapField.Type.Name == schema.QuesmaTypeMap.Name &&
+							(strings.HasPrefix(mapField.InternalPropertyType, "Map(String") ||
+								strings.HasPrefix(mapField.InternalPropertyType, "Map(LowCardinality(String")) {
+							return model.NewFunction("arrayElement", model.NewColumnRef(elements[0]), model.NewLiteral(fmt.Sprintf("'%s'", strings.Join(elements[1:], "."))))
+						}
+					}
+				}
+			}
+
+			// 2. maybe we should use attributes
+
 			if hasAttributesValuesColumn {
 				return model.NewArrayAccess(model.NewColumnRef(clickhouse.AttributesValuesColumn), model.NewLiteral(fmt.Sprintf("'%s'", e.ColumnName)))
 			} else {
@@ -1041,6 +1074,46 @@ func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *m
 			}
 		}
 
+		if s.isFieldMapSyntaxEnabled(query) {
+			// special case where left side is arrayElement,
+			// arrayElement comes from applyFieldEncoding function
+			arrayElementFn, ok := e.Left.(model.FunctionExpr)
+			if ok && arrayElementFn.Name == "arrayElement" && e.Op == model.MatchOperator {
+
+				if len(arrayElementFn.Args) == 2 {
+					if col, ok := arrayElementFn.Args[0].(model.ColumnRef); ok {
+						field, found := indexSchema.ResolveFieldByInternalName(col.ColumnName)
+
+						if found {
+							internalType := field.InternalPropertyType
+
+							// we support Map(K,V) type only
+							if strings.HasPrefix(internalType, "Map(") {
+								types := strings.TrimPrefix(strings.TrimSuffix(internalType, ")"), "Map(")
+								types = strings.ReplaceAll(types, " ", "")
+								kvTypes := strings.Split(types, ",")
+
+								// sanity check for map type with two elements
+								if len(kvTypes) == 2 {
+									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))
+									} else {
+										return model.NewInfixExpr(arrayElementFn.Accept(b).(model.Expr), "=", e.Right.Accept(b).(model.Expr))
+									}
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+
 		return model.NewInfixExpr(e.Left.Accept(b).(model.Expr), e.Op, e.Right.Accept(b).(model.Expr))
 	}
 

platform/frontend_connectors/schema_transformer_test.go

@@ -1344,3 +1344,147 @@ func Test_checkAggOverUnsupportedType(t *testing.T) {
 		})
 	}
 }
+
+func Test_mapKeys(t *testing.T) {
+
+	indexConfig := map[string]config.IndexConfiguration{
+		"test":  {EnableFieldMapSyntax: true},
+		"test2": {EnableFieldMapSyntax: false},
+	}
+
+	fields := map[schema.FieldName]schema.Field{
+		"@timestamp": {PropertyName: "@timestamp", InternalPropertyName: "@timestamp", InternalPropertyType: "DateTime64", Type: schema.QuesmaTypeDate},
+		"foo":        {PropertyName: "foo", InternalPropertyName: "foo", InternalPropertyType: "Map(String, String)", Type: schema.QuesmaTypeMap},
+		"sizes":      {PropertyName: "sizes", InternalPropertyName: "sizes", InternalPropertyType: "Map(String, Int64)", Type: schema.QuesmaTypeMap},
+	}
+
+	indexSchema := schema.Schema{
+		Fields: fields,
+	}
+
+	tableMap := clickhouse.NewTableMap()
+
+	tableDiscovery := clickhouse.NewEmptyTableDiscovery()
+	tableDiscovery.TableMap = tableMap
+	for indexName := range indexConfig {
+		tableMap.Store(indexName, clickhouse.NewEmptyTable(indexName))
+	}
+
+	transform := NewSchemaCheckPass(&config.QuesmaConfiguration{IndexConfig: indexConfig}, tableDiscovery, defaultSearchAfterStrategy)
+
+	tests := []struct {
+		name     string
+		query    *model.Query
+		expected *model.Query
+	}{
+
+		{
+			name: "match operator transformation for String (ILIKE)",
+			query: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					FromClause: model.NewTableRef("test"),
+					Columns:    []model.Expr{model.NewColumnRef("foo")},
+					WhereClause: model.NewInfixExpr(
+						model.NewColumnRef("foo.bar"),
+						model.MatchOperator,
+						model.NewLiteral("'baz'"),
+					),
+				},
+			},
+			expected: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					FromClause: model.NewTableRef("test"),
+					Columns:    []model.Expr{model.NewColumnRef("foo")},
+					WhereClause: model.NewInfixExpr(
+						model.NewFunction("arrayElement", model.NewColumnRef("foo"), model.NewLiteral("'bar'")),
+						"iLIKE",
+						model.NewLiteral("'%baz%'"),
+					),
+				},
+			},
+		},
+
+		{
+			name: "match operator transformation for int (=)",
+			query: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					FromClause: model.NewTableRef("test"),
+					Columns:    []model.Expr{model.NewColumnRef("foo")},
+					WhereClause: model.NewInfixExpr(
+						model.NewColumnRef("sizes.bar"),
+						model.MatchOperator,
+						model.NewLiteral("1"),
+					),
+				},
+			},
+			expected: &model.Query{
+				TableName: "test",
+				SelectCommand: model.SelectCommand{
+					FromClause: model.NewTableRef("test"),
+					Columns:    []model.Expr{model.NewColumnRef("foo")},
+					WhereClause: model.NewInfixExpr(
+						model.NewFunction("arrayElement", model.NewColumnRef("sizes"), model.NewLiteral("'bar'")),
+						"=",
+						model.NewLiteral("1"),
+					),
+				},
+			},
+		},
+
+		{
+			name: "not enabled opt-in flag, we do not transform at all",
+			query: &model.Query{
+				TableName: "test2",
+				SelectCommand: model.SelectCommand{
+					FromClause: model.NewTableRef("test2"),
+					Columns:    []model.Expr{model.NewColumnRef("foo")},
+					WhereClause: model.NewInfixExpr(
+						model.NewColumnRef("foo.bar"),
+						model.MatchOperator,
+						model.NewLiteral("'baz'"),
+					),
+				},
+			},
+			expected: &model.Query{
+				TableName: "test2",
+				SelectCommand: model.SelectCommand{
+					FromClause: model.NewTableRef("test2"),
+					Columns:    []model.Expr{model.NewColumnRef("foo")},
+					WhereClause: model.NewInfixExpr(
+						model.NewLiteral("NULL"),
+						model.MatchOperator,
+						model.NewLiteral("'baz'"),
+					),
+				},
+			},
+		},
+	}
+
+	asString := func(query *model.Query) string {
+		return query.SelectCommand.String()
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.query.Schema = indexSchema
+			tt.query.Indexes = []string{tt.query.TableName}
+			actual, err := transform.Transform([]*model.Query{tt.query})
+			assert.NoError(t, err)
+
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			assert.True(t, len(actual) == 1, "len queries == 1")
+
+			expectedJson := asString(tt.expected)
+			actualJson := asString(actual[0])
+
+			assert.Equal(t, expectedJson, actualJson)
+		})
+	}
+
+}