From 71feafb120da3a499f15427d5e9c4f41800cf6c6 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Fri, 19 Apr 2024 11:03:19 +0000 Subject: [PATCH v7 4/4] Add force_alter option --- doc/src/sgml/ref/alter_subscription.sgml | 9 +++-- src/backend/commands/subscriptioncmds.c | 33 ++++++++++++++++++- src/test/regress/expected/subscription.out | 3 ++ src/test/regress/sql/subscription.sql | 3 ++ src/test/subscription/t/099_twophase_added.pl | 23 ++++++++++--- 5 files changed, 62 insertions(+), 9 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 848e4af649..bbfaa72229 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -257,9 +257,12 @@ ALTER SUBSCRIPTION name RENAME TO < The two_phase parameter can only be altered when the - subscription is disabled. When altering the parameter from true - to false, the backend process checks prepared - transactions done by the logical replication worker and aborts them. + subscription is disabled. Altering the parameter from true + to false will be failed when there are prepared + transactions done by the logical replication worker. If you want to alter + the parameter forcibly in this case, force_alter + option must be set to true at the same time. If + specified, the backend process aborts prepared transactions. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 27be6299e0..512357f9a4 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), @@ -1148,7 +1161,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); @@ -1194,6 +1207,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"))); + /* Abort all listed transactions */ foreach(cell, prepared_xacts) FinishPreparedTransaction((char *) lfirst(cell), @@ -1323,6 +1346,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 51fa4b9690..f607045b28 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -370,6 +370,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 a3886d79ca..80ab4dd9bc 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -255,6 +255,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