Add the ValuesJoin operator and SQL_builder test

Signed-off-by: Florent Poinsard <[email protected]>

go/vt/vtgate/planbuilder/operators/SQL_builder.go

@@ -101,13 +101,7 @@ func buildQuery(op Operator, qb *queryBuilder) {
 	case *Union:
 		buildUnion(op, qb)
 	case *Distinct:
-		buildQuery(op.Source, qb)
-		statement := qb.asSelectStatement()
-		d, ok := statement.(sqlparser.Distinctable)
-		if !ok {
-			panic(vterrors.VT13001("expected a select statement with distinct"))
-		}
-		d.MakeDistinct()
+		buildDistinct(op, qb)
 	case *Update:
 		buildUpdate(op, qb)
 	case *Delete:
@@ -118,19 +112,42 @@ func buildQuery(op Operator, qb *queryBuilder) {
 		buildRecursiveCTE(op, qb)
 	case *Values:
 		buildValues(op, qb)
+	case *ValuesJoin:
+		buildValuesJoin(op, qb)
 	default:
 		panic(vterrors.VT13001(fmt.Sprintf("unknown operator to convert to SQL: %T", op)))
 	}
 }
 
+func buildDistinct(op *Distinct, qb *queryBuilder) {
+	buildQuery(op.Source, qb)
+	statement := qb.asSelectStatement()
+	d, ok := statement.(sqlparser.Distinctable)
+	if !ok {
+		panic(vterrors.VT13001("expected a select statement with distinct"))
+	}
+	d.MakeDistinct()
+}
+
+func buildValuesJoin(op *ValuesJoin, qb *queryBuilder) {
+	qb.ctx.SkipValuesArgument(op.bindVarName)
+	buildQuery(op.LHS, qb)
+	qbR := &queryBuilder{ctx: qb.ctx}
+	buildQuery(op.RHS, qbR)
+	qb.joinWith(qbR, nil, sqlparser.NormalJoinType)
+}
+
 func buildValues(op *Values, qb *queryBuilder) {
 	buildQuery(op.Source, qb)
+	if qb.ctx.IsValuesArgumentSkipped(op.Arg) {
+		return
+	}
+
 	qb.addTableExpr(op.Name, op.Name, TableID(op), &sqlparser.DerivedTable{
 		Select: &sqlparser.ValuesStatement{
 			ListArg: sqlparser.NewListArg(op.Arg),
 		},
 	}, nil, op.Columns)
-
 }
 
 func buildDelete(op *Delete, qb *queryBuilder) {

go/vt/vtgate/planbuilder/operators/SQL_builder_test.go

@@ -26,11 +26,10 @@ import (
 )
 
 func TestToSQLValues(t *testing.T) {
-	ctx := plancontext.PlanningContext{}
+	ctx := plancontext.CreateEmptyPlanningContext()
 
 	tableName := sqlparser.NewTableName("x")
 	tableColumn := sqlparser.NewColName("id")
-	valuesColumn := sqlparser.NewIdentifierCI("user_id")
 	op := &Values{
 		unaryOperator: newUnaryOp(&Table{
 			QTable: &QueryTable{
@@ -39,20 +38,81 @@ func TestToSQLValues(t *testing.T) {
 			},
 			Columns: []*sqlparser.ColName{tableColumn},
 		}),
-		Columns: sqlparser.Columns{valuesColumn},
+		Columns: sqlparser.Columns{sqlparser.NewIdentifierCI("user_id")},
 		Name:    "t",
 		Arg:     "toto",
 	}
 
-	stmt, _, err := ToSQL(&ctx, op)
+	stmt, _, err := ToSQL(ctx, op)
 	require.NoError(t, err)
 	require.Equal(t, "select id from x, (values ::toto) as t(user_id)", sqlparser.String(stmt))
 
+	// Now do the same test but with a projection on top
 	proj := newAliasedProjection(op)
 	proj.addUnexploredExpr(sqlparser.NewAliasedExpr(tableColumn, ""), tableColumn)
-	proj.addUnexploredExpr(sqlparser.NewAliasedExpr(sqlparser.NewColNameWithQualifier("user_id", sqlparser.NewTableName("t")), ""), sqlparser.NewColNameWithQualifier("user_id", sqlparser.NewTableName("t")))
 
-	stmt, _, err = ToSQL(&ctx, proj)
+	userIdColName := sqlparser.NewColNameWithQualifier("user_id", sqlparser.NewTableName("t"))
+	proj.addUnexploredExpr(
+		sqlparser.NewAliasedExpr(userIdColName, ""),
+		userIdColName,
+	)
+
+	stmt, _, err = ToSQL(ctx, proj)
 	require.NoError(t, err)
 	require.Equal(t, "select id, t.user_id from x, (values ::toto) as t(user_id)", sqlparser.String(stmt))
 }
+
+func TestToSQLValuesJoin(t *testing.T) {
+	ctx := plancontext.CreateEmptyPlanningContext()
+	parser := sqlparser.NewTestParser()
+
+	lhsTableName := sqlparser.NewTableName("x")
+	lhsTableColumn := sqlparser.NewColName("id")
+	lhsFilterPred, err := parser.ParseExpr("x.id = 42")
+	require.NoError(t, err)
+
+	LHS := &Filter{
+		unaryOperator: newUnaryOp(&Table{
+			QTable: &QueryTable{
+				Table: lhsTableName,
+				Alias: sqlparser.NewAliasedTableExpr(lhsTableName, ""),
+			},
+			Columns: []*sqlparser.ColName{lhsTableColumn},
+		}),
+		Predicates: []sqlparser.Expr{lhsFilterPred},
+	}
+
+	const argumentName = "v"
+
+	rhsTableName := sqlparser.NewTableName("y")
+	rhsTableColumn := sqlparser.NewColName("tata")
+	rhsFilterPred, err := parser.ParseExpr("y.tata = 42")
+	require.NoError(t, err)
+	rhsJoinFilterPred, err := parser.ParseExpr("y.tata = x.id")
+	require.NoError(t, err)
+
+	RHS := &Filter{
+		unaryOperator: newUnaryOp(&Values{
+			unaryOperator: newUnaryOp(&Table{
+				QTable: &QueryTable{
+					Table: rhsTableName,
+					Alias: sqlparser.NewAliasedTableExpr(rhsTableName, ""),
+				},
+				Columns: []*sqlparser.ColName{rhsTableColumn},
+			}),
+			Columns: sqlparser.Columns{sqlparser.NewIdentifierCI("id")},
+			Name:    lhsTableName.Name.String(),
+			Arg:     argumentName,
+		}),
+		Predicates: []sqlparser.Expr{rhsFilterPred, rhsJoinFilterPred},
+	}
+
+	vj := &ValuesJoin{
+		binaryOperator: newBinaryOp(LHS, RHS),
+		bindVarName:    argumentName,
+	}
+
+	stmt, _, err := ToSQL(ctx, vj)
+	require.NoError(t, err)
+	require.Equal(t, "select id, tata from x, y where x.id = 42 and y.tata = 42 and y.tata = x.id", sqlparser.String(stmt))
+}

go/vt/vtgate/planbuilder/operators/values.go

@@ -31,6 +31,9 @@ type Values struct {
 	Columns sqlparser.Columns
 	Name    string
 	Arg     string
+
+	// TODO: let's see if we want to have noColumns or no
+	// noColumns
 }
 
 func (v *Values) Clone(inputs []Operator) Operator {

go/vt/vtgate/planbuilder/operators/values_join.go

@@ -0,0 +1,46 @@
+/*
+Copyright 2025 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package operators
+
+import (
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
+)
+
+type ValuesJoin struct {
+	binaryOperator
+
+	bindVarName string
+
+	noColumns
+}
+
+func (v *ValuesJoin) Clone(inputs []Operator) Operator {
+	clone := *v
+	clone.LHS = inputs[0]
+	clone.RHS = inputs[1]
+	return &clone
+}
+
+func (v *ValuesJoin) ShortDescription() string {
+	return ""
+}
+
+func (v *ValuesJoin) GetOrdering(ctx *plancontext.PlanningContext) []OrderBy {
+	return v.RHS.GetOrdering(ctx)
+}
+
+var _ Operator = (*ValuesJoin)(nil)

go/vt/vtgate/planbuilder/plancontext/planning_context.go

@@ -43,6 +43,12 @@ type PlanningContext struct {
 	// a join predicate is reverted to its original form during planning.
 	skipPredicates map[sqlparser.Expr]any
 
+	// skipValuesArgument tracks Values operator that should be skipped when
+	// rewriting the operator tree to an AST tree.
+	// This happens when a ValuesJoin is pushed under a route and we do not
+	// need to have a Values operator anymore on its RHS.
+	skipValuesArgument map[string]any
+
 	PlannerVersion querypb.ExecuteOptions_PlannerVersion
 
 	// If we during planning have turned this expression into an argument name,
@@ -81,6 +87,15 @@ type PlanningContext struct {
 	constantCfg *evalengine.Config
 }
 
+func CreateEmptyPlanningContext() *PlanningContext {
+	return &PlanningContext{
+		joinPredicates:     make(map[sqlparser.Expr][]sqlparser.Expr),
+		skipPredicates:     make(map[sqlparser.Expr]any),
+		skipValuesArgument: make(map[string]any),
+		ReservedArguments:  make(map[sqlparser.Expr]string),
+	}
+}
+
 // CreatePlanningContext initializes a new PlanningContext with the given parameters.
 // It analyzes the SQL statement within the given virtual schema context,
 // handling default keyspace settings and semantic analysis.
@@ -104,14 +119,15 @@ func CreatePlanningContext(stmt sqlparser.Statement,
 	vschema.PlannerWarning(semTable.Warning)
 
 	return &PlanningContext{
-		ReservedVars:      reservedVars,
-		SemTable:          semTable,
-		VSchema:           vschema,
-		joinPredicates:    map[sqlparser.Expr][]sqlparser.Expr{},
-		skipPredicates:    map[sqlparser.Expr]any{},
-		PlannerVersion:    version,
-		ReservedArguments: map[sqlparser.Expr]string{},
-		Statement:         stmt,
+		ReservedVars:       reservedVars,
+		SemTable:           semTable,
+		VSchema:            vschema,
+		joinPredicates:     map[sqlparser.Expr][]sqlparser.Expr{},
+		skipPredicates:     map[sqlparser.Expr]any{},
+		skipValuesArgument: map[string]any{},
+		PlannerVersion:     version,
+		ReservedArguments:  map[sqlparser.Expr]string{},
+		Statement:          stmt,
 	}, nil
 }
 
@@ -176,6 +192,15 @@ func (ctx *PlanningContext) SkipJoinPredicates(joinPred sqlparser.Expr) error {
 	return vterrors.VT13001("predicate does not exist: " + sqlparser.String(joinPred))
 }
 
+func (ctx *PlanningContext) SkipValuesArgument(name string) {
+	ctx.skipValuesArgument[name] = ""
+}
+
+func (ctx *PlanningContext) IsValuesArgumentSkipped(name string) bool {
+	_, ok := ctx.skipValuesArgument[name]
+	return ok
+}
+
 // KeepPredicateInfo transfers join predicate information from another context.
 // This is useful when nesting queries, ensuring consistent predicate handling across contexts.
 func (ctx *PlanningContext) KeepPredicateInfo(other *PlanningContext) {