Improve validation of Array types (#1248)

Before this change, `validateValueAgainstType` would just compare the
type name string in case of arrays. This could cause problems in case of
arrays of numbers, for example if the user had an `Array(Float64)`
ClickHouse column, but tried to insert an array of ints (e.g. `[3, 5,
7]`).

After this fix, the `validateValueAgainstType` logic now works
recursively, validating the element type of `Array` (e.g. `Float64` of
`Array(Float64)`) and comparing that with an element of incoming array
from Elastic API.

Author: avelanarius

quesma/ingest/ingest_validator.go

@@ -169,11 +169,41 @@ func validateNumericType(columnType string, incomingValueType string, value inte
 	return false
 }
 
-func validateValueAgainstType(fieldName string, value interface{}, column *clickhouse.Column) types.JSON {
+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])
+
+			// 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
+					}
+				} else {
+					if getTypeName(e) != innerTypeName {
+						innerTypesAreCompatible = false
+						break
+					}
+				}
+			}
+
+			if innerTypesAreCompatible {
+				return validateValueAgainstType(fieldName, valueAsArray[0], compoundType.BaseType)
+			}
+		}
+	}
+
 	const DateTimeType = "DateTime64"
 	const StringType = "String"
 	deletedFields := make(types.JSON, 0)
-	columnType := column.Type.String()
+	columnType := targetColumnType.String()
 	columnType = removeLowCardinality(columnType)
 	incomingValueType := getTypeName(value)
 
@@ -217,7 +247,7 @@ func (ip *IngestProcessor) validateIngest(tableName string, document types.JSON)
 			if value == nil {
 				continue
 			}
-			for k, v := range validateValueAgainstType(columnName, value, column) {
+			for k, v := range validateValueAgainstType(columnName, value, column.Type) {
 				deletedFields[k] = v
 			}
 		}

quesma/ingest/ingest_validator_test.go

@@ -46,14 +46,14 @@ func TestValidateIngest(t *testing.T) {
 		GoType: clickhouse.NewBaseType("float64").GoType,
 	}}
 
-	invalidJson := validateValueAgainstType("float", 1, floatCol)
+	invalidJson := validateValueAgainstType("float", 1, floatCol.Type)
 	assert.Equal(t, 0, len(invalidJson))
 	StringCol := &clickhouse.Column{Name: "float_field", Type: clickhouse.BaseType{
 		Name:   "String",
 		GoType: clickhouse.NewBaseType("string").GoType,
 	}}
 
-	invalidJson = validateValueAgainstType("string", 1, StringCol)
+	invalidJson = validateValueAgainstType("string", 1, StringCol.Type)
 	assert.Equal(t, 1, len(invalidJson))
 
 }
@@ -97,6 +97,9 @@ func TestIngestValidation(t *testing.T) {
 		`{"uint8_field":-1}`,
 		`{"uint8_field":255}`,
 		`{"uint8_field":1000}`,
+
+		`{"float_array_field":[3.14, 6.28, 0.99]}`,
+		`{"float_array_field":[1, 2, 3]}`,
 	}
 	expectedInsertJsons := []string{
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"attributes_values":{"string_field":"10"},"attributes_metadata":{"string_field":"v1;Int64"}}`, tableName),
@@ -124,6 +127,9 @@ func TestIngestValidation(t *testing.T) {
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"attributes_values":{"uint8_field":"-1"},"attributes_metadata":{"uint8_field":"v1;Int64"}}`, tableName),
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"uint8_field":255}`, tableName),
 		fmt.Sprintf(`INSERT INTO "%s" FORMAT JSONEachRow {"attributes_values":{"uint8_field":"1000"},"attributes_metadata":{"uint8_field":"v1;Int64"}}`, tableName),
+
+		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),
 	}
 	tableMap := util.NewSyncMapWith(tableName, &clickhouse.Table{
 		Name:   tableName,
@@ -163,6 +169,13 @@ func TestIngestValidation(t *testing.T) {
 					GoType: clickhouse.NewBaseType("Int64").GoType,
 				},
 			}},
+			"float_array_field": {Name: "float_array_field", Type: clickhouse.CompoundType{
+				Name: "Array",
+				BaseType: clickhouse.BaseType{
+					Name:   "Float64",
+					GoType: clickhouse.NewBaseType("Float64").GoType,
+				},
+			}},
 		},
 		Created: true,
 	})