plpgsql: add execution support for OPEN statements

This patch adds support for executing PLpgSQL OPEN statements, which
open a SQL cursor in the current transaction. The name of the cursor
is supplied through a PLpgSQL variable. Since the `REFCURSOR` type
hasn't been implemented yet, this patch uses `STRING` in the
meantime.

Limitations that will be lifted in future PRs:
1. Unnamed cursor declarations are not supported. If a cursor is opened
   with no name supplied, a name should be automatically generated.
2. Bound cursors are not yet supported. It should be possible to declare
   a cursor in the `DECLARE` block with the query already defined, at
   which point it can be opened with `OPEN <cursor>;`.
3. A cursor cannot be opened in a routine with an exception block. This
   is because correct handling of this case is waiting on separate work
   to implement rollback of changes to database state on exceptions.

Informs cockroachdb#109709

Release note (sql change): Added initial support for executing the
PLpgSQL `OPEN` statement, which allows a PLpgSQL routine to create a
cursor. Currently, opening bound or unnamed cursors is not supported.
In addition, `OPEN` statements cannot be used in a routine with an
exception block.

pkg/sql/conn_executor.go

@@ -1221,6 +1221,11 @@ func (ex *connExecutor) close(ctx context.Context, closeType closeType) {
 		ctx, &ex.extraTxnState.prepStmtsNamespaceMemAcc,
 	)
 
+	// Close all cursors.
+	if err := ex.extraTxnState.sqlCursors.closeAll(false /* errorOnWithHold */); err != nil {
+		log.Warningf(ctx, "error closing cursors: %v", err)
+	}
+
 	if closeType == normalClose {
 		// We'll cleanup the SQL txn by creating a non-retriable (commit:true) event.
 		// This event is guaranteed to be accepted in every state.
@@ -1277,9 +1282,6 @@ func (ex *connExecutor) close(ctx context.Context, closeType closeType) {
 			ctx, &ex.extraTxnState.prepStmtsNamespaceMemAcc,
 		)
 		ex.extraTxnState.prepStmtsNamespaceMemAcc.Close(ctx)
-		if err := ex.extraTxnState.sqlCursors.closeAll(false /* errorOnWithHold */); err != nil {
-			log.Warningf(ctx, "error closing cursors: %v", err)
-		}
 	}
 
 	if ex.sessionTracing.Enabled() {
@@ -3864,6 +3866,10 @@ func (ex *connExecutor) txnStateTransitionsApplyWrapper(
 		ex.extraTxnState.prepStmtsNamespaceAtTxnRewindPos.closeAllPortals(
 			ex.Ctx(), &ex.extraTxnState.prepStmtsNamespaceMemAcc,
 		)
+		// Close all cursors.
+		if err := ex.extraTxnState.sqlCursors.closeAll(false /* errorOnWithHold */); err != nil {
+			log.Warningf(ex.Ctx(), "error closing cursors: %v", err)
+		}
 		ex.resetPlanner(ex.Ctx(), &ex.planner, nil, ex.server.cfg.Clock.PhysicalTime())
 	case txnRestart:
 		// In addition to resetting the extraTxnState, the restart event may

pkg/sql/logictest/testdata/logic_test/udf_plpgsql

@@ -1914,3 +1914,354 @@ CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
     RETURN i;
   END
 $$ LANGUAGE PLpgSQL;
+
+# Testing OPEN statements.
+statement ok
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query I
+FETCH FORWARD 3 FROM foo;
+----
+1
+
+statement ok
+ABORT;
+
+statement error pgcode 34000 pq: cursor \"foo\" does not exist
+FETCH FORWARD 3 FROM foo;
+
+statement ok
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    x INT := 10;
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT x;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query I
+FETCH FORWARD 3 FROM foo;
+----
+10
+
+# TODO(drewk): postgres returns an ambiguous column error here by default,
+# although it can be configured to prefer either the variable or the column.
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    x INT := 10;
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT * FROM xy WHERE xy.x = x;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query II rowsort
+FETCH FORWARD 10 FROM foo;
+----
+1    2
+3    4
+100  200
+101  201
+102  202
+103  203
+104  204
+
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    i INT := 3;
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT * FROM xy WHERE x = i;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query II
+FETCH FORWARD 3 FROM foo;
+----
+3  4
+
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs NO SCROLL FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query I
+FETCH FORWARD 3 FROM foo;
+----
+1
+
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := '';
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query I
+FETCH FORWARD 3 FROM "";
+----
+1
+
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+    curs2 STRING := 'bar';
+    curs3 STRING := 'baz';
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    OPEN curs2 FOR SELECT 2;
+    OPEN curs3 FOR SELECT 3;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query I
+FETCH FORWARD 3 FROM foo;
+----
+1
+
+query I
+FETCH FORWARD 3 FROM bar;
+----
+2
+
+query I
+FETCH FORWARD 3 FROM baz;
+----
+3
+
+# The cursor should reflect changes to the database state that occur before
+# it is opened, but not those that happen after it is opened.
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+    curs2 STRING := 'bar';
+  BEGIN
+    OPEN curs FOR SELECT * FROM xy WHERE x = 99;
+    INSERT INTO xy VALUES (99, 99);
+    OPEN curs2 FOR SELECT * FROM xy WHERE x = 99;
+    DELETE FROM xy WHERE x = 99;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+SELECT f();
+
+query II
+FETCH FORWARD 3 FROM foo;
+----
+
+query II
+FETCH FORWARD 3 FROM bar;
+----
+99  99
+
+statement ok
+ABORT;
+
+# It is possible to use the OPEN statement in an implicit transaction, but the
+# cursor is closed at the end of the transaction when the statement execution
+# finishes. So, until FETCH is implemented, we can't actually read from the
+# cursor.
+statement ok
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+SELECT f();
+
+statement error pgcode 34000 pq: cursor \"foo\" does not exist
+FETCH FORWARD 5 FROM foo;
+
+statement error pgcode 0A000 pq: unimplemented: DECLARE SCROLL CURSOR
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs SCROLL FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+
+statement error pgcode 0A000 pq: unimplemented: bound cursor declarations are not yet supported.
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs CURSOR FOR SELECT 1;
+  BEGIN
+    OPEN curs;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+
+statement error pgcode 42P11 pq: cannot open INSERT query as cursor
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR INSERT INTO xy VALUES (1, 1);
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+
+statement error pgcode 0A000 pq: unimplemented: CTE usage inside a function definition
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    i INT := 3;
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR WITH foo AS (SELECT * FROM xy WHERE x = i) SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+
+# TODO(drewk): once CTEs in routines are supported, the error should be:
+# pgcode 0A000 pq: DECLARE CURSOR must not contain data-modifying statements in WITH
+statement error pgcode 0A000 pq: unimplemented: CTE usage inside a function definition
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    i INT := 3;
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR WITH foo AS (INSERT INTO xy VALUES (1, 1) RETURNING x) SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+
+statement error pgcode 42601 pq: \"curs\" is not a known variable
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    i INT := 3;
+  BEGIN
+    OPEN curs FOR WITH foo AS (SELECT * FROM xy WHERE x = i) SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+
+statement ok
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING;
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+
+statement error pgcode 0A000 pq: unimplemented: opening an unnamed cursor is not yet supported
+SELECT f();
+
+statement ok
+ABORT;
+
+statement error pgcode 0A000 pq: unimplemented: opening a cursor in a routine with an exception block is not yet supported
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING;
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    RETURN 0;
+  EXCEPTION
+    WHEN division_by_zero THEN
+      RETURN -1;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+
+statement ok
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT 1 // 0;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+
+statement error pgcode 22012 pq: division by zero
+SELECT f();
+
+# Conflict with an existing cursor.
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+
+statement ok
+DECLARE foo CURSOR FOR SELECT 100;
+
+statement error pgcode 42P03 pq: cursor \"foo\" already exists
+SELECT f();
+
+# Conflict between OPEN statements within the same routine.
+statement ok
+ABORT;
+CREATE OR REPLACE FUNCTION f() RETURNS INT AS $$
+  DECLARE
+    curs STRING := 'foo';
+    curs2 STRING := 'foo';
+  BEGIN
+    OPEN curs FOR SELECT 1;
+    OPEN curs2 FOR SELECT 2;
+    RETURN 0;
+  END
+$$ LANGUAGE PLpgSQL;
+BEGIN;
+
+statement error pgcode 42P03 pq: cursor \"foo\" already exists
+SELECT f();

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

@@ -3150,6 +3150,7 @@ func (b *Builder) buildCall(c *memo.CallExpr) (execPlan, error) {
 		udf.TailCall,
 		true, /* procedure */
 		nil,  /* exceptionHandler */
+		nil,  /* cursorDeclaration */
 	)
 
 	var ep execPlan