Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Operator for join-values #17641

Draft
wants to merge 13 commits into
base: main
Choose a base branch
from
Draft

Operator for join-values #17641

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
455fe86
Cherry-picked the engine primitive for values join
frouioui Jan 14, 2025
bd6d67f
Enhance ValuesJoin engine and improve unit test
frouioui Jan 15, 2025
491f190
Merge remote-tracking branch 'origin/main' into values-join-engine
frouioui Jan 22, 2025
b4f62e3
wip - dirty implementation of findRoute for Values
frouioui Jan 22, 2025
9dda778
Finalize the Values OpCode and TestFindRouteValuesJoin
frouioui Jan 23, 2025
b265101
Use MultiEqual instead of Values OpCode
frouioui Jan 23, 2025
16a30f3
Add missing headers
frouioui Jan 28, 2025
e9fb2c7
Remove routing multi equal optimization
frouioui Jan 28, 2025
5793d60
Add VALUES operator and SQL_builder bit
frouioui Jan 28, 2025
2e7ae91
Add the ValuesJoin operator and SQL_builder test
frouioui Jan 29, 2025
7a695b0
Rename SQL_builder -> op_to_ast
frouioui Jan 29, 2025
ba53da7
add logic for splitting expressions
systay Jan 30, 2025
57f12cf
wip
frouioui Jan 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions go/vt/vtgate/engine/cached_size.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

36 changes: 18 additions & 18 deletions go/vt/vtgate/engine/delete_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ func TestDeleteUnsharded(t *testing.T) {
},
}

vc := newDMLTestVCursor("0")
vc := newTestVCursor("0")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -80,7 +80,7 @@ func TestDeleteEqual(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -112,7 +112,7 @@ func TestDeleteEqualMultiCol(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -148,7 +148,7 @@ func TestDeleteEqualNoRoute(t *testing.T) {
},
}

vc := newDMLTestVCursor("0")
vc := newTestVCursor("0")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -181,7 +181,7 @@ func TestDeleteEqualNoScatter(t *testing.T) {
},
}

vc := newDMLTestVCursor("0")
vc := newTestVCursor("0")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.EqualError(t, err, "cannot map vindex to unique keyspace id: DestinationKeyRange(-)")
}
Expand Down Expand Up @@ -213,7 +213,7 @@ func TestDeleteOwnedVindex(t *testing.T) {
"1|4|5|6",
)}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.results = results

_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
Expand All @@ -231,7 +231,7 @@ func TestDeleteOwnedVindex(t *testing.T) {
})

// No rows changing
vc = newDMLTestVCursor("-20", "20-")
vc = newTestVCursor("-20", "20-")
_, err = del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand All @@ -252,7 +252,7 @@ func TestDeleteOwnedVindex(t *testing.T) {
"1|4|5|6",
"1|7|8|9",
)}
vc = newDMLTestVCursor("-20", "20-")
vc = newTestVCursor("-20", "20-")
vc.results = results

_, err = del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
Expand Down Expand Up @@ -300,7 +300,7 @@ func TestDeleteOwnedVindexMultiCol(t *testing.T) {
"1|2|4",
)}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.results = results

_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
Expand Down Expand Up @@ -371,7 +371,7 @@ func TestDeleteSharded(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -399,7 +399,7 @@ func TestDeleteShardedStreaming(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
err := del.TryStreamExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false, func(result *sqltypes.Result) error {
return nil
})
Expand Down Expand Up @@ -435,7 +435,7 @@ func TestDeleteScatterOwnedVindex(t *testing.T) {
"1|4|5|6",
)}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.results = results

_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
Expand All @@ -453,7 +453,7 @@ func TestDeleteScatterOwnedVindex(t *testing.T) {
})

// No rows changing
vc = newDMLTestVCursor("-20", "20-")
vc = newTestVCursor("-20", "20-")

_, err = del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
Expand All @@ -475,7 +475,7 @@ func TestDeleteScatterOwnedVindex(t *testing.T) {
"1|4|5|6",
"1|7|8|9",
)}
vc = newDMLTestVCursor("-20", "20-")
vc = newTestVCursor("-20", "20-")
vc.results = results

_, err = del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
Expand Down Expand Up @@ -528,7 +528,7 @@ func TestDeleteInChangedVindexMultiCol(t *testing.T) {
"1|3|6",
"2|3|7",
)}
vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.results = results

_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
Expand Down Expand Up @@ -565,7 +565,7 @@ func TestDeleteEqualSubshard(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.shardForKsid = []string{"-20", "20-"}
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
Expand Down Expand Up @@ -602,7 +602,7 @@ func TestDeleteMultiEqual(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.shardForKsid = []string{"-20", "20-"}
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
Expand Down Expand Up @@ -635,7 +635,7 @@ func TestDeleteInUnique(t *testing.T) {
Type: querypb.Type_TUPLE,
Values: append([]*querypb.Value{sqltypes.ValueToProto(sqltypes.NewInt64(1))}, sqltypes.ValueToProto(sqltypes.NewInt64(2)), sqltypes.ValueToProto(sqltypes.NewInt64(4))),
}
vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.shardForKsid = []string{"-20", "20-"}
_, err := upd.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{"__vals": tupleBV}, false)
require.NoError(t, err)
Expand Down
8 changes: 4 additions & 4 deletions go/vt/vtgate/engine/dml_with_input_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ func TestDeleteWithInputSingleOffset(t *testing.T) {
OutputCols: [][]int{{0}},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -95,7 +95,7 @@ func TestDeleteWithInputMultiOffset(t *testing.T) {
OutputCols: [][]int{{1, 0}},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -160,7 +160,7 @@ func TestDeleteWithMultiTarget(t *testing.T) {
OutputCols: [][]int{{0}, {1, 2}},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
_, err := del.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, false)
require.NoError(t, err)
vc.ExpectLog(t, []string{
Expand Down Expand Up @@ -210,7 +210,7 @@ func TestUpdateWithInputNonLiteral(t *testing.T) {
},
}

vc := newDMLTestVCursor("-20", "20-")
vc := newTestVCursor("-20", "20-")
vc.results = []*sqltypes.Result{
{RowsAffected: 1}, {RowsAffected: 1}, {RowsAffected: 1},
}
Expand Down
13 changes: 10 additions & 3 deletions go/vt/vtgate/engine/fake_primitive_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,8 @@ type fakePrimitive struct {
allResultsInOneCall bool

async bool

useNewPrintBindVars bool
}

func (f *fakePrimitive) Inputs() ([]Primitive, []map[string]any) {
Expand All @@ -72,7 +74,12 @@ func (f *fakePrimitive) GetTableName() string {
}

func (f *fakePrimitive) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
f.log = append(f.log, fmt.Sprintf("Execute %v %v", printBindVars(bindVars), wantfields))
if f.useNewPrintBindVars {
f.log = append(f.log, fmt.Sprintf("Execute %v %v", printBindVars(bindVars), wantfields))
} else {
f.log = append(f.log, fmt.Sprintf("Execute %v %v", deprecatedPrintBindVars(bindVars), wantfields))
}

if f.results == nil {
return nil, f.sendErr
}
Expand All @@ -87,7 +94,7 @@ func (f *fakePrimitive) TryExecute(ctx context.Context, vcursor VCursor, bindVar

func (f *fakePrimitive) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
if !f.noLog {
f.log = append(f.log, fmt.Sprintf("StreamExecute %v %v", printBindVars(bindVars), wantfields))
f.log = append(f.log, fmt.Sprintf("StreamExecute %v %v", deprecatedPrintBindVars(bindVars), wantfields))
}
if f.results == nil {
return f.sendErr
Expand Down Expand Up @@ -171,7 +178,7 @@ func (f *fakePrimitive) asyncCall(callback func(*sqltypes.Result) error) error {
}

func (f *fakePrimitive) GetFields(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
f.log = append(f.log, fmt.Sprintf("GetFields %v", printBindVars(bindVars)))
f.log = append(f.log, fmt.Sprintf("GetFields %v", deprecatedPrintBindVars(bindVars)))
return f.TryExecute(ctx, vcursor, bindVars, true /* wantfields */)
}

Expand Down
47 changes: 43 additions & 4 deletions go/vt/vtgate/engine/fake_vcursor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -597,7 +597,7 @@ func (f *loggingVCursor) Execute(ctx context.Context, method string, query strin
case vtgatepb.CommitOrder_AUTOCOMMIT:
name = "ExecuteAutocommit"
}
f.log = append(f.log, fmt.Sprintf("%s %s %v %v", name, query, printBindVars(bindvars), rollbackOnError))
f.log = append(f.log, fmt.Sprintf("%s %s %v %v", name, query, deprecatedPrintBindVars(bindvars), rollbackOnError))
return f.nextResult()
}

Expand All @@ -621,7 +621,7 @@ func (f *loggingVCursor) AutocommitApproval() bool {
}

func (f *loggingVCursor) ExecuteStandalone(ctx context.Context, _ Primitive, query string, bindvars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard, fetchLastInsertID bool) (*sqltypes.Result, error) {
f.log = append(f.log, fmt.Sprintf("ExecuteStandalone %s %v %s %s", query, printBindVars(bindvars), rs.Target.Keyspace, rs.Target.Shard))
f.log = append(f.log, fmt.Sprintf("ExecuteStandalone %s %v %s %s", query, deprecatedPrintBindVars(bindvars), rs.Target.Keyspace, rs.Target.Shard))
return f.nextResult()
}

Expand Down Expand Up @@ -943,6 +943,24 @@ func expectResultAnyOrder(t *testing.T, result, want *sqltypes.Result) {
}
}

// deprecatedPrintBindVars does not print bind variables, specifically tuples, correctly.
// We should use printBindVars instead.
func deprecatedPrintBindVars(bindvars map[string]*querypb.BindVariable) string {
var keys []string
for k := range bindvars {
keys = append(keys, k)
}
sort.Strings(keys)
buf := &bytes.Buffer{}
for i, k := range keys {
if i > 0 {
fmt.Fprintf(buf, " ")
}
fmt.Fprintf(buf, "%s: %v", k, bindvars[k])
}
return buf.String()
}

func printBindVars(bindvars map[string]*querypb.BindVariable) string {
var keys []string
for k := range bindvars {
Expand All @@ -954,6 +972,27 @@ func printBindVars(bindvars map[string]*querypb.BindVariable) string {
if i > 0 {
fmt.Fprintf(buf, " ")
}

if bindvars[k].Type == querypb.Type_TUPLE {
fmt.Fprintf(buf, "%s: [", k)
for _, val := range bindvars[k].Values {
if val.Type != querypb.Type_TUPLE {
fmt.Fprintf(buf, "[%s]", val.String())
continue
}
var s []string
v := sqltypes.ProtoToValue(val)
err := v.ForEachValue(func(bv sqltypes.Value) {
s = append(s, bv.String())
})
if err != nil {
panic(err)
}
fmt.Fprintf(buf, "[%s]", strings.Join(s, " "))
}
fmt.Fprintf(buf, "]")
continue
}
fmt.Fprintf(buf, "%s: %v", k, bindvars[k])
}
return buf.String()
Expand All @@ -962,15 +1001,15 @@ func printBindVars(bindvars map[string]*querypb.BindVariable) string {
func printResolvedShardQueries(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery) string {
buf := &bytes.Buffer{}
for i, rs := range rss {
fmt.Fprintf(buf, "%s.%s: %s {%s} ", rs.Target.Keyspace, rs.Target.Shard, queries[i].Sql, printBindVars(queries[i].BindVariables))
fmt.Fprintf(buf, "%s.%s: %s {%s} ", rs.Target.Keyspace, rs.Target.Shard, queries[i].Sql, deprecatedPrintBindVars(queries[i].BindVariables))
}
return buf.String()
}

func printResolvedShardsBindVars(rss []*srvtopo.ResolvedShard, bvs []map[string]*querypb.BindVariable) string {
buf := &bytes.Buffer{}
for i, rs := range rss {
fmt.Fprintf(buf, "%s.%s: {%v} ", rs.Target.Keyspace, rs.Target.Shard, printBindVars(bvs[i]))
fmt.Fprintf(buf, "%s.%s: {%v} ", rs.Target.Keyspace, rs.Target.Shard, deprecatedPrintBindVars(bvs[i]))
}
return buf.String()
}
6 changes: 3 additions & 3 deletions go/vt/vtgate/engine/fk_cascade_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ func TestDeleteCascade(t *testing.T) {
Parent: parentP,
}

vc := newDMLTestVCursor("0")
vc := newTestVCursor("0")
vc.results = []*sqltypes.Result{fakeRes}
_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
require.NoError(t, err)
Expand Down Expand Up @@ -123,7 +123,7 @@ func TestUpdateCascade(t *testing.T) {
Parent: parentP,
}

vc := newDMLTestVCursor("0")
vc := newTestVCursor("0")
vc.results = []*sqltypes.Result{fakeRes}
_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
require.NoError(t, err)
Expand Down Expand Up @@ -195,7 +195,7 @@ func TestNonLiteralUpdateCascade(t *testing.T) {
Parent: parentP,
}

vc := newDMLTestVCursor("0")
vc := newTestVCursor("0")
vc.results = []*sqltypes.Result{fakeRes}
_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
require.NoError(t, err)
Expand Down
Loading
Loading