v5-0004-Add-force_alter-option.patch

From f92c5166b808e70d7907b099c847dbffed8136c7 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v5 4/4] Add force_alter option

---
 src/backend/commands/subscriptioncmds.c       | 36 +++++++++++++++++--
 src/test/regress/expected/subscription.out    |  3 ++
 src/test/regress/sql/subscription.sql         |  3 ++
 src/test/subscription/t/099_twophase_added.pl | 23 +++++++++---
 4 files changed, 58 insertions(+), 7 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index cf3e5c64b0..33b789b730 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -73,6 +73,7 @@
 #define SUBOPT_FAILOVER				0x00002000
 #define SUBOPT_LSN					0x00004000
 #define SUBOPT_ORIGIN				0x00008000
+#define SUBOPT_FORCE_ALTER			0x00010000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -100,6 +101,7 @@ typedef struct SubOpts
 	bool		failover;
 	char	   *origin;
 	XLogRecPtr	lsn;
+	bool		twophase_force;
 } SubOpts;
 
 static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -162,6 +164,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		opts->failover = false;
 	if (IsSet(supported_opts, SUBOPT_ORIGIN))
 		opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+	if (IsSet(supported_opts, SUBOPT_FORCE_ALTER))
+		opts->twophase_force = false;
 
 	/* Parse options */
 	foreach(lc, stmt_options)
@@ -367,6 +371,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_LSN;
 			opts->lsn = lsn;
 		}
+		else if (IsSet(supported_opts, SUBOPT_FORCE_ALTER) &&
+				 strcmp(defel->defname, "force_alter") == 0)
+		{
+			if (IsSet(opts->specified_opts, SUBOPT_FORCE_ALTER))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_FORCE_ALTER;
+			opts->twophase_force = defGetBoolean(defel);
+		}
 		else
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -1170,7 +1183,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								  SUBOPT_DISABLE_ON_ERR |
 								  SUBOPT_PASSWORD_REQUIRED |
 								  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
-								  SUBOPT_ORIGIN);
+								  SUBOPT_ORIGIN | SUBOPT_FORCE_ALTER);
 
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
@@ -1206,6 +1219,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					{
 						ListCell	*cell;
 
+						/*
+						 * Abort prepared transactions if force option is also
+						 * specified. Otherwise raise an ERROR.
+						 */
+						if (!opts.twophase_force)
+							ereport(ERROR,
+									(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+									 errmsg("cannot alter %s when there are prepared transactions",
+											"two_phase = false")));
+
 						/* Must not be in the transaction */
 						PreventInTransactionBlock(isTopLevel,
 												  "ALTER SUBSCRIPTION ... SET (two_phase = ...)");
@@ -1215,8 +1238,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 						{
 							FinishPreparedTransaction((char *) lfirst(cell),
 													  false);
-							prepared_xacts = list_delete_cell(prepared_xacts, cell);
 						}
+
+						list_free(prepared_xacts);
 					}
 
 					/* Change system catalog acoordingly */
@@ -1333,6 +1357,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					replaces[Anum_pg_subscription_suborigin - 1] = true;
 				}
 
+				/* force_alter cannot be used standalone */
+				if (IsSet(opts.specified_opts, SUBOPT_FORCE_ALTER) &&
+					!IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+					ereport(ERROR,
+							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							 errmsg("%s must be specified with %s",
+									"force_alter", "two_phase")));
+
 				update_tuple = true;
 				break;
 			}
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 9bba656e00..16154d1bee 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -372,6 +372,9 @@ ERROR:  two_phase requires a Boolean value
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+ERROR:  force_alter must be specified with two_phase
 \dRs+
                                                                                                                 List of subscriptions
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 9ff151f806..8d8f5bb670 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -256,6 +256,9 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 -- now it works
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
 
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+
 \dRs+
 -- We can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index a8135b671c..7c73a58f2a 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -85,16 +85,29 @@ $result = $node_subscriber->safe_psql('postgres',
     "SELECT count(*) FROM pg_prepared_xacts;");
 is($result, q(1), "transaction has been prepared on subscriber");
 
-$node_subscriber->safe_psql(
-    'postgres', "
-    ALTER SUBSCRIPTION sub DISABLE;
-    ALTER SUBSCRIPTION sub SET (two_phase = off);
-    ALTER SUBSCRIPTION sub ENABLE;");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub DISABLE;");
+
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql(
+	'postgres', "ALTER SUBSCRIPTION sub SET (two_phase = off);");
+ok($stderr =~ /cannot alter two_phase = false when there are prepared transactions/,
+	'ALTER SUBSCRIPTION failed');
+
+$result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exits");
+
+$node_subscriber->safe_psql('postgres',
+    "ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter = on);");
 
 $result = $node_subscriber->safe_psql('postgres',
     "SELECT count(*) FROM pg_prepared_xacts;");
 is($result, q(0), "prepared transaction done by worker is aborted");
 
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE;");
+
 $node_publisher->safe_psql( 'postgres',
     "COMMIT PREPARED 'test_prepared_tab_full';");
 $node_publisher->wait_for_catchup('sub');
-- 
2.43.0