Adding siblings queries

Author: pdelewski

cmd/go.mod

@@ -3,7 +3,7 @@ module github.com/QuesmaOrg/quesma/quesma-cli
 go 1.24.0
 
 require (
-	github.com/QuesmaOrg/quesma/platform v0.0.0-20250303135300-04f0a4897c30
+	github.com/QuesmaOrg/quesma/platform v0.0.0-20250519105918-0f6942f1a3dd
 	github.com/goccy/go-json v0.10.5
 	github.com/stretchr/testify v1.10.0
 )

platform/frontend_connectors/search.go

@@ -694,15 +694,15 @@ type QueryJob func(ctx context.Context) (*model.ExecutionPlan, []model.QueryResu
 func (q *QueryRunner) runQueryJobsSequence(ctx context.Context, jobs []QueryJob) ([][]model.QueryResultRow, []clickhouse.PerformanceResult, error) {
 	var results = make([][]model.QueryResultRow, 0)
 	var performance = make([]clickhouse.PerformanceResult, 0)
-	for _, job := range jobs {
+	for n, job := range jobs {
 		plan, rows, perf, err := job(ctx)
 		performance = append(performance, perf)
 		if err != nil {
 			return nil, performance, err
 		}
 
 		results = append(results, rows)
-		if plan.Interrupt(rows) {
+		if plan.Interrupt(n, rows) {
 			break
 		}
 	}
@@ -753,7 +753,7 @@ func (q *QueryRunner) runQueryJobsParallel(ctx context.Context, jobs []QueryJob)
 			return nil, performances, res.err
 		}
 		logger.InfoWithCtx(ctx).Msg("Collected result")
-		if res.plan != nil && res.plan.Interrupt(res.rows) {
+		if res.plan != nil && res.plan.Interrupt(res.jobId, res.rows) {
 			logger.InfoWithCtx(ctx).Msg("Interrupting")
 			break
 		}

platform/go.mod

@@ -22,7 +22,6 @@ 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,8 +84,6 @@ 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,8 +72,7 @@ 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
-		ChildQueries *[]Query
+		Metadata JsonMap
 	}
 	QueryType interface {
 		// TranslateSqlResponseToJson
@@ -115,10 +114,11 @@ type ExecutionPlan struct {
 	// Interrupt function to stop the execution of the plan
 	// if for some reason we need to stop the execution
 	// e.g., there is some condition like enough results
-	Interrupt func(rows []QueryResultRow) bool
+	Interrupt func(queryId int, 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)
+	Siblings       map[int][]int // map of query id to list of sibling query ids
 }
 
 func NewQueryExecutionHints() *QueryOptimizeHints {
@@ -177,3 +177,46 @@ type HitsCountInfo struct {
 func NewEmptyHitsCountInfo() HitsCountInfo {
 	return HitsCountInfo{Type: Normal}
 }
+
+func (q *Query) Clone() *Query {
+	// Create a new Query object
+	clone := &Query{
+		SelectCommand:         q.SelectCommand, // Assuming SelectCommand has its own copy logic if needed
+		OptimizeHints:         nil,
+		TransformationHistory: q.TransformationHistory, // Assuming TransformationHistory is immutable or shallow copy is sufficient
+		Type:                  q.Type,
+		TableName:             q.TableName,
+		Indexes:               append([]string{}, q.Indexes...), // Deep copy of slice
+		Schema:                q.Schema,                         // Assuming schema.Schema is immutable or shallow copy is sufficient
+		Highlighter:           q.Highlighter,                    // Assuming Highlighter is immutable or shallow copy is sufficient
+		SearchAfter:           q.SearchAfter,                    // Assuming `any` is immutable or shallow copy is sufficient
+		RuntimeMappings:       make(map[string]RuntimeMapping),
+		Metadata:              nil,
+	}
+
+	// Deep copy OptimizeHints if it exists
+	if q.OptimizeHints != nil {
+		clone.OptimizeHints = &QueryOptimizeHints{
+			ClickhouseQuerySettings: make(map[string]any),
+			OptimizationsPerformed:  append([]string{}, q.OptimizeHints.OptimizationsPerformed...),
+		}
+		for k, v := range q.OptimizeHints.ClickhouseQuerySettings {
+			clone.OptimizeHints.ClickhouseQuerySettings[k] = v
+		}
+	}
+
+	// Deep copy RuntimeMappings
+	for k, v := range q.RuntimeMappings {
+		clone.RuntimeMappings[k] = v
+	}
+
+	// Deep copy Metadata if it exists
+	if q.Metadata != nil {
+		clone.Metadata = make(JsonMap)
+		for k, v := range q.Metadata {
+			clone.Metadata[k] = v
+		}
+	}
+
+	return clone
+}

platform/optimize/split_time_range_ext.go

@@ -5,8 +5,6 @@ package optimize
 import (
 	"github.com/QuesmaOrg/quesma/platform/logger"
 	"github.com/QuesmaOrg/quesma/platform/model"
-	"github.com/jinzhu/copier"
-	"log"
 	"sort"
 	"strconv"
 	"strings"
@@ -227,46 +225,54 @@ func (s splitTimeRangeExt) transformQuery(query *model.Query, properties map[str
 
 func (s splitTimeRangeExt) Transform(plan *model.ExecutionPlan, properties map[string]string) (*model.ExecutionPlan, error) {
 
-	var newQueries []*model.Query
-
-	for _, query := range plan.Queries {
+	queriesSubqueriesMapping := make(map[int][]*model.Query)
+	for i, query := range plan.Queries {
 		subqueries, err := s.transformQuery(query, properties)
 		if err != nil {
 			return nil, err
 		}
-		newQueries = append(newQueries, subqueries...)
+		queriesSubqueriesMapping[i] = subqueries
 	}
 
-	if len(newQueries) > 0 {
-		plan.Queries[0].SelectCommand = newQueries[0].SelectCommand
+	var newQueries []*model.Query
 
-		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
+	for i := range plan.Queries {
+		subqueries := queriesSubqueriesMapping[i]
+		plan.Queries[i].SelectCommand = subqueries[0].SelectCommand
+		for j := 1; j < len(subqueries); j++ {
+			newQuery := plan.Queries[0].Clone()
+			newQuery.SelectCommand = subqueries[j].SelectCommand
+			newQueries = append(newQueries, newQuery)
+			plan.Siblings[i] = append(plan.Siblings[i], j)
 		}
-		plan.ShouldBeMerged = true
 	}
-	for _, subquery := range newQueries {
-		sql := subquery.SelectCommand.String()
-		logger.Info().Msgf("@@@@@@Transformed query: %s", sql)
+
+	for i, subqueryPerQuery := range queriesSubqueriesMapping {
+		querySQL := plan.Queries[i].SelectCommand
+		logger.Info().Msgf("@@@@@@Original query: %s", querySQL.String())
+		for j := 0; j < len(subqueryPerQuery); j++ {
+			logger.Info().Msgf("@@@@@@Transformed query: %s", subqueryPerQuery[j].SelectCommand.String())
+		}
 	}
 
-	plan.Interrupt = func(rows []model.QueryResultRow) bool {
-		return len(rows) >= 500
+	plan.Interrupt = func(queryId int, rows []model.QueryResultRow) bool {
+		if _, ok := plan.Siblings[queryId]; ok {
+			return len(rows) >= 500
+		}
+		return false
 	}
 	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) {
+			// merge the results of the siblings queries
+			mergedResults = append(mergedResults, results[0])
+			// remove siblings queries from the plan
 			plan.Queries = plan.Queries[:len(plan.Queries)-1]
+			return plan, mergedResults
 		}
-		return plan, mergedResults
+		return plan, results
 	}
 	return plan, nil
 

platform/parsers/elastic_query_dsl/query_parser.go

@@ -101,7 +101,8 @@ func (cw *ClickhouseQueryTranslator) ParseQuery(body types.JSON) (*model.Executi
 	plan := &model.ExecutionPlan{
 		Queries:               queries,
 		QueryRowsTransformers: queryResultTransformers,
-		Interrupt: func(rows []model.QueryResultRow) bool {
+		Siblings:              make(map[int][]int),
+		Interrupt: func(queryId int, rows []model.QueryResultRow) bool {
 			return false
 		},
 	}