Fix cascading Delete failure while using Prepared statements

go/test/vschemawrapper/vschema_wrapper.go

@@ -67,6 +67,11 @@ func (vw *VSchemaWrapper) GetPrepareData(stmtName string) *vtgatepb.PrepareData
 			PrepareStatement: "select 1 from user",
 			ParamsCount:      0,
 		}
+	case "prep_delete":
+		return &vtgatepb.PrepareData{
+			PrepareStatement: "delete from tbl5 where id = :v1",
+			ParamsCount:      1,
+		}
 	}
 	return nil
 }

go/vt/vtgate/engine/exec_prepared_statement.go

@@ -31,8 +31,10 @@ var _ Primitive = (*ExecStmt)(nil)
 type ExecStmt struct {
 	Params []*sqlparser.Variable
 	Input  Primitive
+}
 
-	noTxNeeded
+func (e *ExecStmt) NeedsTransaction() bool {
+	return e.Input.NeedsTransaction()
 }
 
 func (e *ExecStmt) RouteType() string {

go/vt/vtgate/engine/fk_cascade_test.go

@@ -148,3 +148,13 @@ func TestUpdateCascade(t *testing.T) {
 		`ExecuteMultiShard ks.0: update parent set cola = 1 where foo = 48 {} true true`,
 	})
 }
+
+// TestNeedsTransactionInExecPrepared tests that if we have a foreign key cascade inside an ExecStmt plan, then we do mark the plan to require a transaction.
+func TestNeedsTransactionInExecPrepared(t *testing.T) {
+	// Even if FkCascade is wrapped in ExecStmt, the plan should be marked such that it requires a transaction.
+	// This is necessary because if we don't run the cascades for DMLs in a transaction, we might end up committing partial writes that should eventually be rolled back.
+	execPrepared := &ExecStmt{
+		Input: &FkCascade{},
+	}
+	require.True(t, execPrepared.NeedsTransaction())
+}

go/vt/vtgate/planbuilder/plan_test.go

@@ -102,7 +102,8 @@ func TestForeignKeyPlanning(t *testing.T) {
 	vschema := loadSchema(t, "vschemas/schema.json", true)
 	setFks(t, vschema)
 	vschemaWrapper := &vschemawrapper.VSchemaWrapper{
-		V: vschema,
+		V:           vschema,
+		TestBuilder: TestBuilder,
 	}
 
 	testOutputTempDir := makeTestOutput(t)
@@ -212,7 +213,8 @@ func TestOne(t *testing.T) {
 	lv := loadSchema(t, "vschemas/schema.json", true)
 	setFks(t, lv)
 	vschema := &vschemawrapper.VSchemaWrapper{
-		V: lv,
+		V:           lv,
+		TestBuilder: TestBuilder,
 	}
 
 	testFile(t, "onecase.json", "", vschema, false)

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

@@ -1389,5 +1389,86 @@
         "unsharded_fk_allow.u_multicol_tbl3"
       ]
     }
+  },
+  {
+    "comment": "Cascaded delete run from prepared statement",
+    "query": "execute prep_delete using @foo",
+    "plan": {
+      "QueryType": "EXECUTE",
+      "Original": "execute prep_delete using @foo",
+      "Instructions": {
+        "OperatorType": "EXECUTE",
+        "Parameters": [
+          "foo"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "FkCascade",
+            "Inputs": [
+              {
+                "InputName": "Selection",
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "sharded_fk_allow",
+                  "Sharded": true
+                },
+                "FieldQuery": "select col5, t5col5 from tbl5 where 1 != 1",
+                "Query": "select col5, t5col5 from tbl5 where id = :v1 for update",
+                "Table": "tbl5"
+              },
+              {
+                "InputName": "CascadeChild-1",
+                "OperatorType": "Delete",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "sharded_fk_allow",
+                  "Sharded": true
+                },
+                "TargetTabletType": "PRIMARY",
+                "BvName": "fkc_vals",
+                "Cols": [
+                  0
+                ],
+                "Query": "delete from tbl4 where (col4) in ::fkc_vals",
+                "Table": "tbl4"
+              },
+              {
+                "InputName": "CascadeChild-2",
+                "OperatorType": "Delete",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "sharded_fk_allow",
+                  "Sharded": true
+                },
+                "TargetTabletType": "PRIMARY",
+                "BvName": "fkc_vals1",
+                "Cols": [
+                  1
+                ],
+                "Query": "delete from tbl4 where (t4col4) in ::fkc_vals1",
+                "Table": "tbl4"
+              },
+              {
+                "InputName": "Parent",
+                "OperatorType": "Delete",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "sharded_fk_allow",
+                  "Sharded": true
+                },
+                "TargetTabletType": "PRIMARY",
+                "Query": "delete from tbl5 where id = :v1",
+                "Table": "tbl5"
+              }
+            ]
+          }
+        ]
+      },
+      "TablesUsed": [
+        "sharded_fk_allow.tbl4",
+        "sharded_fk_allow.tbl5"
+      ]
+    }
   }
 ]