public inbox for [email protected]
help / color / mirror / Atom feedFrom: Zhijie Hou (Fujitsu) <[email protected]>
To: Hayato Kuroda (Fujitsu) <[email protected]>
To: [email protected] <[email protected]>
Cc: Vitaly Davydov <[email protected]>
Cc: Ajin Cherian <[email protected]>
Cc: 'Amit Kapila' <[email protected]>
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
Date: Sat, 13 Jul 2024 10:48:55 +0000
Message-ID: <OS3PR01MB57184E0995521300AC06CB4B94A72@OS3PR01MB5718.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OSBPR01MB255215A39F66932323295854F5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.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>
<OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<CAA4eK1KY=uwXXuVMtuNTYHGFbbXgDveoFoP3UbxNXqxCAx8GBQ@mail.gmail.com>
<OSBPR01MB25528F4B0B8178D3AA8DE2BFF5082@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<CAA4eK1KNrm8xE-ZOAS249m+DOk=YQ_YaoM-GsZEvC8rTMtdW8w@mail.gmail.com>
<CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com>
<CAFPTHDZNwMWfJZprw9jsoa4-19-h337KRY9Pt41PdNiQTTYaiQ@mail.gmail.com>
<OSBPR01MB2552083E59F366B465DF0EF8F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<CAA4eK1Jpj2Ebk2C_9meEDfZKen38cxpzOkxZ7818f1cfsT8=3w@mail.gmail.com>
<OSBPR01MB2552A7C652D452185785A3C2F5DE2@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<CAA4eK1LhYsNq_0XKspfWKx4oXZpXk_q-A1Oa-tbA-wqQUudhCQ@mail.gmail.com>
<OSBPR01MB25527C8A6257443FAE7391C3F5DF2@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<OSBPR01MB2552E1B03A76FE7D2D7CC748F5DF2@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<e49a0-6687fe00-33-2d9a93c0@151011886>
<OSBPR01MB2552AF02B039F00EF7791401F5DA2@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<CAA4eK1+FRrL_fLWLsWQGHZRESg39ixzDX_S9hU8D7aFtU+a8uQ@mail.gmail.com>
<CAA4eK1Khy_YWFoQ1HOF_tGtiixD8YoTg86coX1-ckxt8vK3U=Q@mail.gmail.com>
<OSBPR01MB2552961D068087129E97048CF5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<OSBPR01MB255215A39F66932323295854F5DB2@OSBPR01MB2552.jpnprd01.prod.outlook.com>
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
view thread (9+ 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: <OS3PR01MB57184E0995521300AC06CB4B94A72@OS3PR01MB5718.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