public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: 'Peter Smith' <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Ajin Cherian <[email protected]>
Cc: Давыдов Виталий <[email protected]>
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
Date: Wed, 8 May 2024 08:26:42 +0000
Message-ID: <OSBPR01MB2552521ABDBC2F995FBACDF7F5E52@OSBPR01MB2552.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAHut+Pt2v3dCvWCE-ng4nxS5qU-Go4T=Q2BBnv73SAAczgrqvg@mail.gmail.com>
References: <8fab8-65d74c80-1-2f28e880@39088166>
<CAA4eK1KOs3s6syZqUgrd2WvjTz64SGf0ToZcRoPMCKKH+M0YFQ@mail.gmail.com>
<99df2-65d8d200-1-63c26880@172687752>
<CAA4eK1KtjjPxe5tjWeGNf5CfF9qfceVQCCC9XGsKzR-ifQD9uA@mail.gmail.com>
<ba4b6-65ddc580-9-5c3e3080@134362348>
<CAA4eK1J6wOMkDDt3iOs+JqkvQvuwSORosTpqdejgvXqFmxex=A@mail.gmail.com>
<badcf-65dde500-3-24ba3c80@28790566>
<cedbe-65e0c000-1-6db17700@133269862>
<[email protected]>
<f9f1b-65e72c80-7-533dda00@76203995>
<CAFPTHDbU4EkVmKL66i5BzrmQ6LFftQPx_TZ1euKt9fvVzQDyjA@mail.gmail.com>
<CAA4eK1+n7M2S1OpoGWDd+YZkDCuURMdVRbvP0eELQUvWgmDneg@mail.gmail.com>
<CAFPTHDa=pJSZ_4dV5DPAOapRSgPcyyUTP0WzGY2Rz_D3-gwraw@mail.gmail.com>
<CAA4eK1K1fSkeK=kc26G5cq87vQG4=1qs_b+no4+ep654SeBy1w@mail.gmail.com>
<23d1ca-66169f80-1f-5a90aa00@207285398>
<26b30e-661d4880-9-62d2df00@231402233>
<OSBPR01MB2552F73968D5A460157B57B9F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<CAHut+Pt2v3dCvWCE-ng4nxS5qU-Go4T=Q2BBnv73SAAczgrqvg@mail.gmail.com>
Dear Peter,
Thanks for reviewing! Here are updated patches.
I updated patches only for HEAD.
> ======
> Commit message
>
> 1.
> This patch allows user to alter two_phase option
>
> /allows user/allows the user/
>
> /to alter two_phase option/to alter the 'two_phase' option/
Fixed.
> ======
> doc/src/sgml/ref/alter_subscription.sgml
>
> 2.
> <literal>two_phase</literal> can be altered only for disabled subscription.
>
> SUGGEST
> The <literal>two_phase</literal> parameter can only be altered when
> the subscription is disabled.
Fixed.
> ======
> src/backend/access/transam/twophase.c
>
> 3. checkGid
> +
> +/*
> + * checkGid
> + */
> +static bool
> +checkGid(char *gid, Oid subid)
> +{
> + int ret;
> + Oid subid_written,
> + xid;
> +
> + ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid);
> +
> + if (ret != 2 || subid != subid_written)
> + return false;
> +
> + return true;
> +}
>
> 3a.
> The function comment should give more explanation of what it does. I
> think this function is the counterpart of the TwoPhaseTransactionGid()
> function of worker.c so the comment can say that too.
Comments were updated.
> 3b.
> Indeed, perhaps the function name should be similar to
> TwoPhaseTransactionGid. e.g. call it like
> IsTwoPhaseTransactionGidForSubid?
Replaced to IsTwoPhaseTransactionGidForSubid().
> 3c.
> Probably 'xid' should be TransactionId instead of Oid.
Right, fixed.
> 3d.
> Why not have a single return?
>
> SUGGESTION
> return (ret == 2 && subid = subid_written);
Fixed.
> 3e.
> I am wondering if the existing TwoPhaseTransactionGid function
> currently in worker.c should be moved here because IMO these 2
> functions belong together and twophase.c seems the right place to put
> them.
+1, moved.
> ~~~
>
> 4.
> +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 && checkGid(gxact->gid, subid))
> + {
> + found = true;
> + break;
> + }
> + }
> + LWLockRelease(TwoPhaseStateLock);
> + return found;
> +}
>
> AFAIK the gxact also has the 'xid' available, so won't it be better to
> pass BOTH the 'xid' and 'subid' to the checkGid so you can do a full
> comparison instead of comparing only the subid part of the gid?
IIUC, the xid written in the gxact means the transaction id on the subscriber,
but formatted GID has xid on the publisher. So the value cannot be used.
> ======
> src/backend/commands/subscriptioncmds.c
>
> 5. AlterSubscription
>
> + /* XXX */
> + if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
> + {
>
> The "XXX" comment looks like it is meant to say something more...
This flag was used only for me, removed.
> ~~~
>
> 6.
> + /*
> + * two_phase can be only changed for disabled
> + * subscriptions
> + */
> + if (form->subenabled)
> + ereport(ERROR,
> + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("cannot set %s for enabled subscription",
> + "two_phase")));
>
> 6a.
> Should this have a more comprehensive comment giving the reason like
> the 'failover' option has?
Modified, but it is almost the same as failover's one.
> 6b.
> Maybe this should include a "translator" comment to say don't
> translate the option name.
Hmm, but other parts in AlterSubscription() does not have.
For now, I kept current style.
> ~~~
>
> 7.
> + /* Check whether the number of prepared transactions */
> + if (!opts.twophase &&
> + form->subtwophasestate == 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")));
> +
>
> 7a.
> The first comment seems to be an incomplete sentence. I think it
> should say something a bit like:
> two_phase cannot be disabled if there are any uncommitted prepared
> transactions present.
Modified, but this part would be replaced by upcoming patches.
> 7b.
> Also, if ereport occurs what is the user supposed to do about it?
> Shouldn't the ereport include some errhint with appropriate advice?
The hint was added, but this part would be replaced by upcoming patches.
> ~~~
>
> 8.
> + /*
> + * The changed failover option of the slot can't be rolled
> + * back.
> + */
> + PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET
> (two_phase)");
> +
> + /* 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;
> + }
>
> Typo I think: /failover option/two_phase option/
Right, fixed.
> ======
> .../libpqwalreceiver/libpqwalreceiver.c
>
> 9.
> static void
> libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
> - bool failover)
> + bool two_phase, bool failover)
>
> Same comment as mentioned elsewhere (#15), IMO the new 'two_phase'
> parameter should be last.
Fixed. Also, some ordering of declarations and if-blocks were also changed.
In later part, I did not reply similar comments but I addressed all of them.
> ======
> src/backend/replication/logical/launcher.c
>
> 10.
> +/*
> + * Stop all the subscription workers.
> + */
> +void
> +logicalrep_workers_stop(Oid subid)
> +{
> + List *subworkers;
> + ListCell *lc;
> +
> + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
> + subworkers = logicalrep_workers_find(subid, false);
> + LWLockRelease(LogicalRepWorkerLock);
> + foreach(lc, subworkers)
> + {
> + LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
> +
> + logicalrep_worker_stop(w->subid, w->relid);
> + }
> + list_free(subworkers);
> +}
>
> I was confused by the logicalrep_workers_find(subid, false). IIUC the
> 'false' means everything (instead of 'only_running') but then I don't
> know why we want to "stop" anything that is NOT running. OTOH I see
> that this code was extracted from where it was previously inlined in
> subscriptioncmds.c, so maybe the 'false' is necessary for another
> reason? At least maybe some explanatory comment is needed for why you
> are passing this flag as false?
Sorry, let me give time for more investigation around here. For now,
I added "XXX" mark.
I think it is listed just in case, but there may be a timing issue.
> ======
> src/backend/replication/logical/worker.c
>
> 11.
> - /* two-phase should not be altered */
> + /* two-phase should not be altered while the worker exists */
> Assert(newsub->twophasestate == MySubscription->twophasestate);
> /should not/cannot/
Fixed.
> ~~~
>
> 13.
> + if (MyReplicationSlot->data.two_phase != two_phase)
> + {
> + SpinLockAcquire(&MyReplicationSlot->mutex);
> + MyReplicationSlot->data.two_phase = two_phase;
> + SpinLockRelease(&MyReplicationSlot->mutex);
> +
> + update_slot = true;
> + }
> +
> +
> if (MyReplicationSlot->data.failover != failover)
> {
> SpinLockAcquire(&MyReplicationSlot->mutex);
> MyReplicationSlot->data.failover = failover;
> SpinLockRelease(&MyReplicationSlot->mutex);
>
> + update_slot = true;
> + }
>
> 13a.
> Doesn't it make more sense for the whole check/set to be "atomic",
> i.e. put the mutex also around the check?
>
> SUGGEST
> SpinLockAcquire(&MyReplicationSlot->mutex);
> if (MyReplicationSlot->data.two_phase != two_phase)
> {
> MyReplicationSlot->data.two_phase = two_phase;
> update_slot = true;
> }
> SpinLockRelease(&MyReplicationSlot->mutex);
>
> ~
>
> Also, (if you agree with the above) why not include both checks
> (two_phase and failover) within the same mutex instead of
> acquiring/releasing it twice:
>
> SUGGEST
> SpinLockAcquire(&MyReplicationSlot->mutex);
> if (MyReplicationSlot->data.two_phase != two_phase)
> {
> MyReplicationSlot->data.two_phase = two_phase;
> update_slot = true;
> }
> if (MyReplicationSlot->data.failover != failover)
> {
> MyReplicationSlot->data.failover = failover;
> update_slot = true;
> }
> SpinLockAcquire(&MyReplicationSlot->mutex);
Hmm. According to comments atop ReplicationSlot, backends which own the slot do
not have to set mutex for reading attributes. Concurrent backends, which do not
acquire the slot, must set the mutex lock before the read. Based on the manner,
I want to keep current style.
```
* - Individual fields are protected by mutex where only the backend owning
* the slot is authorized to update the fields from its own slot. The
* backend owning the slot does not need to take this lock when reading its
* own fields, while concurrent backends not owning this slot should take the
* lock when reading this slot's data.
*/
typedef struct ReplicationSlot
```
> 13b.
> There are double blank lines after the first if-block.
Removed.
> ======
> src/test/regress/sql/subscription.sql
>
> 16.
> I know you do this already in the TAP test, but doesn't the test case
> to demonstrate that 'two-phase' option can be altered when the
> subscription is disabled actually belong here in the regression
> instead?
Actually it cannot be done at main regression test. Because altering two_phase
requires the connection between pub/sub, but it is not established in subscription.sql
file. Succeeded case for altering failover has not been tested neither, and
I think they have same reason.
> src/test/subscription/t/021_twophase.pl
>
> 17.
> +# Disable the subscription and alter it to two_phase = false,
> +# verify that the altered subscription reflects the two_phase option.
>
> /verify/then verify/
Fixed.
> 18.
> +# Now do a prepare on publisher and make sure that it is not replicated.
> +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
> +$node_publisher->safe_psql(
> + 'postgres', qq{
> + BEGIN;
> + INSERT INTO tab_copy VALUES (100);
> + PREPARE TRANSACTION 'newgid';
> + });
> +
>
> 18a.
> /on publisher/on the publisher/
Fixed.
> 18b.
> What is that "DROP SUBSCRIPTION tap_sub" doing here? It seems
> misplaced under this comment.
The subscription must be dropped because it also prepares a transaction.
Moved before the test case and added comments.
> 19.
> +# Make sure that there is 0 prepared transaction on the subscriber
> +$result = $node_subscriber->safe_psql('postgres',
> + "SELECT count(*) FROM pg_prepared_xacts;");
> +is($result, qq(0), 'transaction is prepared on subscriber');
>
> 19a.
> SUGGESTION
> Make sure there are no prepared transactions on the subscriber
Fixed.
> 19b.
> /'transaction is prepared on subscriber'/'should be no prepared
> transactions on subscriber'/
Replaced/
> 20.
> +# Made sure that the commited transaction is replicated.
>
> /Made sure/Make sure/
>
> /commited/committed/
Fixed.
> 21.
> +# 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 is disabled');
>
> The 'two-phase is disabled' is the identical message used in the
> opposite case earlier, so something is amiss. Maybe this one should
> say 'two-phase should be enabled' and the earlier counterpart should
> say 'two-phase should be disabled'.
Both of them were fixed.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v7-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (25.5K, ../OSBPR01MB2552521ABDBC2F995FBACDF7F5E52@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v7-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From e5e7a64c5521c4603490042f451228b32233a72a Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v7 1/4] 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 | 61 ++++++++++++++++
src/backend/commands/subscriptioncmds.c | 68 ++++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 9 +--
src/backend/replication/logical/launcher.c | 22 ++++++
src/backend/replication/logical/worker.c | 21 +-----
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 | 1 +
src/test/regress/expected/subscription.out | 5 +-
src/test/regress/sql/subscription.sql | 5 +-
src/test/subscription/t/021_twophase.pl | 71 ++++++++++++++++++-
16 files changed, 269 insertions(+), 63 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index a78c1c3a47..88e9a72147 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 = on|off)</command>
+ with <literal>refresh</literal> option as <literal>true</literal>,
+ <command>ALTER SUBSCRIPTION ... SET (failover = on|off)</command> and
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = on|off)</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 8090ac9fc1..a67a8c48aa 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,64 @@ 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 is formed by TwoPhaseTransactionGid.
+ */
+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..1a2f0c1e64 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"
@@ -1143,7 +1144,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);
@@ -1151,6 +1153,53 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /*
+ * Do not allow changing the two_phase option if the
+ * subscription is enabled. This is because the two_phase
+ * option of the slot on the publisher cannot be modified
+ * if the slot is currently acquired by the apply worker.
+ */
+ if (form->subenabled)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for enabled subscription",
+ "two_phase")));
+
+ /*
+ * Stop all the subscription workers, just in case. Workers
+ * may still survive even if the subscription is disabled.
+ */
+ logicalrep_workers_stop(subid);
+
+ /*
+ * two_phase cannot be disabled if there are any
+ * uncommitted prepared transactions present.
+ */
+ if (!opts.twophase &&
+ form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ LookupGXactBySubid(subid))
+ /* Add error message */
+ 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")));
+
+ /*
+ * The changed two_phase option of the slot can't be rolled
+ * back.
+ */
+ PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+
+ /* 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_SLOT_NAME))
{
/*
@@ -1505,7 +1554,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 +1575,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();
{
@@ -1562,7 +1612,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1672,16 +1721,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);
- foreach(lc, subworkers)
- {
- LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
-
- logicalrep_worker_stop(w->subid, w->relid);
- }
- list_free(subworkers);
+ logicalrep_workers_stop(subid);
/*
* Remove the no-longer-useful entry in the launcher's table of apply
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 3c2b1bb496..998bbd517a 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 66070e9131..548f6e0edb 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,28 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockRelease(LogicalRepWorkerLock);
}
+/*
+ * Stop all the subscription workers.
+ */
+void
+logicalrep_workers_stop(Oid subid)
+{
+ List *subworkers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ /* XXX clarify the reason why not only running workers are listed. */
+ subworkers = logicalrep_workers_find(subid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+ foreach(lc, subworkers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+ logicalrep_worker_stop(w->subid, w->relid);
+ }
+ list_free(subworkers);
+}
+
/*
* Stop the given logical replication parallel apply worker.
*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..dcf656fd45 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -402,7 +402,6 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata,
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);
@@ -3911,7 +3910,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 +4395,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.
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index aa4ea387da..d0c8d5a4df 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 c623b07cf0..2e6ca35049 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,9 +1411,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)
@@ -1427,6 +1429,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);
}
@@ -1439,9 +1450,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 1bc80960ef..014e216cf5 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..d5428263c1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -246,6 +246,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
Oid userid, Oid relid,
dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_workers_stop(Oid subid);
extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
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..72df258000 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,75 @@ $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->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)");
+$node_subscriber->safe_psql('postgres',
+ "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->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)");
+$node_subscriber->safe_psql('postgres',
+ "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 +443,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] v7-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch (8.7K, ../OSBPR01MB2552521ABDBC2F995FBACDF7F5E52@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v7-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch)
download | inline diff:
From 1f65aaeacb63502cf6f2eb0d6f3889037a1bff9f Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v7 2/4] Alter slot option two_phase only when altering true to
false
---
doc/src/sgml/ref/alter_subscription.sgml | 2 +-
src/backend/commands/subscriptioncmds.c | 23 +++++-
.../libpqwalreceiver/libpqwalreceiver.c | 21 ++++--
src/include/replication/walreceiver.h | 5 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 72 +++++++++++++++++++
6 files changed, 113 insertions(+), 11 deletions(-)
create mode 100644 src/test/subscription/t/099_twophase_added.pl
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 88e9a72147..0c2894a94e 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 = on|off)</command> and
- <command>ALTER SUBSCRIPTION ... SET (two_phase = on|off)</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 1a2f0c1e64..71b058b385 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1190,7 +1190,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* The changed two_phase option of the slot can't be rolled
* back.
*/
- PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+ if (!opts.twophase)
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = off)");
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
@@ -1560,6 +1562,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
bool must_use_password;
char *err;
WalReceiverConn *wrconn;
+ bool failover_needs_to_be_updated;
+ bool two_phase_needs_to_be_updated;
/* Load the library providing us libpq calls. */
load_file("libpqwalreceiver", false);
@@ -1573,9 +1577,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
(errcode(ERRCODE_CONNECTION_FAILURE),
errmsg("could not connect to the publisher: %s", err)));
+ /*
+ * Consider which slot option must be altered.
+ *
+ * We must alter the failover option whenever subfailover is updated.
+ * Two_phase, however, is altered only when changing true to false.
+ */
+ failover_needs_to_be_updated =
+ replaces[Anum_pg_subscription_subfailover - 1];
+ two_phase_needs_to_be_updated =
+ (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase);
+
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
+ if (two_phase_needs_to_be_updated || failover_needs_to_be_updated)
+ walrcv_alter_slot(wrconn, sub->slotname,
+ failover_needs_to_be_updated ? &opts.failover : NULL,
+ two_phase_needs_to_be_updated ? &opts.twophase : NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 998bbd517a..c383767096 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,25 @@ 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 (two_phase)
+ appendStringInfo(&cmd, "TWO_PHASE %s%s ",
+ (*two_phase) ? "true" : "false",
+ failover ? ", " : "");
+
+ 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/meson.build b/src/test/subscription/meson.build
index c591cd7d61..b4bd522c3d 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
't/031_column_list.pl',
't/032_subscribe_use_index.pl',
't/033_run_as_table_owner.pl',
+ 't/099_twophase_added.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
new file mode 100644
index 0000000000..c13a37675a
--- /dev/null
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -0,0 +1,72 @@
+# Copyright (c) 2021-2024, PostgreSQL Global Development Group
+
+# Additional tests for altering two_phase option
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10));
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10));
+$node_subscriber->start;
+
+# Define pre-existing tables on both nodes
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication, with two_phase = off
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION pub FOR ALL TABLES");
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION sub
+ CONNECTION '$publisher_connstr' PUBLICATION pub
+ WITH (two_phase = off, copy_data = off)");
+
+######
+# Check the case that prepared transactions exist on publisher node
+######
+
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(1, 5));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('sub');
+
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "transaction is not prepared on subscriber");
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION sub DISABLE;
+ ALTER SUBSCRIPTION sub SET (two_phase = on);
+ ALTER SUBSCRIPTION sub ENABLE;");
+
+$node_publisher->safe_psql( 'postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM tab_full;");
+is($result, q(5),
+ "prepared transactions done before altering can be replicated");
+
+done_testing();
--
2.43.0
[application/octet-stream] v7-0003-Abort-prepared-transactions-while-altering-two_ph.patch (7.4K, ../OSBPR01MB2552521ABDBC2F995FBACDF7F5E52@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v7-0003-Abort-prepared-transactions-while-altering-two_ph.patch)
download | inline diff:
From c23330158b9e7407922d766d4c7ea6113eda14b2 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v7 3/4] Abort prepared transactions while altering two_phase
to false
---
doc/src/sgml/ref/alter_subscription.sgml | 9 ++++-
src/backend/access/transam/twophase.c | 17 ++++----
src/backend/commands/subscriptioncmds.c | 39 ++++++++++++-------
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 35 +++++++++++++++++
5 files changed, 75 insertions(+), 28 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0c2894a94e..848e4af649 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,13 @@ 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 prepared
+ transactions done by the logical replication worker and aborts them.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index a67a8c48aa..e0cc0e9b21 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2719,13 +2719,13 @@ IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
}
/*
- * LookupGXactBySubid
- * Check if the prepared transaction done by apply worker exists.
+ * GetGidListBySubid
+ * Get a list of GIDs which is PREPARE'd by the given subscription.
*/
-bool
-LookupGXactBySubid(Oid subid)
+List *
+GetGidListBySubid(Oid subid)
{
- bool found = false;
+ List *list = NIL;
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
@@ -2735,11 +2735,8 @@ LookupGXactBySubid(Oid subid)
/* Ignore not-yet-valid GIDs. */
if (gxact->valid &&
IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
- {
- found = true;
- break;
- }
+ list = lappend(list, pstrdup(gxact->gid));
}
LWLockRelease(TwoPhaseStateLock);
- return found;
+ return list;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 71b058b385..27be6299e0 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1155,6 +1155,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
{
+ List *prepared_xacts = NIL;
+
/*
* Do not allow changing the two_phase option if the
* subscription is enabled. This is because the two_phase
@@ -1174,26 +1176,33 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
logicalrep_workers_stop(subid);
/*
- * two_phase cannot be disabled if there are any
- * uncommitted prepared transactions present.
- */
- if (!opts.twophase &&
- form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
- LookupGXactBySubid(subid))
- /* Add error message */
- 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")));
-
- /*
- * The changed two_phase option of the slot can't be rolled
- * back.
+ * If two_phase was enabled, there is a possibility the
+ * transactions has already been PREPARE'd. They must be
+ * checked and rolled back.
*/
if (!opts.twophase)
+ {
+ /*
+ * The changed two_phase option (true->false) of the
+ * slot can't be rolled back.
+ */
PreventInTransactionBlock(isTopLevel,
"ALTER SUBSCRIPTION ... SET (two_phase = off)");
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ ListCell *cell;
+
+ /* Abort all listed transactions */
+ foreach(cell, prepared_xacts)
+ FinishPreparedTransaction((char *) lfirst(cell),
+ false);
+
+ list_free(prepared_xacts);
+ }
+ }
+
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index d37e06fdee..f7a5cf0c12 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"
+#include "nodes/pg_list.h"
/*
* GlobalTransactionData is defined in twophase.c; other places have no
@@ -65,6 +66,6 @@ extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
int szgid);
-extern bool LookupGXactBySubid(Oid subid);
+extern List *GetGidListBySubid(Oid subid);
#endif /* TWOPHASE_H */
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index c13a37675a..a8135b671c 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -69,4 +69,39 @@ $result = $node_subscriber->safe_psql('postgres',
is($result, q(5),
"prepared transactions done before altering can be replicated");
+######
+# Check the case that prepared transactions exist on subscriber node
+######
+
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(6, 10));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "transaction has been prepared on subscriber");
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION sub DISABLE;
+ ALTER SUBSCRIPTION sub SET (two_phase = off);
+ ALTER SUBSCRIPTION sub ENABLE;");
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "prepared transaction done by worker is aborted");
+
+$node_publisher->safe_psql( 'postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(10) FROM tab_full;");
+is($result, q(10),
+ "prepared transactions on publisher can be replicated");
+
done_testing();
--
2.43.0
[application/octet-stream] v7-0004-Add-force_alter-option.patch (8.2K, ../OSBPR01MB2552521ABDBC2F995FBACDF7F5E52@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v7-0004-Add-force_alter-option.patch)
download | inline diff:
From 71feafb120da3a499f15427d5e9c4f41800cf6c6 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v7 4/4] Add force_alter option
---
doc/src/sgml/ref/alter_subscription.sgml | 9 +++--
src/backend/commands/subscriptioncmds.c | 33 ++++++++++++++++++-
src/test/regress/expected/subscription.out | 3 ++
src/test/regress/sql/subscription.sql | 3 ++
src/test/subscription/t/099_twophase_added.pl | 23 ++++++++++---
5 files changed, 62 insertions(+), 9 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 848e4af649..bbfaa72229 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -257,9 +257,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<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 prepared
- transactions done by the logical replication worker and aborts them.
+ subscription is disabled. Altering the parameter from <literal>true</literal>
+ to <literal>false</literal> will be failed when there are prepared
+ transactions done by the logical replication worker. If you want to alter
+ the parameter forcibly in this case, <literal>force_alter</literal>
+ option must be set to <literal>true</literal> at the same time. If
+ specified, the backend process aborts prepared transactions.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 27be6299e0..512357f9a4 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -73,6 +73,7 @@
#define SUBOPT_FAILOVER 0x00002000
#define SUBOPT_LSN 0x00004000
#define SUBOPT_ORIGIN 0x00008000
+#define SUBOPT_FORCE_ALTER 0x00010000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -100,6 +101,7 @@ typedef struct SubOpts
bool failover;
char *origin;
XLogRecPtr lsn;
+ bool twophase_force;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -162,6 +164,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->failover = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_FORCE_ALTER))
+ opts->twophase_force = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -367,6 +371,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_FORCE_ALTER) &&
+ strcmp(defel->defname, "force_alter") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_FORCE_ALTER))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_FORCE_ALTER;
+ opts->twophase_force = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -1148,7 +1161,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1194,6 +1207,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
ListCell *cell;
+ /*
+ * Abort prepared transactions if force option is also
+ * specified. Otherwise raise an ERROR.
+ */
+ if (!opts.twophase_force)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot alter %s when there are prepared transactions",
+ "two_phase = false")));
+
/* Abort all listed transactions */
foreach(cell, prepared_xacts)
FinishPreparedTransaction((char *) lfirst(cell),
@@ -1323,6 +1346,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_suborigin - 1] = true;
}
+ /* force_alter cannot be used standalone */
+ if (IsSet(opts.specified_opts, SUBOPT_FORCE_ALTER) &&
+ !IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("%s must be specified with %s",
+ "force_alter", "two_phase")));
+
update_tuple = true;
break;
}
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 51fa4b9690..f607045b28 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -370,6 +370,9 @@ ERROR: two_phase requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+ERROR: force_alter must be specified with two_phase
\dRs+
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index a3886d79ca..80ab4dd9bc 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -255,6 +255,9 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+
\dRs+
-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index a8135b671c..7c73a58f2a 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -85,16 +85,29 @@ $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(1), "transaction has been prepared on subscriber");
-$node_subscriber->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION sub DISABLE;
- ALTER SUBSCRIPTION sub SET (two_phase = off);
- ALTER SUBSCRIPTION sub ENABLE;");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub DISABLE;");
+
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql(
+ 'postgres', "ALTER SUBSCRIPTION sub SET (two_phase = off);");
+ok($stderr =~ /cannot alter two_phase = false when there are prepared transactions/,
+ 'ALTER SUBSCRIPTION failed');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exits");
+
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter = on);");
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(0), "prepared transaction done by worker is aborted");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE;");
+
$node_publisher->safe_psql( 'postgres',
"COMMIT PREPARED 'test_prepared_tab_full';");
$node_publisher->wait_for_catchup('sub');
--
2.43.0
view thread (25+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
In-Reply-To: <OSBPR01MB2552521ABDBC2F995FBACDF7F5E52@OSBPR01MB2552.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox