wip

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

go/vt/vtgate/planbuilder/operators/expressions.go

@@ -133,14 +133,18 @@ func getFirstSelect(selStmt sqlparser.TableStatement) *sqlparser.Select {
 func breakValuesJoinExpressionInLHS(ctx *plancontext.PlanningContext,
 	expr sqlparser.Expr,
 	lhs semantics.TableSet,
-) (results []*sqlparser.ColName) {
+) (result valuesJoinColumn) {
+	result.Original = expr
+	result.PureLHS = true
 	_ = sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 		col, ok := node.(*sqlparser.ColName)
 		if !ok {
 			return true, nil
 		}
 		if ctx.SemTable.RecursiveDeps(col) == lhs {
-			results = append(results, col)
+			result.LHS = append(result.LHS, col)
+		} else {
+			result.PureLHS = false
 		}
 		return true, nil
 	}, expr)

go/vt/vtgate/planbuilder/operators/expressions_test.go

@@ -49,8 +49,8 @@ func TestSplitComplexPredicateToLHS(t *testing.T) {
 		return false, nil
 	}, ast)
 
-	lhsExprs := breakValuesJoinExpressionInLHS(ctx, ast, lID)
-	nodes := slice.Map(lhsExprs, func(from *sqlparser.ColName) string {
+	valuesJoinCols := breakValuesJoinExpressionInLHS(ctx, ast, lID)
+	nodes := slice.Map(valuesJoinCols.LHS, func(from *sqlparser.ColName) string {
 		return sqlparser.String(from)
 	})
 

go/vt/vtgate/planbuilder/operators/query_planning.go

@@ -104,7 +104,8 @@ func runRewriters(ctx *plancontext.PlanningContext, root Operator) Operator {
 			return tryPushUpdate(in)
 		case *RecurseCTE:
 			return tryMergeRecurse(ctx, in)
-
+		case *Values:
+			return tryPushValues(in)
 		default:
 			return in, NoRewrite
 		}
@@ -120,6 +121,13 @@ func runRewriters(ctx *plancontext.PlanningContext, root Operator) Operator {
 	return FixedPointBottomUp(root, TableID, visitor, stopAtRoute)
 }
 
+func tryPushValues(in *Values) (Operator, *ApplyResult) {
+	if src, ok := in.Source.(*Route); ok {
+		return Swap(in, src, "pushed values under route")
+	}
+	return in, NoRewrite
+}
+
 func tryPushDelete(in *Delete) (Operator, *ApplyResult) {
 	if src, ok := in.Source.(*Route); ok {
 		return pushDMLUnderRoute(in, src, "pushed delete under route")

go/vt/vtgate/planbuilder/operators/route_planning.go

@@ -287,6 +287,31 @@ func requiresSwitchingSides(ctx *plancontext.PlanningContext, op Operator) (requ
 	return
 }
 
+func newJoin(ctx *plancontext.PlanningContext, lhs, rhs Operator, joinType sqlparser.JoinType) JoinOp {
+	lhsID := TableID(lhs)
+	if lhsID.NumberOfTables() > 1 || !joinType.IsInner() {
+		return NewApplyJoin(ctx, lhs, rhs, nil, joinType)
+	}
+	lhsTableInfo, err := ctx.SemTable.TableInfoFor(lhsID)
+	if err != nil {
+		panic(vterrors.VT13001(err.Error()))
+	}
+	lhsTableName, err := lhsTableInfo.Name()
+	if err != nil {
+		panic(vterrors.VT13001(err.Error()))
+	}
+	bindVariableName := ctx.ReservedVars.ReserveVariable("values")
+	v := &Values{
+		unaryOperator: newUnaryOp(rhs),
+		Name:          lhsTableName.Name.String(),
+		Arg:           bindVariableName,
+	}
+	return &ValuesJoin{
+		binaryOperator: newBinaryOp(lhs, v),
+		bindVarName:    bindVariableName,
+	}
+}
+
 func mergeOrJoin(ctx *plancontext.PlanningContext, lhs, rhs Operator, joinPredicates []sqlparser.Expr, joinType sqlparser.JoinType) (Operator, *ApplyResult) {
 	jm := newJoinMerge(joinPredicates, joinType)
 	newPlan := jm.mergeJoinInputs(ctx, lhs, rhs, joinPredicates)
@@ -305,14 +330,14 @@ func mergeOrJoin(ctx *plancontext.PlanningContext, lhs, rhs Operator, joinPredic
 			return join, Rewrote("use a hash join because we have LIMIT on the LHS")
 		}
 
-		join := NewApplyJoin(ctx, Clone(rhs), Clone(lhs), nil, joinType)
+		join := newJoin(ctx, Clone(rhs), Clone(lhs), joinType)
 		for _, pred := range joinPredicates {
 			join.AddJoinPredicate(ctx, pred)
 		}
 		return join, Rewrote("logical join to applyJoin, switching side because LIMIT")
 	}
 
-	join := NewApplyJoin(ctx, Clone(lhs), Clone(rhs), nil, joinType)
+	join := newJoin(ctx, Clone(lhs), Clone(rhs), joinType)
 	for _, pred := range joinPredicates {
 		join.AddJoinPredicate(ctx, pred)
 	}

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

@@ -27,13 +27,62 @@ type (
 
 		bindVarName string
 
-		noColumns
+		JoinColumns    []valuesJoinColumn
+		JoinPredicates []valuesJoinColumn
+
+		// After offset planning
+		Columns    []int
+		ColumnName []string
+	}
+
+	valuesJoinColumn struct {
+		Original sqlparser.Expr
+		LHS      []*sqlparser.ColName
+		PureLHS  bool
 	}
 )
 
 var _ Operator = (*ValuesJoin)(nil)
 var _ JoinOp = (*ValuesJoin)(nil)
 
+func (vj *ValuesJoin) AddColumn(ctx *plancontext.PlanningContext, reuseExisting bool, addToGroupBy bool, expr *sqlparser.AliasedExpr) int {
+	if reuseExisting {
+		if offset := vj.FindCol(ctx, expr.Expr, false); offset >= 0 {
+			return offset
+		}
+	}
+
+	vj.JoinColumns = append(vj.JoinColumns, breakValuesJoinExpressionInLHS(ctx, expr.Expr, TableID(vj.LHS)))
+	vj.ColumnName = append(vj.ColumnName, expr.ColumnName())
+	return len(vj.JoinColumns) - 1
+}
+
+// AddWSColumn is used to add a weight_string column to the operator
+func (vj *ValuesJoin) AddWSColumn(ctx *plancontext.PlanningContext, offset int, underRoute bool) int {
+	panic("oh no")
+}
+
+func (vj *ValuesJoin) FindCol(ctx *plancontext.PlanningContext, expr sqlparser.Expr, underRoute bool) int {
+	for offset, column := range vj.JoinColumns {
+		if ctx.SemTable.EqualsExpr(column.Original, expr) {
+			return offset
+		}
+	}
+	return -1
+}
+
+func (vj *ValuesJoin) GetColumns(ctx *plancontext.PlanningContext) []*sqlparser.AliasedExpr {
+	results := make([]*sqlparser.AliasedExpr, len(vj.JoinColumns))
+	for i, column := range vj.JoinColumns {
+		results = append(results, sqlparser.NewAliasedExpr(column.Original, vj.ColumnName[i]))
+	}
+	return results
+}
+
+func (vj *ValuesJoin) GetSelectExprs(ctx *plancontext.PlanningContext) sqlparser.SelectExprs {
+	return transformColumnsToSelectExprs(ctx, vj)
+}
+
 func (vj *ValuesJoin) GetLHS() Operator {
 	return vj.LHS
 }
@@ -63,23 +112,9 @@ func (vj *ValuesJoin) AddJoinPredicate(ctx *plancontext.PlanningContext, expr sq
 		return
 	}
 	lID := TableID(vj.LHS)
-	lhsCols := breakValuesJoinExpressionInLHS(ctx, expr, lID)
+	lhsJoinCols := breakValuesJoinExpressionInLHS(ctx, expr, lID)
 	vj.RHS = vj.RHS.AddPredicate(ctx, expr)
-
-	columns := ctx.ValuesJoinColumns[vj.bindVarName]
-
-outer:
-	for _, lhsCol := range lhsCols {
-		for _, ci := range columns {
-			if ci.Equal(lhsCol.Name) {
-				// already there, no need to add it again
-				continue outer
-			}
-		}
-		columns = append(columns, lhsCol.Name)
-	}
-
-	ctx.ValuesJoinColumns[vj.bindVarName] = columns
+	vj.JoinPredicates = append(vj.JoinPredicates, lhsJoinCols)
 }
 
 func (vj *ValuesJoin) Clone(inputs []Operator) Operator {
@@ -98,5 +133,29 @@ func (vj *ValuesJoin) GetOrdering(ctx *plancontext.PlanningContext) []OrderBy {
 }
 
 func (vj *ValuesJoin) planOffsets(ctx *plancontext.PlanningContext) Operator {
-	panic("implement me")
+	valuesColumns := ctx.ValuesJoinColumns[vj.bindVarName]
+	for i, jc := range vj.JoinColumns {
+		if jc.PureLHS {
+			offset := vj.LHS.AddColumn(ctx, true, false, sqlparser.NewAliasedExpr(jc.Original, vj.ColumnName[i]))
+			vj.Columns = append(vj.Columns, ToLeftOffset(offset))
+		} else {
+		outer:
+			for _, lh := range jc.LHS {
+				_ = vj.LHS.AddColumn(ctx, true, false, aeWrap(lh))
+				// TODO: can be optimized
+				for _, ci := range valuesColumns {
+					if ci.Equal(lh.Name) {
+						// already there, no need to add it again
+						continue outer
+					}
+				}
+				valuesColumns = append(valuesColumns, lh.Name)
+			}
+
+			vj.RHS.AddColumn(ctx, true, false)
+		}
+
+	}
+	ctx.ValuesJoinColumns[vj.bindVarName] = valuesColumns
+	return vj
 }

go/vt/vtgate/planbuilder/testdata/onecase.json

@@ -1,7 +1,7 @@
 [
   {
     "comment": "Add your test case here for debugging and run go test -run=One.",
-    "query": "",
+    "query": "select user.id, user_extra.user_id from user, user_extra where user.id = user_extra.toto",
     "plan": {
     }
   }