From f92c5166b808e70d7907b099c847dbffed8136c7 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda 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