public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 4/9] Silence 'mixed declarations and code' compiler warnings 9+ messages / 5 participants [nested] [flat]
* [PATCH 4/9] Silence 'mixed declarations and code' compiler warnings @ 2019-09-26 20:49 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Alvaro Herrera @ 2019-09-26 20:49 UTC (permalink / raw) --- src/backend/utils/fmgr/funcapi.c | 98 +++++++++++++++++++------------- 1 file changed, 59 insertions(+), 39 deletions(-) diff --git a/src/backend/utils/fmgr/funcapi.c b/src/backend/utils/fmgr/funcapi.c index e6e82fda63..701664bd3b 100644 --- a/src/backend/utils/fmgr/funcapi.c +++ b/src/backend/utils/fmgr/funcapi.c @@ -541,18 +541,21 @@ resolve_polymorphic_tupdesc(TupleDesc tupdesc, oidvector *declared_args, if (OidIsValid(anymultirange_type)) { - Oid rngtype = resolve_generic_type(ANYRANGEOID, - anymultirange_type, - ANYMULTIRANGEOID); + Oid rngtype; + Oid subtype; + + rngtype = resolve_generic_type(ANYRANGEOID, + anymultirange_type, + ANYMULTIRANGEOID); /* check for inconsistent range and multirange results */ if (OidIsValid(anyrange_type) && anyrange_type != rngtype) return false; - anyrange_type = rngtype; - Oid subtype = resolve_generic_type(ANYELEMENTOID, - anyrange_type, - ANYRANGEOID); + anyrange_type = rngtype; + subtype = resolve_generic_type(ANYELEMENTOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent array and multirange results */ if (OidIsValid(anyelement_type) && anyelement_type != subtype) @@ -596,18 +599,21 @@ resolve_polymorphic_tupdesc(TupleDesc tupdesc, oidvector *declared_args, { if (OidIsValid(anymultirange_type)) { - Oid rngtype = resolve_generic_type(ANYRANGEOID, - anymultirange_type, - ANYMULTIRANGEOID); + Oid rngtype; + Oid subtype; + + rngtype = resolve_generic_type(ANYRANGEOID, + anymultirange_type, + ANYMULTIRANGEOID); /* check for inconsistent range and multirange results */ if (OidIsValid(anyrange_type) && anyrange_type != rngtype) return false; anyrange_type = rngtype; - Oid subtype = resolve_generic_type(ANYELEMENTOID, - anyrange_type, - ANYRANGEOID); + subtype = resolve_generic_type(ANYELEMENTOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent array and multirange results */ if (OidIsValid(anyelement_type) && anyelement_type != subtype) @@ -628,21 +634,25 @@ resolve_polymorphic_tupdesc(TupleDesc tupdesc, oidvector *declared_args, { if (OidIsValid(anyrange_type)) { - Oid subtype = resolve_generic_type(ANYELEMENTOID, - anyrange_type, - ANYRANGEOID); + Oid subtype; + Oid mltrngtype; + Oid rngtype; + + subtype = resolve_generic_type(ANYELEMENTOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent array and range results */ if (OidIsValid(anyelement_type) && anyelement_type != subtype) return false; anyelement_type = subtype; - Oid mltrngtype = resolve_generic_type(ANYMULTIRANGEOID, - anyrange_type, - ANYRANGEOID); + mltrngtype = resolve_generic_type(ANYMULTIRANGEOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent range and multirange results */ - Oid rngtype = get_multirange_subtype(mltrngtype); + rngtype = get_multirange_subtype(mltrngtype); if (OidIsValid(anyrange_type) && anyrange_type != rngtype) return false; @@ -868,18 +878,21 @@ resolve_polymorphic_argtypes(int numargs, Oid *argtypes, char *argmodes, if (OidIsValid(anymultirange_type)) { - Oid rngtype = resolve_generic_type(ANYRANGEOID, - anymultirange_type, - ANYMULTIRANGEOID); + Oid rngtype; + Oid subtype; + + rngtype = resolve_generic_type(ANYRANGEOID, + anymultirange_type, + ANYMULTIRANGEOID); /* check for inconsistent range and multirange results */ if (OidIsValid(anyrange_type) && anyrange_type != rngtype) return false; anyrange_type = rngtype; - Oid subtype = resolve_generic_type(ANYELEMENTOID, - anyrange_type, - ANYRANGEOID); + subtype = resolve_generic_type(ANYELEMENTOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent array and multirange results */ if (OidIsValid(anyelement_type) && anyelement_type != subtype) @@ -923,18 +936,21 @@ resolve_polymorphic_argtypes(int numargs, Oid *argtypes, char *argmodes, { if (OidIsValid(anymultirange_type)) { - Oid rngtype = resolve_generic_type(ANYRANGEOID, - anymultirange_type, - ANYMULTIRANGEOID); + Oid rngtype; + Oid subtype; + + rngtype = resolve_generic_type(ANYRANGEOID, + anymultirange_type, + ANYMULTIRANGEOID); /* check for inconsistent range and multirange results */ if (OidIsValid(anyrange_type) && anyrange_type != rngtype) return false; anyrange_type = rngtype; - Oid subtype = resolve_generic_type(ANYELEMENTOID, - anyrange_type, - ANYRANGEOID); + subtype = resolve_generic_type(ANYELEMENTOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent array and multirange results */ if (OidIsValid(anyelement_type) && anyelement_type != subtype) @@ -955,21 +971,25 @@ resolve_polymorphic_argtypes(int numargs, Oid *argtypes, char *argmodes, { if (OidIsValid(anyrange_type)) { - Oid subtype = resolve_generic_type(ANYELEMENTOID, - anyrange_type, - ANYRANGEOID); + Oid subtype; + Oid mltrngtype; + Oid rngtype; + + subtype = resolve_generic_type(ANYELEMENTOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent array and range results */ if (OidIsValid(anyelement_type) && anyelement_type != subtype) return false; anyelement_type = subtype; - Oid mltrngtype = resolve_generic_type(ANYMULTIRANGEOID, - anyrange_type, - ANYRANGEOID); + mltrngtype = resolve_generic_type(ANYMULTIRANGEOID, + anyrange_type, + ANYRANGEOID); /* check for inconsistent range and multirange results */ - Oid rngtype = get_multirange_subtype(mltrngtype); + rngtype = get_multirange_subtype(mltrngtype); if (OidIsValid(anyrange_type) && anyrange_type != rngtype) return false; -- 2.17.1 --G4iJoqBmSsgzjUCe Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="0005-Silence-mixed-declarations-and-code-compiler-warning.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-09 05:05 Amit Kapila <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Amit Kapila @ 2024-07-09 05:05 UTC (permalink / raw) To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]> On Mon, Jul 8, 2024 at 5:25 PM Amit Kapila <[email protected]> wrote: > > > I see that in 0003/0004, the patch first aborts pending prepared > transactions, update's catalog, and then change slot's property via > walrcv_alter_slot. What if there is any ERROR (say the remote node is > not reachable or there is an error while updating the catalog) after > we abort the pending prepared transaction? Won't we end up with lost > prepared transactions in such a case? > Considering the above is a problem the other possibility I thought of is to change the order like abort prepared xacts after slot update. That is also dangerous because any failure while aborting could make a slot change permanent whereas the subscription option will still be old value. Now, because the slot's two_phase property is off, at commit, it can resend the entire transaction which can create a problem because the corresponding prepared transaction will already be present. One more thing to think about in this regard is what if we fail after aborting a few prepared transactions and not all? At this stage, I am not able to think of a good solution for these problems. So, if we don't get a solution for these, we can document that users can first manually abort prepared transactions and then switch off the two_phase option using Alter Subscription command. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-09 11:49 Hayato Kuroda (Fujitsu) <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Hayato Kuroda (Fujitsu) @ 2024-07-09 11:49 UTC (permalink / raw) To: 'Amit Kapila' <[email protected]>; +Cc: Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]> Dear Amit, > > I see that in 0003/0004, the patch first aborts pending prepared > > transactions, update's catalog, and then change slot's property via > > walrcv_alter_slot. What if there is any ERROR (say the remote node is > > not reachable or there is an error while updating the catalog) after > > we abort the pending prepared transaction? Won't we end up with lost > > prepared transactions in such a case? Yes, v16 could happen the case, becasue FinishPreparedTransaction() itself is not the transactional operation. In below example, the subscription was altered after stopping the publisher. You could see that prepared transaction were rollbacked. ``` subscriber=# SELECT gid FROM pg_prepared_xacts ; gid ------------------ pg_gid_16390_741 pg_gid_16390_742 (2 rows) subscriber=# ALTER SUBSCRIPTION sub SET (TWO_PHASE = off, FORCE_ALTER = on); NOTICE: requested altering to two_phase = false but there are prepared transactions done by the subscription DETAIL: Such transactions are being rollbacked. ERROR: could not connect to the publisher: connection to server on socket "/tmp/.s.PGSQL.5431" failed: No such file or directory Is the server running locally and accepting connections on that socket? subscriber=# SELECT gid FROM pg_prepared_xacts ; gid ----- (0 rows) ``` > Considering the above is a problem the other possibility I thought of > is to change the order like abort prepared xacts after slot update. > That is also dangerous because any failure while aborting could make a > slot change permanent whereas the subscription option will still be > old value. Now, because the slot's two_phase property is off, at > commit, it can resend the entire transaction which can create a > problem because the corresponding prepared transaction will already be > present. I feel it is rare case but still possible. E.g., race condition by TwoPhaseStateLock locking, oom, disk failures and so on. And since prepared transactions hold locks, duplicated arrival of transactions may cause table-lock failures. > One more thing to think about in this regard is what if we fail after > aborting a few prepared transactions and not all? It's bit hard to emulate, but I imagine part of prepared transactions remains. > At this stage, I am not able to think of a good solution for these > problems. So, if we don't get a solution for these, we can document > that users can first manually abort prepared transactions and then > switch off the two_phase option using Alter Subscription command. I'm also not sure what should we do. Ideally, it may be happy to make FinishPreparedTransaction() transactional, but not sure it is realistic. So changes for aborting prepared txns are removed, documentation patch was added instead. Here is a summary of updates for patches. Dropping-prepared-transaction patch was removed for now. 0001 - Codes for SUBOPT_TWOPHASE_COMMIT are moved per requirement [1]. Also, checks for failover and two_phase are unified into one function. 0002 - updated accordingly. An argument for the check function is added. 0003 - this contains documentation changes required in [2]. [1]: https://www.postgresql.org/message-id/CAA4eK1%2BFRrL_fLWLsWQGHZRESg39ixzDX_S9hU8D7aFtU%2Ba8uQ%40mail... [2]: https://www.postgresql.org/message-id/CAA4eK1Khy_YWFoQ1HOF_tGtiixD8YoTg86coX1-ckxt8vK3U%3DQ%40mail.g... Best Regards, Hayato Kuroda FUJITSU LIMITED https://www.fujitsu.com/ Attachments: [application/octet-stream] v17-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (30.2K, ../../OSBPR01MB2552961D068087129E97048CF5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v17-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch) download | inline diff: From 0006608c5b65941f43c5ce622f64144f2030adff Mon Sep 17 00:00:00 2001 From: Ajin Cherian <[email protected]> Date: Fri, 5 Apr 2024 06:47:18 -0400 Subject: [PATCH v17 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/ref/alter_subscription.sgml | 12 +- src/backend/access/transam/twophase.c | 62 ++++++++ src/backend/commands/subscriptioncmds.c | 135 +++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 9 +- src/backend/replication/logical/launcher.c | 12 +- src/backend/replication/logical/worker.c | 25 +--- src/backend/replication/slot.c | 18 ++- src/backend/replication/walsender.c | 18 ++- 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 | 77 +++++++++- 16 files changed, 308 insertions(+), 94 deletions(-) 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..35bce6809d 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2681,3 +2681,65 @@ 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; + + ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + + return (ret == 2 && subid == subid_written); +} + +/* + * 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 e407428dbc..255628c396 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,9 @@ 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 CommonChecksForFailoverAndTwophase(Subscription *sub, + const char *option, + bool isTopLevel); /* @@ -259,21 +263,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); @@ -1077,6 +1069,45 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, table_close(rel, NoLock); } +/* + * Common checks for altering failover and two_phase option + */ +static void +CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, + bool isTopLevel) +{ + StringInfoData cmd; + + 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: prevent we are in + * the transaction state. + */ + PreventInTransactionBlock(isTopLevel, cmd.data); + + pfree(cmd.data); +} + /* * Alter the existing subscription. */ @@ -1143,7 +1174,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); @@ -1227,33 +1259,61 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { - if (!sub->slotname) + CommonChecksForFailoverAndTwophase(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)) + { + CommonChecksForFailoverAndTwophase(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 twophase parameter is + * updated, which causes the assertion failure. 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 uncommitted prepared transactions present"), + errhint("Resolve these transactions and try again"))); + + /* Change system catalog acoordingly */ + 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)) @@ -1505,7 +1565,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; @@ -1525,7 +1586,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(); { @@ -1672,9 +1733,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 02f12f2921..2f035a0c3c 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..45744b771f 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -272,12 +272,15 @@ 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 require_lock) { int i; List *res = NIL; - Assert(LWLockHeldByMe(LogicalRepWorkerLock)); + if (require_lock) + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + else + Assert(LWLockHeldByMe(LogicalRepWorkerLock)); /* Search for attached worker for a given subscription id. */ for (i = 0; i < max_logical_replication_workers; i++) @@ -288,6 +291,9 @@ logicalrep_workers_find(Oid subid, bool only_running) res = lappend(res, w); } + if (require_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 3b285894db..33bfeb1fb2 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. @@ -5013,7 +4992,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..2ad6dca993 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -804,8 +804,10 @@ 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); ReplicationSlotAcquire(name, false); @@ -854,6 +856,20 @@ ReplicationSlotAlter(const char *name, bool failover) MyReplicationSlot->data.failover = failover; SpinLockRelease(&MyReplicationSlot->mutex); + update_slot = true; + } + + if (MyReplicationSlot->data.two_phase != two_phase) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + 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 754f505c13..af776fccb8 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1409,9 +1409,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) * Process extra options given to ALTER_REPLICATION_SLOT. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) +ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, + bool *failover, bool *two_phase) { bool failover_given = false; + bool two_phase_given = false; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) @@ -1425,6 +1427,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) failover_given = true; *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); } @@ -1437,9 +1448,10 @@ static void AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { bool failover = false; + bool two_phase = false; - ParseAlterReplSlotOptions(cmd, &failover); - ReplicationSlotAlter(cmd->slotname, failover); + ParseAlterReplSlotOptions(cmd, &failover, &two_phase); + ReplicationSlotAlter(cmd->slotname, failover, two_phase); } /* 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..163a4a911a 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..990f5242f9 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 require_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 0f2a25cdc1..51fa4b9690 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..a3886d79ca 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..4e8f627f7b 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -367,6 +367,81 @@ $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"); + +############################### +# Disable the subscription and alter it to two_phase = false, +# then 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 and make sure 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 and 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 subscription two_phase 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); + 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 +449,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] v17-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (13.7K, ../../OSBPR01MB2552961D068087129E97048CF5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v17-0002-Alter-slot-option-two_phase-only-when-altering-t.patch) download | inline diff: From 5df654692988e3880b6132c53f218e511cf6f51d Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Wed, 17 Apr 2024 06:18:23 +0000 Subject: [PATCH v17 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 | 76 ++++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 23 ++++-- src/include/replication/walreceiver.h | 5 +- src/test/subscription/t/021_twophase.pl | 41 ++++++---- 5 files changed, 102 insertions(+), 45 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 0b23df1b77..475a42a2e3 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 = off)</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 255628c396..3729956147 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -112,6 +112,7 @@ static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); static void CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, + bool needs_update, bool isTopLevel); @@ -1074,11 +1075,9 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, */ static void CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, - bool isTopLevel) + bool needs_update, bool isTopLevel) { - StringInfoData cmd; - - if (!sub->slotname) + if (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", @@ -1096,16 +1095,20 @@ CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, 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: prevent we are in * the transaction state. */ - PreventInTransactionBlock(isTopLevel, cmd.data); + if (needs_update) + { + StringInfoData cmd; - pfree(cmd.data); + initStringInfo(&cmd); + appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); + + PreventInTransactionBlock(isTopLevel, cmd.data); + pfree(cmd.data); + } } /* @@ -1127,6 +1130,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, Form_pg_subscription form; bits32 supported_opts; SubOpts opts = {0}; + bool update_failover; + bool update_two_phase; rel = table_open(SubscriptionRelationId, RowExclusiveLock); @@ -1259,8 +1264,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { + /* + * 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; + CommonChecksForFailoverAndTwophase(sub, "failover", - isTopLevel); + update_failover, isTopLevel); values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover); @@ -1269,16 +1282,36 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) { + /* + * 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; + CommonChecksForFailoverAndTwophase(sub, "two_phase", - isTopLevel); + 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. + * If the wo_phase slot option must be altered, this cannot + * be altered with slot_name 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)) + 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"))); @@ -1300,7 +1333,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, @@ -1559,14 +1592,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; @@ -1586,7 +1618,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 2f035a0c3c..07dfec947d 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 4e8f627f7b..f56dff4b12 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"); # then 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 and make sure that it is not replicated. $node_publisher->safe_psql( @@ -411,6 +423,19 @@ $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 and verify that it is replicated $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); @@ -422,20 +447,6 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(3), 'replicated data in subscriber table'); -# Alter subscription two_phase 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); - 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] v17-0003-Notify-users-to-roll-back-prepared-transactions.patch (2.2K, ../../OSBPR01MB2552961D068087129E97048CF5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v17-0003-Notify-users-to-roll-back-prepared-transactions.patch) download | inline diff: From e425e241341e61eed3ab2e1969d53990cb033c84 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Tue, 9 Jul 2024 08:01:43 +0000 Subject: [PATCH v17 3/3] Notify users to roll back prepared transactions --- doc/src/sgml/ref/alter_subscription.sgml | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 475a42a2e3..5263fe7c68 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,19 @@ 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 <literal>two_phase</literal> 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. Alter 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] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-09 12:52 Hayato Kuroda (Fujitsu) <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 0 siblings, 2 replies; 9+ messages in thread From: Hayato Kuroda (Fujitsu) @ 2024-07-09 12:52 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]>; 'Amit Kapila' <[email protected]> > 0001 - Codes for SUBOPT_TWOPHASE_COMMIT are moved per requirement [1]. > Also, checks for failover and two_phase are unified into one function. > 0002 - updated accordingly. An argument for the check function is added. > 0003 - this contains documentation changes required in [2]. Previous patch set could not be accepted due to the initialization miss. PSA new version. Best Regards, Hayato Kuroda FUJITSU LIMITED https://www.fujitsu.com/ Attachments: [application/octet-stream] v17-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (30.2K, ../../OSBPR01MB255215A39F66932323295854F5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v17-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch) download | inline diff: From 0006608c5b65941f43c5ce622f64144f2030adff Mon Sep 17 00:00:00 2001 From: Ajin Cherian <[email protected]> Date: Fri, 5 Apr 2024 06:47:18 -0400 Subject: [PATCH v17 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/ref/alter_subscription.sgml | 12 +- src/backend/access/transam/twophase.c | 62 ++++++++ src/backend/commands/subscriptioncmds.c | 135 +++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 9 +- src/backend/replication/logical/launcher.c | 12 +- src/backend/replication/logical/worker.c | 25 +--- src/backend/replication/slot.c | 18 ++- src/backend/replication/walsender.c | 18 ++- 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 | 77 +++++++++- 16 files changed, 308 insertions(+), 94 deletions(-) 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..35bce6809d 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2681,3 +2681,65 @@ 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; + + ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + + return (ret == 2 && subid == subid_written); +} + +/* + * 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 e407428dbc..255628c396 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,9 @@ 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 CommonChecksForFailoverAndTwophase(Subscription *sub, + const char *option, + bool isTopLevel); /* @@ -259,21 +263,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); @@ -1077,6 +1069,45 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, table_close(rel, NoLock); } +/* + * Common checks for altering failover and two_phase option + */ +static void +CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, + bool isTopLevel) +{ + StringInfoData cmd; + + 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: prevent we are in + * the transaction state. + */ + PreventInTransactionBlock(isTopLevel, cmd.data); + + pfree(cmd.data); +} + /* * Alter the existing subscription. */ @@ -1143,7 +1174,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); @@ -1227,33 +1259,61 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { - if (!sub->slotname) + CommonChecksForFailoverAndTwophase(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)) + { + CommonChecksForFailoverAndTwophase(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 twophase parameter is + * updated, which causes the assertion failure. 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 uncommitted prepared transactions present"), + errhint("Resolve these transactions and try again"))); + + /* Change system catalog acoordingly */ + 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)) @@ -1505,7 +1565,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; @@ -1525,7 +1586,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(); { @@ -1672,9 +1733,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 02f12f2921..2f035a0c3c 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..45744b771f 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -272,12 +272,15 @@ 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 require_lock) { int i; List *res = NIL; - Assert(LWLockHeldByMe(LogicalRepWorkerLock)); + if (require_lock) + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + else + Assert(LWLockHeldByMe(LogicalRepWorkerLock)); /* Search for attached worker for a given subscription id. */ for (i = 0; i < max_logical_replication_workers; i++) @@ -288,6 +291,9 @@ logicalrep_workers_find(Oid subid, bool only_running) res = lappend(res, w); } + if (require_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 3b285894db..33bfeb1fb2 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. @@ -5013,7 +4992,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..2ad6dca993 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -804,8 +804,10 @@ 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); ReplicationSlotAcquire(name, false); @@ -854,6 +856,20 @@ ReplicationSlotAlter(const char *name, bool failover) MyReplicationSlot->data.failover = failover; SpinLockRelease(&MyReplicationSlot->mutex); + update_slot = true; + } + + if (MyReplicationSlot->data.two_phase != two_phase) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + 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 754f505c13..af776fccb8 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1409,9 +1409,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) * Process extra options given to ALTER_REPLICATION_SLOT. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) +ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, + bool *failover, bool *two_phase) { bool failover_given = false; + bool two_phase_given = false; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) @@ -1425,6 +1427,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) failover_given = true; *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); } @@ -1437,9 +1448,10 @@ static void AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { bool failover = false; + bool two_phase = false; - ParseAlterReplSlotOptions(cmd, &failover); - ReplicationSlotAlter(cmd->slotname, failover); + ParseAlterReplSlotOptions(cmd, &failover, &two_phase); + ReplicationSlotAlter(cmd->slotname, failover, two_phase); } /* 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..163a4a911a 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..990f5242f9 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 require_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 0f2a25cdc1..51fa4b9690 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..a3886d79ca 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..4e8f627f7b 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -367,6 +367,81 @@ $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"); + +############################### +# Disable the subscription and alter it to two_phase = false, +# then 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 and make sure 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 and 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 subscription two_phase 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); + 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 +449,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] v17-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (13.8K, ../../OSBPR01MB255215A39F66932323295854F5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v17-0002-Alter-slot-option-two_phase-only-when-altering-t.patch) download | inline diff: From e4fe6cde1f4a123668b0a2c1396526bee816c536 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Wed, 17 Apr 2024 06:18:23 +0000 Subject: [PATCH v17 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 | 76 ++++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 23 ++++-- src/include/replication/walreceiver.h | 5 +- src/test/subscription/t/021_twophase.pl | 41 ++++++---- 5 files changed, 102 insertions(+), 45 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 0b23df1b77..475a42a2e3 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 = off)</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 255628c396..d109fb2e84 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -112,6 +112,7 @@ static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); static void CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, + bool needs_update, bool isTopLevel); @@ -1074,11 +1075,9 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, */ static void CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, - bool isTopLevel) + bool needs_update, bool isTopLevel) { - StringInfoData cmd; - - if (!sub->slotname) + if (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", @@ -1096,16 +1095,20 @@ CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, 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: prevent we are in * the transaction state. */ - PreventInTransactionBlock(isTopLevel, cmd.data); + if (needs_update) + { + StringInfoData cmd; - pfree(cmd.data); + initStringInfo(&cmd); + appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); + + PreventInTransactionBlock(isTopLevel, cmd.data); + pfree(cmd.data); + } } /* @@ -1127,6 +1130,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); @@ -1259,8 +1264,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_FAILOVER)) { + /* + * 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; + CommonChecksForFailoverAndTwophase(sub, "failover", - isTopLevel); + update_failover, isTopLevel); values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover); @@ -1269,16 +1282,36 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) { + /* + * 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; + CommonChecksForFailoverAndTwophase(sub, "two_phase", - isTopLevel); + 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. + * If the wo_phase slot option must be altered, this cannot + * be altered with slot_name 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)) + 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"))); @@ -1300,7 +1333,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, @@ -1559,14 +1592,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; @@ -1586,7 +1618,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 2f035a0c3c..07dfec947d 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 4e8f627f7b..f56dff4b12 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"); # then 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 and make sure that it is not replicated. $node_publisher->safe_psql( @@ -411,6 +423,19 @@ $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 and verify that it is replicated $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); @@ -422,20 +447,6 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(3), 'replicated data in subscriber table'); -# Alter subscription two_phase 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); - 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] v17-0003-Notify-users-to-roll-back-prepared-transactions.patch (2.2K, ../../OSBPR01MB255215A39F66932323295854F5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v17-0003-Notify-users-to-roll-back-prepared-transactions.patch) download | inline diff: From a2a93925c9e05ea82739cb16dc7bd06582099f11 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Tue, 9 Jul 2024 08:01:43 +0000 Subject: [PATCH v17 3/3] Notify users to roll back prepared transactions --- doc/src/sgml/ref/alter_subscription.sgml | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 475a42a2e3..5263fe7c68 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,19 @@ 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 <literal>two_phase</literal> 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. Alter 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] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-13 10:48 Zhijie Hou (Fujitsu) <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 1 sibling, 0 replies; 9+ messages in thread From: Zhijie Hou (Fujitsu) @ 2024-07-13 10:48 UTC (permalink / raw) To: Hayato Kuroda (Fujitsu) <[email protected]>; [email protected] <[email protected]>; +Cc: Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]>; 'Amit Kapila' <[email protected]> On Tuesday, July 9, 2024 8:53 PM Hayato Kuroda (Fujitsu) <[email protected]> wrote: > > > 0001 - Codes for SUBOPT_TWOPHASE_COMMIT are moved per requirement > [1]. > > Also, checks for failover and two_phase are unified into one function. > > 0002 - updated accordingly. An argument for the check function is added. > > 0003 - this contains documentation changes required in [2]. > > Previous patch set could not be accepted due to the initialization miss. > PSA new version. Thanks for the patches ! I initially reviewed the 0001 and found that the implementation of ALTER_REPLICATION_SLOT has a issue, e.g. it doesn't handle the case when there is only one specified option in the replication command: ALTER_REPLICATION_SLOT slot (two_phase) In this case, it always overwrites the un-specified option(failover) to false even when the failover was set to true. I tried to make a small fix which is on top of 0001 (please see the attachment). I also added the doc of the new two_phase option of the replication command and a missing period of errhint in the topup patch. Best Regards, Hou zj From 3bbaaba53a0cb3db43cc893acbd3ffbedd61bff1 Mon Sep 17 00:00:00 2001 From: Hou Zhijie <[email protected]> Date: Sat, 13 Jul 2024 18:31:28 +0800 Subject: [PATCH] fix alter replication slot --- doc/src/sgml/protocol.sgml | 16 ++++++++++++++ src/backend/commands/subscriptioncmds.c | 2 +- src/backend/replication/slot.c | 16 ++++++++------ src/backend/replication/walsender.c | 29 +++++++++++++++---------- src/include/replication/slot.h | 4 ++-- 5 files changed, 46 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 1b27d0a547..3ac4a4be28 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -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/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 7604e228e8..c48b6d0549 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1308,7 +1308,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot disable two_phase when uncommitted prepared transactions present"), - errhint("Resolve these transactions and try again"))); + errhint("Resolve these transactions and try again."))); /* Change system catalog acoordingly */ values[Anum_pg_subscription_subtwophasestate - 1] = diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2ad6dca993..2f167a2adc 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -804,11 +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, bool two_phase) +ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase) { bool update_slot = false; Assert(MyReplicationSlot == NULL); + Assert(failover || two_phase); ReplicationSlotAcquire(name, false); @@ -834,7 +835,7 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase) * 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" @@ -845,24 +846,25 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase) * 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) + if (failover && *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) + if (failover && MyReplicationSlot->data.failover != *failover) { SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.failover = failover; + MyReplicationSlot->data.failover = *failover; SpinLockRelease(&MyReplicationSlot->mutex); update_slot = true; } - if (MyReplicationSlot->data.two_phase != two_phase) + if (two_phase && MyReplicationSlot->data.two_phase != *two_phase) { SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.two_phase = two_phase; + MyReplicationSlot->data.two_phase = *two_phase; SpinLockRelease(&MyReplicationSlot->mutex); update_slot = true; diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 5224ea6c2c..f3b5068d95 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1411,30 +1411,31 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) */ static void ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, - bool *failover, bool *two_phase) + bool *failover_given, bool *failover, + bool *two_phase_given, bool *two_phase) { - bool failover_given = false; - bool two_phase_given = false; + *failover_given = false; + *two_phase_given = false; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) { if (strcmp(defel->defname, "failover") == 0) { - if (failover_given) + if (*failover_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - failover_given = true; + *failover_given = true; *failover = defGetBoolean(defel); } else if (strcmp(defel->defname, "two_phase") == 0) { - if (two_phase_given) + if (*two_phase_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - two_phase_given = true; + *two_phase_given = true; *two_phase = defGetBoolean(defel); } else @@ -1448,11 +1449,17 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, static void AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { - bool failover = false; - bool two_phase = false; + bool failover_given; + bool two_phase_given; + bool failover; + bool two_phase; + + ParseAlterReplSlotOptions(cmd, &failover_given, &failover, + &two_phase_given, &two_phase); - ParseAlterReplSlotOptions(cmd, &failover, &two_phase); - ReplicationSlotAlter(cmd->slotname, failover, two_phase); + ReplicationSlotAlter(cmd->slotname, + failover_given ? &failover : NULL, + two_phase_given ? &two_phase : NULL); } /* diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 163a4a911a..cde164472a 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -243,8 +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, - bool two_phase); +extern void ReplicationSlotAlter(const char *name, bool *failover, + bool *two_phase); extern void ReplicationSlotAcquire(const char *name, bool nowait); extern void ReplicationSlotRelease(void); -- 2.30.0.windows.2 Attachments: [text/plain] 0001-fix-alter-replication-slot.patch.txt (7.1K, ../../OS3PR01MB57184E0995521300AC06CB4B94A72@OS3PR01MB5718.jpnprd01.prod.outlook.com/2-0001-fix-alter-replication-slot.patch.txt) download | inline diff: From 3bbaaba53a0cb3db43cc893acbd3ffbedd61bff1 Mon Sep 17 00:00:00 2001 From: Hou Zhijie <[email protected]> Date: Sat, 13 Jul 2024 18:31:28 +0800 Subject: [PATCH] fix alter replication slot --- doc/src/sgml/protocol.sgml | 16 ++++++++++++++ src/backend/commands/subscriptioncmds.c | 2 +- src/backend/replication/slot.c | 16 ++++++++------ src/backend/replication/walsender.c | 29 +++++++++++++++---------- src/include/replication/slot.h | 4 ++-- 5 files changed, 46 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 1b27d0a547..3ac4a4be28 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -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/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 7604e228e8..c48b6d0549 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1308,7 +1308,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot disable two_phase when uncommitted prepared transactions present"), - errhint("Resolve these transactions and try again"))); + errhint("Resolve these transactions and try again."))); /* Change system catalog acoordingly */ values[Anum_pg_subscription_subtwophasestate - 1] = diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2ad6dca993..2f167a2adc 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -804,11 +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, bool two_phase) +ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase) { bool update_slot = false; Assert(MyReplicationSlot == NULL); + Assert(failover || two_phase); ReplicationSlotAcquire(name, false); @@ -834,7 +835,7 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase) * 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" @@ -845,24 +846,25 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase) * 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) + if (failover && *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) + if (failover && MyReplicationSlot->data.failover != *failover) { SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.failover = failover; + MyReplicationSlot->data.failover = *failover; SpinLockRelease(&MyReplicationSlot->mutex); update_slot = true; } - if (MyReplicationSlot->data.two_phase != two_phase) + if (two_phase && MyReplicationSlot->data.two_phase != *two_phase) { SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.two_phase = two_phase; + MyReplicationSlot->data.two_phase = *two_phase; SpinLockRelease(&MyReplicationSlot->mutex); update_slot = true; diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 5224ea6c2c..f3b5068d95 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1411,30 +1411,31 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) */ static void ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, - bool *failover, bool *two_phase) + bool *failover_given, bool *failover, + bool *two_phase_given, bool *two_phase) { - bool failover_given = false; - bool two_phase_given = false; + *failover_given = false; + *two_phase_given = false; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) { if (strcmp(defel->defname, "failover") == 0) { - if (failover_given) + if (*failover_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - failover_given = true; + *failover_given = true; *failover = defGetBoolean(defel); } else if (strcmp(defel->defname, "two_phase") == 0) { - if (two_phase_given) + if (*two_phase_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - two_phase_given = true; + *two_phase_given = true; *two_phase = defGetBoolean(defel); } else @@ -1448,11 +1449,17 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, static void AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { - bool failover = false; - bool two_phase = false; + bool failover_given; + bool two_phase_given; + bool failover; + bool two_phase; + + ParseAlterReplSlotOptions(cmd, &failover_given, &failover, + &two_phase_given, &two_phase); - ParseAlterReplSlotOptions(cmd, &failover, &two_phase); - ReplicationSlotAlter(cmd->slotname, failover, two_phase); + ReplicationSlotAlter(cmd->slotname, + failover_given ? &failover : NULL, + two_phase_given ? &two_phase : NULL); } /* diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 163a4a911a..cde164472a 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -243,8 +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, - bool two_phase); +extern void ReplicationSlotAlter(const char *name, bool *failover, + bool *two_phase); extern void ReplicationSlotAcquire(const char *name, bool nowait); extern void ReplicationSlotRelease(void); -- 2.30.0.windows.2 ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-15 11:39 Amit Kapila <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 1 sibling, 1 reply; 9+ messages in thread From: Amit Kapila @ 2024-07-15 11:39 UTC (permalink / raw) To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: [email protected] <[email protected]>; Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]> On Tue, Jul 9, 2024 at 6:23 PM Hayato Kuroda (Fujitsu) <[email protected]> wrote: > > Previous patch set could not be accepted due to the initialization miss. > PSA new version. > Few minor comments: ================= 0001-patch 1. .git/rebase-apply/patch:253: space before tab in indent. errmsg("slot_name and two_phase cannot be altered at the same time"))); warning: 1 line adds whitespace errors. White space issue as shown by git am command. 2. +/* + * Common checks for altering failover and two_phase option + */ +static void +CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option, + bool isTopLevel) The function name looks odd to me. How about something along the lines of CheckAlterSubOption()? 3. + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot disable two_phase when uncommitted prepared transactions present"), We can slightly change the above error message to: "cannot disable two_phase when prepared transactions are present". 0003-patch Alter the altering from + <literal>true</literal> to <literal>false</literal>, the publisher will + replicate transactions again when they are committed. The beginning of the sentence sounds awkward. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-16 05:17 Hayato Kuroda (Fujitsu) <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 2 replies; 9+ messages in thread From: Hayato Kuroda (Fujitsu) @ 2024-07-16 05:17 UTC (permalink / raw) To: 'Amit Kapila' <[email protected]>; Zhijie Hou (Fujitsu) <[email protected]>; +Cc: [email protected] <[email protected]>; Vitaly Davydov <[email protected]>; Ajin Cherian <[email protected]> Dear Amit, Hou, Thanks for giving comments! PSA new versions. What's new: 0001: included Hou's patch [1] not to overwrite slot options. Some other comments were also addressed. 0002: not so changed, just rebased. 0003: Typo was fixed, s/Alter/After/. [1]: https://www.postgresql.org/message-id/OS3PR01MB57184E0995521300AC06CB4B94A72%40OS3PR01MB5718.jpnprd0... Best regards, Hayato Kuroda FUJITSU LIMITED Attachments: [application/octet-stream] v18-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (32.7K, ../../TYCPR01MB56938F06B84DD92A88234667F5A22@TYCPR01MB5693.jpnprd01.prod.outlook.com/2-v18-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch) download | inline diff: From e7a2051d90b236989707d33236f0ab840c982283 Mon Sep 17 00:00:00 2001 From: Ajin Cherian <[email protected]> Date: Fri, 5 Apr 2024 06:47:18 -0400 Subject: [PATCH v18 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 | 16 +++ doc/src/sgml/ref/alter_subscription.sgml | 12 +- src/backend/access/transam/twophase.c | 62 +++++++++ src/backend/commands/subscriptioncmds.c | 131 +++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 9 +- src/backend/replication/logical/launcher.c | 12 +- src/backend/replication/logical/worker.c | 25 +--- src/backend/replication/slot.c | 28 +++- src/backend/replication/walsender.c | 33 ++++- 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 | 77 +++++++++- 17 files changed, 337 insertions(+), 102 deletions(-) diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 1b27d0a547..3ac4a4be28 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -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..35bce6809d 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2681,3 +2681,65 @@ 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; + + ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + + return (ret == 2 && subid == subid_written); +} + +/* + * 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..d7e2b141b3 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,44 @@ 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; + + 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: prevent we are in + * the transaction state. + */ + PreventInTransactionBlock(isTopLevel, cmd.data); + + pfree(cmd.data); +} + /* * Alter the existing subscription. */ @@ -1145,7 +1174,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 +1259,59 @@ 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 twophase parameter is + * updated, which causes the assertion failure. 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 acoordingly */ + 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 +1563,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 +1585,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 +1732,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..45744b771f 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -272,12 +272,15 @@ 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 require_lock) { int i; List *res = NIL; - Assert(LWLockHeldByMe(LogicalRepWorkerLock)); + if (require_lock) + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + else + Assert(LWLockHeldByMe(LogicalRepWorkerLock)); /* Search for attached worker for a given subscription id. */ for (i = 0; i < max_logical_replication_workers; i++) @@ -288,6 +291,9 @@ logicalrep_workers_find(Oid subid, bool only_running) res = lappend(res, w); } + if (require_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..2f167a2adc 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,7 +835,7 @@ 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" @@ -843,17 +846,32 @@ ReplicationSlotAlter(const char *name, bool 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) + if (failover && *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) + if (failover && MyReplicationSlot->data.failover != *failover) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + MyReplicationSlot->data.failover = *failover; + SpinLockRelease(&MyReplicationSlot->mutex); + + update_slot = true; + } + + 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..f3b5068d95 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1410,22 +1410,34 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) * Process extra options given to ALTER_REPLICATION_SLOT. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) +ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, + bool *failover_given, bool *failover, + bool *two_phase_given, bool *two_phase) { - bool failover_given = false; + *failover_given = false; + *two_phase_given = false; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) { if (strcmp(defel->defname, "failover") == 0) { - if (failover_given) + if (*failover_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - failover_given = true; + *failover_given = true; *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); } @@ -1437,10 +1449,17 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) static void AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { - bool failover = false; + bool failover_given; + bool two_phase_given; + bool failover; + bool two_phase; + + ParseAlterReplSlotOptions(cmd, &failover_given, &failover, + &two_phase_given, &two_phase); - 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..990f5242f9 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 require_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..52ccb160fa 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..a3886d79ca 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..4e8f627f7b 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -367,6 +367,81 @@ $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"); + +############################### +# Disable the subscription and alter it to two_phase = false, +# then 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 and make sure 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 and 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 subscription two_phase 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); + 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 +449,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] v18-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (13.9K, ../../TYCPR01MB56938F06B84DD92A88234667F5A22@TYCPR01MB5693.jpnprd01.prod.outlook.com/3-v18-0002-Alter-slot-option-two_phase-only-when-altering-t.patch) download | inline diff: From e63c14572a34efa4639e038a3c50e930f710d06c Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Wed, 17 Apr 2024 06:18:23 +0000 Subject: [PATCH v18 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 | 80 ++++++++++++++----- .../libpqwalreceiver/libpqwalreceiver.c | 23 ++++-- src/include/replication/walreceiver.h | 5 +- src/test/subscription/t/021_twophase.pl | 41 ++++++---- 5 files changed, 105 insertions(+), 46 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 d7e2b141b3..f09a6bb290 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 needs_update, bool isTopLevel); /* @@ -1074,11 +1074,9 @@ 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 needs_update, bool isTopLevel) { - StringInfoData cmd; - - if (!sub->slotname) + if (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", @@ -1096,16 +1094,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: prevent we are in * the transaction state. */ - PreventInTransactionBlock(isTopLevel, cmd.data); + if (needs_update) + { + StringInfoData cmd; - pfree(cmd.data); + initStringInfo(&cmd); + appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); + + PreventInTransactionBlock(isTopLevel, cmd.data); + pfree(cmd.data); + } } /* @@ -1127,6 +1129,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); @@ -1259,7 +1263,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); @@ -1268,15 +1281,37 @@ 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. + * If the wo_phase slot option must be altered, this + * cannot be altered with slot_name 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)) + 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"))); @@ -1298,7 +1333,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, @@ -1557,14 +1592,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; @@ -1585,7 +1619,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 4e8f627f7b..f56dff4b12 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"); # then 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 and make sure that it is not replicated. $node_publisher->safe_psql( @@ -411,6 +423,19 @@ $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 and verify that it is replicated $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); @@ -422,20 +447,6 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(3), 'replicated data in subscriber table'); -# Alter subscription two_phase 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); - 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] v18-0003-Notify-users-to-roll-back-prepared-transactions.patch (2.2K, ../../TYCPR01MB56938F06B84DD92A88234667F5A22@TYCPR01MB5693.jpnprd01.prod.outlook.com/4-v18-0003-Notify-users-to-roll-back-prepared-transactions.patch) download | inline diff: From 17eba2b13eb220f4885e4dcdadefc4edd6409844 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda <[email protected]> Date: Tue, 9 Jul 2024 08:01:43 +0000 Subject: [PATCH v18 3/3] Notify users to roll back prepared transactions --- doc/src/sgml/ref/alter_subscription.sgml | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index df44415661..e021f8729b 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,19 @@ 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 <literal>two_phase</literal> 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 + <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] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-16 10:02 Peter Smith <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 1 sibling, 0 replies; 9+ messages in thread From: Peter Smith @ 2024-07-16 10:02 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 are some review comments for patch v18-0001. ====== doc/src/sgml/protocol.sgml nitpick - Although it is no fault of your patch, IMO it would be nicer for the TWO_PHASE description (of CREATE REPLICATION SLOT) to also be in the same consistent order as what you have (e.g. below FAILOVER). So I moved it. ====== src/backend/access/transam/twophase.c LookupGXactBySubid: nitpick - add a blank line before return ====== src/backend/commands/subscriptioncmds.c CommonChecksForFailoverAndTwophase: nitpick - added Assert for the generic-looking "option" parameter name nitpick - modified comment about transaction block ~~~ 1. AlterSubscription + * Workers may still survive even if the subscription has + * been disabled. They may read the pg_subscription + * catalog and detect that the twophase parameter is + * updated, which causes the assertion failure. Ensure + * workers have already been exited to avoid it. "which causes the assertion failure" -- what assertion failure is that? The comment is not very clear. ~ nitpick - in comment /twophase/two_phase/ nitpick - typo /acoordingly/accordingly/ ====== src/backend/replication/logical/launcher.c logicalrep_workers_find: nitpick - /require_lock/acquire_lock/ nitpick - take the Assert out of the else. ====== src/backend/replication/slot.c nitpick - refactor the code to check (failover) only one time. See the nitpicks attachment. ~ 2. ParseAlterReplSlotOptions nitpick -- IMO the ParseAlterReplSlotOptions(). function does more harm than good here by adding the unnecessary complexity of messing around with multiple parameters that are passed-by-reference. All this would be simpler if it was just coded inline in the AlterReplicationSlot() function, which is the only caller. I've refactored all this to demonstrate (see nitpicks attachment) ====== src/include/replication/worker_internal.h nitpick - /require_lock/acquire_lock/ ====== src/test/regress/sql/subscription.sql nitpick - tweak comments ====== src/test/subscription/t/021_twophase.pl nitpick - change comment style to indicate each test part better. ====== 99. Please also see the attached diffs patch which implements any nitpicks mentioned above. ====== Kind Regards, Peter Smith. Fujitsu Australia diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 3ac4a4b..cba6661 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2050,21 +2050,6 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" <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> <para> @@ -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> diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 35bce68..f3c6e1f 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2741,5 +2741,6 @@ LookupGXactBySubid(Oid subid) } } LWLockRelease(TwoPhaseStateLock); + return found; } diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 6995a62..3703cf6 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1076,6 +1076,8 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) { StringInfoData cmd; + Assert(strstr("two_phase,failover", option)); + if (!sub->slotname) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), @@ -1098,8 +1100,8 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); /* - * The changed option of the slot can't be rolled back: prevent we are in - * the transaction state. + * The changed option of the slot can't be rolled back, so disallow if we + * are in a transaction block. */ PreventInTransactionBlock(isTopLevel, cmd.data); @@ -1282,7 +1284,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, /* * Workers may still survive even if the subscription has * been disabled. They may read the pg_subscription - * catalog and detect that the twophase parameter is + * catalog and detect that the two_phase parameter is * updated, which causes the assertion failure. Ensure * workers have already been exited to avoid it. */ @@ -1304,7 +1306,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, errmsg("cannot disable two_phase when prepared transactions are present"), errhint("Resolve these transactions and try again."))); - /* Change system catalog acoordingly */ + /* Change system catalog accordingly */ values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(opts.twophase ? LOGICALREP_TWOPHASE_STATE_PENDING : diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 45744b7..c566d50 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -272,15 +272,15 @@ 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, bool require_lock) +logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock) { int i; List *res = NIL; - if (require_lock) + if (acquire_lock) LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - else - Assert(LWLockHeldByMe(LogicalRepWorkerLock)); + + Assert(LWLockHeldByMe(LogicalRepWorkerLock)); /* Search for attached worker for a given subscription id. */ for (i = 0; i < max_logical_replication_workers; i++) @@ -291,7 +291,7 @@ logicalrep_workers_find(Oid subid, bool only_running, bool require_lock) res = lappend(res, w); } - if (require_lock) + if (acquire_lock) LWLockRelease(LogicalRepWorkerLock); return res; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2f167a2..e75f24b 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -842,23 +842,25 @@ ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase) " 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 && *failover && - MyReplicationSlot->data.persistency == RS_TEMPORARY) + 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 (failover && MyReplicationSlot->data.failover != *failover) - { - SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.failover = *failover; - SpinLockRelease(&MyReplicationSlot->mutex); + if (MyReplicationSlot->data.failover != *failover) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + MyReplicationSlot->data.failover = *failover; + SpinLockRelease(&MyReplicationSlot->mutex); - update_slot = true; + update_slot = true; + } } if (two_phase && MyReplicationSlot->data.two_phase != *two_phase) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 04f65e0..af8e958 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1405,56 +1405,42 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) ReplicationSlotDrop(cmd->slotname, !cmd->wait); } + /* - * Process extra options given to ALTER_REPLICATION_SLOT. + * Change the definition of a replication slot. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, - bool *failover_given, bool *failover, - bool *two_phase_given, bool *two_phase) +AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { - *failover_given = false; - *two_phase_given = false; + bool failover_given = false; + bool two_phase_given = false; + bool failover; + bool two_phase; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) { if (strcmp(defel->defname, "failover") == 0) { - if (*failover_given) + if (failover_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - *failover_given = true; - *failover = defGetBoolean(defel); + failover_given = true; + failover = defGetBoolean(defel); } else if (strcmp(defel->defname, "two_phase") == 0) { - if (*two_phase_given) + if (two_phase_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - *two_phase_given = true; - *two_phase = defGetBoolean(defel); + 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_given; - bool two_phase_given; - bool failover; - bool two_phase; - - ParseAlterReplSlotOptions(cmd, &failover_given, &failover, - &two_phase_given, &two_phase); ReplicationSlotAlter(cmd->slotname, failover_given ? &failover : NULL, diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 990f524..9646261 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -241,7 +241,7 @@ 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, - bool require_lock); + 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 51fa4b9..40e1a07 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -377,7 +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) --- We can alter streaming when two_phase enabled +-- we can alter streaming when two_phase is 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 a3886d7..b64f419 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -256,7 +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+ --- We can alter streaming when two_phase enabled +-- we can alter streaming when two_phase is 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 4e8f627..66265c7 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -371,8 +371,8 @@ is($result, qq(2), 'replicated data in subscriber table'); $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); ############################### -# Disable the subscription and alter it to two_phase = false, -# then verify that the altered subscription reflects the two_phase option. +# Alter the subscription to two_phase = false. +# Verify that the altered subscription reflects the two_phase option. ############################### # Alter subscription two_phase to false @@ -395,7 +395,10 @@ $result = $node_subscriber->safe_psql('postgres', ); is($result, qq(d), 'two-phase should be disabled'); -# Now do a prepare on the publisher and make sure that it is not replicated. +############################### +# Now do a prepare on the publisher. +# Verify that it is not replicated. +############################### $node_publisher->safe_psql( 'postgres', qq{ BEGIN; @@ -411,7 +414,10 @@ $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 and verify that it is replicated +############################### +# Now commit the insert. +# Verify that it is replicated. +############################### $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); # Wait for the subscriber to catchup @@ -422,7 +428,10 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(3), 'replicated data in subscriber table'); -# Alter subscription two_phase to true +############################### +# 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', Attachments: [text/plain] PS_NITPICKS_20240716_2PC_v180001.txt (13.0K, ../../CAHut+PuDccJ_HsNcPajDLC1O1LsJ6SeZfB5=EGGLjTKBs-9R2A@mail.gmail.com/3-PS_NITPICKS_20240716_2PC_v180001.txt) download | inline diff: diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 3ac4a4b..cba6661 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2050,21 +2050,6 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" <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> <para> @@ -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> diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 35bce68..f3c6e1f 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2741,5 +2741,6 @@ LookupGXactBySubid(Oid subid) } } LWLockRelease(TwoPhaseStateLock); + return found; } diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 6995a62..3703cf6 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1076,6 +1076,8 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) { StringInfoData cmd; + Assert(strstr("two_phase,failover", option)); + if (!sub->slotname) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), @@ -1098,8 +1100,8 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel) appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option); /* - * The changed option of the slot can't be rolled back: prevent we are in - * the transaction state. + * The changed option of the slot can't be rolled back, so disallow if we + * are in a transaction block. */ PreventInTransactionBlock(isTopLevel, cmd.data); @@ -1282,7 +1284,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, /* * Workers may still survive even if the subscription has * been disabled. They may read the pg_subscription - * catalog and detect that the twophase parameter is + * catalog and detect that the two_phase parameter is * updated, which causes the assertion failure. Ensure * workers have already been exited to avoid it. */ @@ -1304,7 +1306,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, errmsg("cannot disable two_phase when prepared transactions are present"), errhint("Resolve these transactions and try again."))); - /* Change system catalog acoordingly */ + /* Change system catalog accordingly */ values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(opts.twophase ? LOGICALREP_TWOPHASE_STATE_PENDING : diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 45744b7..c566d50 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -272,15 +272,15 @@ 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, bool require_lock) +logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock) { int i; List *res = NIL; - if (require_lock) + if (acquire_lock) LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - else - Assert(LWLockHeldByMe(LogicalRepWorkerLock)); + + Assert(LWLockHeldByMe(LogicalRepWorkerLock)); /* Search for attached worker for a given subscription id. */ for (i = 0; i < max_logical_replication_workers; i++) @@ -291,7 +291,7 @@ logicalrep_workers_find(Oid subid, bool only_running, bool require_lock) res = lappend(res, w); } - if (require_lock) + if (acquire_lock) LWLockRelease(LogicalRepWorkerLock); return res; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2f167a2..e75f24b 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -842,23 +842,25 @@ ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase) " 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 && *failover && - MyReplicationSlot->data.persistency == RS_TEMPORARY) + 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 (failover && MyReplicationSlot->data.failover != *failover) - { - SpinLockAcquire(&MyReplicationSlot->mutex); - MyReplicationSlot->data.failover = *failover; - SpinLockRelease(&MyReplicationSlot->mutex); + if (MyReplicationSlot->data.failover != *failover) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + MyReplicationSlot->data.failover = *failover; + SpinLockRelease(&MyReplicationSlot->mutex); - update_slot = true; + update_slot = true; + } } if (two_phase && MyReplicationSlot->data.two_phase != *two_phase) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 04f65e0..af8e958 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1405,56 +1405,42 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) ReplicationSlotDrop(cmd->slotname, !cmd->wait); } + /* - * Process extra options given to ALTER_REPLICATION_SLOT. + * Change the definition of a replication slot. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, - bool *failover_given, bool *failover, - bool *two_phase_given, bool *two_phase) +AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { - *failover_given = false; - *two_phase_given = false; + bool failover_given = false; + bool two_phase_given = false; + bool failover; + bool two_phase; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) { if (strcmp(defel->defname, "failover") == 0) { - if (*failover_given) + if (failover_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - *failover_given = true; - *failover = defGetBoolean(defel); + failover_given = true; + failover = defGetBoolean(defel); } else if (strcmp(defel->defname, "two_phase") == 0) { - if (*two_phase_given) + if (two_phase_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - *two_phase_given = true; - *two_phase = defGetBoolean(defel); + 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_given; - bool two_phase_given; - bool failover; - bool two_phase; - - ParseAlterReplSlotOptions(cmd, &failover_given, &failover, - &two_phase_given, &two_phase); ReplicationSlotAlter(cmd->slotname, failover_given ? &failover : NULL, diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 990f524..9646261 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -241,7 +241,7 @@ 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, - bool require_lock); + 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 51fa4b9..40e1a07 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -377,7 +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) --- We can alter streaming when two_phase enabled +-- we can alter streaming when two_phase is 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 a3886d7..b64f419 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -256,7 +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+ --- We can alter streaming when two_phase enabled +-- we can alter streaming when two_phase is 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 4e8f627..66265c7 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -371,8 +371,8 @@ is($result, qq(2), 'replicated data in subscriber table'); $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); ############################### -# Disable the subscription and alter it to two_phase = false, -# then verify that the altered subscription reflects the two_phase option. +# Alter the subscription to two_phase = false. +# Verify that the altered subscription reflects the two_phase option. ############################### # Alter subscription two_phase to false @@ -395,7 +395,10 @@ $result = $node_subscriber->safe_psql('postgres', ); is($result, qq(d), 'two-phase should be disabled'); -# Now do a prepare on the publisher and make sure that it is not replicated. +############################### +# Now do a prepare on the publisher. +# Verify that it is not replicated. +############################### $node_publisher->safe_psql( 'postgres', qq{ BEGIN; @@ -411,7 +414,10 @@ $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 and verify that it is replicated +############################### +# Now commit the insert. +# Verify that it is replicated. +############################### $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); # Wait for the subscriber to catchup @@ -422,7 +428,10 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(3), 'replicated data in subscriber table'); -# Alter subscription two_phase to true +############################### +# 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', ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR @ 2024-07-17 03:31 Peter Smith <[email protected]> parent: Hayato Kuroda (Fujitsu) <[email protected]> 1 sibling, 0 replies; 9+ messages in thread From: Peter Smith @ 2024-07-17 03:31 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]> Here are some review comments for patch v18-0002. ====== src/backend/commands/subscriptioncmds.c 1. CheckAlterSubOption 1a. It's not obvious why we are only checking the 'slot name' when needs_update==true, but OTOH is always checking the 'enabled' state. ~ 1b. Param 'needs_update' is a vague name. It needs more explanatory comments or a better name. e.g. First impression was "Why are we calling 'Alter' function if needs_update is false?". I know it encapsulates some common code, but if special cases cause the logic to be more confusing then that cost may outweigh the benefit of this function. ~ 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. ~~~ AlterSubscription: nitpick - typo /needs/need/ nitpick - typo /wo_phase/two_phase/ nitpick - The comment wording "the later part...", was confusing. I've reworded the whole comment. But this belongs in patch 0001. ====== src/test/subscription/t/021_twophase.pl nitpick - Use the same "###############################" comment style as in patch 0001 to indicate each main TEST scenario. ====== 99. Please refer to the diffs attachment patch, which implements any nitpicks mentioned above. ====== Kind Regards, Peter Smith. Fujitsu Australia diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index d80d60c..5c6f0a5 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1262,7 +1262,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). @@ -1280,7 +1280,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) { /* - * First check the need to alter the replication slot. + * 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 @@ -1302,11 +1302,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, isTopLevel); /* - * If the wo_phase slot option must be altered, this - * cannot be altered with slot_name 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 (update_two_phase && IsSet(opts.specified_opts, SUBOPT_SLOT_NAME)) diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index f56dff4..26b9e2c 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -423,9 +423,12 @@ $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". +############################### +# 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', Attachments: [text/plain] PS_NITPICKS_20240717_2PC_V180002.txt (2.7K, ../../CAHut+PsqMRS3dcijo5jsTSbgV1-9So-YBC7PH7xg0+Z8hA7fDQ@mail.gmail.com/3-PS_NITPICKS_20240717_2PC_V180002.txt) download | inline diff: diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index d80d60c..5c6f0a5 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1262,7 +1262,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). @@ -1280,7 +1280,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) { /* - * First check the need to alter the replication slot. + * 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 @@ -1302,11 +1302,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, isTopLevel); /* - * If the wo_phase slot option must be altered, this - * cannot be altered with slot_name 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 (update_two_phase && IsSet(opts.specified_opts, SUBOPT_SLOT_NAME)) diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index f56dff4..26b9e2c 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -423,9 +423,12 @@ $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". +############################### +# 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', ^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2024-07-17 03:31 UTC | newest] Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2019-09-26 20:49 [PATCH 4/9] Silence 'mixed declarations and code' compiler warnings Alvaro Herrera <[email protected]> 2024-07-09 05:05 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]> 2024-07-09 11:49 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]> 2024-07-09 12:52 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]> 2024-07-13 10:48 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Zhijie Hou (Fujitsu) <[email protected]> 2024-07-15 11:39 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]> 2024-07-16 05:17 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]> 2024-07-16 10:02 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Peter Smith <[email protected]> 2024-07-17 03:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Peter Smith <[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