Some intermediate changes

Author: pdelewski

platform/frontend_connectors/search.go

@@ -395,6 +395,9 @@ func (q *QueryRunner) runExecutePlanAsync(ctx context.Context, plan *model.Execu
 			doneCh <- asyncSearchWithError{translatedQueryBody: translatedQueryBody, err: err}
 		}
 
+		if plan.Merge != nil && plan.ShouldBeMerged {
+			plan, results = plan.Merge(plan, results)
+		}
 		searchResponse := queryTranslator.MakeSearchResponse(plan.Queries, results)
 
 		doneCh <- asyncSearchWithError{response: searchResponse, translatedQueryBody: translatedQueryBody, err: err}
@@ -516,10 +519,15 @@ func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern strin
 		logger.ErrorWithCtxAndReason(ctx, "Quesma generated invalid SQL query").Msg(queriesBodyConcat)
 		goto logErrorAndReturn
 	}
+	if len(plan.Queries) > 0 {
+		for i, query := range plan.Queries {
+			logger.InfoWithCtx(ctx).Msgf("Input SQL query %d: %s", i, query.SelectCommand.String())
+		}
+	}
 	err = q.transformQueries(plan)
 
 	// TODO only for debug purposes, remove it
-	if len(plan.Queries) > 1 {
+	if len(plan.Queries) > 0 {
 		logger.InfoWithCtx(ctx).Msgf("Parsed queries: %d", len(plan.Queries))
 		bytes, _ := body.Bytes()
 		logger.InfoWithCtx(ctx).Msgf("Body: %s", string(bytes))

platform/go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/huandu/go-clone v1.7.2
 	github.com/ip2location/ip2location-go/v9 v9.7.1
 	github.com/jackc/pgx/v5 v5.7.2
+	github.com/jinzhu/copier v0.4.0
 	github.com/k0kubun/pp v3.0.1+incompatible
 	github.com/knadh/koanf/parsers/json v0.1.0
 	github.com/knadh/koanf/parsers/yaml v0.1.0

platform/go.sum

@@ -84,6 +84,8 @@ github.com/jackc/pgx/v5 v5.7.2 h1:mLoDLV6sonKlvjIEsV56SkWNCnuNv531l94GaIzO+XI=
 github.com/jackc/pgx/v5 v5.7.2/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/jinzhu/copier v0.4.0 h1:w3ciUoD19shMCRargcpm0cm91ytaBhDvuRpz1ODO/U8=
+github.com/jinzhu/copier v0.4.0/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
 github.com/k0kubun/pp v3.0.1+incompatible h1:3tqvf7QgUnZ5tXO6pNAZlrvHgl6DvifjDrd9g2S9Z40=

platform/model/query.go

@@ -72,7 +72,8 @@ type (
 		// WARNING: it's probably not passed everywhere where it's needed, just in one place.
 		// But it works for the test + our dashboards, so let's fix it later if necessary.
 		// NoMetadataField (nil) is a valid option and means no meta field in the response.
-		Metadata JsonMap
+		Metadata     JsonMap
+		ChildQueries *[]Query
 	}
 	QueryType interface {
 		// TranslateSqlResponseToJson
@@ -115,6 +116,9 @@ type ExecutionPlan struct {
 	// if for some reason we need to stop the execution
 	// e.g., there is some condition like enough results
 	Interrupt func(rows []QueryResultRow) bool
+
+	ShouldBeMerged bool // if true, we should merge the results of this plan with the main plan
+	Merge          func(plan *ExecutionPlan, results [][]QueryResultRow) (*ExecutionPlan, [][]QueryResultRow)
 }
 
 func NewQueryExecutionHints() *QueryOptimizeHints {

platform/optimize/split_time_range_ext.go

@@ -5,6 +5,8 @@ package optimize
 import (
 	"github.com/QuesmaOrg/quesma/platform/logger"
 	"github.com/QuesmaOrg/quesma/platform/model"
+	"github.com/jinzhu/copier"
+	"log"
 	"sort"
 	"strconv"
 	"strings"
@@ -237,6 +239,17 @@ func (s splitTimeRangeExt) Transform(plan *model.ExecutionPlan, properties map[s
 
 	if len(newQueries) > 0 {
 		plan.Queries[0].SelectCommand = newQueries[0].SelectCommand
+
+		for i := 1; i < len(newQueries); i++ {
+			var queryCopy model.Query
+			err := copier.Copy(&queryCopy, &plan.Queries[0])
+			if err != nil {
+				log.Println("copier.Copy failed:", err)
+			}
+			plan.Queries = append(plan.Queries, &queryCopy)
+			plan.Queries[i].SelectCommand = newQueries[i].SelectCommand
+		}
+		plan.ShouldBeMerged = true
 	}
 	for _, subquery := range newQueries {
 		sql := subquery.SelectCommand.String()
@@ -246,7 +259,15 @@ func (s splitTimeRangeExt) Transform(plan *model.ExecutionPlan, properties map[s
 	plan.Interrupt = func(rows []model.QueryResultRow) bool {
 		return len(rows) >= 500
 	}
-
+	plan.Merge = func(plan *model.ExecutionPlan, results [][]model.QueryResultRow) (*model.ExecutionPlan, [][]model.QueryResultRow) {
+		var mergedResults [][]model.QueryResultRow
+		mergedResults = make([][]model.QueryResultRow, 0)
+		mergedResults = append(mergedResults, results[0])
+		if len(plan.Queries) > len(mergedResults) {
+			plan.Queries = plan.Queries[:len(plan.Queries)-1]
+		}
+		return plan, mergedResults
+	}
 	return plan, nil
 
 }

platform/parsers/elastic_query_dsl/query_translator.go

@@ -229,6 +229,20 @@ func (cw *ClickhouseQueryTranslator) makeTotalCount(queries []*model.Query, resu
 func (cw *ClickhouseQueryTranslator) MakeSearchResponse(queries []*model.Query, ResultSets [][]model.QueryResultRow) *model.SearchResp {
 	var hits *model.SearchHits
 	var total *model.Total
+
+	if len(queries) != len(ResultSets) {
+		logger.ErrorWithCtx(cw.Ctx).Msgf("queries and resultsets have different length: %d != %d", len(queries), len(ResultSets))
+		return &model.SearchResp{
+			Aggregations: nil,
+			Timeout:      false,
+			Shards: model.ResponseShards{
+				Total:      1,
+				Successful: 1,
+				Failed:     0,
+			},
+		}
+	}
+
 	queries, ResultSets, total = cw.makeTotalCount(queries, ResultSets) // get hits and remove it from queries
 	queries, ResultSets, hits = cw.makeHits(queries, ResultSets)        // get hits and remove it from queries