Operator for join-values

go/test/endtoend/vtgate/queries/misc/misc_test.go

@@ -19,6 +19,7 @@ package misc
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"testing"
@@ -34,7 +35,7 @@ import (
 	"vitess.io/vitess/go/test/endtoend/utils"
 )
 
-func start(t *testing.T) (utils.MySQLCompare, func()) {
+func start(t testing.TB) (utils.MySQLCompare, func()) {
 	mcmp, err := utils.NewMySQLCompare(t, vtParams, mysqlParams)
 	require.NoError(t, err)
 
@@ -53,6 +54,96 @@ func start(t *testing.T) (utils.MySQLCompare, func()) {
 	}
 }
 
+func BenchmarkValuesJoin(b *testing.B) {
+	mcmp, closer := start(b)
+	defer closer()
+
+	type Rep struct {
+		QueriesRouted map[string]int `json:"QueriesRouted"`
+	}
+
+	getQueriesRouted := func(thisB *testing.B) int {
+		_, response, _ := clusterInstance.VtgateProcess.MakeAPICall("/debug/vars")
+		r := Rep{}
+		err := json.Unmarshal([]byte(response), &r)
+		require.NoError(thisB, err)
+
+		var res int
+		for _, c := range r.QueriesRouted {
+			res += c
+		}
+		return res
+	}
+
+	b.ReportAllocs()
+
+	lhsRowCount := 0
+	rhsRowCount := 0
+
+	insertLHS := func(count int) {
+		for ; lhsRowCount < count; lhsRowCount++ {
+			mcmp.Exec(fmt.Sprintf("insert into t1(id1, id2) values (%d, %d)", lhsRowCount, lhsRowCount))
+		}
+	}
+	insertRHS := func(count int) {
+		for ; rhsRowCount < count; rhsRowCount++ {
+			mcmp.Exec(fmt.Sprintf("insert into tbl(id, unq_col, nonunq_col) values (%d, %d, %d)", rhsRowCount, rhsRowCount, rhsRowCount))
+		}
+	}
+
+	testCases := []struct {
+		lhsRowCount int
+		rhsRowCount int
+	}{
+		{
+			lhsRowCount: 20,
+			rhsRowCount: 10,
+		},
+		{
+			lhsRowCount: 50,
+			rhsRowCount: 25,
+		},
+		{
+			lhsRowCount: 100,
+			rhsRowCount: 50,
+		},
+		{
+			lhsRowCount: 200,
+			rhsRowCount: 100,
+		},
+		{
+			lhsRowCount: 500,
+			rhsRowCount: 250,
+		},
+		{
+			lhsRowCount: 1000,
+			rhsRowCount: 500,
+		},
+		{
+			lhsRowCount: 2000,
+			rhsRowCount: 1000,
+		},
+	}
+
+	var previousQueriesRoutedSum int
+	for _, testCase := range testCases {
+		insertLHS(testCase.lhsRowCount)
+		insertRHS(testCase.rhsRowCount)
+
+		b.Run(fmt.Sprintf("LHS(%d) RHS(%d)", testCase.lhsRowCount, testCase.rhsRowCount), func(b *testing.B) {
+			for range b.N {
+				mcmp.Exec("select t1.id1, tbl.id from t1, tbl where t1.id2 = tbl.nonunq_col")
+			}
+			b.StopTimer()
+
+			totalQueriesRouted := getQueriesRouted(b)
+			queriesRouted := totalQueriesRouted - previousQueriesRoutedSum
+			previousQueriesRoutedSum = totalQueriesRouted
+			b.ReportMetric(float64(queriesRouted/b.N), "queries_routed/op")
+		})
+	}
+}
+
 func TestBitVals(t *testing.T) {
 	mcmp, closer := start(t)
 	defer closer()

go/vt/sqlparser/comments.go

@@ -49,6 +49,8 @@ const (
 	DirectiveAllowHashJoin = "ALLOW_HASH_JOIN"
 	// DirectiveQueryPlanner lets the user specify per query which planner should be used
 	DirectiveQueryPlanner = "PLANNER"
+	// DirectiveAllowValuesJoin allows the planner to use VALUES JOINS when possible.
+	DirectiveAllowValuesJoin = "ALLOW_VALUES_JOIN"
 	// DirectiveVExplainRunDMLQueries tells vexplain queries/all that it is okay to also run the query.
 	DirectiveVExplainRunDMLQueries = "EXECUTE_DML_QUERIES"
 	// DirectiveConsolidator enables the query consolidator.
@@ -554,6 +556,10 @@ func AllowScatterDirective(stmt Statement) bool {
 	return checkDirective(stmt, DirectiveAllowScatter)
 }
 
+func AllowValuesJoinDirective(stmt Statement) bool {
+	return checkDirective(stmt, DirectiveAllowValuesJoin)
+}
+
 func checkDirective(stmt Statement, key string) bool {
 	cmt, ok := stmt.(Commented)
 	if ok {

go/vt/vtgate/engine/delete_test.go

@@ -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{
@@ -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{
@@ -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{
@@ -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{
@@ -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(-)")
 }
@@ -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)
@@ -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{
@@ -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)
@@ -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)
@@ -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{
@@ -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
 	})
@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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)

go/vt/vtgate/engine/dml_with_input_test.go

@@ -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{
@@ -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{
@@ -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{
@@ -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},
 	}