public inbox for [email protected]help / color / mirror / Atom feed
Re: Slow catchup of 2PC (twophase) transactions on replica in LR 5+ messages / 3 participants [nested] [flat]
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-17 04:28 Peter Smith <[email protected]> 0 siblings, 1 reply; 5+ messages in thread From: Peter Smith @ 2024-07-17 04:28 UTC (permalink / raw) To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Amit Kapila <[email protected]>; Zhijie Hou (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]> Hi, here is my review of the v18-0003 patch. ====== sgml/ref/alter_subscription.sgml nitpick - some minor tweaks to the documentation text. I also added a link back to the two_phase parameter. Please see the attached diffs file. ====== Kind Regards, Peter Smith. Fujitsu Australia diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index e021f87..58db97f 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -256,14 +256,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < </para> <para> - The <literal>two_phase</literal> parameter can only be altered when the - subscription is disabled. When altering the parameter from <literal>true</literal> + The <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link> + parameter can only be altered when the subscription is disabled. + When altering the parameter from <literal>true</literal> to <literal>false</literal>, the backend process checks for any incomplete prepared transactions done by the logical replication worker (from when <literal>two_phase</literal> parameter was still <literal>true</literal>) - and, if any are found, this command is failed with an error. In this - case, you can resolve prepared transactions on the publisher node or - manually roll back them on the subscriber. After the altering from + and, if any are found, an error is reported. If this happens, you can + resolve prepared transactions on the publisher node or + manually roll back them on the subscriber, then try again. After the altering from <literal>true</literal> to <literal>false</literal>, the publisher will replicate transactions again when they are committed. </para> Attachments: [text/plain] PS_NITPICKS_20240717_2PC_V180003.txt (1.6K, ../../CAHut+Pv-up=MFqYXSVzCszdxU-n2kdcs3asbrAiKn4_z0VdrrA@mail.gmail.com/3-PS_NITPICKS_20240717_2PC_V180003.txt) download | inline diff: diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index e021f87..58db97f 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -256,14 +256,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < </para> <para> - The <literal>two_phase</literal> parameter can only be altered when the - subscription is disabled. When altering the parameter from <literal>true</literal> + The <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link> + parameter can only be altered when the subscription is disabled. + When altering the parameter from <literal>true</literal> to <literal>false</literal>, the backend process checks for any incomplete prepared transactions done by the logical replication worker (from when <literal>two_phase</literal> parameter was still <literal>true</literal>) - and, if any are found, this command is failed with an error. In this - case, you can resolve prepared transactions on the publisher node or - manually roll back them on the subscriber. After the altering from + and, if any are found, an error is reported. If this happens, you can + resolve prepared transactions on the publisher node or + manually roll back them on the subscriber, then try again. After the altering from <literal>true</literal> to <literal>false</literal>, the publisher will replicate transactions again when they are committed. </para> ^ permalink raw reply [nested|flat] 5+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-17 05:13 Hayato Kuroda (Fujitsu) <[email protected]> parent: Peter Smith <[email protected]> 0 siblings, 2 replies; 5+ messages in thread From: Hayato Kuroda (Fujitsu) @ 2024-07-17 05:13 UTC (permalink / raw) To: 'Peter Smith' <[email protected]>; Zhijie Hou (Fujitsu) <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]> Dear Hou, Peter, Thanks for giving comments! PSA new version. Almost comments were addressed. What's new: 0001 - IsTwoPhaseTransactionGidForSubid() was updated per comment from Hou-san [1]. Some nitpicks were accepted. 0002 - An argument in CheckAlterSubOption() was renamed to " slot_needs_update " Some nitpicks were accepted. 0003 - Some nitpicks were accepted. Below part contains the reason why I rejected some comments. > CommonChecksForFailoverAndTwophase: > nitpick - added Assert for the generic-looking "option" parameter name The style looks strange for me, using multiple strcmp() is more straightforward. Added like that. > 1c. > If the error checks can be moved to be done up-front, then all the 'needs_update' > can be combined. Avoiding multiple checks to 'needs_update' will make this function simpler. This style was needed to preserve error condition for failover option. Not changed. [1]: https://www.postgresql.org/message-id/OS3PR01MB571834FBD3E6D3804484038F94A32%40OS3PR01MB5718.jpnprd0... Best regards, Hayato Kuroda FUJITSU LIMITED Attachments: [application/octet-stream] v19-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (35.2K, ../../TYAPR01MB569240C3946E019FB419B6A3F5A32@TYAPR01MB5692.jpnprd01.prod.outlook.com/2-v19-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch) download | inline diff: From 3bb372f1f02554f36738abc75e650cde79649c69 Mon Sep 17 00:00:00 2001 From: Ajin Cherian <[email protected]> Date: Fri, 5 Apr 2024 06:47:18 -0400 Subject: [PATCH v19 1/3] Allow altering of two_phase option of a SUBSCRIPTION This patch allows the user to alter the 'two_phase' option of a subscriber provided no uncommitted prepared transactions are pending on that subscription. Author: Cherian Ajin, Hayato Kuroda --- doc/src/sgml/protocol.sgml | 46 ++++-- doc/src/sgml/ref/alter_subscription.sgml | 12 +- src/backend/access/transam/twophase.c | 72 ++++++++++ src/backend/commands/subscriptioncmds.c | 136 +++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 9 +- src/backend/replication/logical/launcher.c | 10 +- src/backend/replication/logical/worker.c | 25 +--- src/backend/replication/slot.c | 44 ++++-- src/backend/replication/walsender.c | 32 +++-- src/bin/psql/tab-complete.c | 2 +- src/include/access/twophase.h | 5 + src/include/replication/slot.h | 3 +- src/include/replication/walreceiver.h | 11 +- src/include/replication/worker_internal.h | 3 +- src/test/regress/expected/subscription.out | 5 +- src/test/regress/sql/subscription.sql | 5 +- src/test/subscription/t/021_twophase.pl | 86 ++++++++++- 17 files changed, 376 insertions(+), 130 deletions(-) diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 1b27d0a547..cba6661cf0 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2049,21 +2049,6 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" <para>The following options are supported:</para> <variablelist> - <varlistentry> - <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term> - <listitem> - <para> - If true, this logical replication slot supports decoding of two-phase - commit. With this option, commands related to two-phase commit such as - <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal> - and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted. - The transaction will be decoded and transmitted at - <literal>PREPARE TRANSACTION</literal> time. - The default is false. - </para> - </listitem> - </varlistentry> - <varlistentry> <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term> <listitem> @@ -2104,6 +2089,21 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" </para> </listitem> </varlistentry> + + <varlistentry> + <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term> + <listitem> + <para> + If true, this logical replication slot supports decoding of two-phase + commit. With this option, commands related to two-phase commit such as + <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal> + and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted. + The transaction will be decoded and transmitted at + <literal>PREPARE TRANSACTION</literal> time. + The default is false. + </para> + </listitem> + </varlistentry> </variablelist> <para> @@ -2206,6 +2206,22 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" </varlistentry> </variablelist> + <variablelist> + <varlistentry> + <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term> + <listitem> + <para> + If true, this logical replication slot supports decoding of two-phase + commit. With this option, commands related to two-phase commit such as + <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal> + and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted. + The transaction will be decoded and transmitted at + <literal>PREPARE TRANSACTION</literal> time. + </para> + </listitem> + </varlistentry> + </variablelist> + </listitem> </varlistentry> diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 476f195622..0b23df1b77 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -68,8 +68,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <para> Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>, <command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command> - with <literal>refresh</literal> option as <literal>true</literal> and - <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> + with <literal>refresh</literal> option as <literal>true</literal>, + <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and + <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command> cannot be executed inside a transaction block. These commands also cannot be executed when the subscription has @@ -228,9 +229,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>, <link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>, <link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>, - <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and - <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>. + <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, + <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and + <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>. Only a superuser can set <literal>password_required = false</literal>. + The <literal>two_phase</literal> parameter can only be altered when the + subscription is disabled. </para> <para> diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 9a8257fcaf..f7100306f7 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2681,3 +2681,75 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, LWLockRelease(TwoPhaseStateLock); return found; } + +/* + * TwoPhaseTransactionGid + * Form the prepared transaction GID for two_phase transactions. + * + * Return the GID in the supplied buffer. + */ +void +TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) +{ + Assert(subid != InvalidRepOriginId); + + if (!TransactionIdIsValid(xid)) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg_internal("invalid two-phase transaction ID"))); + + snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); +} + +/* + * IsTwoPhaseTransactionGidForSubid + * Check whether the given GID (as formed by TwoPhaseTransactionGid) is + * for the specified 'subid'. + */ +static bool +IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid) +{ + int ret; + Oid subid_written; + TransactionId xid; + char gid_generated[GIDSIZE]; + + ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + + /* Return false if the given GID has different format */ + if (ret != 2 || subid != subid_written) + return false; + + /* Construct the format GID based on the got xid */ + TwoPhaseTransactionGid(subid, xid, gid_generated, sizeof(gid)); + + /* ...And check whether the given GID is exact same as the format GID */ + return strcmp(gid, gid_generated) == 0; +} + +/* + * LookupGXactBySubid + * Check if the prepared transaction done by apply worker exists. + */ +bool +LookupGXactBySubid(Oid subid) +{ + bool found = false; + + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + for (int i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + + /* Ignore not-yet-valid GIDs. */ + if (gxact->valid && + IsTwoPhaseTransactionGidForSubid(subid, gxact->gid)) + { + found = true; + break; + } + } + LWLockRelease(TwoPhaseStateLock); + + return found; +} diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 16d83b3253..a8e4faacbe 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -16,6 +16,7 @@ #include "access/htup_details.h" #include "access/table.h" +#include "access/twophase.h" #include "access/xact.h" #include "catalog/catalog.h" #include "catalog/dependency.h" @@ -109,6 +110,8 @@ static void check_publications_origin(WalReceiverConn *wrconn, static void check_duplicates_in_publist(List *publist, Datum *datums); static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname); static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); +static void CheckAlterSubOption(Subscription *sub, const char *option, + bool isTopLevel); /* @@ -259,21 +262,9 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->specified_opts |= SUBOPT_STREAMING; opts->streaming = defGetStreamingMode(defel); } - else if (strcmp(defel->defname, "two_phase") == 0) + else if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT) && + strcmp(defel->defname, "two_phase") == 0) { - /* - * Do not allow toggling of two_phase option. Doing so could cause - * missing of transactions and lead to an inconsistent replica. - * See comments atop worker.c - * - * Note: Unsupported twophase indicates that this call originated - * from AlterSubscription. - */ - if (!IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT)) - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("unrecognized subscription parameter: \"%s\"", defel->defname))); - if (IsSet(opts->specified_opts, SUBOPT_TWOPHASE_COMMIT)) errorConflictingDefElem(defel, pstate); @@ -1079,6 +1070,47 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, table_close(rel, NoLock); } +/* + * Common checks for altering failover and two_phase option + */ +static void +CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) +{ + StringInfoData cmd; + + Assert(strcmp(option, "failover") == 0 || + strcmp(option, "two_phase") == 0); + + if (!sub->slotname) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot set %s for a subscription that does not have a slot name", + option))); + + /* + * Do not allow changing the option if the subscription is enabled. This + * is because both failover and two_phase options of the slot on the + * publisher cannot be modified if the slot is currently acquired by the + * apply worker. + */ + if (sub->enabled) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot set %s for enabled subscription", + option))); + + initStringInfo(&cmd); + appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); + + /* + * The changed option of the slot can't be rolled back, so disallow if we + * are in a transaction block. + */ + PreventInTransactionBlock(isTopLevel, cmd.data); + + pfree(cmd.data); +} + /* * Alter the existing subscription. */ @@ -1145,7 +1177,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, { supported_opts = (SUBOPT_SLOT_NAME | SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY | - SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR | + SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT | + SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED | SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN); @@ -1229,33 +1262,61 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { - if (!sub->slotname) + CheckAlterSubOption(sub, "failover", isTopLevel); + + values[Anum_pg_subscription_subfailover - 1] = + BoolGetDatum(opts.failover); + replaces[Anum_pg_subscription_subfailover - 1] = true; + } + + if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) + { + CheckAlterSubOption(sub, "two_phase", isTopLevel); + + /* + * slot_name and two_phase cannot be altered + * simultaneously. The latter part refers to the pre-set + * slot name and tries to modify the slot option, so + * changing both does not make sense. + */ + if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME)) ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot set %s for a subscription that does not have a slot name", - "failover"))); + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("slot_name and two_phase cannot be altered at the same time"))); /* - * Do not allow changing the failover state if the - * subscription is enabled. This is because the failover - * state of the slot on the publisher cannot be modified - * if the slot is currently acquired by the apply worker. + * Workers may still survive even if the subscription has + * been disabled. They may read the pg_subscription + * catalog and detect that the two_phase parameter is + * updated, which causes an assertion failure that + * two_phase should not be updated while the worker + * exists. Ensure workers have already been exited to + * avoid it. */ - if (sub->enabled) + if (logicalrep_workers_find(subid, true, true)) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot set %s for enabled subscription", - "failover"))); + errmsg("cannot alter two_phase when logical replication worker is still running"), + errhint("Wait certain time and try again."))); /* - * The changed failover option of the slot can't be rolled - * back. + * two_phase cannot be disabled if there are any + * uncommitted prepared transactions present. */ - PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (failover)"); - - values[Anum_pg_subscription_subfailover - 1] = - BoolGetDatum(opts.failover); - replaces[Anum_pg_subscription_subfailover - 1] = true; + if (!opts.twophase && + sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && + LookupGXactBySubid(subid)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot disable two_phase when prepared transactions are present"), + errhint("Resolve these transactions and try again."))); + + /* Change system catalog accordingly */ + values[Anum_pg_subscription_subtwophasestate - 1] = + CharGetDatum(opts.twophase ? + LOGICALREP_TWOPHASE_STATE_PENDING : + LOGICALREP_TWOPHASE_STATE_DISABLED); + replaces[Anum_pg_subscription_subtwophasestate - 1] = true; } if (IsSet(opts.specified_opts, SUBOPT_ORIGIN)) @@ -1507,7 +1568,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, * doing the database operations we won't be able to rollback altered * slot. */ - if (replaces[Anum_pg_subscription_subfailover - 1]) + if (replaces[Anum_pg_subscription_subfailover - 1] || + replaces[Anum_pg_subscription_subtwophasestate - 1]) { bool must_use_password; char *err; @@ -1528,7 +1590,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, PG_TRY(); { - walrcv_alter_slot(wrconn, sub->slotname, opts.failover); + walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase); } PG_FINALLY(); { @@ -1675,9 +1737,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * New workers won't be started because we hold an exclusive lock on the * subscription till the end of the transaction. */ - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - subworkers = logicalrep_workers_find(subid, false); - LWLockRelease(LogicalRepWorkerLock); + subworkers = logicalrep_workers_find(subid, false, true); foreach(lc, subworkers) { LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc); diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 6c42c209d2..1cb601a148 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn); static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, - bool failover); + bool failover, bool two_phase); static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn); static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn, const char *query, @@ -1121,15 +1121,16 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, */ static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, - bool failover) + bool failover, bool two_phase) { StringInfoData cmd; PGresult *res; initStringInfo(&cmd); - appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )", + appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )", quote_identifier(slotname), - failover ? "true" : "false"); + failover ? "true" : "false", + two_phase ? "true" : "false"); res = libpqrcv_PQexec(conn->streamConn, cmd.data); pfree(cmd.data); diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 27c3a91fb7..c566d50a07 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -272,11 +272,14 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running) * the subscription, instead of just one. */ List * -logicalrep_workers_find(Oid subid, bool only_running) +logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock) { int i; List *res = NIL; + if (acquire_lock) + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + Assert(LWLockHeldByMe(LogicalRepWorkerLock)); /* Search for attached worker for a given subscription id. */ @@ -288,6 +291,9 @@ logicalrep_workers_find(Oid subid, bool only_running) res = lappend(res, w); } + if (acquire_lock) + LWLockRelease(LogicalRepWorkerLock); + return res; } @@ -759,7 +765,7 @@ logicalrep_worker_detach(void) LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true); + workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true, false); foreach(lc, workers) { LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc); diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index c0bda6269b..6c798cd5b4 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -401,9 +401,6 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata, LogicalRepTupleData *newtup, CmdType operation); -/* Compute GID for two_phase transactions */ -static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid); - /* Functions for skipping changes */ static void maybe_start_skipping_changes(XLogRecPtr finish_lsn); static void stop_skipping_changes(void); @@ -3911,7 +3908,7 @@ maybe_reread_subscription(void) /* !slotname should never happen when enabled is true. */ Assert(newsub->slotname); - /* two-phase should not be altered */ + /* two-phase cannot be altered while the worker exists */ Assert(newsub->twophasestate == MySubscription->twophasestate); /* @@ -4396,24 +4393,6 @@ cleanup_subxact_info() subxact_data.nsubxacts_max = 0; } -/* - * Form the prepared transaction GID for two_phase transactions. - * - * Return the GID in the supplied buffer. - */ -static void -TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) -{ - Assert(subid != InvalidRepOriginId); - - if (!TransactionIdIsValid(xid)) - ereport(ERROR, - (errcode(ERRCODE_PROTOCOL_VIOLATION), - errmsg_internal("invalid two-phase transaction ID"))); - - snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); -} - /* * Common function to run the apply loop with error handling. Disable the * subscription, if necessary. @@ -5014,7 +4993,7 @@ AtEOXact_LogicalRepWorkers(bool isCommit) List *workers; ListCell *lc2; - workers = logicalrep_workers_find(subid, true); + workers = logicalrep_workers_find(subid, true, false); foreach(lc2, workers) { LogicalRepWorker *worker = (LogicalRepWorker *) lfirst(lc2); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index baf9b89dc4..90494cb858 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -804,9 +804,12 @@ ReplicationSlotDrop(const char *name, bool nowait) * Change the definition of the slot identified by the specified name. */ void -ReplicationSlotAlter(const char *name, bool failover) +ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase) { + bool update_slot = false; + Assert(MyReplicationSlot == NULL); + Assert(failover || two_phase); ReplicationSlotAcquire(name, false); @@ -832,28 +835,45 @@ ReplicationSlotAlter(const char *name, bool failover) * Do not allow users to enable failover on the standby as we do not * support sync to the cascading standby. */ - if (failover) + if (failover && *failover) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot enable failover for a replication slot" " on the standby")); } - /* - * Do not allow users to enable failover for temporary slots as we do not - * support syncing temporary slots to the standby. - */ - if (failover && MyReplicationSlot->data.persistency == RS_TEMPORARY) - ereport(ERROR, - errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot enable failover for a temporary replication slot")); + if (failover) + { + /* + * Do not allow users to enable failover for temporary slots as we do + * not support syncing temporary slots to the standby. + */ + if (*failover && MyReplicationSlot->data.persistency == RS_TEMPORARY) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot enable failover for a temporary replication slot")); + + if (MyReplicationSlot->data.failover != *failover) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + MyReplicationSlot->data.failover = *failover; + SpinLockRelease(&MyReplicationSlot->mutex); + + update_slot = true; + } + } - if (MyReplicationSlot->data.failover != failover) + if (two_phase && MyReplicationSlot->data.two_phase != *two_phase) { SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.failover = failover; + MyReplicationSlot->data.two_phase = *two_phase; SpinLockRelease(&MyReplicationSlot->mutex); + update_slot = true; + } + + if (update_slot) + { ReplicationSlotMarkDirty(); ReplicationSlotSave(); } diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d1a9ec900..9b40b1cf7d 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1407,12 +1407,15 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) } /* - * Process extra options given to ALTER_REPLICATION_SLOT. + * Change the definition of a replication slot. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) +AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { bool failover_given = false; + bool two_phase_given = false; + bool failover; + bool two_phase; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) @@ -1424,23 +1427,24 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); failover_given = true; - *failover = defGetBoolean(defel); + failover = defGetBoolean(defel); + } + else if (strcmp(defel->defname, "two_phase") == 0) + { + if (two_phase_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + two_phase_given = true; + two_phase = defGetBoolean(defel); } else elog(ERROR, "unrecognized option: %s", defel->defname); } -} - -/* - * Change the definition of a replication slot. - */ -static void -AlterReplicationSlot(AlterReplicationSlotCmd *cmd) -{ - bool failover = false; - ParseAlterReplSlotOptions(cmd, &failover); - ReplicationSlotAlter(cmd->slotname, failover); + ReplicationSlotAlter(cmd->slotname, + failover_given ? &failover : NULL, + two_phase_given ? &two_phase : NULL); } /* diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index d453e224d9..891face1b6 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end) else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "(")) COMPLETE_WITH("binary", "disable_on_error", "failover", "origin", "password_required", "run_as_owner", "slot_name", - "streaming", "synchronous_commit"); + "streaming", "synchronous_commit", "two_phase"); /* ALTER SUBSCRIPTION <name> SKIP ( */ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "(")) COMPLETE_WITH("lsn"); diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index 56248c0006..d37e06fdee 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -62,4 +62,9 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning); extern void restoreTwoPhaseData(void); extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp); + +extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, + int szgid); +extern bool LookupGXactBySubid(Oid subid); + #endif /* TWOPHASE_H */ diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index c9675ee87c..cde164472a 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific, extern void ReplicationSlotPersist(void); extern void ReplicationSlotDrop(const char *name, bool nowait); extern void ReplicationSlotDropAcquired(void); -extern void ReplicationSlotAlter(const char *name, bool failover); +extern void ReplicationSlotAlter(const char *name, bool *failover, + bool *two_phase); extern void ReplicationSlotAcquire(const char *name, bool nowait); extern void ReplicationSlotRelease(void); diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 12f71fa99b..31fa1257ec 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -372,12 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn, /* * walrcv_alter_slot_fn * - * Change the definition of a replication slot. Currently, it only supports - * changing the failover property of the slot. + * Change the definition of a replication slot. Currently, it supports + * changing the failover and the two_phase property of the slot. */ typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn, const char *slotname, - bool failover); + bool failover, + bool two_phase); /* * walrcv_get_backend_pid_fn @@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions; WalReceiverFunctions->walrcv_send(conn, buffer, nbytes) #define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \ WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) -#define walrcv_alter_slot(conn, slotname, failover) \ - WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover) +#define walrcv_alter_slot(conn, slotname, failover, two_phase) \ + WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase) #define walrcv_get_backend_pid(conn) \ WalReceiverFunctions->walrcv_get_backend_pid(conn) #define walrcv_exec(conn, exec, nRetTypes, retTypes) \ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 515aefd519..9646261d7e 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -240,7 +240,8 @@ extern PGDLLIMPORT bool InitializingApplyWorker; extern void logicalrep_worker_attach(int slot); extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid, bool only_running); -extern List *logicalrep_workers_find(Oid subid, bool only_running); +extern List *logicalrep_workers_find(Oid subid, bool only_running, + bool acquire_lock); extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype, Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 5c2f1ee517..17d48b1685 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -377,10 +377,7 @@ HINT: To initiate replication, you must manually create the replication slot, e regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 (1 row) ---fail - alter of two_phase option not supported. -ALTER SUBSCRIPTION regress_testsub SET (two_phase = false); -ERROR: unrecognized subscription parameter: "two_phase" --- but can alter streaming when two_phase enabled +-- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ List of subscriptions diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 3e5ba4cb8c..007c9e7037 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -256,10 +256,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true); \dRs+ ---fail - alter of two_phase option not supported. -ALTER SUBSCRIPTION regress_testsub SET (two_phase = false); - --- but can alter streaming when two_phase enabled +-- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index 9437cd4c3b..66265c729f 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -367,6 +367,90 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(2), 'replicated data in subscriber table'); +# Clean up +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); + +############################### +# Alter the subscription to two_phase = false. +# Verify that the altered subscription reflects the two_phase option. +############################### + +# Alter subscription two_phase to false +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication worker'" +); +$node_subscriber->safe_psql( + 'postgres', " + ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false); + ALTER SUBSCRIPTION tap_sub_copy ENABLE;"); + +# Wait for subscription startup +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy); + +# Make sure that the two-phase is disabled on the subscriber +$result = $node_subscriber->safe_psql('postgres', + "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';" +); +is($result, qq(d), 'two-phase should be disabled'); + +############################### +# Now do a prepare on the publisher. +# Verify that it is not replicated. +############################### +$node_publisher->safe_psql( + 'postgres', qq{ + BEGIN; + INSERT INTO tab_copy VALUES (100); + PREPARE TRANSACTION 'newgid'; + }); + +# Wait for the subscriber to catchup +$node_publisher->wait_for_catchup($appname_copy); + +# Make sure there are no prepared transactions on the subscriber +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*) FROM pg_prepared_xacts;"); +is($result, qq(0), 'should be no prepared transactions on subscriber'); + +############################### +# Now commit the insert. +# Verify that it is replicated. +############################### +$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); + +# Wait for the subscriber to catchup +$node_publisher->wait_for_catchup($appname_copy); + +# Make sure that the committed transaction is replicated. +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); +is($result, qq(3), 'replicated data in subscriber table'); + +############################### +# Alter the subscription to two_phase = true. +# Verify that the altered subscription reflects the two_phase option. +############################### +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication worker'" +); +$node_subscriber->safe_psql( + 'postgres', " + ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true); + ALTER SUBSCRIPTION tap_sub_copy ENABLE;"); + +# Wait for subscription startup +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy); + +# Make sure that the two-phase is enabled on the subscriber +$result = $node_subscriber->safe_psql('postgres', + "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';" +); +is($result, qq(e), 'two-phase should be enabled'); + $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;"); $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;"); @@ -374,8 +458,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;"); # check all the cleanup ############################### -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); - $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_subscription"); is($result, qq(0), 'check subscription was dropped on subscriber'); -- 2.43.0 [application/octet-stream] v19-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (14.2K, ../../TYAPR01MB569240C3946E019FB419B6A3F5A32@TYAPR01MB5692.jpnprd01.prod.outlook.com/3-v19-0002-Alter-slot-option-two_phase-only-when-altering-t.patch) download | inline diff: From 1fef6db98d4bc479c75f2ac5bebae4c659617a1c Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Wed, 17 Apr 2024 06:18:23 +0000 Subject: [PATCH v19 2/3] Alter slot option two_phase only when altering "true" to "false" Since the two_phase option is controlled by both the publisher (as a slot option) and the subscriber (as a subscription option), the slot option must also be modified. Regarding the false->true case, the backend process alters the subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once. After the subscription is enabled, a new logical replication worker requests to change the two_phase option of its slot from pending to true after the initial data synchronization is done. The code path is the same as the case in which two_phase is initially set to true, so there is no need to do something remarkable. However, for the true->false case, the backend must connect to the publisher and expressly change the parameter because the apply worker does not alter the option to false. Because this operation cannot be rolled back, altering the two_phase parameter from "true" to "false" within a transaction is prohibited. --- doc/src/sgml/ref/alter_subscription.sgml | 2 +- src/backend/commands/subscriptioncmds.c | 84 ++++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 23 +++-- src/include/replication/walreceiver.h | 5 +- src/test/subscription/t/021_twophase.pl | 47 +++++++---- 5 files changed, 112 insertions(+), 49 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 0b23df1b77..df44415661 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -70,7 +70,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command> with <literal>refresh</literal> option as <literal>true</literal>, <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and - <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command> + <command>ALTER SUBSCRIPTION ... SET (two_phase = false)</command> cannot be executed inside a transaction block. These commands also cannot be executed when the subscription has diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index a8e4faacbe..9fed49f88c 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -111,7 +111,7 @@ static void check_duplicates_in_publist(List *publist, Datum *datums); static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname); static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); static void CheckAlterSubOption(Subscription *sub, const char *option, - bool isTopLevel); + bool slot_needs_update, bool isTopLevel); /* @@ -1074,14 +1074,18 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, * Common checks for altering failover and two_phase option */ static void -CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) +CheckAlterSubOption(Subscription *sub, const char *option, + bool slot_needs_update, bool isTopLevel) { - StringInfoData cmd; - Assert(strcmp(option, "failover") == 0 || strcmp(option, "two_phase") == 0); - if (!sub->slotname) + /* + * If the slot needs to be updated, the backend must connect to the + * publisher and request the alteration. slot_name must be required at that + * time. + */ + if (slot_needs_update && !sub->slotname) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot set %s for a subscription that does not have a slot name", @@ -1099,16 +1103,20 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) errmsg("cannot set %s for enabled subscription", option))); - initStringInfo(&cmd); - appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); - /* * The changed option of the slot can't be rolled back, so disallow if we * are in a transaction block. */ - PreventInTransactionBlock(isTopLevel, cmd.data); + if (slot_needs_update) + { + StringInfoData cmd; - pfree(cmd.data); + initStringInfo(&cmd); + appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); + + PreventInTransactionBlock(isTopLevel, cmd.data); + pfree(cmd.data); + } } /* @@ -1130,6 +1138,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, Form_pg_subscription form; bits32 supported_opts; SubOpts opts = {0}; + bool update_failover = false; + bool update_two_phase = false; rel = table_open(SubscriptionRelationId, RowExclusiveLock); @@ -1262,7 +1272,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { - CheckAlterSubOption(sub, "failover", isTopLevel); + /* + * First, mark the needs to alter the replication slot. + * Failover option is controlled by both the publisher (as + * a slot option) and the subscriber (as a subscription + * option). + */ + update_failover = true; + + CheckAlterSubOption(sub, "failover", update_failover, + isTopLevel); values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover); @@ -1271,15 +1290,35 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) { - CheckAlterSubOption(sub, "two_phase", isTopLevel); + /* + * First, check the need to alter the replication slot. + * Two_phase option is controlled by both the publisher + * (as a slot option) and the subscriber (as a + * subscription option). The slot option must be altered + * only when changing "true" to "false". + * + * There is no need to do something remarkable regarding + * the "false" to "true" case; the backend process alters + * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once. + * After the subscription is enabled, a new logical + * replication worker requests to change the two_phase + * option of its slot from pending to true when the + * initial data synchronization is done. The code path is + * the same as the case in which two_phase is initially + * set to true. + */ + update_two_phase = !opts.twophase; + + CheckAlterSubOption(sub, "two_phase", update_two_phase, + isTopLevel); /* - * slot_name and two_phase cannot be altered - * simultaneously. The latter part refers to the pre-set - * slot name and tries to modify the slot option, so - * changing both does not make sense. + * Modifying the two_phase slot option requires a slot + * lookup by slot name, so changing the slot name + * at the same time is not allowed. */ - if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME)) + if (update_two_phase && + IsSet(opts.specified_opts, SUBOPT_SLOT_NAME)) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("slot_name and two_phase cannot be altered at the same time"))); @@ -1303,7 +1342,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, * two_phase cannot be disabled if there are any * uncommitted prepared transactions present. */ - if (!opts.twophase && + if (update_two_phase && sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && LookupGXactBySubid(subid)) ereport(ERROR, @@ -1562,14 +1601,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, } /* - * Try to acquire the connection necessary for altering slot. + * Try to acquire the connection necessary for altering slot, if needed. * * This has to be at the end because otherwise if there is an error while * doing the database operations we won't be able to rollback altered * slot. */ - if (replaces[Anum_pg_subscription_subfailover - 1] || - replaces[Anum_pg_subscription_subtwophasestate - 1]) + if (update_failover || update_two_phase) { bool must_use_password; char *err; @@ -1590,7 +1628,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, PG_TRY(); { - walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase); + walrcv_alter_slot(wrconn, sub->slotname, + update_failover ? &opts.failover : NULL, + update_two_phase ? &opts.twophase : NULL); } PG_FINALLY(); { diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 1cb601a148..97f957cd87 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn); static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, - bool failover, bool two_phase); + const bool *failover, const bool *two_phase); static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn); static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn, const char *query, @@ -1121,16 +1121,27 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, */ static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, - bool failover, bool two_phase) + const bool *failover, const bool *two_phase) { StringInfoData cmd; PGresult *res; initStringInfo(&cmd); - appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )", - quote_identifier(slotname), - failover ? "true" : "false", - two_phase ? "true" : "false"); + appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ", + quote_identifier(slotname)); + + if (failover) + appendStringInfo(&cmd, "FAILOVER %s", + *failover ? "true" : "false"); + + if (failover && two_phase) + appendStringInfo(&cmd, ", "); + + if (two_phase) + appendStringInfo(&cmd, "TWO_PHASE %s", + *two_phase ? "true" : "false"); + + appendStringInfoString(&cmd, " );"); res = libpqrcv_PQexec(conn->streamConn, cmd.data); pfree(cmd.data); diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 31fa1257ec..7ffa5a58b3 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -377,8 +377,9 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn, */ typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn, const char *slotname, - bool failover, - bool two_phase); + const bool *failover, + const bool *two_phase); + /* * walrcv_get_backend_pid_fn diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index 66265c729f..c8fb3b7f17 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -375,6 +375,12 @@ $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); # Verify that the altered subscription reflects the two_phase option. ############################### +# Confirm two-phase slot option is enabled before altering +$result = $node_publisher->safe_psql('postgres', + "SELECT two_phase FROM pg_replication_slots WHERE slot_name = 'tap_sub_copy';" +); +is($result, qq(t), 'two-phase is enabled'); + # Alter subscription two_phase to false $node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); @@ -393,7 +399,13 @@ $node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy); $result = $node_subscriber->safe_psql('postgres', "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';" ); -is($result, qq(d), 'two-phase should be disabled'); +is($result, qq(d), 'two-phase subscription option should be disabled'); + +# Make sure that the two-phase slot option is also disabled +$result = $node_publisher->safe_psql('postgres', + "SELECT two_phase FROM pg_replication_slots WHERE slot_name = 'tap_sub_copy';" +); +is($result, qq(f), 'two-phase slot option should be disabled'); ############################### # Now do a prepare on the publisher. @@ -414,6 +426,22 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;"); is($result, qq(0), 'should be no prepared transactions on subscriber'); +############################### +# Toggle the two_phase to "true" before the COMMIT PREPARED. +# +# Since we are the special path for the case where both two_phase +# and failover are altered, it is also set to "true". +############################### +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication worker'" +); +$node_subscriber->safe_psql( + 'postgres', " + ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true, failover = true); + ALTER SUBSCRIPTION tap_sub_copy ENABLE;"); + ############################### # Now commit the insert. # Verify that it is replicated. @@ -428,23 +456,6 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(3), 'replicated data in subscriber table'); -############################### -# Alter the subscription to two_phase = true. -# Verify that the altered subscription reflects the two_phase option. -############################### -$node_subscriber->safe_psql('postgres', - "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); -$node_subscriber->poll_query_until('postgres', - "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication worker'" -); -$node_subscriber->safe_psql( - 'postgres', " - ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true); - ALTER SUBSCRIPTION tap_sub_copy ENABLE;"); - -# Wait for subscription startup -$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy); - # Make sure that the two-phase is enabled on the subscriber $result = $node_subscriber->safe_psql('postgres', "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';" -- 2.43.0 [application/octet-stream] v19-0003-Notify-users-to-roll-back-prepared-transactions.patch (2.2K, ../../TYAPR01MB569240C3946E019FB419B6A3F5A32@TYAPR01MB5692.jpnprd01.prod.outlook.com/4-v19-0003-Notify-users-to-roll-back-prepared-transactions.patch) download | inline diff: From 34cd7afbeafb90da3c65a71cf12362c980019061 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Tue, 9 Jul 2024 08:01:43 +0000 Subject: [PATCH v19 3/3] Notify users to roll back prepared transactions --- doc/src/sgml/ref/alter_subscription.sgml | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index df44415661..58db97f415 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -233,8 +233,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>. Only a superuser can set <literal>password_required = false</literal>. - The <literal>two_phase</literal> parameter can only be altered when the - subscription is disabled. </para> <para> @@ -256,6 +254,20 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link> option is enabled. </para> + + <para> + The <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link> + parameter can only be altered when the subscription is disabled. + When altering the parameter from <literal>true</literal> + to <literal>false</literal>, the backend process checks for any incomplete + prepared transactions done by the logical replication worker (from when + <literal>two_phase</literal> parameter was still <literal>true</literal>) + and, if any are found, an error is reported. If this happens, you can + resolve prepared transactions on the publisher node or + manually roll back them on the subscriber, then try again. After the altering from + <literal>true</literal> to <literal>false</literal>, the publisher will + replicate transactions again when they are committed. + </para> </listitem> </varlistentry> -- 2.43.0 ^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-17 07:53 Peter Smith <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 1 sibling, 0 replies; 5+ messages in thread From: Peter Smith @ 2024-07-17 07:53 UTC (permalink / raw) To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Zhijie Hou (Fujitsu) <[email protected]>; Amit Kapila <[email protected]>; [email protected] <[email protected]>; Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]> Hi, here are my review comments for v19-0001. ====== doc/src/sgml/protocol.sgml nitpick - Now there is >1 option. /The following option is supported:/The following options are supported:/ ====== src/backend/access/transam/twophase.c TwoPhaseTransactionGid: nitpick - renamed parameter /gid/gid_res/ to emphasize that this is returned by reference ~~~ 1. IsTwoPhaseTransactionGidForSubid + /* Construct the format GID based on the got xid */ + TwoPhaseTransactionGid(subid, xid, gid_generated, sizeof(gid)); I think the wrong size is being passed here. It should be the buffer size -- e.g. sizeof(gid_generated). ~ nitpick - renamed a couple of vars for readability nitpick - expanded some comments. ====== src/backend/commands/subscriptioncmds.c 2. AlterSubscription + /* + * slot_name and two_phase cannot be altered + * simultaneously. The latter part refers to the pre-set + * slot name and tries to modify the slot option, so + * changing both does not make sense. + */ I had given a v18-0002 nitpick suggestion to re-word all of this comment. But, as I wrote before [1], that fix belongs here in patch 0001 where the comment was first added. ====== [1] https://www.postgresql.org/message-id/CAHut%2BPsqMRS3dcijo5jsTSbgV1-9So-YBC7PH7xg0%2BZ8hA7fDQ%40mail... Kind Regards, Peter Smith. Fujitsu Australia diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index cba6661..24c57fb 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2192,7 +2192,7 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" </varlistentry> </variablelist> - <para>The following option is supported:</para> + <para>The following options are supported:</para> <variablelist> <varlistentry> diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index f710030..b426f0b 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2689,7 +2689,7 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, * Return the GID in the supplied buffer. */ void -TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) +TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid_res, int szgid) { Assert(subid != InvalidRepOriginId); @@ -2698,7 +2698,7 @@ TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg_internal("invalid two-phase transaction ID"))); - snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); + snprintf(gid_res, szgid, "pg_gid_%u_%u", subid, xid); } /* @@ -2710,20 +2710,26 @@ static bool IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid) { int ret; - Oid subid_written; - TransactionId xid; + Oid subid_from_gid; + TransactionId xid_from_gid; char gid_generated[GIDSIZE]; - ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + /* Extract the subid and xid from the given GID. */ + ret = sscanf(gid, "pg_gid_%u_%u", &subid_from_gid, &xid_from_gid); - /* Return false if the given GID has different format */ - if (ret != 2 || subid != subid_written) + /* + * Check that the given GID has expected format, and at least the subid + * matches. + */ + if (ret != 2 || subid != subid_from_gid) return false; - /* Construct the format GID based on the got xid */ - TwoPhaseTransactionGid(subid, xid, gid_generated, sizeof(gid)); - - /* ...And check whether the given GID is exact same as the format GID */ + /* + * Reconstruct a tmp GID based on the subid and xid extracted from + * the given GID. Check that the tmp GID and the given GID match. + */ + TwoPhaseTransactionGid(subid_from_gid, xid_from_gid, + gid_generated, sizeof(gid_generated)); return strcmp(gid, gid_generated) == 0; } Attachments: [text/plain] PS_NITPICKS_20240717_v190001.txt (2.4K, ../../CAHut+PuEiJSXNv_ixk2+jdxhK4Q3pcfALHoQe7kVnFyCBDdP9Q@mail.gmail.com/3-PS_NITPICKS_20240717_v190001.txt) download | inline diff: diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index cba6661..24c57fb 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2192,7 +2192,7 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" </varlistentry> </variablelist> - <para>The following option is supported:</para> + <para>The following options are supported:</para> <variablelist> <varlistentry> diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index f710030..b426f0b 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2689,7 +2689,7 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, * Return the GID in the supplied buffer. */ void -TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) +TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid_res, int szgid) { Assert(subid != InvalidRepOriginId); @@ -2698,7 +2698,7 @@ TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg_internal("invalid two-phase transaction ID"))); - snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); + snprintf(gid_res, szgid, "pg_gid_%u_%u", subid, xid); } /* @@ -2710,20 +2710,26 @@ static bool IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid) { int ret; - Oid subid_written; - TransactionId xid; + Oid subid_from_gid; + TransactionId xid_from_gid; char gid_generated[GIDSIZE]; - ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + /* Extract the subid and xid from the given GID. */ + ret = sscanf(gid, "pg_gid_%u_%u", &subid_from_gid, &xid_from_gid); - /* Return false if the given GID has different format */ - if (ret != 2 || subid != subid_written) + /* + * Check that the given GID has expected format, and at least the subid + * matches. + */ + if (ret != 2 || subid != subid_from_gid) return false; - /* Construct the format GID based on the got xid */ - TwoPhaseTransactionGid(subid, xid, gid_generated, sizeof(gid)); - - /* ...And check whether the given GID is exact same as the format GID */ + /* + * Reconstruct a tmp GID based on the subid and xid extracted from + * the given GID. Check that the tmp GID and the given GID match. + */ + TwoPhaseTransactionGid(subid_from_gid, xid_from_gid, + gid_generated, sizeof(gid_generated)); return strcmp(gid, gid_generated) == 0; } ^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-17 23:24 Peter Smith <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 1 sibling, 0 replies; 5+ messages in thread From: Peter Smith @ 2024-07-17 23:24 UTC (permalink / raw) To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Zhijie Hou (Fujitsu) <[email protected]>; Amit Kapila <[email protected]>; [email protected] <[email protected]>; Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]> Hi, here are my review comments for patch v19-0002. ====== src/backend/commands/subscriptioncmds.c CheckAlterSubOption: nitpick - tweak some comment wording ~ On Wed, Jul 17, 2024 at 3:13 PM Hayato Kuroda (Fujitsu) <[email protected]> wrote: > > > 1c. > > If the error checks can be moved to be done up-front, then all the 'needs_update' > > can be combined. Avoiding multiple checks to 'needs_update' will make this function simpler. > > This style was needed to preserve error condition for failover option. Not changed. > nitpick - Hmm. I think you might be trying to preserve the ordering of errors when that order is of no consequence. AFAICT which error comes first here is neither documented nor regression tested. e.g. reordering them gives no problem for testing, but OTOH reordering them does simplify the code. Anyway, I have modified the code in my attached nitpicks diffs to demonstrate this suggestion in case you change your mind. ~~~ AlterSubscription: nitpick - let's keep all the variables called 'update_xxx' together. nitpick - comment typo /needs/need/ nitpick - tweak some comment wording ====== src/test/subscription/t/021_twophase.pl nitpick - didn't quite understand the "Since we are..." comment. I reworded it according to what I thought was the intention. ====== Kind Regards, Peter Smith. Fujitsu Australia diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index f770e9c..a826be9 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1079,17 +1079,6 @@ CheckAlterSubOption(Subscription *sub, const char *option, strcmp(option, "two_phase") == 0); /* - * If the slot needs to be updated, the backend must connect to the - * publisher and request the alteration. slot_name must be required at that - * time. - */ - if (slot_needs_update && !sub->slotname) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot set %s for a subscription that does not have a slot name", - option))); - - /* * Do not allow changing the option if the subscription is enabled. This * is because both failover and two_phase options of the slot on the * publisher cannot be modified if the slot is currently acquired by the @@ -1101,17 +1090,27 @@ CheckAlterSubOption(Subscription *sub, const char *option, errmsg("cannot set %s for enabled subscription", option))); - /* - * The changed option of the slot can't be rolled back, so disallow if we - * are in a transaction block. - */ if (slot_needs_update) { StringInfoData cmd; + /* + * If the slot needs to be updated, the backend must connect to the + * publisher and request the alteration. A slot_name is required at + * that time. + */ + if (!sub->slotname) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot set %s for a subscription that does not have a slot name", + option))); + + /* + * The changed option of the slot can't be rolled back, so disallow if we + * are in a transaction block. + */ initStringInfo(&cmd); appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); - PreventInTransactionBlock(isTopLevel, cmd.data); pfree(cmd.data); } @@ -1132,12 +1131,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, HeapTuple tup; Oid subid; bool update_tuple = false; + bool update_failover = false; + bool update_two_phase = false; Subscription *sub; Form_pg_subscription form; bits32 supported_opts; SubOpts opts = {0}; - bool update_failover = false; - bool update_two_phase = false; rel = table_open(SubscriptionRelationId, RowExclusiveLock); @@ -1271,7 +1270,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { /* - * First, mark the needs to alter the replication slot. + * First, mark the need to alter the replication slot. * Failover option is controlled by both the publisher (as * a slot option) and the subscriber (as a subscription * option). @@ -1295,15 +1294,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, * subscription option). The slot option must be altered * only when changing "true" to "false". * - * There is no need to do something remarkable regarding + * There is no need to do anything remarkable for * the "false" to "true" case; the backend process alters * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once. * After the subscription is enabled, a new logical * replication worker requests to change the two_phase * option of its slot from pending to true when the - * initial data synchronization is done. The code path is - * the same as the case in which two_phase is initially - * set to true. + * initial data synchronization is done. That code path is + * the same as when two_phase is initially set to true. */ update_two_phase = !opts.twophase; diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index c8fb3b7..91e2e17 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -428,9 +428,8 @@ is($result, qq(0), 'should be no prepared transactions on subscriber'); ############################### # Toggle the two_phase to "true" before the COMMIT PREPARED. -# -# Since we are the special path for the case where both two_phase -# and failover are altered, it is also set to "true". +# Also, set failover to "true" to test the code path where +# both two_phase and failover are altered at the same time. ############################### $node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); Attachments: [text/plain] PS_NITPICKS_20240718_2PC_V190002.txt (4.5K, ../../CAHut+PvmMQHRog+N+hkFCd4g9kHPSJMBfb+NqEbQFJY9dug0bg@mail.gmail.com/2-PS_NITPICKS_20240718_2PC_V190002.txt) download | inline diff: diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index f770e9c..a826be9 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1079,17 +1079,6 @@ CheckAlterSubOption(Subscription *sub, const char *option, strcmp(option, "two_phase") == 0); /* - * If the slot needs to be updated, the backend must connect to the - * publisher and request the alteration. slot_name must be required at that - * time. - */ - if (slot_needs_update && !sub->slotname) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot set %s for a subscription that does not have a slot name", - option))); - - /* * Do not allow changing the option if the subscription is enabled. This * is because both failover and two_phase options of the slot on the * publisher cannot be modified if the slot is currently acquired by the @@ -1101,17 +1090,27 @@ CheckAlterSubOption(Subscription *sub, const char *option, errmsg("cannot set %s for enabled subscription", option))); - /* - * The changed option of the slot can't be rolled back, so disallow if we - * are in a transaction block. - */ if (slot_needs_update) { StringInfoData cmd; + /* + * If the slot needs to be updated, the backend must connect to the + * publisher and request the alteration. A slot_name is required at + * that time. + */ + if (!sub->slotname) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot set %s for a subscription that does not have a slot name", + option))); + + /* + * The changed option of the slot can't be rolled back, so disallow if we + * are in a transaction block. + */ initStringInfo(&cmd); appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); - PreventInTransactionBlock(isTopLevel, cmd.data); pfree(cmd.data); } @@ -1132,12 +1131,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, HeapTuple tup; Oid subid; bool update_tuple = false; + bool update_failover = false; + bool update_two_phase = false; Subscription *sub; Form_pg_subscription form; bits32 supported_opts; SubOpts opts = {0}; - bool update_failover = false; - bool update_two_phase = false; rel = table_open(SubscriptionRelationId, RowExclusiveLock); @@ -1271,7 +1270,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { /* - * First, mark the needs to alter the replication slot. + * First, mark the need to alter the replication slot. * Failover option is controlled by both the publisher (as * a slot option) and the subscriber (as a subscription * option). @@ -1295,15 +1294,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, * subscription option). The slot option must be altered * only when changing "true" to "false". * - * There is no need to do something remarkable regarding + * There is no need to do anything remarkable for * the "false" to "true" case; the backend process alters * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once. * After the subscription is enabled, a new logical * replication worker requests to change the two_phase * option of its slot from pending to true when the - * initial data synchronization is done. The code path is - * the same as the case in which two_phase is initially - * set to true. + * initial data synchronization is done. That code path is + * the same as when two_phase is initially set to true. */ update_two_phase = !opts.twophase; diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index c8fb3b7..91e2e17 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -428,9 +428,8 @@ is($result, qq(0), 'should be no prepared transactions on subscriber'); ############################### # Toggle the two_phase to "true" before the COMMIT PREPARED. -# -# Since we are the special path for the case where both two_phase -# and failover are altered, it is also set to "true". +# Also, set failover to "true" to test the code path where +# both two_phase and failover are altered at the same time. ############################### $node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub_copy DISABLE;"); ^ permalink raw reply [nested|flat] 5+ messages in thread
* [PATCH v46 5/7] Fix a few problems in index build progress reporting. @ 2026-03-27 15:50 Álvaro Herrera <[email protected]> 0 siblings, 0 replies; 5+ messages in thread From: Álvaro Herrera @ 2026-03-27 15:50 UTC (permalink / raw) First, index_build() should not update the progress when being driven by REPACK, because the progress reporting infractructure cannot handle status of two commands at the same time. So far, REPACK with the CONCURRENTLY option neglected this problem altogether, but even the existing REPACK wasn't consistent enough: even if the 'progress' variable in repack_index() was false, it didn't pass the value to index_build(). Second, REPACK (CONCURRENTLY) should not set PROGRESS_REPACK_PHASE to PROGRESS_REPACK_PHASE_FINAL_CLEANUP in rebuild_relation() because it calls finish_heap_swap() anyway (via rebuild_relation_finish_concurrent()), which does the same thing. --- src/backend/bootstrap/bootstrap.c | 2 +- src/backend/catalog/heap.c | 3 ++- src/backend/catalog/index.c | 22 ++++++++++++++++++---- src/backend/catalog/toasting.c | 3 ++- src/backend/commands/indexcmds.c | 1 + src/include/catalog/index.h | 4 +++- 6 files changed, 27 insertions(+), 8 deletions(-) diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c index 68a42de0889..1c8226c6d67 100644 --- a/src/backend/bootstrap/bootstrap.c +++ b/src/backend/bootstrap/bootstrap.c @@ -1181,7 +1181,7 @@ build_indices(void) heap = table_open(ILHead->il_heap, NoLock); ind = index_open(ILHead->il_ind, NoLock); - index_build(heap, ind, ILHead->il_info, false, false); + index_build(heap, ind, ILHead->il_info, false, false, false); index_close(ind, NoLock); table_close(heap, NoLock); diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c index 5748aa9a1a9..ae6b7cda3dd 100644 --- a/src/backend/catalog/heap.c +++ b/src/backend/catalog/heap.c @@ -3570,7 +3570,8 @@ RelationTruncateIndexes(Relation heapRelation) /* Initialize the index and rebuild */ /* Note: we do not need to re-establish pkey setting */ - index_build(heapRelation, currentIndex, indexInfo, true, false); + index_build(heapRelation, currentIndex, indexInfo, true, false, + true); /* We're done with this index */ index_close(currentIndex, NoLock); diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index de7182a85a9..cfd33f73f48 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -715,6 +715,9 @@ UpdateIndexRelation(Oid indexoid, * already exists. * INDEX_CREATE_PARTITIONED: * create a partitioned index (table must be partitioned) + * INDEX_CREATE_REPORT_PROGRESS: + * update the backend's progress information during index build. + * constr_flags: flags passed to index_constraint_create * (only if INDEX_CREATE_ADD_CONSTRAINT is set) * allow_system_table_mods: allow table to be a system catalog @@ -760,6 +763,7 @@ index_create(Relation heapRelation, bool invalid = (flags & INDEX_CREATE_INVALID) != 0; bool concurrent = (flags & INDEX_CREATE_CONCURRENT) != 0; bool partitioned = (flags & INDEX_CREATE_PARTITIONED) != 0; + bool progress = (flags & INDEX_CREATE_REPORT_PROGRESS) != 0; char relkind; TransactionId relfrozenxid; MultiXactId relminmxid; @@ -1276,7 +1280,8 @@ index_create(Relation heapRelation, } else { - index_build(heapRelation, indexRelation, indexInfo, false, true); + index_build(heapRelation, indexRelation, indexInfo, false, true, + progress); } /* @@ -1438,6 +1443,12 @@ index_create_copy(Relation heapRelation, bool concurrently, stattargets[i].isnull = isnull; } + /* + * Note: The current callers do not need INDEX_CREATE_REPORT_PROGRESS. If + * 'concurrently' is true, there is no build at all. Otherwise the index + * build is a sub-command of REPACK. The current infrastructure does not + * allow two commands to report their progress at the same time. + */ if (concurrently) flags = INDEX_CREATE_SKIP_BUILD | INDEX_CREATE_CONCURRENT; @@ -1528,7 +1539,7 @@ index_concurrently_build(Oid heapRelationId, indexInfo->ii_BrokenHotChain = false; /* Now build the index */ - index_build(heapRel, indexRelation, indexInfo, false, true); + index_build(heapRel, indexRelation, indexInfo, false, true, true); /* Roll back any GUC changes executed by index functions */ AtEOXact_GUC(false, save_nestlevel); @@ -3001,6 +3012,7 @@ index_update_stats(Relation rel, * * isreindex indicates we are recreating a previously-existing index. * parallel indicates if parallelism may be useful. + * progress indicates if the backend should update its progress info. * * Note: before Postgres 8.2, the passed-in heap and index Relations * were automatically closed by this routine. This is no longer the case. @@ -3011,7 +3023,8 @@ index_build(Relation heapRelation, Relation indexRelation, IndexInfo *indexInfo, bool isreindex, - bool parallel) + bool parallel, + bool progress) { IndexBuildResult *stats; Oid save_userid; @@ -3062,6 +3075,7 @@ index_build(Relation heapRelation, RestrictSearchPath(); /* Set up initial progress report status */ + if (progress) { const int progress_index[] = { PROGRESS_CREATEIDX_PHASE, @@ -3819,7 +3833,7 @@ reindex_index(const ReindexStmt *stmt, Oid indexId, /* Initialize the index and rebuild */ /* Note: we do not need to re-establish pkey setting */ - index_build(heapRelation, iRel, indexInfo, true, true); + index_build(heapRelation, iRel, indexInfo, true, true, progress); /* Re-allow use of target index */ ResetReindexProcessing(); diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c index 078a1cf5127..73c41360ae9 100644 --- a/src/backend/catalog/toasting.c +++ b/src/backend/catalog/toasting.c @@ -331,7 +331,8 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid, BTREE_AM_OID, rel->rd_rel->reltablespace, collationIds, opclassIds, NULL, coloptions, NULL, (Datum) 0, - INDEX_CREATE_IS_PRIMARY, 0, true, true, NULL); + INDEX_CREATE_IS_PRIMARY | INDEX_CREATE_REPORT_PROGRESS, 0, + true, true, NULL); table_close(toast_rel, NoLock); diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c index 83edab38760..c9e22e95515 100644 --- a/src/backend/commands/indexcmds.c +++ b/src/backend/commands/indexcmds.c @@ -1233,6 +1233,7 @@ DefineIndex(ParseState *pstate, flags |= INDEX_CREATE_PARTITIONED; if (stmt->primary) flags |= INDEX_CREATE_IS_PRIMARY; + flags |= INDEX_CREATE_REPORT_PROGRESS; /* * If the table is partitioned, and recursion was declined but partitions diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index 56a064ef444..a03238304a0 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -71,6 +71,7 @@ extern void index_check_primary_key(Relation heapRel, #define INDEX_CREATE_IF_NOT_EXISTS (1 << 4) #define INDEX_CREATE_PARTITIONED (1 << 5) #define INDEX_CREATE_INVALID (1 << 6) +#define INDEX_CREATE_REPORT_PROGRESS (1 << 7) extern Oid index_create(Relation heapRelation, const char *indexRelationName, @@ -148,7 +149,8 @@ extern void index_build(Relation heapRelation, Relation indexRelation, IndexInfo *indexInfo, bool isreindex, - bool parallel); + bool parallel, + bool progress); extern void validate_index(Oid heapId, Oid indexId, Snapshot snapshot); -- 2.47.3 --mdsv5bdgfjjj6d77 Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v46-0006-Error-out-any-process-that-would-block-at-REPACK.patch" ^ permalink raw reply [nested|flat] 5+ messages in thread
end of thread, other threads:[~2026-03-27 15:50 UTC | newest] Thread overview: 5+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2024-07-17 04:28 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Peter Smith <[email protected]> 2024-07-17 05:13 ` Hayato Kuroda (Fujitsu) <[email protected]> 2024-07-17 07:53 ` Peter Smith <[email protected]> 2024-07-17 23:24 ` Peter Smith <[email protected]> 2026-03-27 15:50 [PATCH v46 5/7] Fix a few problems in index build progress reporting. Álvaro Herrera <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox