[BACKPORT 2024.2.3][#27464] YSQL: Speculatively execute PL statements that do not have non-transactional side effects

Summary:
**Merge Notes (2025.1 --> 2024.2)**
  - In `pl_exec.c`
    - Parameter type change to function `YbIsFlushRequiredForCommandTag`
        - Upstream postgres commit (`2f9661311b83dc481fc19f6e3bda015392010a40`) introduces a mechanism to represent command tags as structs/enums.
        - This commit is not present in 2024.2.
        - Resolve by using the stringified version of the command tag.
    - Changes in body of function `YbIsFlushRequiredForCommandTag`
        - Since switch-case statements cannot be used with strings, use if-else blocks.
        - Since if-else statements are used, introduce an early return when `yb_speculatively_execute_pl_statements` is false.
        - 2024.2 does not have the MERGE command. Remove from list of whitelisted statements.
        - The stringified command tags also do not distinguish between different types of SELECTs (SELECT, SELECT INTO, SELECT FOR <lock>). Retain only "SELECT" which encompasses all of the SELECTs. Notably, this makes the behavior different from that of master when `yb_whitelist_extra_statements_for_pl_speculative_execution`. A follow-up diff will be used to reconcile this difference.
    - In function `exec_stmt_execsql`
        - A potential bug ([#27552](#27552)) involving visibility rules for temporary tables in Read Committed was discovered after the parent commit was merged with master.
        - As a result, this revision disables speculative execution for a statement if the statement involves a temp table. A follow-up diff will be used to reconcile this difference with master.
  - In `spi.c` / `spi.h`
    - Struct `SPIExecuteOptions`
        - Upstream postgres commit (`ee895a655ce4341546facd6f23e3e8f2931b96bf`) introduces struct `SPIExecuteOptions`.
        - This commit is not present in 2024.2.
        - master adds a new field `yb_reuse_existing_snapshot_in_read_committed` to this struct.
        - Resolve by:
            - Adding `yb_reuse_existing_snapshot_in_read_committed` as a parameter to internal function `_SPI_execute_plan` which uses the struct.
            - Rework all existing callers of `_SPI_execute_plan` (which are all public functions) to supply `yb_reuse_existing_snapshot_in_read_committed` as false.
            - Introduce a new public function `SPI_yb_execute_plan_with_paramlist` which accepts `yb_reuse_existing_snapshot_in_read_committed` as a param in addition to the params in `SPI_execute_plan_with_paramlist`.
    - Function `SPI_execute_plan_extended`
        - Upstream postgres commit (`d5a83d79c9f9b660a6a5a77afafe146d3c8c6f46`) introduces function `SPI_execute_plan_extended`.
        - This commit is not present in 2024.2.
        - Resolve by making `exec_stmt_execsql` invoke newly defined function `SPI_yb_execute_plan_with_paramlist` instead of `SPI_execute_plan_extended`.
    - Function `_SPI_execute_plan`
        - Upstream postgres commit (`84f5c2908dad81e8622b0406beea580e40bb03ac`) reworks parts of the postgres snapshot management logic.
        - This commit is not present in 2024.2.
        - Resolve by ensuring that command counter is incremented only when a new snapshot is requested. This is in line with the logic introduced in `84f5c2908dad81e8622b0406beea580e40bb03ac`.

### The problem
Consider the following statement:
```
WITH
  cte1 AS (INSERT INTO t1 (a, b, c) VALUES (a1, b1, c1) RETURNING a),
  cte2 AS (INSERT INTO t2 (a, b, c) VALUES (a2, b2, c2) RETURNING a),
SELECT * FROM cte1, cte2;
```

If the tables `t1` and `t2` do not have any relevant triggers, then the inserts to `t1` and `t2` are flushed together in a single batch.
By structuring the query in the form of CTEs, this concept can be used to batch related writes to N tables.
On the other hand, consider the case where the tables have relevant triggers which execute the following function:
```
CREATE sample_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
	INSERT INTO audit VALUES (NEW.a);
	RETURN NEW;
END; $$;
```

Since buffered writes are flushed at statement boundaries (except when the next statement is a DML ie. INSERT/UPDATE/DELETE/MERGE), the `RETURN` statement in the function will incur a flush.
This loses the performance benefit of grouping together multiple writes in the form of the CTEs.
This results in the behavior that writes can only be buffered within a PL function but not across functions.

### The fix
To workaround the above problem, this revision defines a framework where certain statements can be speculatively executed by skipping a flush at the statement boundary of preceding statements.
The requirements to skip flushing are as follows:
 - Statements that have (persistent) side effects must not be executed.
 - Conditional statements whose execution is dependent on the result/execution of buffered operations must not be executed.
 - Exceptions must be processed in the same order that they would have been had all statements been flushed at their respective boundaries.

The above requirements resulted in whitelisting the following control PL statements (from being flushed):
 - Statements denoting statement blocks (like BEGIN)
 - Variable assignments
 - Return statements
 - non-DDL SQL statements (reads/SELECTs)

Further, the following statements can be whitelisted by toggling a secondary GUC:
 - PERFORM statements
 - Stored procedure invocations (including CALL and DO statements)
 - SELECT for lock statements
 - EXPLAIN statements
 - SHOW statements

### Read Committed Behavior
In Read Committed mode, every statement is associated with a new postgres snapshot. Moving across snapshots requires flushing the buffered operations of the previous snapshot.
In order to preserve the performance benefit, multiple PL statements share the same snapshot. A new snapshot is requested only when the next statement violates one of the three requirements above.
In case of a serialization error, the entire top level statement is retried and not just individual statements that share the snapshot. So, skipping the snapshot does not alter the retry logic.

### GUCs
This revision introduces the following GUCs:
 - `yb_speculatively_execute_pl_statements` (bool, default false): Set to true to enable speculative execution of pl/pgSQL statements.
 - `yb_whitelist_extra_statements_for_pl_speculative_execution` (bool, default false): Enable on an experimental basis to whitelist statements in the secondary list above.

### Results
```
yugabyte=# EXPLAIN (ANALYZE, DIST)
WITH
cte1 AS (INSERT INTO t1 VALUES (1, 1), (2, 2) RETURNING k),
cte2 AS (INSERT INTO t2 VALUES (1, 1), (2, 2) RETURNING k)
SELECT * FROM cte1, cte2;
                                             QUERY PLAN
-----------------------------------------------------------------------------------------------------
 Nested Loop  (cost=0.05..0.19 rows=4 width=8) (actual rows=4 loops=1)
   CTE cte1
     ->  Insert on t1  (cost=0.00..0.03 rows=2 width=8) (actual rows=2 loops=1)
           Storage Table Write Requests: 4
           ->  Values Scan on "*VALUES*"  (cost=0.00..0.03 rows=2 width=8) (actual rows=2 loops=1)
   CTE cte2
     ->  Insert on t2  (cost=0.00..0.03 rows=2 width=8) (actual rows=2 loops=1)
           Storage Table Write Requests: 4
           ->  Values Scan on "*VALUES*_1"  (cost=0.00..0.03 rows=2 width=8) (actual rows=2 loops=1)
   ->  CTE Scan on cte1  (cost=0.00..0.04 rows=2 width=4) (actual rows=2 loops=1)
   ->  CTE Scan on cte2  (cost=0.00..0.04 rows=2 width=4) (actual rows=2 loops=2)
 Trigger trigger_t1 on t1: calls=2
 Trigger trigger_t1 on t2: calls=2
 Storage Read Requests: 0
 Storage Rows Scanned: 0
 Storage Write Requests: 8
 Storage Flush Requests: 1
(17 rows)
```

 - [#27507](#27507) - Add more test scenarios (including ones in the unaddressed comments)
Jira: DB-17005

Original commit: aadf2c9 / D44454

Test Plan:
Run the following tests:
```
./yb_build.sh --java-test 'org.yb.pgsql.TestPgRegressBufferingWithSQLErrors'
./yb_build.sh --java-test 'org.yb.pgsql.TestPgRegressPlpgsql'
```

Jenkins: urgent

Reviewers: #db-approvers, pjain, mihnea, patnaik.balivada

Reviewed By: pjain

Subscribers: svc_phabricator, smishra, yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D44610

Author: karthik-ramanathan-3006

src/postgres/src/backend/executor/spi.c

@@ -64,7 +64,8 @@ static void _SPI_prepare_oneshot_plan(const char *src, SPIPlanPtr plan);
 
 static int _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 				  Snapshot snapshot, Snapshot crosscheck_snapshot,
-				  bool read_only, bool fire_triggers, uint64 tcount);
+				  bool read_only, bool fire_triggers, uint64 tcount,
+				  bool yb_reuse_existing_snapshot_in_read_committed);
 
 static ParamListInfo _SPI_convert_params(int nargs, Oid *argtypes,
 					Datum *Values, const char *Nulls);
@@ -462,7 +463,8 @@ SPI_execute(const char *src, bool read_only, long tcount)
 
 	res = _SPI_execute_plan(&plan, NULL,
 							InvalidSnapshot, InvalidSnapshot,
-							read_only, true, tcount);
+							read_only, true, tcount,
+							false /* yb_reuse_existing_snapshot_in_read_committed */ );
 
 	_SPI_end_call(true);
 	return res;
@@ -496,7 +498,8 @@ SPI_execute_plan(SPIPlanPtr plan, Datum *Values, const char *Nulls,
 							_SPI_convert_params(plan->nargs, plan->argtypes,
 												Values, Nulls),
 							InvalidSnapshot, InvalidSnapshot,
-							read_only, true, tcount);
+							read_only, true, tcount,
+							false /* yb_reuse_existing_snapshot_in_read_committed */ );
 
 	_SPI_end_call(true);
 	return res;
@@ -513,6 +516,15 @@ SPI_execp(SPIPlanPtr plan, Datum *Values, const char *Nulls, long tcount)
 int
 SPI_execute_plan_with_paramlist(SPIPlanPtr plan, ParamListInfo params,
 								bool read_only, long tcount)
+{
+	return SPI_yb_execute_plan_with_paramlist(plan, params, read_only, tcount,
+											  false /* yb_reuse_existing_snapshot_in_read_committed */ );
+}
+
+int
+SPI_yb_execute_plan_with_paramlist(SPIPlanPtr plan, ParamListInfo params,
+								   bool read_only, long tcount,
+								   bool yb_reuse_existing_snapshot_in_read_committed)
 {
 	int			res;
 
@@ -525,7 +537,8 @@ SPI_execute_plan_with_paramlist(SPIPlanPtr plan, ParamListInfo params,
 
 	res = _SPI_execute_plan(plan, params,
 							InvalidSnapshot, InvalidSnapshot,
-							read_only, true, tcount);
+							read_only, true, tcount,
+							yb_reuse_existing_snapshot_in_read_committed);
 
 	_SPI_end_call(true);
 	return res;
@@ -566,7 +579,8 @@ SPI_execute_snapshot(SPIPlanPtr plan,
 							_SPI_convert_params(plan->nargs, plan->argtypes,
 												Values, Nulls),
 							snapshot, crosscheck_snapshot,
-							read_only, fire_triggers, tcount);
+							read_only, fire_triggers, tcount,
+							false /* yb_reuse_existing_snapshot_in_read_committed */ );
 
 	_SPI_end_call(true);
 	return res;
@@ -613,7 +627,8 @@ SPI_execute_with_args(const char *src,
 
 	res = _SPI_execute_plan(&plan, paramLI,
 							InvalidSnapshot, InvalidSnapshot,
-							read_only, true, tcount);
+							read_only, true, tcount,
+							false /* yb_reuse_existing_snapshot_in_read_committed */ );
 
 	_SPI_end_call(true);
 	return res;
@@ -2051,7 +2066,8 @@ _SPI_prepare_oneshot_plan(const char *src, SPIPlanPtr plan)
 static int
 _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 				  Snapshot snapshot, Snapshot crosscheck_snapshot,
-				  bool read_only, bool fire_triggers, uint64 tcount)
+				  bool read_only, bool fire_triggers, uint64 tcount,
+				  bool yb_reuse_existing_snapshot_in_read_committed)
 {
 	int			my_res = 0;
 	uint64		my_processed = 0;
@@ -2179,8 +2195,20 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 		/*
 		 * In the default non-read-only case, get a new snapshot, replacing
 		 * any that we pushed in a previous cycle.
+		 * YB: When batching of writes across queries is requested in Read
+		 * Committed isolation, skip creating a new snapshot (and
+		 * consequently a read point) as this would cause previously
+		 * buffered writes to be flushed. As a result, all the statements in
+		 * the batch share the same snapshot. In case of a serialization
+		 * error, the entire top level statement will be retried and not just
+		 * individual statements in the batch. So, skipping the snapshot does
+		 * not alter the retry logic.
+		 * TODO(kramanathan): Use this as a workaround until we can explicitly
+		 * specify that multiple statements share a read point in RC mode if
+		 * they do not perform any reads.
 		 */
-		if (snapshot == InvalidSnapshot && !read_only && !plan->no_snapshots)
+		if (snapshot == InvalidSnapshot && !read_only && !plan->no_snapshots &&
+			!yb_reuse_existing_snapshot_in_read_committed)
 		{
 			if (pushed_active_snap)
 				PopActiveSnapshot();
@@ -2229,9 +2257,10 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 
 			/*
 			 * If not read-only mode, advance the command counter before each
-			 * command and update the snapshot.
+			 * command and update the snapshot. (But skip it if the snapshot
+			 * isn't under our control.)
 			 */
-			if (!read_only && !plan->no_snapshots)
+			if (!read_only && pushed_active_snap)
 			{
 				CommandCounterIncrement();
 				UpdateActiveSnapshotCommandId();

src/postgres/src/backend/utils/misc/guc.c

@@ -2809,6 +2809,33 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"yb_speculatively_execute_pl_statements", PGC_SUSET, CUSTOM_OPTIONS,
+			gettext_noop("If enabled, procedural language statements may be speculatively executed "
+						 "when it is safe to do so without waiting for the successful completion "
+						 "of previous statements. This allows any writes produced by triggers to "
+						 "be batched alongside their parent data-modifying writes such that the "
+						 "number of storages flushes may be minimized."),
+			NULL,
+			GUC_NOT_IN_SAMPLE
+		},
+		&yb_speculatively_execute_pl_statements,
+		false,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"yb_whitelist_extra_statements_for_pl_speculative_execution", PGC_SUSET, CUSTOM_OPTIONS,
+			gettext_noop("If enabled, additional procedural language constructs are whitelisted "
+						 "for use in speculative execution."),
+			NULL,
+			GUC_NOT_IN_SAMPLE
+		},
+		&yb_whitelist_extra_stmts_for_pl_speculative_execution,
+		false,
+		NULL, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, false, NULL, NULL, NULL

src/postgres/src/backend/utils/misc/pg_yb_utils.c

@@ -1501,6 +1501,9 @@ YBUpdateOptimizationOptions yb_update_optimization_options = {
 	.max_cols_size_to_compare = 10 * 1024
 };
 
+bool yb_speculatively_execute_pl_statements = false;
+bool yb_whitelist_extra_stmts_for_pl_speculative_execution = false;
+
 //------------------------------------------------------------------------------
 // YB Debug utils.
 

src/postgres/src/include/executor/spi.h

@@ -88,6 +88,10 @@ extern int SPI_execute_plan(SPIPlanPtr plan, Datum *Values, const char *Nulls,
 extern int SPI_execute_plan_with_paramlist(SPIPlanPtr plan,
 								ParamListInfo params,
 								bool read_only, long tcount);
+extern int SPI_yb_execute_plan_with_paramlist(SPIPlanPtr plan,
+											  ParamListInfo params,
+											  bool read_only, long tcount,
+											  bool yb_reuse_existing_snapshot_in_read_committed);
 extern int	SPI_exec(const char *src, long tcount);
 extern int SPI_execp(SPIPlanPtr plan, Datum *Values, const char *Nulls,
 		  long tcount);

src/postgres/src/include/pg_yb_utils.h

@@ -712,6 +712,10 @@ typedef struct YBUpdateOptimizationOptions
 /* GUC variables to control the behavior of optimizing update queries. */
 extern YBUpdateOptimizationOptions yb_update_optimization_options;
 
+/* GUC variables to control the speculative executive of PL statements. */
+extern bool yb_speculatively_execute_pl_statements;
+extern bool yb_whitelist_extra_stmts_for_pl_speculative_execution;
+
 /*
  * GUC to allow user to silence the error saying that advisory locks are not
  * supported.