Combine schema and JSON types for Array fields

quesma/ingest/ingest_validator.go

@@ -169,11 +169,36 @@ 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 {
+	// TODO: comment this logic
+	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)
+				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 +242,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

@@ -45,14 +45,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))
 
 }

quesma/ingest/parser.go

@@ -57,11 +57,24 @@ func columnsToString(columnsFromJson []CreateTableEntry,
 		columnMetadata.Values[comment_metadata.ElasticFieldName] = propertyName
 		comment := columnMetadata.Marshall()
 
-		if columnFromSchema, found := columnsFromSchema[schema.FieldName(columnFromJson.ClickHouseColumnName)]; found && !strings.Contains(columnFromJson.ClickHouseType, "Array") {
-			// Schema takes precedence over JSON (except for Arrays which are not currently handled)
-			result.WriteString(fmt.Sprintf("\"%s\" %s '%s'", columnFromSchema.ClickHouseColumnName, columnFromSchema.ClickHouseType+" COMMENT ", comment))
+		if columnFromSchema, found := columnsFromSchema[schema.FieldName(columnFromJson.ClickHouseColumnName)]; found {
+			// Schema takes precedence over inferred type from JSON
+			if strings.Contains(columnFromJson.ClickHouseType, "Array") {
+				// The schema (e.g. PUT /:index/_mapping) doesn't contain information about whether a field is an array or not.
+				// Therefore, we have to combine the information from the schema and the JSON in such case.
+				// For example: in the mapping we have a field "products.name" with type "keyword" (String)
+				// and in the JSON "products.name" is an array of strings (Array(String)).
+
+				if strings.Count(columnFromJson.ClickHouseType, "Array") > 1 {
+					logger.Warn().Msgf("Column '%s' has type '%s' - an array nested multiple times. Such case might not be handled correctly.", columnFromJson.ClickHouseColumnName, columnFromJson.ClickHouseType)
+				}
+
+				result.WriteString(fmt.Sprintf("\"%s\" Array(%s) COMMENT '%s'", columnFromSchema.ClickHouseColumnName, columnFromSchema.ClickHouseType, comment))
+			} else {
+				result.WriteString(fmt.Sprintf("\"%s\" %s COMMENT '%s'", columnFromSchema.ClickHouseColumnName, columnFromSchema.ClickHouseType, comment))
+			}
 		} else {
-			result.WriteString(fmt.Sprintf("\"%s\" %s '%s'", columnFromJson.ClickHouseColumnName, columnFromJson.ClickHouseType+" COMMENT ", comment))
+			result.WriteString(fmt.Sprintf("\"%s\" %s COMMENT '%s'", columnFromJson.ClickHouseColumnName, columnFromJson.ClickHouseType, comment))
 		}
 
 		delete(columnsFromSchema, schema.FieldName(columnFromJson.ClickHouseColumnName))