Array mapping parser improvements (#1427)

<!-- A note on testing your PR -->
<!-- Basic unit test run is executed against each commit in the PR.
If you want to run a full integration test suite, you can trigger it by
commenting
     with '/run-integration-tests' or '/run-it' -->

This is follow up to #976

Author: pdelewski

platform/ingest/parser.go

@@ -56,9 +56,22 @@ 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 '%s'", columnFromSchema.ClickHouseColumnName, columnFromSchema.ClickHouseType+" COMMENT ", comment))
+			}
 		} else {
 			result.WriteString(fmt.Sprintf("\"%s\" %s '%s'", columnFromJson.ClickHouseColumnName, columnFromJson.ClickHouseType+" COMMENT ", comment))
 		}