Ingest parser fallback (#1428)

<!-- 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' -->

---------

Co-authored-by: przemyslaw <[email protected]>

Author: pdelewski

.github/workflows/integration-tests.yml

@@ -46,6 +46,7 @@ jobs:
     uses: ./.github/workflows/build-quesma-docker-image.yml
     with:
       REF: ${{ github.event.inputs.GIT_REF || needs.check-comment.outputs.ref }}
+      VERSION: ${{ github.event.inputs.GIT_REF || needs.check-comment.outputs.ref }}
 
   integration-test-run:
     runs-on: ubuntu-latest
@@ -75,7 +76,9 @@ jobs:
           docker image ls -a  
 
       - name: Set environment variable
-        run: echo "EXECUTING_ON_GITHUB_CI=true" >> $GITHUB_ENV
+        run: | 
+          echo "EXECUTING_ON_GITHUB_CI=true" >> $GITHUB_ENV
+          echo "QUESMA_IT_VERSION=${{ github.event.inputs.GIT_REF || needs.check-comment.outputs.ref }}" >> $GITHUB_ENV
 
       - name: Get last commit author
         id: get_author

ci/it/testcases/utils.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/docker/docker/api/types/container"
+	"github.com/docker/go-connections/nat"
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/wait"
 	"io"
@@ -181,8 +182,15 @@ func setupQuesma(ctx context.Context, quesmaConfig string) (testcontainers.Conta
 	if err != nil {
 		return nil, err
 	}
+
+	quesmaVersion := os.Getenv("QUESMA_IT_VERSION")
+	if quesmaVersion == "" {
+		log.Println("No QUESMA_IT_VERSION environment variable set, watch out for stale images!")
+		quesmaVersion = "nightly"
+	}
+
 	quesmaReq := testcontainers.ContainerRequest{
-		Image:        "quesma/quesma:nightly",
+		Image:        fmt.Sprintf("quesma/quesma:%s", quesmaVersion),
 		ExposedPorts: []string{"0.0.0.0::9999/tcp", "0.0.0.0::8080/tcp"},
 		Env: map[string]string{
 			"QUESMA_CONFIG_FILE": "/configuration/conf.yaml",
@@ -245,7 +253,10 @@ func setupClickHouse(ctx context.Context) (testcontainers.Container, error) {
 		HostConfigModifier: func(hc *container.HostConfig) {
 			hc.ExtraHosts = []string{"localhost-for-github-ci:host-gateway"}
 		},
-		WaitingFor: wait.ForExposedPort().WithStartupTimeout(2 * time.Minute),
+		WaitingFor: wait.ForSQL("9000", "clickhouse",
+			func(host string, port nat.Port) string {
+				return fmt.Sprintf("clickhouse://%s:%d", host, port.Int())
+			}).WithStartupTimeout(2 * time.Minute),
 	}
 	return testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
 		ContainerRequest: req,

platform/ingest/parser.go

@@ -55,24 +55,25 @@ func columnsToString(columnsFromJson []CreateTableEntry,
 		columnMetadata := comment_metadata.NewCommentMetadata()
 		columnMetadata.Values[comment_metadata.ElasticFieldName] = propertyName
 		comment := columnMetadata.Marshall()
-
 		if columnFromSchema, found := columnsFromSchema[schema.FieldName(columnFromJson.ClickHouseColumnName)]; found {
-			// Schema takes precedence over inferred type from JSON
+			// Check if the type is an Array – if so, fallback to JSON type
 			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))
+				result.WriteString(fmt.Sprintf("\"%s\" %s '%s'", columnFromJson.ClickHouseColumnName, columnFromJson.ClickHouseType+" COMMENT ", comment))
+				// TODO this should be changed to use the schema type, but needs further investigation
+				//result.WriteString(fmt.Sprintf("\"%s\" Array(%s) COMMENT '%s'", columnFromSchema.ClickHouseColumnName, columnFromSchema.ClickHouseType, comment))
 			} else {
+				// Use schema type
 				result.WriteString(fmt.Sprintf("\"%s\" %s '%s'", columnFromSchema.ClickHouseColumnName, columnFromSchema.ClickHouseType+" COMMENT ", comment))
 			}
 		} else {
+			// Not found in schema – fallback to JSON type
 			result.WriteString(fmt.Sprintf("\"%s\" %s '%s'", columnFromJson.ClickHouseColumnName, columnFromJson.ClickHouseType+" COMMENT ", comment))
 		}