Skip to content
This repository was archived by the owner on Nov 7, 2025. It is now read-only.

Commit 8931985

Browse files
pdelewskipdelewski
committed
Changing Transform interface to take model.ExecutionPlan
1 parent 7614a7e commit 8931985

File tree

12 files changed

+75
-49
lines changed

12 files changed

+75
-49
lines changed

‎cmd/v2_test_objects.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -422,11 +422,11 @@ func (p *QueryTransformationPipeline) ComposeResult(results [][]model.QueryResul
422422
type QueryTransformer1 struct {
423423
}
424424

425-
func (p *QueryTransformer1) Transform(queries []*model.Query) ([]*model.Query, error) {
425+
func (p *QueryTransformer1) Transform(plan *model.ExecutionPlan) (*model.ExecutionPlan, error) {
426426
logger.Debug().Msg("SimpleQueryTransformationPipeline: Transform")
427427
// Do basic transformation
428428

429-
return queries, nil
429+
return plan, nil
430430
}
431431

432432
func NewQueryTransformer1() *QueryTransformer1 {

‎platform/frontend_connectors/schema_transformer.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -974,7 +974,7 @@ func (s *SchemaCheckPass) applyAliasColumns(indexSchema schema.Schema, query *mo
974974
return query, nil
975975
}
976976

977-
func (s *SchemaCheckPass) Transform(queries []*model.Query) ([]*model.Query, error) {
977+
func (s *SchemaCheckPass) Transform(plan *model.ExecutionPlan) (*model.ExecutionPlan, error) {
978978

979979
transformationChain := []struct {
980980
TransformationName string
@@ -1014,7 +1014,7 @@ func (s *SchemaCheckPass) Transform(queries []*model.Query) ([]*model.Query, err
10141014
{TransformationName: "BooleanLiteralTransformation", Transformation: s.applyBooleanLiteralLowering},
10151015
}
10161016

1017-
for k, query := range queries {
1017+
for k, query := range plan.Queries {
10181018
var err error
10191019

10201020
if !s.cfg.Logging.EnableSQLTracing {
@@ -1043,9 +1043,9 @@ func (s *SchemaCheckPass) Transform(queries []*model.Query) ([]*model.Query, err
10431043
}
10441044
}
10451045

1046-
queries[k] = query
1046+
plan.Queries[k] = query
10471047
}
1048-
return queries, nil
1048+
return plan, nil
10491049
}
10501050

10511051
func (s *SchemaCheckPass) applyMatchOperator(indexSchema schema.Schema, query *model.Query) (*model.Query, error) {

‎platform/frontend_connectors/schema_transformer_test.go‎

Lines changed: 26 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -499,10 +499,13 @@ func Test_ipRangeTransform(t *testing.T) {
499499
q.Schema = currentSchema
500500
q.Indexes = []string{q.TableName}
501501
}
502+
plan := &model.ExecutionPlan{
503+
Queries: queries[k],
504+
}
502505

503-
resultQueries, err := transform.Transform(queries[k])
506+
resultPlan, err := transform.Transform(plan)
504507
assert.NoError(t, err)
505-
assert.Equal(t, expectedQueries[k].SelectCommand.String(), resultQueries[0].SelectCommand.String())
508+
assert.Equal(t, expectedQueries[k].SelectCommand.String(), resultPlan.Queries[0].SelectCommand.String())
506509
})
507510
}
508511
}
@@ -739,17 +742,20 @@ func Test_arrayType(t *testing.T) {
739742
t.Run(util.PrettyTestName(tt.name, i), func(t *testing.T) {
740743
tt.query.Schema = indexSchema
741744
tt.query.Indexes = []string{tt.query.TableName}
742-
actual, err := transform.Transform([]*model.Query{tt.query})
745+
plan := &model.ExecutionPlan{
746+
Queries: []*model.Query{tt.query},
747+
}
748+
actual, err := transform.Transform(plan)
743749
assert.NoError(t, err)
744750

745751
if err != nil {
746752
t.Fatal(err)
747753
}
748754

749-
assert.True(t, len(actual) == 1, "len queries == 1")
755+
assert.True(t, len(actual.Queries) == 1, "len queries == 1")
750756

751757
expectedJson := asString(tt.expected)
752-
actualJson := asString(actual[0])
758+
actualJson := asString(actual.Queries[0])
753759

754760
assert.Equal(t, expectedJson, actualJson)
755761
})
@@ -1802,23 +1808,26 @@ func Test_mapKeys(t *testing.T) {
18021808
t.Run(util.PrettyTestName(tt.name, i), func(t *testing.T) {
18031809
tt.query.Schema = indexSchema
18041810
tt.query.Indexes = []string{tt.query.TableName}
1805-
var actual []*model.Query
1811+
var actual *model.ExecutionPlan
18061812
var err error
1813+
plan := &model.ExecutionPlan{
1814+
Queries: []*model.Query{tt.query},
1815+
}
18071816
if indexConfig[tt.query.TableName].EnableFieldMapSyntax {
1808-
actual, err = transformPass.Transform([]*model.Query{tt.query})
1817+
actual, err = transformPass.Transform(plan)
18091818
} else {
1810-
actual, err = noTransformPass.Transform([]*model.Query{tt.query})
1819+
actual, err = noTransformPass.Transform(plan)
18111820
}
18121821
assert.NoError(t, err)
18131822

18141823
if err != nil {
18151824
t.Fatal(err)
18161825
}
18171826

1818-
assert.True(t, len(actual) == 1, "len queries == 1")
1827+
assert.True(t, len(actual.Queries) == 1, "len queries == 1")
18191828

18201829
expectedJson := asString(tt.expected)
1821-
actualJson := asString(actual[0])
1830+
actualJson := asString(actual.Queries[0])
18221831

18231832
assert.Equal(t, expectedJson, actualJson)
18241833
})
@@ -1895,17 +1904,21 @@ func Test_cluster(t *testing.T) {
18951904
t.Run(util.PrettyTestName(tt.name, i), func(t *testing.T) {
18961905
tt.query.Schema = indexSchema
18971906
tt.query.Indexes = []string{tt.query.TableName}
1898-
actual, err := transform.Transform([]*model.Query{tt.query})
1907+
1908+
plan := &model.ExecutionPlan{
1909+
Queries: []*model.Query{tt.query},
1910+
}
1911+
actual, err := transform.Transform(plan)
18991912
assert.NoError(t, err)
19001913

19011914
if err != nil {
19021915
t.Fatal(err)
19031916
}
19041917

1905-
assert.True(t, len(actual) == 1, "len queries == 1")
1918+
assert.True(t, len(actual.Queries) == 1, "len queries == 1")
19061919

19071920
expectedJson := asString(tt.expected)
1908-
actualJson := asString(actual[0])
1921+
actualJson := asString(actual.Queries[0])
19091922

19101923
assert.Equal(t, expectedJson, actualJson)
19111924
})

‎platform/frontend_connectors/search.go‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -364,7 +364,7 @@ type AsyncQuery struct {
364364

365365
func (q *QueryRunner) transformQueries(plan *model.ExecutionPlan) error {
366366
var err error
367-
plan.Queries, err = q.transformationPipeline.Transform(plan.Queries)
367+
plan, err = q.transformationPipeline.Transform(plan)
368368
if err != nil {
369369
return fmt.Errorf("error transforming queries: %v", err)
370370
}
@@ -517,6 +517,18 @@ func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern strin
517517
goto logErrorAndReturn
518518
}
519519
err = q.transformQueries(plan)
520+
521+
// TODO only for debug purposes, remove it
522+
if len(plan.Queries) > 1 {
523+
logger.InfoWithCtx(ctx).Msgf("Parsed queries: %d", len(plan.Queries))
524+
bytes, _ := body.Bytes()
525+
logger.InfoWithCtx(ctx).Msgf("Body: %s", string(bytes))
526+
527+
for i, query := range plan.Queries {
528+
logger.InfoWithCtx(ctx).Msgf("Parsed query %d: %s", i, query.SelectCommand.String())
529+
}
530+
logger.InfoWithCtx(ctx).Msg("----------------------------------------")
531+
}
520532
if err != nil {
521533
goto logErrorAndReturn
522534
}
@@ -790,6 +802,7 @@ func (q *QueryRunner) searchWorkerCommon(
790802
var jobs []QueryJob
791803
var jobHitsPosition []int // it keeps the position of the hits array for each job
792804

805+
logger.InfoWithCtx(ctx).Msgf("search worker with query %d %v", len(queries), queries)
793806
for i, query := range queries {
794807
sql := query.SelectCommand.String()
795808

‎platform/model/transformation_pipeline.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,15 @@ func NewTransformationPipeline() *TransformationPipeline {
1010
return &TransformationPipeline{}
1111
}
1212

13-
func (o *TransformationPipeline) Transform(queries []*Query) ([]*Query, error) {
13+
func (o *TransformationPipeline) Transform(plan *ExecutionPlan) (*ExecutionPlan, error) {
1414
var err error
1515
for _, transformer := range o.transformers {
16-
queries, err = transformer.Transform(queries)
16+
plan, err = transformer.Transform(plan)
1717
if err != nil {
1818
return nil, err
1919
}
2020
}
21-
return queries, nil
21+
return plan, nil
2222
}
2323

2424
func (o *TransformationPipeline) AddTransformer(transformer QueryTransformer) {

‎platform/model/transformers.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ package model
55
import "context"
66

77
type QueryTransformer interface {
8-
Transform(query []*Query) ([]*Query, error)
8+
Transform(plan *ExecutionPlan) (*ExecutionPlan, error)
99
}
1010

1111
type ResultTransformer interface {

‎platform/optimize/cache_queries.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,9 +31,9 @@ func (s *cacheQueries) IsEnabledByDefault() bool {
3131
return false
3232
}
3333

34-
func (s *cacheQueries) Transform(queries []*model.Query, properties map[string]string) ([]*model.Query, error) {
34+
func (s *cacheQueries) Transform(plan *model.ExecutionPlan, properties map[string]string) (*model.ExecutionPlan, error) {
3535

36-
for _, query := range queries {
36+
for _, query := range plan.Queries {
3737

3838
var hasGroupBy bool
3939
var hasWindowFunction bool
@@ -83,5 +83,5 @@ func (s *cacheQueries) Transform(queries []*model.Query, properties map[string]s
8383
}
8484

8585
}
86-
return queries, nil
86+
return plan, nil
8787
}

‎platform/optimize/materialized_view_replace.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -202,14 +202,14 @@ func (s *materializedViewReplace) IsEnabledByDefault() bool {
202202
return false
203203
}
204204

205-
func (s *materializedViewReplace) Transform(queries []*model.Query, properties map[string]string) ([]*model.Query, error) {
205+
func (s *materializedViewReplace) Transform(plan *model.ExecutionPlan, properties map[string]string) (*model.ExecutionPlan, error) {
206206

207207
//
208208
// TODO add list of rules maybe
209209
//
210210
rule := s.readRule(properties)
211211

212-
for k, query := range queries {
212+
for k, query := range plan.Queries {
213213

214214
result, replaced := s.replace(rule, query.SelectCommand)
215215

@@ -218,9 +218,9 @@ func (s *materializedViewReplace) Transform(queries []*model.Query, properties m
218218
logger.Info().Msgf(s.Name()+" triggered, input query: %s", query.SelectCommand.String())
219219
logger.Info().Msgf(s.Name()+" triggered, output query: %s", (*result).String())
220220

221-
queries[k].SelectCommand = *result
221+
plan.Queries[k].SelectCommand = *result
222222
query.OptimizeHints.OptimizationsPerformed = append(query.OptimizeHints.OptimizationsPerformed, s.Name())
223223
}
224224
}
225-
return queries, nil
225+
return plan, nil
226226
}

‎platform/optimize/pipeline.go‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ import (
1111

1212
// OptimizeTransformer - an interface for query transformers that have a name.
1313
type OptimizeTransformer interface {
14-
Transform(queries []*model.Query, properties map[string]string) ([]*model.Query, error)
14+
Transform(plan *model.ExecutionPlan, properties map[string]string) (*model.ExecutionPlan, error)
1515

1616
Name() string // this name is used to enable/disable the transformer in the configuration
1717
IsEnabledByDefault() bool // should return true for "not aggressive" transformers only
@@ -74,14 +74,14 @@ func (s *OptimizePipeline) findConfig(transformer OptimizeTransformer, queries [
7474
return !transformer.IsEnabledByDefault(), make(map[string]string)
7575
}
7676

77-
func (s *OptimizePipeline) Transform(queries []*model.Query) ([]*model.Query, error) {
77+
func (s *OptimizePipeline) Transform(plan *model.ExecutionPlan) (*model.ExecutionPlan, error) {
7878

79-
if len(queries) == 0 {
80-
return queries, nil
79+
if len(plan.Queries) == 0 {
80+
return plan, nil
8181
}
8282

8383
// add hints if not present
84-
for _, query := range queries {
84+
for _, query := range plan.Queries {
8585
if query.OptimizeHints == nil {
8686
query.OptimizeHints = model.NewQueryExecutionHints()
8787
}
@@ -90,18 +90,18 @@ func (s *OptimizePipeline) Transform(queries []*model.Query) ([]*model.Query, er
9090
// run optimizations on queries
9191
for _, optimization := range s.optimizations {
9292

93-
disabled, properties := s.findConfig(optimization, queries)
93+
disabled, properties := s.findConfig(optimization, plan.Queries)
9494

9595
if disabled {
9696
continue
9797
}
9898

9999
var err error
100-
queries, err = optimization.Transform(queries, properties)
100+
plan, err = optimization.Transform(plan, properties)
101101
if err != nil {
102102
return nil, err
103103
}
104104
}
105105

106-
return queries, nil
106+
return plan, nil
107107
}

‎platform/optimize/split_time_range.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -267,15 +267,15 @@ func (s splitTimeRange) transformQuery(query *model.Query, properties map[string
267267
return query, nil
268268
}
269269

270-
func (s splitTimeRange) Transform(queries []*model.Query, properties map[string]string) ([]*model.Query, error) {
271-
for i, query := range queries {
270+
func (s splitTimeRange) Transform(plan *model.ExecutionPlan, properties map[string]string) (*model.ExecutionPlan, error) {
271+
for i, query := range plan.Queries {
272272
transformedQuery, err := s.transformQuery(query, properties)
273273
if err != nil {
274274
return nil, err
275275
}
276-
queries[i] = transformedQuery
276+
plan.Queries[i] = transformedQuery
277277
}
278-
return queries, nil
278+
return plan, nil
279279
}
280280

281281
func (s splitTimeRange) Name() string {

0 commit comments

Comments
 (0)