Fix Delete with multi-tables related by foreign keys (#15218)

Signed-off-by: Manan Gupta <[email protected]>

go/test/endtoend/vtgate/foreignkey/fk_test.go

@@ -967,6 +967,16 @@ func TestFkQueries(t *testing.T) {
 				"update fk_t16 set col = col * (col - (col)) where id = 4",
 			},
 		},
+		{
+			name: "Multi table delete that uses two tables related by foreign keys",
+			queries: []string{
+				"insert /*+ SET_VAR(foreign_key_checks=0) */ into fk_t10 (id, col) values (1, '5'), (2, NULL), (3, NULL), (4, '4'), (6, '1'), (7, '2')",
+				"insert /*+ SET_VAR(foreign_key_checks=0) */ into fk_t11 (id, col) values (4, '1'), (5, '3'), (7, '22'), (8, '5'), (9, NULL), (10, '3')",
+				"insert /*+ SET_VAR(foreign_key_checks=0) */ into fk_t12 (id, col) values (2, NULL), (3, NULL), (4, '1'), (6, '6'), (8, NULL), (10, '1')",
+				"insert /*+ SET_VAR(foreign_key_checks=0) */ into fk_t13 (id, col) values (2, '1'), (5, '5'), (7, '5')",
+				"delete fk_t11 from fk_t11 join fk_t12 using (id) where fk_t11.id = 4",
+			},
+		},
 	}
 
 	for _, testcase := range testcases {

go/vt/sqlparser/ast_format.go

@@ -177,7 +177,7 @@ func (node *Delete) Format(buf *TrackedBuffer) {
 	if node.Ignore {
 		buf.literal("ignore ")
 	}
-	if node.Targets != nil && !node.isSingleAliasExpr() {
+	if node.Targets != nil && !node.IsSingleAliasExpr() {
 		buf.astPrintf(node, "%v ", node.Targets)
 	}
 	prefix := "from "

go/vt/sqlparser/ast_funcs.go

@@ -2587,7 +2587,7 @@ func (ct *ColumnType) Invisible() bool {
 	return ct.Options.Invisible != nil && *ct.Options.Invisible
 }
 
-func (node *Delete) isSingleAliasExpr() bool {
+func (node *Delete) IsSingleAliasExpr() bool {
 	if len(node.Targets) > 1 {
 		return false
 	}

go/vt/vtgate/planbuilder/operators/delete.go

@@ -82,10 +82,11 @@ func (d *Delete) ShortDescription() string {
 func createOperatorFromDelete(ctx *plancontext.PlanningContext, deleteStmt *sqlparser.Delete) (op Operator) {
 	childFks := ctx.SemTable.GetChildForeignKeysForTable(deleteStmt.Targets[0])
 
-	// If the delete statement has a limit and has foreign keys, we will use a DMLWithInput
-	// operator wherein we do a selection first and use that output for the subsequent deletes.
-	if len(childFks) > 0 && deleteStmt.Limit != nil {
-		return deletePlanningForLimitFk(ctx, deleteStmt)
+	// We check if delete with input plan is required. DML with input planning is generally
+	// slower, because it does a selection and then creates a delete statement wherein we have to
+	// list all the primary key values.
+	if deleteWithInputPlanningRequired(childFks, deleteStmt) {
+		return deleteWithInputPlanningForFk(ctx, deleteStmt)
 	}
 
 	delClone := sqlparser.CloneRefOfDelete(deleteStmt)
@@ -107,7 +108,23 @@ func createOperatorFromDelete(ctx *plancontext.PlanningContext, deleteStmt *sqlp
 	return createFkCascadeOpForDelete(ctx, op, delClone, childFks, delOp.Target.VTable)
 }
 
-func deletePlanningForLimitFk(ctx *plancontext.PlanningContext, del *sqlparser.Delete) Operator {
+func deleteWithInputPlanningRequired(childFks []vindexes.ChildFKInfo, deleteStmt *sqlparser.Delete) bool {
+	// If there are no foreign keys, we don't need to use delete with input.
+	if len(childFks) == 0 {
+		return false
+	}
+	// Limit requires delete with input.
+	if deleteStmt.Limit != nil {
+		return true
+	}
+	// If there are no limit clauses, and it is not a multi-delete, we don't need delete with input.
+	// TODO: In the future, we can check if the tables involved in the multi-table delete are related by foreign keys or not.
+	// If they aren't then we don't need the multi-table delete. But this check isn't so straight-forward. We need to check if the two
+	// tables are connected in the undirected graph built from the tables related by foreign keys.
+	return !deleteStmt.IsSingleAliasExpr()
+}
+
+func deleteWithInputPlanningForFk(ctx *plancontext.PlanningContext, del *sqlparser.Delete) Operator {
 	delClone := ctx.SemTable.Clone(del).(*sqlparser.Delete)
 	del.Limit = nil
 	del.OrderBy = nil
@@ -129,7 +146,7 @@ func deletePlanningForLimitFk(ctx *plancontext.PlanningContext, del *sqlparser.D
 	var leftComp sqlparser.ValTuple
 	cols := make([]*sqlparser.ColName, 0, len(vTbl.PrimaryKey))
 	for _, col := range vTbl.PrimaryKey {
-		colName := sqlparser.NewColName(col.String())
+		colName := sqlparser.NewColNameWithQualifier(col.String(), vTbl.GetTableName())
 		selectStmt.SelectExprs = append(selectStmt.SelectExprs, aeWrap(colName))
 		cols = append(cols, colName)
 		leftComp = append(leftComp, colName)
@@ -141,6 +158,9 @@ func deletePlanningForLimitFk(ctx *plancontext.PlanningContext, del *sqlparser.D
 		lhs = leftComp[0]
 	}
 	compExpr := sqlparser.NewComparisonExpr(sqlparser.InOp, lhs, sqlparser.ListArg(engine.DmlVals), nil)
+
+	del.Targets = sqlparser.TableNames{del.Targets[0]}
+	del.TableExprs = sqlparser.TableExprs{ti.GetAliasedTableExpr()}
 	del.Where = sqlparser.NewWhere(sqlparser.WhereClause, compExpr)
 
 	return &DMLWithInput{