public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: [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: Tue, 23 Apr 2024 12:15:32 +0000
Message-ID: <OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OSBPR01MB25521E993657E754B0C51A98F5122@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>
<23d1ca-66169f80-1f-5a90aa00@207285398>
<26b30e-661d4880-9-62d2df00@231402233>
<OSBPR01MB2552F73968D5A460157B57B9F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
<OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
Dear hackers,
Per recent commit (b29cbd3da), our patch needed to be rebased.
Here is an updated version.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v6-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (23.0K, ../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v6-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From d4bb208e621c0f47500fd1b4542a5c10ebd2ec59 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v6 1/4] Allow altering of two_phase option of a SUBSCRIPTION
This patch allows user to alter 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 | 11 +--
src/backend/access/transam/twophase.c | 43 ++++++++++++
src/backend/commands/subscriptioncmds.c | 62 +++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 7 +-
src/backend/replication/logical/launcher.c | 21 ++++++
src/backend/replication/logical/worker.c | 2 +-
src/backend/replication/slot.c | 19 +++++-
src/backend/replication/walsender.c | 20 ++++--
src/bin/psql/tab-complete.c | 2 +-
src/include/access/twophase.h | 3 +
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 5 +-
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 | 67 ++++++++++++++++++-
16 files changed, 235 insertions(+), 41 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index a78c1c3a47..e69132c39d 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,11 @@ 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>.
+ <literal>two_phase</literal> can be altered only for disabled subscription.
</para>
<para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 8090ac9fc1..495f99a357 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,46 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
LWLockRelease(TwoPhaseStateLock);
return found;
}
+
+/*
+ * 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;
+}
+
+/*
+ * 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 && checkGid(gxact->gid, subid))
+ {
+ found = true;
+ break;
+ }
+ }
+ LWLockRelease(TwoPhaseStateLock);
+ return found;
+}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e407428dbc..aa8a8e1f84 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,47 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ /* XXX */
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /*
+ * 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")));
+
+ /*
+ * Stop all the subscription workers, just in case. Workers
+ * may still survive even if the subscription is disabled.
+ */
+ logicalrep_workers_stop(subid);
+
+ /* 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")));
+
+ /*
+ * 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;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
{
/*
@@ -1505,7 +1548,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_subtwophasestate - 1] ||
+ replaces[Anum_pg_subscription_subfailover - 1])
{
bool must_use_password;
char *err;
@@ -1525,7 +1569,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+ walrcv_alter_slot(wrconn, sub->slotname, opts.twophase, opts.failover);
}
PG_FINALLY();
{
@@ -1562,7 +1606,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1672,16 +1715,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..baef3bdec0 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, bool failover);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,14 +1121,15 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool two_phase, bool failover)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s, FAILOVER %s )",
quote_identifier(slotname),
+ two_phase ? "true" : "false",
failover ? "true" : "false");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 66070e9131..94b73f3085 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,27 @@ 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);
+ 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..374aa22091 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3911,7 +3911,7 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
- /* two-phase should not be altered */
+ /* two-phase should not be altered while the worker exists */
Assert(newsub->twophasestate == MySubscription->twophasestate);
/*
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index cebf44bb0f..621f35ab1e 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -800,8 +800,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 two_phase, bool failover)
{
+ bool update_slot = false;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -844,12 +846,27 @@ ReplicationSlotAlter(const char *name, bool failover)
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot enable failover for a temporary replication slot"));
+ 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;
+ }
+
+ if (update_slot)
+ {
ReplicationSlotMarkDirty();
ReplicationSlotSave();
}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 9bf7c67f37..c45881554b 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,14 +1411,25 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
+ bool *two_phase, bool *failover)
{
+ bool two_phase_given = false;
bool failover_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
{
- if (strcmp(defel->defname, "failover") == 0)
+ 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 if (strcmp(defel->defname, "failover") == 0)
{
if (failover_given)
ereport(ERROR,
@@ -1438,10 +1449,11 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
+ bool two_phase = false;
bool failover = false;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &two_phase, &failover);
+ ReplicationSlotAlter(cmd->slotname, two_phase, failover);
}
/*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 6fee3160f0..5ff84301cd 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..d493ed24c5 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -62,4 +62,7 @@ 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 bool LookupGXactBySubid(Oid subid);
+
#endif /* TWOPHASE_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 7b937d1a0c..2fcb11418f 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 two_phase,
+ bool failover);
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..a443f402f5 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -377,6 +377,7 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
+ bool two_phase,
bool failover);
/*
@@ -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, two_phase, failover) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, two_phase, failover)
#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..e710f3c4c0 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,71 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
is($result, qq(2), 'replicated data in subscriber table');
+# Disable the subscription and alter it to two_phase = false,
+# verify that the altered subscription reflects the two_phase option.
+
+# Alter subscription two_phase to false
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->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 is disabled');
+
+# 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';
+ });
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# 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');
+
+# 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);
+
+# Made sure that the commited 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 is disabled');
+
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;");
$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
@@ -374,8 +439,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] v6-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch (9.0K, ../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v6-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch)
download | inline diff:
From 458bdbfeabae2e22dae84386a8b8e54126d101a3 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v6 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 | 8 +--
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 72 +++++++++++++++++++
6 files changed, 114 insertions(+), 13 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 e69132c39d..e54aa1b128 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 aa8a8e1f84..b02e21f535 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1184,7 +1184,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* The changed failover 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] =
@@ -1554,6 +1556,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
bool must_use_password;
char *err;
WalReceiverConn *wrconn;
+ bool two_phase_needs_to_be_updated;
+ bool failover_needs_to_be_updated;
/* Load the library providing us libpq calls. */
load_file("libpqwalreceiver", false);
@@ -1567,9 +1571,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.
+ */
+ two_phase_needs_to_be_updated =
+ (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase);
+ failover_needs_to_be_updated =
+ replaces[Anum_pg_subscription_subfailover - 1];
+
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.twophase, opts.failover);
+ if (two_phase_needs_to_be_updated || failover_needs_to_be_updated)
+ walrcv_alter_slot(wrconn, sub->slotname,
+ two_phase_needs_to_be_updated ? &opts.twophase : NULL,
+ failover_needs_to_be_updated ? &opts.failover : NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index baef3bdec0..546b599848 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 two_phase, bool failover);
+ const bool *two_phase, const bool *failover);
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 two_phase, bool failover)
+ const bool *two_phase, const bool *failover)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s, FAILOVER %s )",
- quote_identifier(slotname),
- two_phase ? "true" : "false",
- failover ? "true" : "false");
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
+ quote_identifier(slotname));
+
+ if (two_phase)
+ appendStringInfo(&cmd, "TWO_PHASE %s%s ",
+ (*two_phase) ? "true" : "false",
+ failover ? ", " : "");
+
+ if (failover)
+ appendStringInfo(&cmd, "FAILOVER %s ",
+ (*failover) ? "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 a443f402f5..f30637aa4a 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -372,13 +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 two_phase and the failover property of the slot.
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool two_phase,
- bool failover);
+ const bool *two_phase,
+ const bool *failover);
/*
* 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] v6-0003-Abort-prepared-transactions-while-altering-two_ph.patch (7.1K, ../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v6-0003-Abort-prepared-transactions-while-altering-two_ph.patch)
download | inline diff:
From d0c8138ccdf19dd9d4395855e5482cce496bda22 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v6 3/4] Abort prepared transactions while altering two_phase
to false
---
doc/src/sgml/ref/alter_subscription.sgml | 8 ++++-
src/backend/access/transam/twophase.c | 19 +++++-----
src/backend/commands/subscriptioncmds.c | 33 +++++++++++------
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 35 +++++++++++++++++++
5 files changed, 76 insertions(+), 22 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index e54aa1b128..926f560566 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -233,7 +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>.
- <literal>two_phase</literal> can be altered only for disabled subscription.
</para>
<para>
@@ -255,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>
+ <literal>two_phase</literal> can be altered only for disabled
+ subscriptions. 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 495f99a357..9121195725 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2702,13 +2702,13 @@ checkGid(char *gid, Oid subid)
}
/*
- * 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++)
@@ -2717,11 +2717,10 @@ LookupGXactBySubid(Oid subid)
/* Ignore not-yet-valid GIDs. */
if (gxact->valid && checkGid(gxact->gid, subid))
- {
- 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 b02e21f535..8a36558b2c 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1156,6 +1156,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
/* XXX */
if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
{
+ List *prepared_xacts = NIL;
+
/*
* two_phase can be only changed for disabled
* subscriptions
@@ -1172,22 +1174,33 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
*/
logicalrep_workers_stop(subid);
- /* 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")));
-
/*
- * The changed failover 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.
*/
if (!opts.twophase)
+ {
+ /*
+ * The changed failover option 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 d493ed24c5..95770bbd69 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
@@ -63,6 +64,6 @@ extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
TimestampTz origin_prepare_timestamp);
-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] v6-0004-Add-force_alter-option.patch (8.2K, ../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v6-0004-Add-force_alter-option.patch)
download | inline diff:
From 79b7d98f71c6bf6676ce868d16f8965709a4251a Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v6 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 926f560566..e6228490a8 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>
<literal>two_phase</literal> can be altered only for disabled
- subscriptions. 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.
+ subscriptions. 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 8a36558b2c..7ea4e85595 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);
@@ -1192,6 +1205,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),
@@ -1321,6 +1344,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 (9+ messages)
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]
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
In-Reply-To: <OSBPR01MB2552936A8C9F9790A4CFD389F5112@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