Unify golang type to ClickHouse type conversion in CREATE TABLE and validation codepaths, improve Tuple(...) validation (#1259)

In the ingest processor we had two separate codepaths for converting a
Golang type into ClickHouse column type: one used in `CREATE TABLE`
codepath (`clickhouse.NewType`) and another in ingest validation
(`getTypeName`). The first one had better support for (nested) arrays
and maps, which were not fully supported in ingest validation.

Therefore this PR unifies the ingest validation codepath to use the
`clickhouse.NewType` from `CREATE TABLE` codepath. As a result, ingest
validation now has a better support for nested arrays, maps (see
`test_ingest.go`) and the code is now simpler.

Author: avelanarius

ci/it/testcases/test_ingest.go

@@ -789,9 +789,7 @@ func (a *IngestTestcase) testIncompleteTypes(ctx context.Context, t *testing.T)
 	assert.Contains(t, results[0].cols[2].(string), "QUESMA_DOC2_2")
 	assert.Contains(t, results[0].cols[2].(string), "QUESMA_DOC2_3")
 	assert.Contains(t, *results[0].cols[5].(*string), "QUESMA_DOC2_4")
-
-	// FIXME: this is not inserted correctly yet!
-	// assert.Contains(t, results[0].cols[8].(string), "QUESMA_DOC2_5")
+	assert.Contains(t, results[0].cols[8].(string), "QUESMA_DOC2_5")
 
 	assert.Contains(t, *results[1].cols[0].(*string), "QUESMA_DOC3_1")
 	assert.Contains(t, *results[1].cols[1].(*string), "QUESMA_DOC3_2")
@@ -802,9 +800,6 @@ func (a *IngestTestcase) testIncompleteTypes(ctx context.Context, t *testing.T)
 	assert.Contains(t, *results[1].cols[5].(*string), "QUESMA_DOC3_7")
 	assert.Contains(t, *results[1].cols[6].(*string), "QUESMA_DOC3_8")
 	assert.Contains(t, *results[1].cols[7].(*string), "QUESMA_DOC3_9")
-
-	// FIXME: this is not inserted correctly yet!
-	// assert.Contains(t, results[1].cols[8].(string), "QUESMA_DOC3_10")
-
+	assert.Contains(t, results[1].cols[8].(string), "QUESMA_DOC3_10")
 	assert.Contains(t, results[1].cols[9].(string), "QUESMA_DOC3_11")
 }

quesma/clickhouse/schema.go

@@ -191,6 +191,16 @@ func (t MultiValueType) CanConvert(v interface{}) bool {
 	return false // TODO for now. For sure can implement tuples easily, maybe some other too
 }
 
+func (t MultiValueType) GetColumn(name string) *Column {
+	// TODO: linear scan, but this will suffice for now (Tuples aren't typically large)
+	for _, col := range t.Cols {
+		if col.Name == name {
+			return col
+		}
+	}
+	return nil
+}
+
 func NewBaseType(clickHouseTypeName string) BaseType {
 	var GoType = ResolveType(clickHouseTypeName)
 	if GoType == nil {
@@ -255,6 +265,8 @@ func NewType(value any, valueOrigin string) (Type, error) {
 		} else {
 			return BaseType{Name: "Float64", GoType: reflect.TypeOf(float64(0))}, nil
 		}
+	case int:
+		return BaseType{Name: "Int64", GoType: reflect.TypeOf(int64(0))}, nil
 	case bool:
 		return BaseType{Name: "Bool", GoType: reflect.TypeOf(true)}, nil
 	case map[string]interface{}:

quesma/ingest/ingest_validator.go

@@ -9,80 +9,8 @@ import (
 	"github.com/QuesmaOrg/quesma/quesma/logger"
 	"github.com/QuesmaOrg/quesma/quesma/quesma/types"
 	"math"
-	"reflect"
 )
 
-func isInt(f float64) bool {
-	return f == float64(int64(f))
-}
-
-func isUnsignedInt(f float64) bool {
-	if f < 0 {
-		return false
-	}
-	return f == float64(uint64(f))
-}
-
-func getTypeName(v interface{}) string {
-	const unknownLiteral = "unknown"
-	const arrayLiteral = "Array"
-	primitiveTypes := map[string]string{
-		"string":  "String",
-		"bool":    "Bool",
-		"int":     "Int64",
-		"float64": "Float64",
-		"uint":    "UInt64",
-	}
-	if v == nil {
-		return unknownLiteral
-	}
-	GoType := reflect.TypeOf(v).String()
-	switch GoType {
-	case "string", "bool":
-		return primitiveTypes[GoType]
-	case "int":
-		if v.(int) < 0 {
-			return primitiveTypes["int"]
-		} else {
-			return primitiveTypes["uint"]
-		}
-	case "float64":
-		if isInt(v.(float64)) {
-			return primitiveTypes["int"]
-		} else if isUnsignedInt(v.(float64)) {
-			return primitiveTypes["uint"]
-		}
-		return primitiveTypes[GoType]
-	}
-	switch elem := v.(type) {
-	case []interface{}:
-		if len(elem) == 0 {
-			return arrayLiteral + "(unknown)"
-		} else {
-			innerTypeName := getTypeName(elem[0])
-			// Make sure that all elements of the array have the same type
-			for _, e := range elem {
-				if getTypeName(e) != innerTypeName {
-					return arrayLiteral + "(unknown)"
-				}
-			}
-			return arrayLiteral + "(" + innerTypeName + ")"
-		}
-	case interface{}:
-		if e := reflect.ValueOf(elem); e.Kind() == reflect.Slice {
-			innerTypeName := getTypeName(e.Index(0).Interface())
-			// Make sure that all elements of the slice have the same type
-			for i := 1; i < e.Len(); i++ {
-				if getTypeName(e.Index(i).Interface()) != innerTypeName {
-					return arrayLiteral + "(unknown)"
-				}
-			}
-			return arrayLiteral + "(" + innerTypeName + ")"
-		}
-	}
-	return GoType
-}
-
 func removeLowCardinality(columnType string) string {
 	if columnType == "LowCardinality(String)" {
 		return "String"
@@ -169,63 +97,66 @@ func validateNumericType(columnType string, incomingValueType string, value inte
 	return false
 }
 
-func validateValueAgainstType(fieldName string, value interface{}, targetColumnType clickhouse.Type) types.JSON {
-	// Validate Array() types by recursing on the inner type:
-	if compoundType, isCompound := targetColumnType.(clickhouse.CompoundType); isCompound && compoundType.Name == "Array" {
-		if valueAsArray, isArray := value.([]interface{}); isArray && len(valueAsArray) > 0 {
-			innerTypesAreCompatible := true
-			innerTypeName := getTypeName(valueAsArray[0])
+func validateValueAgainstType(fieldName string, value interface{}, columnType clickhouse.Type) (isValid bool) {
+	switch columnType := columnType.(type) {
+	case clickhouse.BaseType:
+		incomingValueType, err := clickhouse.NewType(value, fieldName)
+		if err != nil {
+			return false
+		}
+
+		columnTypeName := removeLowCardinality(columnType.Name)
+
+		if isNumericType(columnTypeName) {
+			if incomingValueType, isBaseType := incomingValueType.(clickhouse.BaseType); isBaseType && validateNumericType(columnTypeName, incomingValueType.Name, value) {
+				// Numeric types match!
+				return true
+			}
+		}
 
-			// Make sure that all elements of the array have the same type
-			for _, e := range valueAsArray {
-				eTypeName := getTypeName(e)
-				// Check if the type names are exactly the same. However, if it's a numeric type, perform more advanced
-				// validation (to handle an array like [3.5, 4, 4.5]).
-				if isNumericType(eTypeName) && isNumericType(innerTypeName) {
-					if !validateNumericType(innerTypeName, eTypeName, e) {
-						innerTypesAreCompatible = false
-						break
+		if incomingValueType, isBaseType := incomingValueType.(clickhouse.BaseType); isBaseType && incomingValueType.Name == columnTypeName {
+			// Types match exactly!
+			return true
+		}
+
+		return false
+	case clickhouse.MultiValueType:
+		if columnType.Name == "Tuple" {
+			if value, isMap := value.(map[string]interface{}); isMap {
+				for key, elem := range value {
+					subtype := columnType.GetColumn(key)
+					if subtype == nil {
+						return false
 					}
-				} else {
-					if getTypeName(e) != innerTypeName {
-						innerTypesAreCompatible = false
-						break
+					if !validateValueAgainstType(fmt.Sprintf("%s.%s", fieldName, key), elem, subtype.Type) {
+						return false
 					}
 				}
+				return true
 			}
+		} else {
+			logger.Error().Msgf("MultiValueType validation is not yet supported for type: %v", columnType)
+		}
 
-			if innerTypesAreCompatible {
-				return validateValueAgainstType(fieldName, valueAsArray[0], compoundType.BaseType)
+		return false
+	case clickhouse.CompoundType:
+		if columnType.Name == "Array" {
+			if value, isArray := value.([]interface{}); isArray {
+				for i, elem := range value {
+					if !validateValueAgainstType(fmt.Sprintf("%s[%d]", fieldName, i), elem, columnType.BaseType) {
+						return false
+					}
+				}
+				return true
 			}
+		} else {
+			logger.Error().Msgf("CompoundType validation is not yet supported for type: %v", columnType)
 		}
-	}
-
-	const DateTimeType = "DateTime64"
-	const StringType = "String"
-	deletedFields := make(types.JSON, 0)
-	columnType := targetColumnType.String()
-	columnType = removeLowCardinality(columnType)
-	incomingValueType := getTypeName(value)
 
-	// Hot path
-	if columnType == incomingValueType {
-		return deletedFields
+		return false
 	}
 
-	if columnType == DateTimeType {
-		// TODO validate date format
-		// For now we store dates as strings
-		if incomingValueType != StringType {
-			deletedFields[fieldName] = value
-		}
-	} else if isNumericType(columnType) {
-		if !validateNumericType(columnType, incomingValueType, value) {
-			deletedFields[fieldName] = value
-		}
-	} else if columnType != incomingValueType {
-		deletedFields[fieldName] = value
-	}
-	return deletedFields
+	return false
 }
 
 // validateIngest validates the document against the table schema
@@ -247,8 +178,8 @@ func (ip *IngestProcessor) validateIngest(tableName string, document types.JSON)
 			if value == nil {
 				continue
 			}
-			for k, v := range validateValueAgainstType(columnName, value, column.Type) {
-				deletedFields[k] = v
+			if !validateValueAgainstType(columnName, value, column.Type) {
+				deletedFields[columnName] = value
 			}
 		}
 	}

quesma/ingest/ingest_validator_test.go

@@ -19,42 +19,21 @@ import (
 	"testing"
 )
 
-func TestGetTypeName(t *testing.T) {
-	values := make(map[string][]interface{})
-	values["UInt64"] = []interface{}{1}
-	values["Float64"] = []interface{}{1.1}
-	values["Int64"] = []interface{}{-1}
-	values["String"] = []interface{}{"string"}
-	values["Bool"] = []interface{}{true}
-	values["Array(UInt64)"] = []interface{}{[]interface{}{1}}
-	values["Array(Int64)"] = []interface{}{[]interface{}{-1}}
-	values["Array(Array(Int64))"] = []interface{}{[][]interface{}{{-1}}}
-	values["Array(Array(Array(Int64)))"] = []interface{}{[][][]interface{}{{{-1}}}}
-	for typeName, values := range values {
-		for _, value := range values {
-			t.Run(typeName, func(t *testing.T) {
-				assert.NotNil(t, value)
-				assert.Equal(t, typeName, getTypeName(value))
-			})
-		}
-	}
-}
-
 func TestValidateIngest(t *testing.T) {
 	floatCol := &clickhouse.Column{Name: "float_field", Type: clickhouse.BaseType{
 		Name:   "Float64",
 		GoType: clickhouse.NewBaseType("float64").GoType,
 	}}
 
 	invalidJson := validateValueAgainstType("float", 1, floatCol.Type)
-	assert.Equal(t, 0, len(invalidJson))
+	assert.True(t, invalidJson)
 	StringCol := &clickhouse.Column{Name: "float_field", Type: clickhouse.BaseType{
 		Name:   "String",
 		GoType: clickhouse.NewBaseType("string").GoType,
 	}}
 
 	invalidJson = validateValueAgainstType("string", 1, StringCol.Type)
-	assert.Equal(t, 1, len(invalidJson))
+	assert.False(t, invalidJson)
 
 }
 
@@ -100,6 +79,45 @@ func TestIngestValidation(t *testing.T) {
 
 		`{"float_array_field":[3.14, 6.28, 0.99]}`,
 		`{"float_array_field":[1, 2, 3]}`,
+
+		`{"nested_array_map_field": [
+			[
+				[
+					{"field1": "value1", "field2": [1, 2, 3]},
+					{"field1": "value2", "field2": [4, 5, 6]}
+				],
+				[
+					{"field1": "value3", "field2": [7, 8, 9]},
+					{"field1": "value4", "field2": [10, 11, 12]}
+				]
+			],
+			[
+				[
+					{"field1": "value1", "field2": [1, 2, 3]}
+				]
+			],
+			[]
+		]}`,
+		`{"nested_array_map_field": [
+			[],
+			[
+				[],
+				[{}],
+				[
+					{"field1": "value1", "field2": [1, 2, 3]},
+					{"field1": "value2", "field2": [4, 5, 6]}
+				],
+				[
+					{"field1": "value3", "field2": [7, 8, 9]},
+					{"field1": "value4", "field2": [10, 11, 12]}
+				]
+			],
+			[
+				[
+					{"field1": "value1", "field2": [1, 2, 3]}
+				]
+			],
+		]}`,
 	}
 	expectedInsertJsons := []string{
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"attributes_values":{"string_field":"10"},"attributes_metadata":{"string_field":"v1;Int64"}}`, tableName),
@@ -130,6 +148,9 @@ func TestIngestValidation(t *testing.T) {
 
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"float_array_field":[3.14,6.28,0.99]}`, tableName),
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"float_array_field":[1,2,3]}`, tableName),
+
+		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"nested_array_map_field":[[[{"field1":"value1","field2":[1,2,3]},{"field1":"value2","field2":[4,5,6]}],[{"field1":"value3","field2":[7,8,9]},{"field1":"value4","field2":[10,11,12]}]],[[{"field1":"value1","field2":[1,2,3]}]],[]]}`, tableName),
+		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"nested_array_map_field":[[],[[],[{}],[{"field1":"value1","field2":[1,2,3]},{"field1":"value2","field2":[4,5,6]}],[{"field1":"value3","field2":[7,8,9]},{"field1":"value4","field2":[10,11,12]}]],[[{"field1":"value1","field2":[1,2,3]}]]]}`, tableName),
 	}
 	tableMap := util.NewSyncMapWith(tableName, &clickhouse.Table{
 		Name:   tableName,
@@ -176,6 +197,38 @@ func TestIngestValidation(t *testing.T) {
 					GoType: clickhouse.NewBaseType("Float64").GoType,
 				},
 			}},
+			// Array(Array(Array(Tuple(field1 String, field2 Array(Int64)))))
+			"nested_array_map_field": {Name: "nested_array_map_field", Type: clickhouse.CompoundType{
+				Name: "Array",
+				BaseType: clickhouse.CompoundType{
+					Name: "Array",
+					BaseType: clickhouse.CompoundType{
+						Name: "Array",
+						BaseType: clickhouse.MultiValueType{
+							Name: "Tuple",
+							Cols: []*clickhouse.Column{
+								{
+									Name: "field1",
+									Type: clickhouse.BaseType{
+										Name:   "String",
+										GoType: clickhouse.NewBaseType("String").GoType,
+									},
+								},
+								{
+									Name: "field2",
+									Type: clickhouse.CompoundType{
+										Name: "Array",
+										BaseType: clickhouse.BaseType{
+											Name:   "Int64",
+											GoType: clickhouse.NewBaseType("Int64").GoType,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			}},
 		},
 		Created: true,
 	})