Add the ValuesJoin engine primitive

go/vt/vtgate/engine/join_values.go

@@ -0,0 +1,120 @@
+/*
+Copyright 2024 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 engine
+
+import (
+	"context"
+
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+var _ Primitive = (*ValuesJoin)(nil)
+
+// ValuesJoin is a primitive that joins two primitives by constructing a table from the rows of the LHS primitive.
+// The table is passed in as a bind variable to the RHS primitive.
+// It's called ValuesJoin because the LHS of the join is sent to the RHS as a VALUES clause.
+type ValuesJoin struct {
+	// Left and Right are the LHS and RHS primitives
+	// of the Join. They can be any primitive.
+	Left, Right Primitive
+
+	Vars              []int
+	RowConstructorArg string
+	Cols              []int
+	ColNames          []string
+}
+
+// TryExecute performs a non-streaming exec.
+func (jv *ValuesJoin) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	lresult, err := vcursor.ExecutePrimitive(ctx, jv.Left, bindVars, wantfields)
+	if err != nil {
+		return nil, err
+	}
+	bv := &querypb.BindVariable{
+		Type: querypb.Type_TUPLE,
+	}
+	if len(lresult.Rows) == 0 && wantfields {
+		// If there are no rows, we still need to construct a single row
+		// to send down to RHS for Values Table to execute correctly.
+		// It will be used to execute the field query to provide the output fields.
+		var vals []sqltypes.Value
+		for _, field := range lresult.Fields {
+			val, _ := sqltypes.NewValue(field.Type, nil)
+			vals = append(vals, val)
+		}
+		bv.Values = append(bv.Values, sqltypes.TupleToProto(vals))
+
+		bindVars[jv.RowConstructorArg] = bv
+		return jv.Right.GetFields(ctx, vcursor, bindVars)
+	}
+
+	for _, row := range lresult.Rows {
+		bv.Values = append(bv.Values, sqltypes.TupleToProto(row))
+	}
+	bindVars[jv.RowConstructorArg] = bv
+	return vcursor.ExecutePrimitive(ctx, jv.Right, bindVars, wantfields)
+}
+
+// TryStreamExecute performs a streaming exec.
+func (jv *ValuesJoin) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	panic("implement me")
+}
+
+// GetFields fetches the field info.
+func (jv *ValuesJoin) GetFields(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	return jv.Right.GetFields(ctx, vcursor, bindVars)
+}
+
+// Inputs returns the input primitives for this join
+func (jv *ValuesJoin) Inputs() ([]Primitive, []map[string]any) {
+	return []Primitive{jv.Left, jv.Right}, nil
+}
+
+// RouteType returns a description of the query routing type used by the primitive
+func (jv *ValuesJoin) RouteType() string {
+	return "ValuesJoin"
+}
+
+// GetKeyspaceName specifies the Keyspace that this primitive routes to.
+func (jv *ValuesJoin) GetKeyspaceName() string {
+	if jv.Left.GetKeyspaceName() == jv.Right.GetKeyspaceName() {
+		return jv.Left.GetKeyspaceName()
+	}
+	return jv.Left.GetKeyspaceName() + "_" + jv.Right.GetKeyspaceName()
+}
+
+// GetTableName specifies the table that this primitive routes to.
+func (jv *ValuesJoin) GetTableName() string {
+	return jv.Left.GetTableName() + "_" + jv.Right.GetTableName()
+}
+
+// NeedsTransaction implements the Primitive interface
+func (jv *ValuesJoin) NeedsTransaction() bool {
+	return jv.Right.NeedsTransaction() || jv.Left.NeedsTransaction()
+}
+
+func (jv *ValuesJoin) description() PrimitiveDescription {
+	return PrimitiveDescription{
+		OperatorType: "Join",
+		Variant:      "Values",
+		Other: map[string]any{
+			"ValuesArg": jv.RowConstructorArg,
+			"Vars":      jv.Vars,
+		},
+	}
+}

go/vt/vtgate/engine/join_values_test.go

@@ -0,0 +1,98 @@
+/*
+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 engine
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+func TestJoinValuesExecute(t *testing.T) {
+
+	/*
+		select col1, col2, col3, col4, col5, col6 from left join right on left.col1 = right.col4
+		LHS: select col1, col2, col3 from left
+		RHS: select id, col5, col6 from (values row(1,2), ...) left(id,col1) join right on left.col1 = right.col4
+	*/
+
+	leftPrim := &fakePrimitive{
+		results: []*sqltypes.Result{
+			sqltypes.MakeTestResult(
+				sqltypes.MakeTestFields(
+					"col1|col2|col3",
+					"int64|varchar|varchar",
+				),
+				"1|a|aa",
+				"2|b|bb",
+				"3|c|cc",
+				"4|d|dd",
+			),
+		},
+	}
+	rightPrim := &fakePrimitive{
+		results: []*sqltypes.Result{
+			sqltypes.MakeTestResult(
+				sqltypes.MakeTestFields(
+					"id|col5|col6",
+					"int64|varchar|varchar",
+				),
+				"1|d|dd",
+				"2|e|ee",
+				"3|f|ff",
+				"4|g|gg",
+			),
+		},
+	}
+
+	bv := map[string]*querypb.BindVariable{
+		"a": sqltypes.Int64BindVariable(10),
+	}
+
+	vjn := &ValuesJoin{
+		Left:              leftPrim,
+		Right:             rightPrim,
+		Vars:              []int{0},
+		RowConstructorArg: "v",
+		Cols:              []int{-1, -2, -3, -1, 2, 3},
+		ColNames:          []string{"col1", "col2", "col3", "col4", "col5", "col6"},
+	}
+
+	r, err := vjn.TryExecute(context.Background(), &noopVCursor{}, bv, true)
+	require.NoError(t, err)
+	leftPrim.ExpectLog(t, []string{
+		`Execute a: type:INT64 value:"10" true`,
+	})
+	rightPrim.ExpectLog(t, []string{
+		`Execute a: type:INT64 value:"10" v: type:TUPLE values:{type:TUPLE value:"\x89\x02\x011\x950\x01a\x950\x02aa"} values:{type:TUPLE value:"\x89\x02\x012\x950\x01b\x950\x02bb"} values:{type:TUPLE value:"\x89\x02\x013\x950\x01c\x950\x02cc"} values:{type:TUPLE value:"\x89\x02\x014\x950\x01d\x950\x02dd"} true`,
+	})
+
+	result := sqltypes.MakeTestResult(
+		sqltypes.MakeTestFields(
+			"col1|col2|col3|col4|col5|col6",
+			"int64|varchar|varchar|int64|varchar|varchar",
+		),
+		"1|a|aa|1|d|dd",
+		"2|b|bb|2|e|ee",
+		"3|c|cc|3|f|ff",
+		"4|d|dd|4|g|gg",
+	)
+	expectResult(t, r, result)
+}

proto/query.proto

@@ -219,7 +219,11 @@ enum Type {
   // Properties: 35, IsQuoted.
   VECTOR = 2083;
   // RAW specifies a type which won't be quoted but the value used as-is while encoding.
+  // Properties: 36, None.
   RAW = 2084;
+  // ROW_TUPLE represents multiple rows.
+  // Properties: 37, None.
+  ROW_TUPLE = 2085;
 }
 
 // Value represents a typed value.