sql: add support for foreign key cascades in udfs

This commit adds testing and makes some fixes to support foreign key
cascades in UDFs.

Epic: CRDB-25388
Informs: #87289

Release note: none

pkg/sql/apply_join.go

@@ -286,6 +286,7 @@ func runPlanInsidePlan(
 
 	plannerCopy := *params.p
 	plannerCopy.curPlan.planComponents = *plan
+
 	// "Pausable portal" execution model is only applicable to the outer
 	// statement since we actually need to execute all inner plans to completion
 	// before we can produce any "outer" rows to be returned to the client, so
@@ -333,6 +334,12 @@ func runPlanInsidePlan(
 		// We don't have "inner" subqueries, so the apply join can only refer to
 		// the "outer" ones.
 		plannerCopy.curPlan.subqueryPlans = params.p.curPlan.subqueryPlans
+		// During cleanup, nil out the inner subquery plans before closing the plan
+		// components. Otherwise, we may inadvertently close nodes that are needed
+		// when executing the outer query.
+		defer func() {
+			plan.subqueryPlans = nil
+		}()
 	}
 
 	distributePlan := getPlanDistribution(
@@ -365,10 +372,13 @@ func runPlanInsidePlan(
 	evalCtxFactory2 := func(usedConcurrently bool) *extendedEvalContext {
 		return evalCtxFactory()
 	}
-
+	plannerCopy.autoCommit = false
 	execCfg.DistSQLPlanner.PlanAndRunCascadesAndChecks(
 		ctx, &plannerCopy, evalCtxFactory2, &plannerCopy.curPlan.planComponents, recv,
 	)
+	// We might have appended some cascades or checks to the plannerCopy, so we
+	// need to update the plan for cleanup purposes before proceeding.
+	*plan = plannerCopy.curPlan.planComponents
 	if recv.commErr != nil {
 		return recv.commErr
 	}

pkg/sql/logictest/testdata/logic_test/udf_fk

@@ -110,6 +110,8 @@ SELECT currval('s');
 ----
 1
 
+subtest end
+
 subtest delete
 
 statement ok
@@ -187,6 +189,7 @@ query II rowsort
 SELECT * FROM child
 ----
 
+subtest end
 
 subtest upsert
 
@@ -226,3 +229,219 @@ SELECT f_fk_c_ups(102,4);
 
 statement error pq: upsert on table "child" violates foreign key constraint "child_p_fkey"
 SELECT f_fk_c_ups(103,4);
+
+subtest end
+
+subtest cascade
+
+statement ok
+CREATE TABLE parent_cascade (p INT PRIMARY KEY);
+
+statement ok
+CREATE TABLE child_cascade (
+  c INT PRIMARY KEY,
+  p INT UNIQUE NOT NULL REFERENCES parent_cascade(p) ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+statement ok
+CREATE FUNCTION f_fk_p_cascade(old INT, new INT) RETURNS RECORD AS $$
+  UPDATE parent_cascade SET p = new WHERE p = old RETURNING *;
+$$ LANGUAGE SQL;
+
+statement ok
+INSERT INTO parent_cascade VALUES (1);
+
+statement ok
+INSERT INTO child_cascade VALUES (100,1);
+
+statement ok
+SELECT f_fk_p_cascade(1, 2);
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+100 2
+
+statement ok
+CREATE TABLE grandchild_cascade (
+  c INT PRIMARY KEY,
+  p INT NOT NULL REFERENCES child_cascade(p) ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+statement ok
+INSERT INTO grandchild_cascade VALUES (1000,2);
+
+statement ok
+SELECT f_fk_p_cascade(2, 3);
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+100 3
+
+query II rowsort
+SELECT * FROM grandchild_cascade;
+----
+1000 3
+
+statement ok
+CREATE FUNCTION f_fk_p_del_cascade(old INT) RETURNS RECORD AS $$
+  DELETE FROM parent_cascade WHERE p = old RETURNING *;
+$$ LANGUAGE SQL;
+
+statement ok
+SELECT f_fk_p_del_cascade(3);
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+
+query II rowsort
+SELECT * FROM grandchild_cascade;
+----
+
+statement ok
+INSERT INTO parent_cascade VALUES (1), (2);
+
+statement ok
+INSERT INTO child_cascade VALUES (1, 1), (2, 2);
+
+statement ok
+INSERT INTO grandchild_cascade VALUES (11, 1), (12, 2);
+
+query TT rowsort
+SELECT f_fk_p_cascade(1, 3), f_fk_p_cascade(2, 4);
+----
+(3) (4)
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+1 3
+2 4
+
+query II rowsort
+SELECT * FROM grandchild_cascade;
+----
+11 3
+12 4
+
+query TTT rowsort
+SELECT f_fk_p_cascade(3, 5), f_fk_p_del_cascade(4), f_fk_p_del_cascade(5);
+----
+(5) (4) (5)
+
+query I rowsort
+SELECT * FROM parent_cascade;
+----
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+
+query II rowsort
+SELECT * FROM grandchild_cascade;
+----
+
+statement ok
+DROP TABLE grandchild_cascade;
+
+statement ok
+DROP TABLE child_cascade;
+
+statement ok
+CREATE TABLE child_cascade (
+  c INT PRIMARY KEY,
+  p INT REFERENCES parent_cascade(p) ON DELETE SET NULL ON UPDATE SET NULL
+);
+
+statement ok
+INSERT INTO parent_cascade VALUES (3);
+
+statement ok
+INSERT INTO child_cascade VALUES (100,3);
+
+statement ok
+SELECT f_fk_p_cascade(3, 4);
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+100 NULL
+
+statement ok
+INSERT INTO child_cascade VALUES(101, 4);
+
+statement ok
+SELECT f_fk_p_del_cascade(4);
+
+query II rowsort
+SELECT * FROM child_cascade;
+----
+100 NULL
+101 NULL
+
+query I rowsort
+SELECT * FROM parent_cascade;
+----
+
+statement ok
+DROP TABLE child_cascade
+
+subtest end
+
+# Test a query with both an apply join and a UDF with a cascade.
+subtest apply_join
+
+statement ok
+CREATE TABLE child_cascade (
+  c INT PRIMARY KEY,
+  p INT UNIQUE NOT NULL REFERENCES parent_cascade(p) ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+statement ok
+CREATE TABLE grandchild_cascade (
+  c INT PRIMARY KEY,
+  p INT NOT NULL REFERENCES child_cascade(p) ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+statement ok
+CREATE OR REPLACE FUNCTION f_fk_p_cascade(old INT, new INT) RETURNS RECORD AS $$
+  UPDATE parent_cascade SET p = new WHERE p = old RETURNING *;
+$$ LANGUAGE SQL;
+
+statement ok
+INSERT INTO parent_cascade VALUES (1), (2), (3);
+
+statement ok
+INSERT INTO child_cascade VALUES (1, 1), (2, 2), (3, 3);
+
+statement ok
+INSERT INTO grandchild_cascade VALUES (11, 1), (12, 2), (13, 3);
+
+query IT rowsort
+SELECT
+	(SELECT * FROM (VALUES ((SELECT x FROM (VALUES (1)) AS s (x)) + y))),
+	f_fk_p_cascade(y, y+10)
+FROM
+	(VALUES (1), (2), (3)) AS t (y)
+----
+2 (11)
+3 (12)
+4 (13)
+
+query I rowsort
+SELECT p FROM child_cascade
+----
+11
+12
+13
+
+query I rowsort
+SELECT p FROM grandchild_cascade
+----
+11
+12
+13
+
+subtest end

pkg/sql/opt/exec/execbuilder/scalar.go

@@ -1159,9 +1159,6 @@ func (b *Builder) buildRoutinePlanGenerator(
 			if len(eb.subqueries) > 0 {
 				return expectedLazyRoutineError("subquery")
 			}
-			if len(eb.cascades) > 0 {
-				return expectedLazyRoutineError("cascade")
-			}
 			isFinalPlan := i == len(stmts)-1
 			err = fn(plan, isFinalPlan)
 			if err != nil {