public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: 'Amit Kapila' <[email protected]>
To: Ajin Cherian <[email protected]>
Cc: Давыдов Виталий <[email protected]>
Cc: Heikki Linnakangas <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
Date: Mon, 15 Apr 2024 07:57:49 +0000
Message-ID: <OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1K1fSkeK=kc26G5cq87vQG4=1qs_b+no4+ep654SeBy1w@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>
Dear Amit,
> Vitaly, does the minimal solution provided by the proposed patch
> (Allow to alter two_phase option of a subscriber provided no
> uncommitted
> prepared transactions are pending on that subscription.) address your use case?
I think we do not have to handle cases which there are prepared transactions on
publisher/subscriber, as the first step. It leads additional complexity and we
do not have smarter solutions, especially for problem 2.
IIUC it meets the Vitaly's condition, right?
> > 1. While toggling two_phase from true to false, we could probably get a list of
> prepared transactions for this subscriber id and rollback/abort the prepared
> transactions. This will allow the transactions to be re-applied like a normal
> transaction when the commit comes. Alternatively, if this isn't appropriate doing it
> in the ALTER SUBSCRIPTION context, we could store the xids of all prepared
> transactions of this subscription in a list and when the corresponding xid is being
> committed by the apply worker, prior to commit, we make sure the previously
> prepared transaction is rolled back. But this would add the overhead of checking
> this list every time a transaction is committed by the apply worker.
> >
>
> In the second solution, if you check at the time of commit whether
> there exists a prior prepared transaction then won't we end up
> applying the changes twice? I think we can first try to achieve it at
> the time of Alter Subscription because the other solution can add
> overhead at each commit?
Yeah, at least the second solution might be problematic. I prototyped
the first one and worked well. However, to make the feature more consistent,
it is prohibit to exist prepared transactions on subscriber for now.
We can ease based on the requirement.
> > 2. No solution yet.
> >
>
> One naive idea is that on the publisher we can remember whether the
> prepare has been sent and if so then only send commit_prepared,
> otherwise send the entire transaction. On the subscriber-side, we
> somehow, need to ensure before applying the first change whether the
> corresponding transaction is already prepared and if so then skip the
> changes and just perform the commit prepared. One drawback of this
> approach is that after restart, the prepare flag wouldn't be saved in
> the memory and we end up sending the entire transaction again. One way
> to avoid this overhead is that the publisher before sending the entire
> transaction checks with subscriber whether it has a prepared
> transaction corresponding to the current commit. I understand that
> this is not a good idea even if it works but I don't have any better
> ideas. What do you think?
I considered but not sure it is good to add such mechanism. Your idea requires
additional wait-loop, which might lead bugs and unexpected behavior. And it may
degrade the performance based on the network environment.
As for the another solution (worker sends a list of prepared transactions), it
is also not so good because list of prepared transactions may be huge.
Based on above, I think we can reject the case for now.
FYI - We also considered the idea which walsender waits until all prepared transactions
are resolved before decoding and sending changes, but it did not work well
- the restarted walsender sent only COMMIT PREPARED record for transactions which
have been prepared before disabling the subscription. This happened because
1) if the two_phase option of slots is false, the confirmed_flush can be ahead of
PREPARE record, and
2) after the altering and restarting, start_decoding_at becomes same as
confirmed_flush and records behind this won't be decoded.
> > 3. We could mandate that the altering of two_phase state only be done after
> disabling the subscription, just like how it is handled for failover option.
> >
>
> makes sense.
OK, this spec was added.
According to above, I updated the patch with Ajin.
0001 - extends ALTER SUBSCRIPTION statement. A tab-completion was added.
0002 - mandates the subscription has been disabled. Since no need to change
AtEOXact_ApplyLauncher(), the change is reverted.
If no objections, this can be included to 0001.
0003 - checks whether there are transactions prepared by the worker. If found,
rejects the ALTER SUBSCRIPTION command.
0004 - checks whether there are transactions prepared on publisher. The backend
connects to the publisher and confirms it. If found, rejects the ALTER
SUBSCRIPTION command.
0005 - adds TAP test for it.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v3-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (17.7K, ../OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v3-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From d6d4b8ac93c60dcaadb53c6cb9a446ae0882511b Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v3 1/5] 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
---
src/backend/access/transam/twophase.c | 43 +++++++++++++++++++
src/backend/commands/subscriptioncmds.c | 42 +++++++++++++++---
.../libpqwalreceiver/libpqwalreceiver.c | 7 +--
src/backend/replication/logical/launcher.c | 21 +++++++--
src/backend/replication/logical/worker.c | 3 --
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/logicallauncher.h | 2 +-
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 5 ++-
src/test/regress/expected/subscription.out | 5 +--
src/test/regress/sql/subscription.sql | 5 +--
14 files changed, 146 insertions(+), 34 deletions(-)
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 5a47fa984d..6643fc08a6 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"
@@ -849,7 +850,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
else if (opts.slot_name &&
(opts.failover || walrcv_server_version(wrconn) >= 170000))
{
- walrcv_alter_slot(wrconn, opts.slot_name, opts.failover);
+ walrcv_alter_slot(wrconn, opts.slot_name, opts.twophase, opts.failover);
}
}
PG_FINALLY();
@@ -868,7 +869,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
pgstat_create_subscription(subid);
if (opts.enabled)
- ApplyLauncherWakeupAtCommit();
+ ApplyLauncherWakeupAtEOXact(true);
ObjectAddressSet(myself, SubscriptionRelationId, subid);
@@ -1165,7 +1166,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);
@@ -1173,6 +1175,31 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ /* XXX */
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /* Stop corresponding worker */
+ logicalrep_worker_stop(subid, InvalidOid);
+
+ /* Request to start worker at the end of transaction */
+ ApplyLauncherWakeupAtEOXact(false);
+
+ /* 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")));
+
+ /* 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))
{
/*
@@ -1299,7 +1326,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_subenabled - 1] = true;
if (opts.enabled)
- ApplyLauncherWakeupAtCommit();
+ ApplyLauncherWakeupAtEOXact(true);
update_tuple = true;
break;
@@ -1521,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;
@@ -1541,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();
{
@@ -1962,7 +1990,7 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
form->oid, 0);
/* Wake up related background processes to handle this change quickly. */
- ApplyLauncherWakeupAtCommit();
+ ApplyLauncherWakeupAtEOXact(true);
LogicalRepWorkersWakeupAtCommit(form->oid);
}
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..3e0e5a77e0 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -89,6 +89,7 @@ static dsa_area *last_start_times_dsa = NULL;
static dshash_table *last_start_times = NULL;
static bool on_commit_launcher_wakeup = false;
+static bool launcher_wakeup = false;
static void ApplyLauncherWakeup(void);
@@ -1085,13 +1086,22 @@ ApplyLauncherForgetWorkerStartTime(Oid subid)
void
AtEOXact_ApplyLauncher(bool isCommit)
{
+ bool kicked = false;
+
if (isCommit)
{
if (on_commit_launcher_wakeup)
+ {
ApplyLauncherWakeup();
+ kicked = true;
+ }
}
+ if (!kicked && launcher_wakeup)
+ ApplyLauncherWakeup();
+
on_commit_launcher_wakeup = false;
+ launcher_wakeup = false;
}
/*
@@ -1102,10 +1112,15 @@ AtEOXact_ApplyLauncher(bool isCommit)
* tuple was added to the pg_subscription catalog.
*/
void
-ApplyLauncherWakeupAtCommit(void)
+ApplyLauncherWakeupAtEOXact(bool on_commit)
{
- if (!on_commit_launcher_wakeup)
- on_commit_launcher_wakeup = true;
+ if (on_commit)
+ {
+ if (!on_commit_launcher_wakeup)
+ on_commit_launcher_wakeup = true;
+ }
+ else if (!launcher_wakeup)
+ launcher_wakeup = true;
}
static void
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..ca3d260fc3 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3911,9 +3911,6 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
- /* two-phase should not be altered */
- Assert(newsub->twophasestate == MySubscription->twophasestate);
-
/*
* Exit if any parameter that affects the remote connection was changed.
* The launcher will start a new worker but note that the parallel apply
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 bc40c454de..be155067ce 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/logicallauncher.h b/src/include/replication/logicallauncher.h
index ff0438b5bb..075842c67e 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -24,7 +24,7 @@ extern void ApplyLauncherShmemInit(void);
extern void ApplyLauncherForgetWorkerStartTime(Oid subid);
-extern void ApplyLauncherWakeupAtCommit(void);
+extern void ApplyLauncherWakeupAtEOXact(bool on_commit);
extern void AtEOXact_ApplyLauncher(bool isCommit);
extern bool IsLogicalLauncher(void);
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/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 1eee6b17b8..9bba656e00 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -379,10 +379,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 1b2a23ba7b..9ff151f806 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -257,10 +257,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+
--
2.43.0
[application/octet-stream] v3-0002-Mandate-the-subscription-has-been-disabled.patch (6.2K, ../OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v3-0002-Mandate-the-subscription-has-been-disabled.patch)
download | inline diff:
From e5da6b142c6bcfb2943cdc24c6d63bf110dcf75e Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 02:32:38 +0000
Subject: [PATCH v3 2/5] Mandate the subscription has been disabled
---
doc/src/sgml/ref/alter_subscription.sgml | 6 ++++--
src/backend/commands/subscriptioncmds.c | 20 ++++++++++++--------
src/backend/replication/logical/launcher.c | 21 +++------------------
src/backend/replication/logical/worker.c | 3 +++
src/include/replication/logicallauncher.h | 2 +-
5 files changed, 23 insertions(+), 29 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 413ce68ce2..20b45e36e0 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -227,9 +227,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/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 6643fc08a6..bfbb2873b1 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -869,7 +869,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
pgstat_create_subscription(subid);
if (opts.enabled)
- ApplyLauncherWakeupAtEOXact(true);
+ ApplyLauncherWakeupAtCommit();
ObjectAddressSet(myself, SubscriptionRelationId, subid);
@@ -1178,11 +1178,15 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
/* XXX */
if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
{
- /* Stop corresponding worker */
- logicalrep_worker_stop(subid, InvalidOid);
-
- /* Request to start worker at the end of transaction */
- ApplyLauncherWakeupAtEOXact(false);
+ /*
+ * 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")));
/* Check whether the number of prepared transactions */
if (!opts.twophase &&
@@ -1326,7 +1330,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_subenabled - 1] = true;
if (opts.enabled)
- ApplyLauncherWakeupAtEOXact(true);
+ ApplyLauncherWakeupAtCommit();
update_tuple = true;
break;
@@ -1990,7 +1994,7 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
form->oid, 0);
/* Wake up related background processes to handle this change quickly. */
- ApplyLauncherWakeupAtEOXact(true);
+ ApplyLauncherWakeupAtCommit();
LogicalRepWorkersWakeupAtCommit(form->oid);
}
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 3e0e5a77e0..66070e9131 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -89,7 +89,6 @@ static dsa_area *last_start_times_dsa = NULL;
static dshash_table *last_start_times = NULL;
static bool on_commit_launcher_wakeup = false;
-static bool launcher_wakeup = false;
static void ApplyLauncherWakeup(void);
@@ -1086,22 +1085,13 @@ ApplyLauncherForgetWorkerStartTime(Oid subid)
void
AtEOXact_ApplyLauncher(bool isCommit)
{
- bool kicked = false;
-
if (isCommit)
{
if (on_commit_launcher_wakeup)
- {
ApplyLauncherWakeup();
- kicked = true;
- }
}
- if (!kicked && launcher_wakeup)
- ApplyLauncherWakeup();
-
on_commit_launcher_wakeup = false;
- launcher_wakeup = false;
}
/*
@@ -1112,15 +1102,10 @@ AtEOXact_ApplyLauncher(bool isCommit)
* tuple was added to the pg_subscription catalog.
*/
void
-ApplyLauncherWakeupAtEOXact(bool on_commit)
+ApplyLauncherWakeupAtCommit(void)
{
- if (on_commit)
- {
- if (!on_commit_launcher_wakeup)
- on_commit_launcher_wakeup = true;
- }
- else if (!launcher_wakeup)
- launcher_wakeup = true;
+ if (!on_commit_launcher_wakeup)
+ on_commit_launcher_wakeup = true;
}
static void
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ca3d260fc3..374aa22091 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3911,6 +3911,9 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
+ /* two-phase should not be altered while the worker exists */
+ Assert(newsub->twophasestate == MySubscription->twophasestate);
+
/*
* Exit if any parameter that affects the remote connection was changed.
* The launcher will start a new worker but note that the parallel apply
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index 075842c67e..ff0438b5bb 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -24,7 +24,7 @@ extern void ApplyLauncherShmemInit(void);
extern void ApplyLauncherForgetWorkerStartTime(Oid subid);
-extern void ApplyLauncherWakeupAtEOXact(bool on_commit);
+extern void ApplyLauncherWakeupAtCommit(void);
extern void AtEOXact_ApplyLauncher(bool isCommit);
extern bool IsLogicalLauncher(void);
--
2.43.0
[application/octet-stream] v3-0003-Prohibit-altering-from-true-to-false-if-there-are.patch (3.6K, ../OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v3-0003-Prohibit-altering-from-true-to-false-if-there-are.patch)
download | inline diff:
From 929515539b0919d560088f97d89b02376ca5492b Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v3 3/5] Prohibit altering from true to false if there are
prepared transactions on subscriber
---
doc/src/sgml/ref/alter_subscription.sgml | 8 +++++++-
src/backend/access/transam/twophase.c | 4 +++-
src/backend/commands/subscriptioncmds.c | 13 ++++++++++---
3 files changed, 20 insertions(+), 5 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 20b45e36e0..4f33769858 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -231,7 +231,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>
@@ -253,6 +252,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. Prepared transactions done by the logical replication
+ worker must not be existed. If found, the <command>ALTER
+ SUBSCRIPTION</command> command will fail.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 495f99a357..34bf6bfb0b 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2703,7 +2703,7 @@ checkGid(char *gid, Oid subid)
/*
* LookupGXactBySubid
- * Check if the prepared transaction done by apply worker exists.
+ * Check if the prepared transaction done by the given subscription.
*/
bool
LookupGXactBySubid(Oid subid)
@@ -2721,7 +2721,9 @@ LookupGXactBySubid(Oid subid)
found = true;
break;
}
+
}
LWLockRelease(TwoPhaseStateLock);
+
return found;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index bfbb2873b1..563c757be5 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1188,13 +1188,20 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errmsg("cannot set %s for enabled subscription",
"two_phase")));
- /* Check whether the number of prepared transactions */
+ /*
+ * If the two_phase is altered from true to false,
+ * prepared transactions shipped from the publisher won't
+ * be resolved anymore. Therefore, reject the ALTER
+ * command if they exists.
+ */
if (!opts.twophase &&
form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
- LookupGXactBySubid(subid))
+ LookupGXactBySubid(form->oid))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("cannot disable two_phase when uncommitted prepared transactions present")));
+ errmsg("cannot alter %s to false if there are prepared transactions by the subscription",
+ "two_phase")));
+
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
--
2.43.0
[application/octet-stream] v3-0004-Prohibit-altering-from-false-to-true-if-there-are.patch (4.9K, ../OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v3-0004-Prohibit-altering-from-false-to-true-if-there-are.patch)
download | inline diff:
From 3a7125c89de3d6fd6c1081e1100476e99d74c9c1 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 15 Apr 2024 04:58:29 +0000
Subject: [PATCH v3 4/5] Prohibit altering from false to true if there are
prepared transactions on publisher
---
doc/src/sgml/ref/alter_subscription.sgml | 9 +--
src/backend/commands/subscriptioncmds.c | 73 ++++++++++++++++++++++++
2 files changed, 78 insertions(+), 4 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 4f33769858..12d6ca2f5e 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -254,10 +254,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
</para>
<para>
- <literal>two_phase</literal> can be altered only for disabled
- subscriptions. Prepared transactions done by the logical replication
- worker must not be existed. If found, the <command>ALTER
- SUBSCRIPTION</command> command will fail.
+ On the publisher side, any prepared transactions must not exist. On the
+ subscriber side, prepared transactions done by the logical replication
+ worker must not exist. Prepared transactions done by users are allowed.
+ The <command>ALTER SUBSCRIPTION</command> command will fail if they are
+ found.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 563c757be5..57d2e615c6 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -110,6 +110,7 @@ static void check_publications_origin(WalReceiverConn *wrconn,
static void check_duplicates_in_publist(List *publist, Datum *datums);
static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
+static bool IsPreparedTransactionExistsOnPublisher(Subscription *sub);
/*
@@ -1202,6 +1203,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errmsg("cannot alter %s to false if there are prepared transactions by the subscription",
"two_phase")));
+ /*
+ * Suppose the two_phase is altering from false to true,
+ * and there have been prepared transactions on the
+ * publisher. In that case, only the COMMIT PREPARED
+ * record may be decoded and sent to the subscriber. It
+ * occurs because confirmed_flush_lsn can be ahead of the
+ * PREPARE record, so decoding all the transactions might
+ * be skipped after enabling the subscription.
+ *
+ * We prohibit the existing prepared transactions on the
+ * publisher to avoid the issue.
+ */
+ else if (opts.twophase &&
+ IsPreparedTransactionExistsOnPublisher(sub))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot alter %s to true if there are prepared transactions on publisher",
+ "two_phase")));
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
@@ -2489,3 +2508,57 @@ defGetStreamingMode(DefElem *def)
def->defname)));
return LOGICALREP_STREAM_OFF; /* keep compiler quiet */
}
+
+/*
+ * Check whether there are prepared transactions on the publisher node. Returns
+ * true if exists, otherwise false.
+ */
+static bool
+IsPreparedTransactionExistsOnPublisher(Subscription *sub)
+{
+ bool must_use_password;
+ bool found = false;
+ WalReceiverConn *wrconn;
+ WalRcvExecResult *res;
+ char *err;
+ StringInfo cmd;
+ Oid tableRow[1] = {INT4OID};
+
+ /* Load the library providing us libpq calls. */
+ load_file("libpqwalreceiver", false);
+ /* Try to connect to the publisher. */
+ must_use_password = (!superuser_arg(GetUserId()) &&
+ sub->passwordrequired);
+ wrconn = walrcv_connect(sub->conninfo, true, true,
+ must_use_password,
+ sub->name, &err);
+ if (!wrconn)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("could not connect to the publisher: %s", err)));
+
+ /* Construct a query and execute it */
+ cmd = makeStringInfo();
+ appendStringInfo(cmd,
+ "SELECT 1 FROM pg_prepared_xacts WHERE database = '%s'",
+ get_database_name(sub->dbid));
+
+ res = walrcv_exec(wrconn, cmd->data, 1, tableRow);
+ destroyStringInfo(cmd);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ errmsg("could not fetch number of prepared transactions from the primary server: %s",
+ res->err));
+
+ /*
+ * We are only interested in the existence of prepared transactions.
+ * Hence, it is sufficient to check the number of returned rows.
+ */
+ if (tuplestore_tuple_count(res->tuplestore) > 1)
+ found = true;
+
+ walrcv_clear_result(res);
+
+ return found;
+}
--
2.43.0
[application/octet-stream] v3-0005-Add-TAP-tests-for-altering-two_phase-option.patch (3.7K, ../OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com/6-v3-0005-Add-TAP-tests-for-altering-two_phase-option.patch)
download | inline diff:
From 0fedd27b5a9d01cc6ea65bedb8bbb3bf765ca1ec Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 15 Apr 2024 05:40:07 +0000
Subject: [PATCH v3 5/5] Add TAP tests for altering two_phase option
---
src/test/subscription/t/021_twophase.pl | 66 ++++++++++++++++++++++++-
1 file changed, 64 insertions(+), 2 deletions(-)
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 9437cd4c3b..1937905493 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,70 @@ $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', "
+ 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 +438,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
view thread (11+ 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: <OSBPR01MB2552707A847936E6803CFAA5F5092@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