public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 7/7] Move code to apply one WAL record to a subroutine.
11+ messages / 4 participants
[nested] [flat]
* [PATCH 7/7] Move code to apply one WAL record to a subroutine.
@ 2021-06-21 21:00 Heikki Linnakangas <[email protected]>
0 siblings, 0 replies; 11+ messages in thread
From: Heikki Linnakangas @ 2021-06-21 21:00 UTC (permalink / raw)
---
src/backend/access/transam/xlogrecovery.c | 283 +++++++++++-----------
1 file changed, 148 insertions(+), 135 deletions(-)
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 44eb425eaf9..d7787c9a082 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -366,6 +366,7 @@ static char recoveryStopName[MAXFNAMELEN];
static bool recoveryStopAfter;
/* prototypes for local functions */
+static void ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record);
static void xlog_block_info(StringInfo buf, XLogReaderState *record);
static void readRecoverySignalFile(void);
@@ -1392,11 +1393,8 @@ PerformWalRecovery(void)
if (record != NULL)
{
- ErrorContextCallback errcallback;
TimestampTz xtime;
PGRUsage ru0;
- XLogRecPtr ReadRecPtr;
- XLogRecPtr EndRecPtr;
pg_rusage_init(&ru0);
@@ -1418,11 +1416,6 @@ PerformWalRecovery(void)
*/
do
{
- bool switchedTLI = false;
-
- ReadRecPtr = xlogreader->ReadRecPtr;
- EndRecPtr = xlogreader->EndRecPtr;
-
#ifdef WAL_DEBUG
if (XLOG_DEBUG ||
(rmid == RM_XACT_ID && trace_recovery_messages <= DEBUG2) ||
@@ -1432,8 +1425,8 @@ PerformWalRecovery(void)
initStringInfo(&buf);
appendStringInfo(&buf, "REDO @ %X/%X; LSN %X/%X: ",
- LSN_FORMAT_ARGS(ReadRecPtr),
- LSN_FORMAT_ARGS(EndRecPtr));
+ LSN_FORMAT_ARGS(xlogreader->ReadRecPtr),
+ LSN_FORMAT_ARGS(xlogreader->EndRecPtr));
xlog_outrec(&buf, xlogreader);
appendStringInfoString(&buf, " - ");
xlog_outdesc(&buf, xlogreader);
@@ -1488,132 +1481,10 @@ PerformWalRecovery(void)
recoveryPausesHere(false);
}
- /* Setup error traceback support for ereport() */
- errcallback.callback = rm_redo_error_callback;
- errcallback.arg = (void *) xlogreader;
- errcallback.previous = error_context_stack;
- error_context_stack = &errcallback;
-
/*
- * ShmemVariableCache->nextXid must be beyond record's xid.
+ * Apply the record
*/
- AdvanceNextFullTransactionIdPastXid(record->xl_xid);
-
- /*
- * Before replaying this record, check if this record causes the
- * current timeline to change. The record is already considered to
- * be part of the new timeline, so we update ThisTimeLineID before
- * replaying it. That's important so that replayEndTLI, which is
- * recorded as the minimum recovery point's TLI if recovery stops
- * after this record, is set correctly.
- */
- if (record->xl_rmid == RM_XLOG_ID)
- {
- TimeLineID newTLI = ThisTimeLineID;
- TimeLineID prevTLI = ThisTimeLineID;
- uint8 info = record->xl_info & ~XLR_INFO_MASK;
-
- if (info == XLOG_CHECKPOINT_SHUTDOWN)
- {
- CheckPoint checkPoint;
-
- memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));
- newTLI = checkPoint.ThisTimeLineID;
- prevTLI = checkPoint.PrevTimeLineID;
- }
- else if (info == XLOG_END_OF_RECOVERY)
- {
- xl_end_of_recovery xlrec;
-
- memcpy(&xlrec, XLogRecGetData(xlogreader), sizeof(xl_end_of_recovery));
- newTLI = xlrec.ThisTimeLineID;
- prevTLI = xlrec.PrevTimeLineID;
- }
-
- if (newTLI != ThisTimeLineID)
- {
- /* Check that it's OK to switch to this TLI */
- checkTimeLineSwitch(EndRecPtr, newTLI, prevTLI);
-
- /* Following WAL records should be run with new TLI */
- ThisTimeLineID = newTLI;
- switchedTLI = true;
- }
- }
-
- /*
- * Update shared replayEndRecPtr before replaying this record, so
- * that XLogFlush will update minRecoveryPoint correctly.
- */
- SpinLockAcquire(&XLogRecCtl->info_lck);
- XLogRecCtl->replayEndRecPtr = EndRecPtr;
- XLogRecCtl->replayEndTLI = ThisTimeLineID;
- SpinLockRelease(&XLogRecCtl->info_lck);
-
- /*
- * If we are attempting to enter Hot Standby mode, process XIDs we
- * see
- */
- if (standbyState >= STANDBY_INITIALIZED &&
- TransactionIdIsValid(record->xl_xid))
- RecordKnownAssignedTransactionIds(record->xl_xid);
-
- /* Now apply the WAL record itself */
- RmgrTable[record->xl_rmid].rm_redo(xlogreader);
-
- /*
- * After redo, check whether the backup pages associated with the
- * WAL record are consistent with the existing pages. This check
- * is done only if consistency check is enabled for this record.
- */
- if ((record->xl_info & XLR_CHECK_CONSISTENCY) != 0)
- checkXLogConsistency(xlogreader);
-
- /* Pop the error context stack */
- error_context_stack = errcallback.previous;
-
- /*
- * Update lastReplayedEndRecPtr after this record has been
- * successfully replayed.
- */
- SpinLockAcquire(&XLogRecCtl->info_lck);
- XLogRecCtl->lastReplayedEndRecPtr = EndRecPtr;
- XLogRecCtl->lastReplayedTLI = ThisTimeLineID;
- SpinLockRelease(&XLogRecCtl->info_lck);
-
- /* Also remember its starting position. */
- LastReplayedReadRecPtr = ReadRecPtr;
-
- /*
- * If rm_redo called XLogRequestWalReceiverReply, then we wake up
- * the receiver so that it notices the updated
- * lastReplayedEndRecPtr and sends a reply to the primary.
- */
- if (doRequestWalReceiverReply)
- {
- doRequestWalReceiverReply = false;
- WalRcvForceReply();
- }
-
- /* Allow read-only connections if we're consistent now */
- CheckRecoveryConsistency();
-
- /* Is this a timeline switch? */
- if (switchedTLI)
- {
- /*
- * Before we continue on the new timeline, clean up any
- * (possibly bogus) future WAL segments on the old timeline.
- */
- RemoveNonParentXlogFiles(EndRecPtr, ThisTimeLineID);
-
- /*
- * Wake up any walsenders to notice that we are on a new
- * timeline.
- */
- if (AllowCascadeReplication())
- WalSndWakeup();
- }
+ ApplyWalRecord(xlogreader, record);
/* Exit loop if we reached inclusive recovery target */
if (recoveryStopsAfter(xlogreader))
@@ -1672,7 +1543,7 @@ PerformWalRecovery(void)
ereport(LOG,
(errmsg("redo done at %X/%X system usage: %s",
- LSN_FORMAT_ARGS(ReadRecPtr),
+ LSN_FORMAT_ARGS(xlogreader->ReadRecPtr),
pg_rusage_show(&ru0))));
xtime = GetLatestXTime();
if (xtime)
@@ -1701,6 +1572,148 @@ PerformWalRecovery(void)
(errmsg("recovery ended before configured recovery target was reached")));
}
+/*
+ * Subroutine of PerformWalRecovery, to apply one WAL record.
+ */
+static void
+ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record)
+{
+ XLogRecPtr ReadRecPtr;
+ XLogRecPtr EndRecPtr;
+ ErrorContextCallback errcallback;
+ bool switchedTLI = false;
+
+ ReadRecPtr = xlogreader->ReadRecPtr;
+ EndRecPtr = xlogreader->EndRecPtr;
+
+ /* Setup error traceback support for ereport() */
+ errcallback.callback = rm_redo_error_callback;
+ errcallback.arg = (void *) xlogreader;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ /*
+ * ShmemVariableCache->nextXid must be beyond record's xid.
+ */
+ AdvanceNextFullTransactionIdPastXid(record->xl_xid);
+
+ /*
+ * Before replaying this record, check if this record causes the
+ * current timeline to change. The record is already considered to
+ * be part of the new timeline, so we update ThisTimeLineID before
+ * replaying it. That's important so that replayEndTLI, which is
+ * recorded as the minimum recovery point's TLI if recovery stops
+ * after this record, is set correctly.
+ */
+ if (record->xl_rmid == RM_XLOG_ID)
+ {
+ TimeLineID newTLI = ThisTimeLineID;
+ TimeLineID prevTLI = ThisTimeLineID;
+ uint8 info = record->xl_info & ~XLR_INFO_MASK;
+
+ if (info == XLOG_CHECKPOINT_SHUTDOWN)
+ {
+ CheckPoint checkPoint;
+
+ memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));
+ newTLI = checkPoint.ThisTimeLineID;
+ prevTLI = checkPoint.PrevTimeLineID;
+ }
+ else if (info == XLOG_END_OF_RECOVERY)
+ {
+ xl_end_of_recovery xlrec;
+
+ memcpy(&xlrec, XLogRecGetData(xlogreader), sizeof(xl_end_of_recovery));
+ newTLI = xlrec.ThisTimeLineID;
+ prevTLI = xlrec.PrevTimeLineID;
+ }
+
+ if (newTLI != ThisTimeLineID)
+ {
+ /* Check that it's OK to switch to this TLI */
+ checkTimeLineSwitch(EndRecPtr, newTLI, prevTLI);
+
+ /* Following WAL records should be run with new TLI */
+ ThisTimeLineID = newTLI;
+ switchedTLI = true;
+ }
+ }
+
+ /*
+ * Update shared replayEndRecPtr before replaying this record, so
+ * that XLogFlush will update minRecoveryPoint correctly.
+ */
+ SpinLockAcquire(&XLogRecCtl->info_lck);
+ XLogRecCtl->replayEndRecPtr = EndRecPtr;
+ XLogRecCtl->replayEndTLI = ThisTimeLineID;
+ SpinLockRelease(&XLogRecCtl->info_lck);
+
+ /*
+ * If we are attempting to enter Hot Standby mode, process XIDs we
+ * see
+ */
+ if (standbyState >= STANDBY_INITIALIZED &&
+ TransactionIdIsValid(record->xl_xid))
+ RecordKnownAssignedTransactionIds(record->xl_xid);
+
+ /* Now apply the WAL record itself */
+ RmgrTable[record->xl_rmid].rm_redo(xlogreader);
+
+ /*
+ * After redo, check whether the backup pages associated with the
+ * WAL record are consistent with the existing pages. This check
+ * is done only if consistency check is enabled for this record.
+ */
+ if ((record->xl_info & XLR_CHECK_CONSISTENCY) != 0)
+ checkXLogConsistency(xlogreader);
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+
+ /*
+ * Update lastReplayedEndRecPtr after this record has been
+ * successfully replayed.
+ */
+ SpinLockAcquire(&XLogRecCtl->info_lck);
+ XLogRecCtl->lastReplayedEndRecPtr = EndRecPtr;
+ XLogRecCtl->lastReplayedTLI = ThisTimeLineID;
+ SpinLockRelease(&XLogRecCtl->info_lck);
+
+ /* Also remember its starting position. */
+ LastReplayedReadRecPtr = ReadRecPtr;
+
+ /*
+ * If rm_redo called XLogRequestWalReceiverReply, then we wake up
+ * the receiver so that it notices the updated
+ * lastReplayedEndRecPtr and sends a reply to the primary.
+ */
+ if (doRequestWalReceiverReply)
+ {
+ doRequestWalReceiverReply = false;
+ WalRcvForceReply();
+ }
+
+ /* Allow read-only connections if we're consistent now */
+ CheckRecoveryConsistency();
+
+ /* Is this a timeline switch? */
+ if (switchedTLI)
+ {
+ /*
+ * Before we continue on the new timeline, clean up any
+ * (possibly bogus) future WAL segments on the old timeline.
+ */
+ RemoveNonParentXlogFiles(EndRecPtr, ThisTimeLineID);
+
+ /*
+ * Wake up any walsenders to notice that we are on a new
+ * timeline.
+ */
+ if (AllowCascadeReplication())
+ WalSndWakeup();
+ }
+}
+
/*
* Error context callback for errors occurring during rm_redo().
*/
--
2.30.2
--------------4DE063CB5604E65545208F19--
^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
@ 2024-04-15 07:57 Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-15 07:57 UTC (permalink / raw)
To: 'Amit Kapila' <[email protected]>; Ajin Cherian <[email protected]>; +Cc: Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
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
^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
@ 2024-04-15 09:46 ` Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Amit Kapila @ 2024-04-15 09:46 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
On Mon, Apr 15, 2024 at 1:28 PM Hayato Kuroda (Fujitsu)
<[email protected]> wrote:
>
> > 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.
>
I don't understand the exact problem you are facing. IIUC, if the
commit is after start_decoding_at point and prepare was before it, we
expect to send the entire transaction followed by a commit record. The
restart_lsn should be before the start of such a transaction and we
should have recorded the changes in the reorder buffer.
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
@ 2024-04-16 02:18 ` Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-16 02:18 UTC (permalink / raw)
To: 'Amit Kapila' <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
Dear Amit,
> > 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.
> >
>
> I don't understand the exact problem you are facing. IIUC, if the
> commit is after start_decoding_at point and prepare was before it, we
> expect to send the entire transaction followed by a commit record. The
> restart_lsn should be before the start of such a transaction and we
> should have recorded the changes in the reorder buffer.
This behavior is right for two_phase = false case. But if the parameter is
altered between PREPARE and COMMIT PREPARED, there is a possibility that only
COMMIT PREPARED is sent. As the first place, the executed workload is below.
1. created a subscription with (two_phase = false)
2. prepared a transaction on publisher
3. disabled the subscription once
4. altered the subscription to two_phase = true
5. enabled the subscription again
6. did COMMIT PREPARED on the publisher
-> Apply worker would raise an ERROR while applying COMMIT PREPARED record:
ERROR: prepared transaction with identifier "pg_gid_XXX_YYY" does not exist
Below part describes why the ERROR occurred.
======
### Regarding 1) the confirmed_flush can be ahead of PREPARE record,
If two_phase is off, as you might know, confirmed_flush can be ahead of PREPARE
record by keepalive mechanism.
Walsender sometimes sends a keepalive message in WalSndKeepalive(). Here the LSN
is written, which is lastly decoded record. Since the PREPARE record is skipped
(just handled by ReorderBufferProcessXid()), sometimes the written LSN in the
message can be ahead of PREPARE record. If the WAL records are aligned like below,
the LSN can point CHECKPOINT_ONLINE.
...
INSERT
PREPARE txn1
CHECKPOINT_ONLINE
...
On worker side, when it receives the keepalive, it compares the LSN in the
message and lastly received LSN, and advance last_received. Then, the worker replies
to the walsender, and at that time it replies that last_recevied record has been
flushed on the subscriber. See send_feedback().
On publisher, when the walsender receives the message from subscriber, it reads
the message and advance the confirmed_flush to the written value. If the walsender
sends LSN which locates ahead PREPARE, the confirmed flush is updated as well.
### Regarding 2) after the altering, records behind the confirmed_flush are not decoded
Then, at decoding phase. The snapshot builder determines the point where decoding
is resumed, as start_decoding_at. After the restart, the value is same as
confirmed_flush of the slot. Since the confiremed_fluish is ahead of PREPARE,
the start_decoding_at becomes ahead as well, so whole of prepared transactions
are not decoded.
======
Attached zip file contains the PoC and used script. You can refer what I really did.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/x-zip-compressed] alter_subscription_patches.zip (13.0K, ../../OSBPR01MB25528F4B0B8178D3AA8DE2BFF5082@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-alter_subscription_patches.zip)
download
^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
@ 2024-04-16 06:25 ` Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Amit Kapila @ 2024-04-16 06:25 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
On Tue, Apr 16, 2024 at 7:48 AM Hayato Kuroda (Fujitsu)
<[email protected]> wrote:
>
> > > 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.
> > >
> >
> > I don't understand the exact problem you are facing. IIUC, if the
> > commit is after start_decoding_at point and prepare was before it, we
> > expect to send the entire transaction followed by a commit record. The
> > restart_lsn should be before the start of such a transaction and we
> > should have recorded the changes in the reorder buffer.
>
> This behavior is right for two_phase = false case. But if the parameter is
> altered between PREPARE and COMMIT PREPARED, there is a possibility that only
> COMMIT PREPARED is sent.
>
Can you please once consider the idea shared by me at [1] (One naive
idea is that on the publisher .....) to solve this problem?
[1] - https://www.postgresql.org/message-id/CAA4eK1K1fSkeK%3Dkc26G5cq87vQG4%3D1qs_b%2Bno4%2Bep654SeBy1w%40...
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
@ 2024-04-18 06:26 ` Ajin Cherian <[email protected]>
2024-04-22 07:34 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Ajin Cherian @ 2024-04-18 06:26 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Hayato Kuroda (Fujitsu) <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
On Tue, Apr 16, 2024 at 4:25 PM Amit Kapila <[email protected]> wrote:
>
> >
>
> Can you please once consider the idea shared by me at [1] (One naive
> idea is that on the publisher .....) to solve this problem?
>
> [1] -
> https://www.postgresql.org/message-id/CAA4eK1K1fSkeK%3Dkc26G5cq87vQG4%3D1qs_b%2Bno4%2Bep654SeBy1w%40...
>
>
>
Expanding on Amit's idea, we found out that there is already a mechanism in
code to fully decode prepared transactions prior to a defined LSN where
two_phase is enabled using the "two_phase_at" LSN in the slot. Look at
ReorderBufferFinishPrepared() on how this is done. This code was not
working as expected in our patch because
we were setting two_phase on the slot to true as soon as the alter command
was received. This was not the correct way, initially when two_phase is
enabled, the two_phase changes to pending state and two_phase option on the
slot should only be set to true when two_phase moves from pending to
enabled. This will happen once the replication is restarted with two_phase
option. Look at code in CreateDecodingContext() on how "two_phase_at" is
set in the slot when done this way. So we changed the code to not remotely
alter two_phase when toggling from false to true. With this change, now
even if there are pending transactions on the publisher when toggling
two_phase from false to true, these pending transactions will be fully
decoded and sent once the commit prepared is decoded as the pending
prepared transactions are prior to the "two_phase_at" LSN. With this patch,
now we are able to handle both pending prepared transactions when altering
two_phase from true to false as well as false to true.
Attaching the patch for your review and comments. Big thanks to Kuroda-san
for also working on the patch.
regards,
Ajin Cherian
Fujitsu Australia.
Attachments:
[application/octet-stream] v4-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch (8.0K, ../../CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com/3-v4-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch)
download | inline diff:
From 5e4e79a5056bae374911d96eef6c4d945e23903e Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v4 2/3] Alter slot option two_phase only when altering true to
false
---
src/backend/commands/subscriptioncmds.c | 21 +++++-
.../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 +++++++++++++++++++
5 files changed, 111 insertions(+), 12 deletions(-)
create mode 100644 src/test/subscription/t/099_twophase_added.pl
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 3299a60fff..0d80d6e110 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -850,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.twophase, opts.failover);
+ walrcv_alter_slot(wrconn, opts.slot_name, NULL, &opts.failover);
}
}
PG_FINALLY();
@@ -1564,6 +1564,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);
@@ -1577,9 +1579,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] v4-0003-Abort-prepared-transactions-while-altering-two_ph.patch (7.0K, ../../CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com/4-v4-0003-Abort-prepared-transactions-while-altering-two_ph.patch)
download | inline diff:
From 35d6a2a4f3ffe42aeebb56f219cd7004777baf10 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v4 3/3] Abort prepared transactions while altering two_phase
to false
---
doc/src/sgml/ref/alter_subscription.sgml | 10 +++++-
src/backend/access/transam/twophase.c | 19 +++++-----
src/backend/commands/subscriptioncmds.c | 27 +++++++++++---
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 35 +++++++++++++++++++
5 files changed, 77 insertions(+), 17 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 20b45e36e0..cfd2a18e2c 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,15 @@ 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. If
+ prepared transactions are found, the parameter cannot be altered to
+ <literal>false</literal> inside a transaction block.
+ </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 0d80d6e110..cf3e5c64b0 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1178,6 +1178,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
@@ -1194,13 +1196,28 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
*/
logicalrep_workers_stop(subid);
- /* Check whether the number of prepared transactions */
+ /*
+ * If two phase was enabled, there is a possibility the
+ * transactions has already been PREPARE'd.
+ */
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")));
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ ListCell *cell;
+
+ /* Must not be in the transaction */
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = ...)");
+
+ /* Abort all listed transactions */
+ foreach(cell, prepared_xacts)
+ {
+ FinishPreparedTransaction((char *) lfirst(cell),
+ false);
+ prepared_xacts = list_delete_cell(prepared_xacts, cell);
+ }
+ }
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
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] v4-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (22.5K, ../../CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com/5-v4-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From 8295540696e9699ac8191f1c2b0e09b6b684cb50 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v4 1/3] 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 | 6 +-
src/backend/access/transam/twophase.c | 43 ++++++++++++
src/backend/commands/subscriptioncmds.c | 58 +++++++++++-----
.../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, 227 insertions(+), 40 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/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..3299a60fff 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();
@@ -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,41 @@ 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")));
+
+ /* 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))
{
/*
@@ -1521,7 +1558,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 +1579,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();
{
@@ -1578,7 +1616,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1688,16 +1725,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 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/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 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+
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
^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
@ 2024-04-22 07:34 ` Ajin Cherian <[email protected]>
2024-04-22 08:56 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Ajin Cherian @ 2024-04-22 07:34 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Hayato Kuroda (Fujitsu) <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
On Thu, Apr 18, 2024 at 4:26 PM Ajin Cherian <[email protected]> wrote:
>
> Attaching the patch for your review and comments. Big thanks to Kuroda-san
> for also working on the patch.
>
>
Looking at this a bit more, maybe rolling back all prepared transactions on
the subscriber when toggling two_phase from true to false might not be
desirable for the customer. Maybe we should have an option for customers to
control whether transactions should be rolled back or not. Maybe
transactions should only be rolled back if a "force" option is also set.
What do people think?
regards,
Ajin Cherian
Fujitsu Australia
^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 07:34 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
@ 2024-04-22 08:56 ` Hayato Kuroda (Fujitsu) <[email protected]>
2024-07-04 06:55 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-22 08:56 UTC (permalink / raw)
To: 'Ajin Cherian' <[email protected]>; Amit Kapila <[email protected]>; +Cc: Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
Dear hackers,
> Looking at this a bit more, maybe rolling back all prepared transactions on the
> subscriber when toggling two_phase from true to false might not be desirable
> for the customer. Maybe we should have an option for customers to control
> whether transactions should be rolled back or not. Maybe transactions should
> only be rolled back if a "force" option is also set. What do people think?
And here is a patch for adds new option "force_alter" (better name is very welcome).
It could be used only when altering two_phase option. Let me share examples.
Assuming that there are logical replication system with two_phase = on, and
there are prepared transactions:
```
subscriber=# SELECT * FROM pg_prepared_xacts ;
transaction | gid | prepared | owner | database
-------------+------------------+-------------------------------+----------+----------
741 | pg_gid_16390_741 | 2024-04-22 08:02:34.727913+00 | postgres | postgres
742 | pg_gid_16390_742 | 2024-04-22 08:02:34.729486+00 | postgres | postgres
(2 rows)
```
At that time, altering two_phase to false alone will be failed:
```
subscriber=# ALTER SUBSCRIPTION sub DISABLE ;
ALTER SUBSCRIPTION
subscriber=# ALTER SUBSCRIPTION sub SET (two_phase = off);
ERROR: cannot alter two_phase = false when there are prepared transactions
```
It succeeds if force_alter is also expressly set. Prepared transactions will be
aborted at that time.
```
subscriber=# ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter = on);
ALTER SUBSCRIPTION
subscriber=# SELECT * FROM pg_prepared_xacts ;
transaction | gid | prepared | owner | database
-------------+-----+----------+-------+----------
(0 rows)
```
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/global/
Attachments:
[application/octet-stream] v5-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (22.5K, ../../OSBPR01MB2552083E59F366B465DF0EF8F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v5-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From 43a5314acbbfdf8210f83c03576cfbcc2faddb80 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v5 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 | 6 +-
src/backend/access/transam/twophase.c | 43 ++++++++++++
src/backend/commands/subscriptioncmds.c | 58 +++++++++++-----
.../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, 227 insertions(+), 40 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/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..3299a60fff 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();
@@ -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,41 @@ 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")));
+
+ /* 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))
{
/*
@@ -1521,7 +1558,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 +1579,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();
{
@@ -1578,7 +1616,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1688,16 +1725,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 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+
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] v5-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch (8.0K, ../../OSBPR01MB2552083E59F366B465DF0EF8F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v5-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch)
download | inline diff:
From 5db0fa5a9f6d7522ec65fb0acacfa630b02bb63c Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v5 2/4] Alter slot option two_phase only when altering true to
false
---
src/backend/commands/subscriptioncmds.c | 21 +++++-
.../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 +++++++++++++++++++
5 files changed, 111 insertions(+), 12 deletions(-)
create mode 100644 src/test/subscription/t/099_twophase_added.pl
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 3299a60fff..0d80d6e110 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -850,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.twophase, opts.failover);
+ walrcv_alter_slot(wrconn, opts.slot_name, NULL, &opts.failover);
}
}
PG_FINALLY();
@@ -1564,6 +1564,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);
@@ -1577,9 +1579,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] v5-0003-Abort-prepared-transactions-while-altering-two_ph.patch (7.0K, ../../OSBPR01MB2552083E59F366B465DF0EF8F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v5-0003-Abort-prepared-transactions-while-altering-two_ph.patch)
download | inline diff:
From bf47ee629d262dae465b3113e946baf6dca9b592 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v5 3/4] Abort prepared transactions while altering two_phase
to false
---
doc/src/sgml/ref/alter_subscription.sgml | 10 +++++-
src/backend/access/transam/twophase.c | 19 +++++-----
src/backend/commands/subscriptioncmds.c | 27 +++++++++++---
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 35 +++++++++++++++++++
5 files changed, 77 insertions(+), 17 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 20b45e36e0..cfd2a18e2c 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,15 @@ 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. If
+ prepared transactions are found, the parameter cannot be altered to
+ <literal>false</literal> inside a transaction block.
+ </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 0d80d6e110..cf3e5c64b0 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1178,6 +1178,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
@@ -1194,13 +1196,28 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
*/
logicalrep_workers_stop(subid);
- /* Check whether the number of prepared transactions */
+ /*
+ * If two phase was enabled, there is a possibility the
+ * transactions has already been PREPARE'd.
+ */
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")));
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ ListCell *cell;
+
+ /* Must not be in the transaction */
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = ...)");
+
+ /* Abort all listed transactions */
+ foreach(cell, prepared_xacts)
+ {
+ FinishPreparedTransaction((char *) lfirst(cell),
+ false);
+ prepared_xacts = list_delete_cell(prepared_xacts, cell);
+ }
+ }
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
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] v5-0004-Add-force_alter-option.patch (7.3K, ../../OSBPR01MB2552083E59F366B465DF0EF8F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v5-0004-Add-force_alter-option.patch)
download | inline diff:
From f92c5166b808e70d7907b099c847dbffed8136c7 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v5 4/4] Add force_alter option
---
src/backend/commands/subscriptioncmds.c | 36 +++++++++++++++++--
src/test/regress/expected/subscription.out | 3 ++
src/test/regress/sql/subscription.sql | 3 ++
src/test/subscription/t/099_twophase_added.pl | 23 +++++++++---
4 files changed, 58 insertions(+), 7 deletions(-)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index cf3e5c64b0..33b789b730 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),
@@ -1170,7 +1183,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);
@@ -1206,6 +1219,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")));
+
/* Must not be in the transaction */
PreventInTransactionBlock(isTopLevel,
"ALTER SUBSCRIPTION ... SET (two_phase = ...)");
@@ -1215,8 +1238,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
FinishPreparedTransaction((char *) lfirst(cell),
false);
- prepared_xacts = list_delete_cell(prepared_xacts, cell);
}
+
+ list_free(prepared_xacts);
}
/* Change system catalog acoordingly */
@@ -1333,6 +1357,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 9bba656e00..16154d1bee 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -372,6 +372,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 9ff151f806..8d8f5bb670 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -256,6 +256,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
^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 07:34 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 08:56 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
@ 2024-07-04 06:55 ` Amit Kapila <[email protected]>
2024-07-04 08:04 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Amit Kapila @ 2024-07-04 06:55 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
On Mon, Apr 22, 2024 at 2:26 PM Hayato Kuroda (Fujitsu)
<[email protected]> wrote:
>
```
>
> It succeeds if force_alter is also expressly set. Prepared transactions will be
> aborted at that time.
>
> ```
> subscriber=# ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter = on);
> ALTER SUBSCRIPTION
>
Isn't it better to give a Notice when force_alter option leads to the
rollback of already prepared transactions?
I have another question on the latest 0001 patch:
+ /*
+ * Stop all the subscription workers, just in case.
+ * Workers may still survive even if the subscription is
+ * disabled.
+ */
+ logicalrep_workers_stop(subid);
In which case the workers will survive when the subscription is disabled?
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 07:34 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 08:56 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-07-04 06:55 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
@ 2024-07-04 08:04 ` Hayato Kuroda (Fujitsu) <[email protected]>
2024-07-04 11:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
0 siblings, 1 reply; 11+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-07-04 08:04 UTC (permalink / raw)
To: 'Amit Kapila' <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
Dear Amit,
> >
> > It succeeds if force_alter is also expressly set. Prepared transactions will be
> > aborted at that time.
> >
> > ```
> > subscriber=# ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter =
> on);
> > ALTER SUBSCRIPTION
> >
>
> Isn't it better to give a Notice when force_alter option leads to the
> rollback of already prepared transactions?
Indeed. I think this can be added for 0003. For now, it says like:
```
postgres=# ALTER SUBSCRIPTION sub SET (TWO_PHASE = off, FORCE_ALTER = on);
WARNING: requested altering to two_phase = false but there are prepared transactions done by the subscription
DETAIL: Such transactions are being rollbacked.
ALTER SUBSCRIPTION
```
> I have another question on the latest 0001 patch:
> + /*
> + * Stop all the subscription workers, just in case.
> + * Workers may still survive even if the subscription is
> + * disabled.
> + */
> + logicalrep_workers_stop(subid);
>
> In which case the workers will survive when the subscription is disabled?
I think both normal and tablesync worker can survive, because ALTER SUBSCRIPTION
DISABLE command does not send signal to workers. It just change the system catalog.
logicalrep_workers_stop() is added to ensure all workers are stopped.
Actually, earlier version (-v3) did not have a mechanism but they sometimes got
assertion failures in maybe_reread_subscription(). This was because the survived
workers read pg_subscription catalog and failed below assertion:
```
/* two-phase cannot be altered while the worker exists */
Assert(newsub->twophasestate == MySubscription->twophasestate);
```
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v14-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (25.5K, ../../OSBPR01MB2552A7C652D452185785A3C2F5DE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v14-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch)
download | inline diff:
From ac8631d7032ccd3a277bc4777f8ff40bd74951d4 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v14 1/4] Allow altering of two_phase option of a SUBSCRIPTION
This patch allows the user to alter the 'two_phase' option of a subscriber provided no
uncommitted prepared transactions are pending on that subscription.
Author: Cherian Ajin, Hayato Kuroda
---
doc/src/sgml/ref/alter_subscription.sgml | 12 ++--
src/backend/access/transam/twophase.c | 62 ++++++++++++++++
src/backend/commands/subscriptioncmds.c | 68 ++++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 9 +--
src/backend/replication/logical/launcher.c | 22 ++++++
src/backend/replication/logical/worker.c | 21 +-----
src/backend/replication/slot.c | 18 ++++-
src/backend/replication/walsender.c | 18 ++++-
src/bin/psql/tab-complete.c | 2 +-
src/include/access/twophase.h | 5 ++
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 11 +--
src/include/replication/worker_internal.h | 1 +
src/test/regress/expected/subscription.out | 5 +-
src/test/regress/sql/subscription.sql | 5 +-
src/test/subscription/t/021_twophase.pl | 71 ++++++++++++++++++-
16 files changed, 270 insertions(+), 63 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 476f195622..0b23df1b77 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -68,8 +68,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>,
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
- with <literal>refresh</literal> option as <literal>true</literal> and
- <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command>
+ with <literal>refresh</literal> option as <literal>true</literal>,
+ <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
@@ -228,9 +229,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
- <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+ <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
+ The <literal>two_phase</literal> parameter can only be altered when the
+ subscription is disabled.
</para>
<para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 9a8257fcaf..35bce6809d 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2681,3 +2681,65 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
LWLockRelease(TwoPhaseStateLock);
return found;
}
+
+/*
+ * TwoPhaseTransactionGid
+ * Form the prepared transaction GID for two_phase transactions.
+ *
+ * Return the GID in the supplied buffer.
+ */
+void
+TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
+{
+ Assert(subid != InvalidRepOriginId);
+
+ if (!TransactionIdIsValid(xid))
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("invalid two-phase transaction ID")));
+
+ snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
+}
+
+/*
+ * IsTwoPhaseTransactionGidForSubid
+ * Check whether the given GID (as formed by TwoPhaseTransactionGid) is
+ * for the specified 'subid'.
+ */
+static bool
+IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
+{
+ int ret;
+ Oid subid_written;
+ TransactionId xid;
+
+ ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid);
+
+ return (ret == 2 && subid == subid_written);
+}
+
+/*
+ * LookupGXactBySubid
+ * Check if the prepared transaction done by apply worker exists.
+ */
+bool
+LookupGXactBySubid(Oid subid)
+{
+ bool found = false;
+
+ LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+ for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
+ {
+ GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+
+ /* Ignore not-yet-valid GIDs. */
+ if (gxact->valid &&
+ IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
+ {
+ found = true;
+ break;
+ }
+ }
+ LWLockRelease(TwoPhaseStateLock);
+ return found;
+}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e407428dbc..e925158a4d 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -16,6 +16,7 @@
#include "access/htup_details.h"
#include "access/table.h"
+#include "access/twophase.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
@@ -1143,7 +1144,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
+ SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+ SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
SUBOPT_ORIGIN);
@@ -1151,6 +1153,53 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /*
+ * Do not allow changing the two_phase option if the
+ * subscription is enabled. This is because the two_phase
+ * option of the slot on the publisher cannot be modified
+ * if the slot is currently acquired by the apply worker.
+ */
+ if (form->subenabled)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for enabled subscription",
+ "two_phase")));
+
+ /*
+ * Stop all the subscription workers, just in case.
+ * Workers may still survive even if the subscription is
+ * disabled.
+ */
+ logicalrep_workers_stop(subid);
+
+ /*
+ * two_phase cannot be disabled if there are any
+ * uncommitted prepared transactions present.
+ */
+ if (!opts.twophase &&
+ form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ LookupGXactBySubid(subid))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
+ errhint("Resolve these transactions and try again")));
+
+ /*
+ * The changed two_phase option of the slot can't be
+ * rolled back.
+ */
+ PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+
+ /* Change system catalog acoordingly */
+ values[Anum_pg_subscription_subtwophasestate - 1] =
+ CharGetDatum(opts.twophase ?
+ LOGICALREP_TWOPHASE_STATE_PENDING :
+ LOGICALREP_TWOPHASE_STATE_DISABLED);
+ replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
{
/*
@@ -1505,7 +1554,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1])
+ if (replaces[Anum_pg_subscription_subfailover - 1] ||
+ replaces[Anum_pg_subscription_subtwophasestate - 1])
{
bool must_use_password;
char *err;
@@ -1525,7 +1575,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+ walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
}
PG_FINALLY();
{
@@ -1562,7 +1612,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1672,16 +1721,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
* New workers won't be started because we hold an exclusive lock on the
* subscription till the end of the transaction.
*/
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
- subworkers = logicalrep_workers_find(subid, false);
- LWLockRelease(LogicalRepWorkerLock);
- foreach(lc, subworkers)
- {
- LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
-
- logicalrep_worker_stop(w->subid, w->relid);
- }
- list_free(subworkers);
+ logicalrep_workers_stop(subid);
/*
* Remove the no-longer-useful entry in the launcher's table of apply
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 02f12f2921..2f035a0c3c 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover);
+ bool failover, bool two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,15 +1121,16 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool failover, bool two_phase)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )",
quote_identifier(slotname),
- failover ? "true" : "false");
+ failover ? "true" : "false",
+ two_phase ? "true" : "false");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 27c3a91fb7..bef65b839b 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,28 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockRelease(LogicalRepWorkerLock);
}
+/*
+ * Stop all the subscription workers.
+ */
+void
+logicalrep_workers_stop(Oid subid)
+{
+ List *subworkers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ /* XXX clarify the reason why not only running workers are listed. */
+ subworkers = logicalrep_workers_find(subid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+ foreach(lc, subworkers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+ logicalrep_worker_stop(w->subid, w->relid);
+ }
+ list_free(subworkers);
+}
+
/*
* Stop the given logical replication parallel apply worker.
*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 3b285894db..33d9549f0a 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -402,7 +402,6 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata,
CmdType operation);
/* Compute GID for two_phase transactions */
-static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid);
/* Functions for skipping changes */
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn);
@@ -3911,7 +3910,7 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
- /* two-phase should not be altered */
+ /* two-phase cannot be altered while the worker exists */
Assert(newsub->twophasestate == MySubscription->twophasestate);
/*
@@ -4396,24 +4395,6 @@ cleanup_subxact_info()
subxact_data.nsubxacts_max = 0;
}
-/*
- * Form the prepared transaction GID for two_phase transactions.
- *
- * Return the GID in the supplied buffer.
- */
-static void
-TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
-{
- Assert(subid != InvalidRepOriginId);
-
- if (!TransactionIdIsValid(xid))
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("invalid two-phase transaction ID")));
-
- snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
-}
-
/*
* Common function to run the apply loop with error handling. Disable the
* subscription, if necessary.
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index baf9b89dc4..2ad6dca993 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -804,8 +804,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
* Change the definition of the slot identified by the specified name.
*/
void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
{
+ bool update_slot = false;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -854,6 +856,20 @@ ReplicationSlotAlter(const char *name, bool failover)
MyReplicationSlot->data.failover = failover;
SpinLockRelease(&MyReplicationSlot->mutex);
+ update_slot = true;
+ }
+
+ if (MyReplicationSlot->data.two_phase != two_phase)
+ {
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.two_phase = two_phase;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ update_slot = true;
+ }
+
+ if (update_slot)
+ {
ReplicationSlotMarkDirty();
ReplicationSlotSave();
}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 754f505c13..af776fccb8 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1409,9 +1409,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
+ bool *failover, bool *two_phase)
{
bool failover_given = false;
+ bool two_phase_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
@@ -1425,6 +1427,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "two_phase") == 0)
+ {
+ if (two_phase_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ two_phase_given = true;
+ *two_phase = defGetBoolean(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1437,9 +1448,10 @@ static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
bool failover = false;
+ bool two_phase = false;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &failover, &two_phase);
+ ReplicationSlotAlter(cmd->slotname, failover, two_phase);
}
/*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index d453e224d9..891face1b6 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
"password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit");
+ "streaming", "synchronous_commit", "two_phase");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 56248c0006..d37e06fdee 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -62,4 +62,9 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
TimestampTz origin_prepare_timestamp);
+
+extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
+ int szgid);
+extern bool LookupGXactBySubid(Oid subid);
+
#endif /* TWOPHASE_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index c9675ee87c..163a4a911a 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool failover,
+ bool two_phase);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 12f71fa99b..31fa1257ec 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -372,12 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
/*
* walrcv_alter_slot_fn
*
- * Change the definition of a replication slot. Currently, it only supports
- * changing the failover property of the slot.
+ * Change the definition of a replication slot. Currently, it supports
+ * changing the failover and the two_phase property of the slot.
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover);
+ bool failover,
+ bool two_phase);
/*
* walrcv_get_backend_pid_fn
@@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
-#define walrcv_alter_slot(conn, slotname, failover) \
- WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
+#define walrcv_alter_slot(conn, slotname, failover, two_phase) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)
#define walrcv_get_backend_pid(conn) \
WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 515aefd519..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..0436cafdb8 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,75 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
is($result, qq(2), 'replicated data in subscriber table');
+# Clean up
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+
+###############################
+# Disable the subscription and alter it to two_phase = false,
+# then verify that the altered subscription reflects the two_phase option.
+###############################
+
+# Alter subscription two_phase to false
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is disabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(d), 'two-phase should be disabled');
+
+# Now do a prepare on the publisher and make sure that it is not replicated.
+$node_publisher->safe_psql(
+ 'postgres', qq{
+ BEGIN;
+ INSERT INTO tab_copy VALUES (100);
+ PREPARE TRANSACTION 'newgid';
+ });
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure there are no prepared transactions on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(0), 'should be no prepared transactions on subscriber');
+
+# Now commit the insert and verify that it IS replicated
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';");
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure that the committed transaction is replicated.
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
+is($result, qq(3), 'replicated data in subscriber table');
+
+# Alter subscription two_phase to true
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is enabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(e), 'two-phase should be enabled');
+
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;");
$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
@@ -374,8 +443,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
# check all the cleanup
###############################
-$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
-
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_subscription");
is($result, qq(0), 'check subscription was dropped on subscriber');
--
2.43.0
[application/octet-stream] v14-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (12.3K, ../../OSBPR01MB2552A7C652D452185785A3C2F5DE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v14-0002-Alter-slot-option-two_phase-only-when-altering-t.patch)
download | inline diff:
From d22eef501989c96d7a49b9f7fd8b60f3e749c22d Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v14 2/4] Alter slot option two_phase only when altering "true"
to "false"
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Since the two_phase option is controlled by both the publisher (as a slot option)
and the subscriber (as a subscription option), the slot option must also be
modified.
Regarding the false->true case, the backend process alters the subtwophase to
LOGICALREP_TWOPHASE_STATE_PENDING once. After the subscription is enabled, a new
logical replication worker requests to change the two_phase option of its slot
from pending to true after the initial data synchronization is done. The code
path is the same as the case in which two_phase is initially set to true, so
there is no need to do something remarkable. However, for the true->false case,
the backend must connect to the publisher and expressly change the parameter
because the apply worker does not alter the option to false. Because this
operation cannot be rolled back, altering the two_phase parameter from "true"
to "false" within a transaction is prohibited.
---
doc/src/sgml/ref/alter_subscription.sgml | 2 +-
src/backend/commands/subscriptioncmds.c | 43 ++++++--
.../libpqwalreceiver/libpqwalreceiver.c | 23 +++--
src/include/replication/walreceiver.h | 5 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 99 +++++++++++++++++++
6 files changed, 157 insertions(+), 16 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 0b23df1b77..475a42a2e3 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -70,7 +70,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
with <literal>refresh</literal> option as <literal>true</literal>,
<command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
- <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command>
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = off)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e925158a4d..996ea6b6de 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1097,6 +1097,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
Form_pg_subscription form;
bits32 supported_opts;
SubOpts opts = {0};
+ bool update_failover;
+ bool update_two_phase;
rel = table_open(SubscriptionRelationId, RowExclusiveLock);
@@ -1187,10 +1189,25 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errhint("Resolve these transactions and try again")));
/*
- * The changed two_phase option of the slot can't be
- * rolled back.
+ * Altering the parameter from "true" to "false" within a
+ * transaction is prohibited. Since the apply worker does
+ * not alter the slot option to false, the backend must
+ * connect to the publisher and expressly change the
+ * parameter.
+ *
+ * There is no need to do something remarkable regarding
+ * the "false" to "true" case; the backend process alters
+ * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once.
+ * After the subscription is enabled, a new logical
+ * replication worker requests to change the two_phase
+ * option of its slot from pending to true when the
+ * initial data synchronization is done. The code path is
+ * the same as the case in which two_phase is initially
+ * set to true.
*/
- PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+ if (!opts.twophase)
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = false)");
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
@@ -1548,14 +1565,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
}
/*
- * Try to acquire the connection necessary for altering slot.
+ * Check the need to alter the replication slot. Failover and two_phase
+ * options are controlled by both the publisher (as a slot option) and the
+ * subscriber (as a subscription option). The slot option must be altered
+ * only when changing "true" to "false". The reason has already been
+ * described in the ALTER_SUBSCRIPTION_OPTIONS section of this function.
+ */
+ update_failover = replaces[Anum_pg_subscription_subfailover - 1];
+ update_two_phase = (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase);
+
+ /*
+ * Try to acquire the connection necessary for altering slot, if needed.
*
* This has to be at the end because otherwise if there is an error while
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1] ||
- replaces[Anum_pg_subscription_subtwophasestate - 1])
+ if (update_failover || update_two_phase)
{
bool must_use_password;
char *err;
@@ -1575,7 +1602,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
+ walrcv_alter_slot(wrconn, sub->slotname,
+ update_failover ? &opts.failover : NULL,
+ update_two_phase ? &opts.twophase : NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 2f035a0c3c..07dfec947d 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover, bool two_phase);
+ const bool *failover, const bool *two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,16 +1121,27 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover, bool two_phase)
+ const bool *failover, const bool *two_phase)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )",
- quote_identifier(slotname),
- failover ? "true" : "false",
- two_phase ? "true" : "false");
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
+ quote_identifier(slotname));
+
+ if (failover)
+ appendStringInfo(&cmd, "FAILOVER %s",
+ *failover ? "true" : "false");
+
+ if (failover && two_phase)
+ appendStringInfo(&cmd, ", ");
+
+ if (two_phase)
+ appendStringInfo(&cmd, "TWO_PHASE %s",
+ *two_phase ? "true" : "false");
+
+ appendStringInfoString(&cmd, " );");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 31fa1257ec..7ffa5a58b3 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -377,8 +377,9 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover,
- bool two_phase);
+ const bool *failover,
+ const bool *two_phase);
+
/*
* walrcv_get_backend_pid_fn
diff --git a/src/test/subscription/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..1124f7fa00
--- /dev/null
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -0,0 +1,99 @@
+# Copyright (c) 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
+ log_min_messages = debug1));
+$node_subscriber->start;
+
+# Define 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 = "false"
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION pub FOR ALL TABLES");
+
+my $log_offset = -s $node_subscriber->logfile;
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION regress_sub
+ CONNECTION '$publisher_connstr' PUBLICATION pub
+ WITH (two_phase = false, copy_data = false, failover = false)");
+
+# Verify the started worker recognized two_phase was disabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is DISABLED',
+ $log_offset);
+
+#####################
+# Check the case that prepared transactions exist on the publisher node.
+#
+# Since the two_phase is "false", then normally, this PREPARE will do nothing
+# until the COMMIT PREPARED, but in this test, we toggle the two_phase to
+# "true" again before the COMMIT PREPARED happens.
+
+# Prepare a transaction to insert some tuples into the table
+$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('regress_sub');
+
+# Verify the prepared transaction is not yet replicated to the subscriber
+# because two_phase is set to "false".
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "transaction is not prepared on subscriber");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Toggle the two_phase to "true" *before* the COMMIT PREPARED. Since we are the
+# special path for the case where both two_phase and failover are altered, it
+# is also set to "true".
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub DISABLE;
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = true, failover = true);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify the started worker recognized two_phase was enabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is ENABLED',
+ $log_offset);
+
+# And do COMMIT PREPARED the prepared transaction
+$node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify inserted tuples are replicated
+$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] v14-0003-Abort-prepared-transactions-while-altering-two_p.patch (10.0K, ../../OSBPR01MB2552A7C652D452185785A3C2F5DE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v14-0003-Abort-prepared-transactions-while-altering-two_p.patch)
download | inline diff:
From be5727394811456e5b1dc3e38c6faccfc390452b Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v14 3/4] Abort prepared transactions while altering two_phase
to off
If we alter the two_phase parameter from "on" to "off" and there are prepared
transactions on the subscriber, they won't be resolved. To avoid this issue, we
allow the backend to abort all prepared transactions while altering the
subscription.
---
doc/src/sgml/ref/alter_subscription.sgml | 11 ++-
src/backend/access/transam/twophase.c | 17 ++---
src/backend/commands/subscriptioncmds.c | 75 ++++++++++++-------
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 44 +++++++++++
5 files changed, 110 insertions(+), 40 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 475a42a2e3..8801f37f0e 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -233,8 +233,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
- The <literal>two_phase</literal> parameter can only be altered when the
- subscription is disabled.
</para>
<para>
@@ -256,6 +254,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>
option is enabled.
</para>
+
+ <para>
+ The <literal>two_phase</literal> parameter can only be altered when the
+ subscription is disabled. When altering the parameter from <literal>true</literal>
+ to <literal>false</literal>, the backend process checks for any incomplete
+ prepared transactions done by the logical replication worker (from when
+ <literal>two_phase</literal> parameter was still <literal>true</literal>)
+ and, if any are found, those are aborted.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 35bce6809d..0be8a15367 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2719,13 +2719,13 @@ IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
}
/*
- * LookupGXactBySubid
- * Check if the prepared transaction done by apply worker exists.
+ * GetGidListBySubid
+ * Get a list of GIDs which is PREPARE'd by the given subscription.
*/
-bool
-LookupGXactBySubid(Oid subid)
+List *
+GetGidListBySubid(Oid subid)
{
- bool found = false;
+ List *list = NIL;
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
@@ -2735,11 +2735,8 @@ LookupGXactBySubid(Oid subid)
/* Ignore not-yet-valid GIDs. */
if (gxact->valid &&
IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
- {
- found = true;
- break;
- }
+ list = lappend(list, pstrdup(gxact->gid));
}
LWLockRelease(TwoPhaseStateLock);
- return found;
+ return list;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 996ea6b6de..54a2c76f37 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1177,38 +1177,59 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
logicalrep_workers_stop(subid);
/*
- * two_phase cannot be disabled if there are any
- * uncommitted prepared transactions present.
- */
- if (!opts.twophase &&
- form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
- LookupGXactBySubid(subid))
- 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")));
-
- /*
- * Altering the parameter from "true" to "false" within a
- * transaction is prohibited. Since the apply worker does
- * not alter the slot option to false, the backend must
- * connect to the publisher and expressly change the
- * parameter.
- *
- * There is no need to do something remarkable regarding
- * the "false" to "true" case; the backend process alters
- * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once.
- * After the subscription is enabled, a new logical
- * replication worker requests to change the two_phase
- * option of its slot from pending to true when the
- * initial data synchronization is done. The code path is
- * the same as the case in which two_phase is initially
- * set to true.
+ * If two_phase was previously enabled, there is a
+ * possibility that transactions have already been
+ * PREPARE'd. They must be checked and rolled back.
*/
if (!opts.twophase)
+ {
+ List *prepared_xacts;
+
+ /*
+ * Altering the parameter from "true" to "false"
+ * within a transaction is prohibited. Since the apply
+ * worker does not alter the slot option to false, the
+ * backend must connect to the publisher and expressly
+ * change the parameter.
+ *
+ * There is no need to do something remarkable
+ * regarding the "false" to "true" case; the backend
+ * process alters subtwophase to
+ * LOGICALREP_TWOPHASE_STATE_PENDING once. After the
+ * subscription is enabled, a new logical replication
+ * worker requests to change the two_phase option of
+ * its slot from pending to true when the initial data
+ * synchronization is done. The code path is the same
+ * as the case in which two_phase is initially
+ * set to true.
+ */
PreventInTransactionBlock(isTopLevel,
"ALTER SUBSCRIPTION ... SET (two_phase = false)");
+ /*
+ * To prevent prepared transactions from being
+ * isolated, they must manually be aborted.
+ */
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ ereport(WARNING,
+ (errmsg_plural("requested altering to %s but there is prepared transaction done by the subscription",
+ "requested altering to %s but there are prepared transactions done by the subscription",
+ list_length(prepared_xacts),
+ "two_phase = false"),
+ errdetail_plural("Such a transaction is being rollbacked.",
+ "Such transactions are being rollbacked.",
+ list_length(prepared_xacts))));
+
+ /* Abort all listed transactions */
+ foreach_ptr(char, gid, prepared_xacts)
+ FinishPreparedTransaction(gid, false);
+
+ list_free_deep(prepared_xacts);
+ }
+ }
+
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index d37e06fdee..f7a5cf0c12 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"
+#include "nodes/pg_list.h"
/*
* GlobalTransactionData is defined in twophase.c; other places have no
@@ -65,6 +66,6 @@ extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
int szgid);
-extern bool LookupGXactBySubid(Oid subid);
+extern List *GetGidListBySubid(Oid subid);
#endif /* TWOPHASE_H */
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index 1124f7fa00..635daf7a78 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -96,4 +96,48 @@ $result =
is($result, q(5),
"prepared transactions done before altering can be replicated");
+#####################
+# Check the case that prepared transactions exist on the subscriber node
+#
+# If the two_phase is altering from "true" to "false" and there are prepared
+# transactions on the subscriber, they must be aborted. This test checks it.
+
+# Prepare a transaction to insert some tuples into the table
+$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('regress_sub');
+
+# Verify the prepared transaction has been replicated to the subscriber because
+# two_phase is set to "true".
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "transaction has been prepared on subscriber");
+
+# Toggle the two_phase to "false" before the COMMIT PREPARED
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub DISABLE;
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = false);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify any prepared transactions are aborted because two_phase is changed to
+# "false".
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "prepared transaction done by worker is aborted");
+
+# Do COMMIT PREPARED the prepared transaction
+$node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify inserted tuples are replicated
+$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] v14-0004-Add-force_alter-option-for-ALTER-SUBSCRIPTION-.-.patch (57.2K, ../../OSBPR01MB2552A7C652D452185785A3C2F5DE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v14-0004-Add-force_alter-option-for-ALTER-SUBSCRIPTION-.-.patch)
download | inline diff:
From 85d2922ab3f3329a8cb4e2975efa930da9bf966d Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v14 4/4] Add force_alter option for ALTER SUBSCRIPTION ... SET
command
Previously, all prepared transactions on the standby were rolled back when
toggling two_phase from "true" to "false". However, this operation may not be
expected by users. To ensure users understand what happens, we added the
"force_alter" parameter. When two_phase is toggling to "false", and there are
prepared transactions, they will be aborted only when "force_alter" is set to
true. Otherwise, an ERROR occurs.
---
doc/src/sgml/catalogs.sgml | 12 ++
doc/src/sgml/ref/alter_subscription.sgml | 16 +-
doc/src/sgml/ref/create_subscription.sgml | 24 +++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 2 +-
src/backend/commands/subscriptioncmds.c | 36 ++++-
src/bin/pg_dump/pg_dump.c | 18 ++-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 7 +-
src/include/catalog/pg_subscription.h | 13 ++
src/test/regress/expected/subscription.out | 152 +++++++++---------
src/test/subscription/t/099_twophase_added.pl | 44 ++++-
12 files changed, 229 insertions(+), 97 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index b654fae1b2..bca2f0c77a 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8035,6 +8035,18 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subforcealter</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, then the <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION</command></link>
+ can sometimes be forced to proceed instead of giving an error. See
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
+ parameter for details about when this might be useful.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 8801f37f0e..ab2cdaeaa3 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -230,8 +230,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<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>,
- <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>.
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
Only a superuser can set <literal>password_required = false</literal>.
</para>
@@ -257,11 +258,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
The <literal>two_phase</literal> parameter can only be altered when the
- subscription is disabled. When altering the parameter from <literal>true</literal>
- to <literal>false</literal>, the backend process checks for any incomplete
- prepared transactions done by the logical replication worker (from when
- <literal>two_phase</literal> parameter was still <literal>true</literal>)
- and, if any are found, those are aborted.
+ subscription is disabled. Altering the parameter from <literal>true</literal>
+ to <literal>false</literal> will give an error when there are prepared
+ transactions done by the logical replication worker. If you want to alter
+ the parameter forcibly in this case,
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
+ option must be set to <literal>true</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 740b7d9421..83ac52f865 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -428,6 +428,30 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-force-alter">
+ <term><literal>force_alter</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies if the <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION</command></link>
+ can be forced to proceed instead of giving an error.
+ </para>
+ <para>
+ There is currently only one scenario where this parameter has any
+ effect: When altering <literal>two_phase</literal> option from
+ <literal>true</literal> to <literal>false</literal> it is possible
+ for there to be incomplete prepared transactions done by the logical
+ replication worker (from when <literal>two_phase</literal> parameter
+ was still <literal>true</literal>). If <literal>force_alter</literal>
+ is <literal>false</literal>, then this will give an error; if
+ <literal>force_alter</literal> is <literal>true</literal>, then the
+ incomplete prepared transactions are aborted and the alter will proceed.
+ </para>
+ <para>
+ The default is <literal>false</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..b568fe3470 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -72,6 +72,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->passwordrequired = subform->subpasswordrequired;
sub->runasowner = subform->subrunasowner;
sub->failover = subform->subfailover;
+ sub->forcealter = subform->subforcealter;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 19cabc9a47..b89eacc96b 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1355,7 +1355,7 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
- subpasswordrequired, subrunasowner, subfailover,
+ subpasswordrequired, subrunasowner, subfailover, subforcealter,
subslotname, subsynccommit, subpublications, suborigin)
ON pg_subscription TO public;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 54a2c76f37..01fe1ee0aa 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 force_alter;
} 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->force_alter = 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->force_alter = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -604,7 +617,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN |
+ SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -711,6 +725,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
+ values[Anum_pg_subscription_subforcealter - 1] = BoolGetDatum(opts.force_alter);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1150,7 +1165,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);
@@ -1213,6 +1228,23 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
(prepared_xacts = GetGidListBySubid(subid)) != NIL)
{
+ bool raise_error =
+ IsSet(opts.specified_opts, SUBOPT_FORCE_ALTER) ?
+ !opts.force_alter : !sub->forcealter;
+
+ /*
+ * Abort prepared transactions only if
+ * 'force_alter' option is true. Otherwise raise
+ * an ERROR.
+ */
+ if (raise_error)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot alter %s when there are prepared transactions",
+ "two_phase = false"),
+ errhint("Resolve these transactions or set %s, and then try again.",
+ "force_alter = true")));
+
ereport(WARNING,
(errmsg_plural("requested altering to %s but there is prepared transaction done by the subscription",
"requested altering to %s but there are prepared transactions done by the subscription",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 5426f1177c..85c69a835b 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4754,6 +4754,7 @@ getSubscriptions(Archive *fout)
int i_suboriginremotelsn;
int i_subenabled;
int i_subfailover;
+ int i_subforcealter;
int i,
ntups;
@@ -4826,10 +4827,17 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 170000)
appendPQExpBufferStr(query,
- " s.subfailover\n");
+ " s.subfailover,\n");
else
appendPQExpBuffer(query,
- " false AS subfailover\n");
+ " false AS subfailover,\n");
+
+ if (fout->remoteVersion >= 170000)
+ appendPQExpBufferStr(query,
+ " s.subforcealter\n");
+ else
+ appendPQExpBuffer(query,
+ " false AS subforcealter\n");
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -4869,6 +4877,7 @@ getSubscriptions(Archive *fout)
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
i_subenabled = PQfnumber(res, "subenabled");
i_subfailover = PQfnumber(res, "subfailover");
+ i_subforcealter = PQfnumber(res, "subforcealter");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4915,6 +4924,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_subenabled));
subinfo[i].subfailover =
pg_strdup(PQgetvalue(res, i, i_subfailover));
+ subinfo[i].subforcealter =
+ pg_strdup(PQgetvalue(res, i, i_subforcealter));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5155,6 +5166,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subfailover, "t") == 0)
appendPQExpBufferStr(query, ", failover = true");
+ if (strcmp(subinfo->subforcealter, "t") == 0)
+ appendPQExpBufferStr(query, ", force_alter = true");
+
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 4b2e5870a9..8f5f9d13b9 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -671,6 +671,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *suboriginremotelsn;
char *subfailover;
+ char *subforcealter;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 7c9a1f234c..abf91a81fa 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6539,7 +6539,7 @@ describeSubscriptions(const char *pattern, bool verbose)
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
false, false, false, false, false, false, false, false, false, false,
- false};
+ false, false};
if (pset.sversion < 100000)
{
@@ -6608,6 +6608,11 @@ describeSubscriptions(const char *pattern, bool verbose)
", subfailover AS \"%s\"\n",
gettext_noop("Failover"));
+ if (pset.sversion >= 170000)
+ appendPQExpBuffer(&buf,
+ ", subforcealter AS \"%s\"\n",
+ gettext_noop("Force alter"));
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a2..deac7aa943 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -98,6 +98,13 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool subforcealter; /* True allows the ALTER SUBSCRIPTION command
+ * to proceed under conditions that would
+ * otherwise result in an error. Currently,
+ * 'force_alter' only has an effect when
+ * altering the two_phase option from "true"
+ * to "false". */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -151,6 +158,12 @@ typedef struct Subscription
* (i.e. the main slot and the table sync
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool forcealter; /* True allows the ALTER SUBSCRIPTION command
+ * to proceed under conditions that would
+ * otherwise result in an error. Currently,
+ * 'force_alter' only has an effect when
+ * altering the two_phase option from "true"
+ * to "false". */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 51fa4b9690..b36fc6b8f7 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
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.
\dRs+ regress_testsub4
- 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
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- 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
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\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
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\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
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\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
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\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
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
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.
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
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.
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\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
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -332,10 +332,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -371,19 +371,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
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.
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -393,10 +393,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
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.
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -409,18 +409,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
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.
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\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
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index 635daf7a78..4da8392962 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -117,15 +117,43 @@ $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(1), "transaction has been prepared on subscriber");
-# Toggle the two_phase to "false" before the COMMIT PREPARED
-$node_subscriber->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION regress_sub DISABLE;
- ALTER SUBSCRIPTION regress_sub SET (two_phase = false);
- ALTER SUBSCRIPTION regress_sub ENABLE;");
+# Disable the subscription to alter the two_phase option
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub DISABLE;");
+
+# Try altering the two_phase option to "false". The command will fail since
+# there is a prepared transaction and the 'force_alter' option is not specified
+# as true.
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub SET (two_phase = false);");
+ok( $stderr =~
+ /cannot alter two_phase = false when there are prepared transactions/,
+ 'ALTER SUBSCRIPTION failed');
+
+# Verify the prepared transaction still exists
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exists");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Alter the two_phase true to false with the force_alter option enabled. This
+# command will succeed after aborting the prepared transaction.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub SET (two_phase = false, force_alter = true);"
+);
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify the started worker recognized two_phase was disabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is DISABLED',
+ $log_offset);
-# Verify any prepared transactions are aborted because two_phase is changed to
-# "false".
+# # Verify the prepared transaction was aborted
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(0), "prepared transaction done by worker is aborted");
--
2.43.0
^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 07:34 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 08:56 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-07-04 06:55 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-07-04 08:04 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
@ 2024-07-04 11:31 ` Amit Kapila <[email protected]>
0 siblings, 0 replies; 11+ messages in thread
From: Amit Kapila @ 2024-07-04 11:31 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected] <[email protected]>
On Thu, Jul 4, 2024 at 1:34 PM Hayato Kuroda (Fujitsu)
<[email protected]> wrote:
>
> > >
> > > It succeeds if force_alter is also expressly set. Prepared transactions will be
> > > aborted at that time.
> > >
> > > ```
> > > subscriber=# ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter =
> > on);
> > > ALTER SUBSCRIPTION
> > >
> >
> > Isn't it better to give a Notice when force_alter option leads to the
> > rollback of already prepared transactions?
>
> Indeed. I think this can be added for 0003. For now, it says like:
>
> ```
> postgres=# ALTER SUBSCRIPTION sub SET (TWO_PHASE = off, FORCE_ALTER = on);
> WARNING: requested altering to two_phase = false but there are prepared transactions done by the subscription
> DETAIL: Such transactions are being rollbacked.
> ALTER SUBSCRIPTION
>
Is it possible to get a NOTICE instead of a WARNING?
>
> > I have another question on the latest 0001 patch:
> > + /*
> > + * Stop all the subscription workers, just in case.
> > + * Workers may still survive even if the subscription is
> > + * disabled.
> > + */
> > + logicalrep_workers_stop(subid);
> >
> > In which case the workers will survive when the subscription is disabled?
>
> I think both normal and tablesync worker can survive, because ALTER SUBSCRIPTION
> DISABLE command does not send signal to workers. It just change the system catalog.
> logicalrep_workers_stop() is added to ensure all workers are stopped.
>
> Actually, earlier version (-v3) did not have a mechanism but they sometimes got
> assertion failures in maybe_reread_subscription(). This was because the survived
> workers read pg_subscription catalog and failed below assertion:
>
> ```
> /* two-phase cannot be altered while the worker exists */
> Assert(newsub->twophasestate == MySubscription->twophasestate);
> ```
>
But that is not a good reason for this operation to stop workers
first. Instead, we should prohibit this operation if any worker is
present. The reason is that there is always a chance that if any
worker is alive, it can prepare a new transaction after we have
checked for the presence of any prepared transactions.
Comments:
=========
1.
There is no need to do something remarkable regarding
+ * the "false" to "true" case; the backend process alters
+ * subtwophase <funny_char> to LOGICALREP_TWOPHASE_STATE_PENDING once.
+ * After the subscription is enabled, a new logical
+ * replication worker requests to change the two_phase
+ * option of its slot from pending to true when the
+ * initial data synchronization is done. The code path is
+ * the same as the case in which two_phase <funny_char> is initially
+ * set <funny_char> to true.
The patch has some funny characters in the above comment at the places
highlighted by me. It seems you have copied from some editor that has
inserted such characters.
2.
/*
* Do not allow toggling of two_phase option. Doing so could cause
* missing of transactions and lead to an inconsistent replica.
* See comments atop worker.c
*
* Note: Unsupported twophase indicates that this call originated
* from AlterSubscription.
*/
if (!IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("unrecognized subscription parameter: \"%s\"", defel->defname)));
This part of the code must either be removed or converted to an assert.
3. The tests added in 099_twophase_added.pl should be part of 021_twophase.pl
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 11+ messages in thread
end of thread, other threads:[~2024-07-04 11:31 UTC | newest]
Thread overview: 11+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-06-21 21:00 [PATCH 7/7] Move code to apply one WAL record to a subroutine. Heikki Linnakangas <[email protected]>
2024-04-15 07:57 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-15 09:46 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-16 02:18 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-16 06:25 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-18 06:26 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 07:34 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-22 08:56 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-07-04 06:55 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-07-04 08:04 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-07-04 11:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox