public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v17 02/10] pg_stat_file and pg_ls_dir_* to use lstat()..
9+ messages / 5 participants
[nested] [flat]
* [PATCH v17 02/10] pg_stat_file and pg_ls_dir_* to use lstat()..
@ 2020-03-30 23:59 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Justin Pryzby @ 2020-03-30 23:59 UTC (permalink / raw)
pg_ls_dir_* will now skip (no longer show) symbolic links, same as other
non-regular file types, as we advertize we do since 8b6d94cf6. That seems to
be the intented behavior, since irregular file types are 1) less portable; and,
2) we don't currently show a file's type except for "bool is_dir".
pg_stat_file will now 1) show metadata of links themselves, rather than their
target; and, 2) specifically, show links to directories with "is_dir=false";
and, 3) not error on broken symlinks.
---
doc/src/sgml/func.sgml | 2 +-
src/backend/utils/adt/genfile.c | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index e0d1eff6b5..d9b3598977 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -25861,7 +25861,7 @@ SELECT convert_from(pg_read_binary_file('file_in_utf8.txt'), 'UTF8');
Returns a record containing the file's size, last access time stamp,
last modification time stamp, last file status change time stamp (Unix
platforms only), file creation time stamp (Windows only), and a flag
- indicating if it is a directory (or a symbolic link to a directory).
+ indicating if it is a directory.
</para>
<para>
This function is restricted to superusers by default, but other users
diff --git a/src/backend/utils/adt/genfile.c b/src/backend/utils/adt/genfile.c
index ceaa6180da..219ac160f8 100644
--- a/src/backend/utils/adt/genfile.c
+++ b/src/backend/utils/adt/genfile.c
@@ -370,7 +370,7 @@ pg_stat_file(PG_FUNCTION_ARGS)
filename = convert_and_check_filename(filename_t);
- if (stat(filename, &fst) < 0)
+ if (lstat(filename, &fst) < 0)
{
if (missing_ok && errno == ENOENT)
PG_RETURN_NULL();
@@ -596,7 +596,7 @@ pg_ls_dir_files(FunctionCallInfo fcinfo, const char *dir, bool missing_ok)
/* Get the file info */
snprintf(path, sizeof(path), "%s/%s", dir, de->d_name);
- if (stat(path, &attrib) < 0)
+ if (lstat(path, &attrib) < 0)
{
/* Ignore concurrently-deleted files, else complain */
if (errno == ENOENT)
--
2.17.0
--4LFBTxd4L5NLO6ly
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v17-0003-Add-tests-on-pg_ls_dir-before-changing-it.patch"
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
@ 2024-04-05 11:29 Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Ajin Cherian @ 2024-04-05 11:29 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected]
On Thu, Apr 4, 2024 at 4:38 PM Amit Kapila <[email protected]> wrote:
>
> I think this would probably be better than the current situation but
> can we think of a solution to allow toggling the value of two_phase
> even when prepared transactions are present? Can you please summarize
> the reason for the problems in doing that and the solutions, if any?
>
> --
> With Regards,
> Amit Kapila.
>
Updated the patch, as it wasn't addressing updating of two-phase in the
remote slot.
Currently the main issue that needs to be handled is the handling of
pending prepared transactions while the two_phase is altered. I see 3
issues with the current approach.
1. Uncommitted prepared transactions when toggling two_phase from true to
false
When two_phase was true, prepared transactions were decoded at PREPARE time
and send to the subscriber, which is then prepared on the subscriber with a
new gid. Once the two_phase is toggled to false, then the COMMIT PREPARED
on the publisher is converted to commit and the entire transaction is
decoded and sent to the subscriber. This will leave the previously
prepared transaction pending.
2. Uncommitted prepared transactions when toggling two_phase form false to
true
When two_phase was false, prepared transactions were ignored and not
decoded at PREPARE time on the publisher. Once the two_phase is toggled to
true, the apply worker and the walsender are restarted and a replication is
restarted from a new "start_decoding_at" LSN. Now, this new
"start_decoding_at" could be past the LSN of the PREPARE record and if so,
the PREPARE record is skipped and not send to the subscriber. Look at
comments in DecodeTXNNeedSkip() for detail. Later when the user issues
COMMIT PREPARED, this is decoded and sent to the subscriber. but there is
no prepared transaction on the subscriber, and this fails because the
corresponding gid of the transaction couldn't be found.
3. While altering the two_phase of the subscription, it is required to also
alter the two_phase field of the slot on the primary. The subscription
cannot remotely alter the two_phase option of the slot when the
subscription is enabled, as the slot is owned by the walsender on the
publisher side.
Possible solutions for the 3 problems:
1. While toggling two_phase from true to false, we could probably get 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.
2. No solution yet.
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.
Let me know your thoughts.
regards,
Ajin Cherian
Fujitsu Australia
Attachments:
[application/octet-stream] v2-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (15.1K, ../../CAFPTHDa=pJSZ_4dV5DPAOapRSgPcyyUTP0WzGY2Rz_D3-gwraw@mail.gmail.com/3-v2-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From 10604538a61c4b015378b5e33af49405ec774366 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v2] 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.
---
src/backend/access/transam/twophase.c | 42 ++++++++++++++++++++++
src/backend/commands/subscriptioncmds.c | 42 ++++++++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 7 ++--
src/backend/replication/logical/launcher.c | 22 ++++++++++--
src/backend/replication/logical/worker.c | 3 --
src/backend/replication/slot.c | 19 +++++++++-
src/backend/replication/walsender.c | 20 ++++++++---
src/include/access/twophase.h | 3 ++
src/include/replication/logicallauncher.h | 2 +-
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 5 +--
11 files changed, 143 insertions(+), 25 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 8090ac9..b0aae25 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,45 @@ 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 5a47fa9..6643fc0 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 761bf0f..9f7a936 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 66070e9..899ec22 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,16 @@ 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 b5a80fe..ca3d260 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 3bddaae..3d24ebb 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 bc40c45..be15506 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/include/access/twophase.h b/src/include/access/twophase.h
index 56248c0..d493ed2 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 ff0438b..075842c 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 7b937d1..2fcb114 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 12f71fa..a443f40 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) \
--
1.8.3.1
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
@ 2024-04-10 11:18 ` Amit Kapila <[email protected]>
2024-04-10 14:16 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-11 02:07 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 2 replies; 9+ messages in thread
From: Amit Kapila @ 2024-04-10 11:18 UTC (permalink / raw)
To: Ajin Cherian <[email protected]>; +Cc: Давыдов Виталий <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected]
On Fri, Apr 5, 2024 at 4:59 PM Ajin Cherian <[email protected]> wrote:
>
> On Thu, Apr 4, 2024 at 4:38 PM Amit Kapila <[email protected]> wrote:
>>
>>
>> I think this would probably be better than the current situation but
>> can we think of a solution to allow toggling the value of two_phase
>> even when prepared transactions are present? Can you please summarize
>> the reason for the problems in doing that and the solutions, if any?
>>
>
>
> Updated the patch, as it wasn't addressing updating of two-phase in the remote slot.
>
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?
> Currently the main issue that needs to be handled is the handling of pending prepared transactions while the two_phase is altered. I see 3 issues with the current approach.
>
> 1. Uncommitted prepared transactions when toggling two_phase from true to false
> When two_phase was true, prepared transactions were decoded at PREPARE time and send to the subscriber, which is then prepared on the subscriber with a new gid. Once the two_phase is toggled to false, then the COMMIT PREPARED on the publisher is converted to commit and the entire transaction is decoded and sent to the subscriber. This will leave the previously prepared transaction pending.
>
> 2. Uncommitted prepared transactions when toggling two_phase form false to true
> When two_phase was false, prepared transactions were ignored and not decoded at PREPARE time on the publisher. Once the two_phase is toggled to true, the apply worker and the walsender are restarted and a replication is restarted from a new "start_decoding_at" LSN. Now, this new "start_decoding_at" could be past the LSN of the PREPARE record and if so, the PREPARE record is skipped and not send to the subscriber. Look at comments in DecodeTXNNeedSkip() for detail. Later when the user issues COMMIT PREPARED, this is decoded and sent to the subscriber. but there is no prepared transaction on the subscriber, and this fails because the corresponding gid of the transaction couldn't be found.
>
> 3. While altering the two_phase of the subscription, it is required to also alter the two_phase field of the slot on the primary. The subscription cannot remotely alter the two_phase option of the slot when the subscription is enabled, as the slot is owned by the walsender on the publisher side.
>
Thanks for summarizing the reasons for not allowing altering the
two_pc property for a subscription.
> Possible solutions for the 3 problems:
>
> 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?
> 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?
> 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.
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
@ 2024-04-10 14:16 ` Давыдов Виталий <[email protected]>
2024-04-15 15:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
1 sibling, 1 reply; 9+ messages in thread
From: Давыдов Виталий @ 2024-04-10 14:16 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Ajin Cherian <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected]
Hi Amit, Ajin, All
Thank you for the patch and the responses. I apologize for my delayed answer due to some curcumstances.
On Wednesday, April 10, 2024 14:18 MSK, Amit Kapila <[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?In general, the idea behind the patch seems to be suitable for my case. Furthermore, the case of two_phase switch from false to true with uncommitted pending prepared transactions probably never happens in my case. The switch from false to true means that the replica completes the catchup from the master and switches to the normal mode when it participates in the multi-node configuration. There should be no uncommitted pending prepared transactions at the moment of the switch to the normal mode.
I'm going to try this patch. Give me please some time to investigate the patch. I will come with some feedback a little bit later.
Thank you for your help!
With best regards,
Vitaly Davydov
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-10 14:16 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
@ 2024-04-15 15:31 ` Давыдов Виталий <[email protected]>
2024-04-22 08:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Давыдов Виталий @ 2024-04-15 15:31 UTC (permalink / raw)
To: Давыдов Виталий <[email protected]>; +Cc: Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; Heikki Linnakangas <[email protected]>; [email protected]
Dear All,
On Wednesday, April 10, 2024 17:16 MSK, Давыдов Виталий <[email protected]> wrote:
Hi Amit, Ajin, All
Thank you for the patch and the responses. I apologize for my delayed answer due to some curcumstances.
On Wednesday, April 10, 2024 14:18 MSK, Amit Kapila <[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?In general, the idea behind the patch seems to be suitable for my case. Furthermore, the case of two_phase switch from false to true with uncommitted pending prepared transactions probably never happens in my case. The switch from false to true means that the replica completes the catchup from the master and switches to the normal mode when it participates in the multi-node configuration. There should be no uncommitted pending prepared transactions at the moment of the switch to the normal mode.
I'm going to try this patch. Give me please some time to investigate the patch. I will come with some feedback a little bit later.
I looked at the patch and realized that I can't try it easily in the near future because the solution I'm working on is based on PG16 or earlier. This patch is not easily applicable to the older releases. I have to port my solution to the master, which is not done yet. I apologize for that - so much work should be done before applying the patch. BTW, I tested the idea with async 2PC commit on my side and it seems to work fine in my case. Anyway, I agree, the idea with altering of subscription seems the best one but much harder to implement.
To summarise my case of a synchronous multimaster where twophase is used to implement global transactions:
* Replica may have prepared but not committed transactions when I toggle subscription twophase from true to false. In this case, all prepared transactions may be aborted on the replica before altering the subscription. * Replica will not have prepared transactions when subscription is toggled from false to true. In this scenario, the replica completes the catchup (with twophase=off) and becomes the part of the multi-nodal cluster and is ready to accept new 2PC transactions. All the new pending transactions will wait until replica responds. But it may work differently for some other solutions. In general, it would be great to allow toggling for all scenarious.Just interested, does anyone tried to reproduce the problem with slow catchup of twophase transactions (pgbench should be used with big number of clients)? I haven't seen any messages from anyone other that me that the problem takes place.
Thank you for your help!
With best regards,
Vitaly
^ permalink raw reply [nested|flat] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-10 14:16 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-15 15:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
@ 2024-04-22 08:54 ` Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-22 12:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-22 08:54 UTC (permalink / raw)
To: 'Давыдов Виталий' <[email protected]>; +Cc: Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; [email protected] <[email protected]>
Dear Vitaly,
> I looked at the patch and realized that I can't try it easily in the near future
> because the solution I'm working on is based on PG16 or earlier. This patch is
> not easily applicable to the older releases. I have to port my solution to the
> master, which is not done yet.
We also tried to port our patch for PG16, but the largest barrier was that a
replication command ALTER_SLOT is not supported. Since the slot option two_phase
can't be modified, it is difficult to skip decoding PREPARE command even when
altering the option from true to false.
IIUC, Adding a new feature (e.g., replication command) for minor updates is generally
prohibited
We must consider another approach for backpatching, but we do not have solutions
for now.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/global/
^ permalink raw reply [nested|flat] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-10 14:16 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-15 15:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-22 08:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
@ 2024-04-22 12:54 ` Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-23 12:15 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-22 12:54 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; 'Давыдов Виталий' <[email protected]>; +Cc: Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; [email protected] <[email protected]>
> Dear Vitaly,
>
> > I looked at the patch and realized that I can't try it easily in the near future
> > because the solution I'm working on is based on PG16 or earlier. This patch is
> > not easily applicable to the older releases. I have to port my solution to the
> > master, which is not done yet.
>
> We also tried to port our patch for PG16, but the largest barrier was that a
> replication command ALTER_SLOT is not supported. Since the slot option
> two_phase
> can't be modified, it is difficult to skip decoding PREPARE command even when
> altering the option from true to false.
> IIUC, Adding a new feature (e.g., replication command) for minor updates is
> generally
> prohibited
>
> We must consider another approach for backpatching, but we do not have
> solutions
> for now.
Attached patch set is a ported version for PG16, which breaks ABI. This can be used
for testing purpose, but it won't be pushed to REL_16_STABLE.
At least, this patchset can pass my github CI.
Can you apply and check whether your issue is solved?
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] REL_16_0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPTION.patch (25.8K, ../../OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-REL_16_0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPTION.patch)
download | inline diff:
From e5c865867983282814c60723499546913fd2d8c1 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH 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
---
src/backend/access/transam/twophase.c | 43 ++++++++++
src/backend/commands/subscriptioncmds.c | 83 ++++++++++++++++---
src/backend/nodes/gen_node_support.pl | 2 +-
.../libpqwalreceiver/libpqwalreceiver.c | 30 +++++++
src/backend/replication/logical/launcher.c | 21 +++++
src/backend/replication/logical/worker.c | 2 +-
src/backend/replication/repl_gram.y | 20 ++++-
src/backend/replication/repl_scanner.l | 2 +
src/backend/replication/slot.c | 25 ++++++
src/backend/replication/walsender.c | 47 +++++++++++
src/bin/psql/tab-complete.c | 2 +-
src/include/access/twophase.h | 3 +
src/include/nodes/replnodes.h | 12 +++
src/include/replication/slot.h | 2 +-
src/include/replication/walreceiver.h | 10 +++
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 ++++++++++++++-
19 files changed, 354 insertions(+), 28 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index c6af8cfd7e..2148daba3c 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2658,3 +2658,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 6fe111e98d..59852fd9af 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"
@@ -1130,13 +1131,49 @@ 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_ORIGIN);
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))
{
/*
@@ -1453,6 +1490,38 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
heap_freetuple(tup);
}
+ /*
+ * Try to acquire the connection necessary for altering slot.
+ */
+ if (replaces[Anum_pg_subscription_subtwophasestate - 1])
+ {
+ bool must_use_password;
+ char *err;
+ WalReceiverConn *wrconn;
+
+ /* Load the library providing us libpq calls. */
+ load_file("libpqwalreceiver", false);
+
+ /* Try to connect to the publisher */
+ must_use_password = (!superuser() && opts.passwordrequired);
+ wrconn = walrcv_connect(sub->conninfo, true, must_use_password,
+ sub->name, &err);
+ if (!wrconn)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("could not connect to the publisher: %s", err)));
+
+ PG_TRY();
+ {
+ walrcv_alter_slot(wrconn, sub->slotname, opts.twophase);
+ }
+ PG_FINALLY();
+ {
+ walrcv_disconnect(wrconn);
+ }
+ PG_END_TRY();
+ }
+
table_close(rel, RowExclusiveLock);
ObjectAddressSet(myself, SubscriptionRelationId, subid);
@@ -1481,7 +1550,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1591,16 +1659,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/nodes/gen_node_support.pl b/src/backend/nodes/gen_node_support.pl
index d67565b925..fcba6f986f 100644
--- a/src/backend/nodes/gen_node_support.pl
+++ b/src/backend/nodes/gen_node_support.pl
@@ -107,7 +107,7 @@ my @nodetag_only_files = qw(
# ABI stability during development.
my $last_nodetag = 'WindowObjectData';
-my $last_nodetag_no = 454;
+my $last_nodetag_no = 455;
# output file names
my @output_files;
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index b4038e114d..815ab8f9df 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -75,6 +75,8 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
bool two_phase,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
+static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
+ bool two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -95,6 +97,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
.walrcv_receive = libpqrcv_receive,
.walrcv_send = libpqrcv_send,
.walrcv_create_slot = libpqrcv_create_slot,
+ .walrcv_alter_slot = libpqrcv_alter_slot,
.walrcv_get_backend_pid = libpqrcv_get_backend_pid,
.walrcv_exec = libpqrcv_exec,
.walrcv_disconnect = libpqrcv_disconnect
@@ -1039,6 +1042,33 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
return snapshot;
}
+/*
+ * Change the definition of the replication slot.
+ */
+static void
+libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
+ bool two_phase)
+{
+ StringInfoData cmd;
+ PGresult *res;
+
+ initStringInfo(&cmd);
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s )",
+ quote_identifier(slotname),
+ two_phase ? "true" : "false");
+
+ res = libpqrcv_PQexec(conn->streamConn, cmd.data);
+ pfree(cmd.data);
+
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg("could not alter replication slot \"%s\": %s",
+ slotname, pchomp(PQerrorMessage(conn->streamConn)))));
+
+ PQclear(res);
+}
+
/*
* Return PID of remote backend process.
*/
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 8395ae7b23..295fe0f2a0 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -608,6 +608,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 832b1cf764..7da6d546dd 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3926,7 +3926,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/repl_gram.y b/src/backend/replication/repl_gram.y
index 0c874e33cf..bff501263e 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -64,6 +64,7 @@ Node *replication_parse_result;
%token K_START_REPLICATION
%token K_CREATE_REPLICATION_SLOT
%token K_DROP_REPLICATION_SLOT
+%token K_ALTER_REPLICATION_SLOT
%token K_TIMELINE_HISTORY
%token K_WAIT
%token K_TIMELINE
@@ -79,8 +80,9 @@ Node *replication_parse_result;
%type <node> command
%type <node> base_backup start_replication start_logical_replication
- create_replication_slot drop_replication_slot identify_system
- read_replication_slot timeline_history show
+ create_replication_slot drop_replication_slot
+ alter_replication_slot identify_system read_replication_slot
+ timeline_history show
%type <list> generic_option_list
%type <defelt> generic_option
%type <uintval> opt_timeline
@@ -111,6 +113,7 @@ command:
| start_logical_replication
| create_replication_slot
| drop_replication_slot
+ | alter_replication_slot
| read_replication_slot
| timeline_history
| show
@@ -257,6 +260,18 @@ drop_replication_slot:
}
;
+/* ALTER_REPLICATION_SLOT slot (options) */
+alter_replication_slot:
+ K_ALTER_REPLICATION_SLOT IDENT '(' generic_option_list ')'
+ {
+ AlterReplicationSlotCmd *cmd;
+ cmd = makeNode(AlterReplicationSlotCmd);
+ cmd->slotname = $2;
+ cmd->options = $4;
+ $$ = (Node *) cmd;
+ }
+ ;
+
/*
* START_REPLICATION [SLOT slot] [PHYSICAL] %X/%X [TIMELINE %d]
*/
@@ -399,6 +414,7 @@ ident_or_keyword:
| K_START_REPLICATION { $$ = "start_replication"; }
| K_CREATE_REPLICATION_SLOT { $$ = "create_replication_slot"; }
| K_DROP_REPLICATION_SLOT { $$ = "drop_replication_slot"; }
+ | K_ALTER_REPLICATION_SLOT { $$ = "alter_replication_slot"; }
| K_TIMELINE_HISTORY { $$ = "timeline_history"; }
| K_WAIT { $$ = "wait"; }
| K_TIMELINE { $$ = "timeline"; }
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index cb467ca46f..6396463cf9 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -125,6 +125,7 @@ TIMELINE { return K_TIMELINE; }
START_REPLICATION { return K_START_REPLICATION; }
CREATE_REPLICATION_SLOT { return K_CREATE_REPLICATION_SLOT; }
DROP_REPLICATION_SLOT { return K_DROP_REPLICATION_SLOT; }
+ALTER_REPLICATION_SLOT { return K_ALTER_REPLICATION_SLOT; }
TIMELINE_HISTORY { return K_TIMELINE_HISTORY; }
PHYSICAL { return K_PHYSICAL; }
RESERVE_WAL { return K_RESERVE_WAL; }
@@ -301,6 +302,7 @@ replication_scanner_is_replication_command(void)
case K_START_REPLICATION:
case K_CREATE_REPLICATION_SLOT:
case K_DROP_REPLICATION_SLOT:
+ case K_ALTER_REPLICATION_SLOT:
case K_READ_REPLICATION_SLOT:
case K_TIMELINE_HISTORY:
case K_SHOW:
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 1f9d0ed9b3..04915590d2 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -648,6 +648,31 @@ ReplicationSlotDrop(const char *name, bool nowait)
ReplicationSlotDropAcquired();
}
+/*
+ * Change the definition of the slot identified by the specified name.
+ */
+void
+ReplicationSlotAlter(const char *name, bool two_phase)
+{
+ Assert(MyReplicationSlot == NULL);
+
+ ReplicationSlotAcquire(name, false);
+
+ if (SlotIsPhysical(MyReplicationSlot))
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot use %s with a physical replication slot",
+ "ALTER_REPLICATION_SLOT"));
+
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.two_phase = two_phase;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+ ReplicationSlotRelease();
+}
+
/*
* Permanently drop the currently acquired replication slot.
*/
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 4c53de08b9..4c18dadaad 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1247,6 +1247,46 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
ReplicationSlotDrop(cmd->slotname, !cmd->wait);
}
+/*
+ * Process extra options given to ALTER_REPLICATION_SLOT.
+ */
+static void
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *two_phase)
+{
+ bool two_phase_given = false;
+ ListCell *lc;
+
+ /* Parse options */
+ foreach(lc, cmd->options)
+ {
+ DefElem *defel = (DefElem *) lfirst(lc);
+
+ 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);
+ }
+}
+
+/*
+ * Change the definition of a replication slot.
+ */
+static void
+AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
+{
+ bool two_phase = false;
+
+ ParseAlterReplSlotOptions(cmd, &two_phase);
+ ReplicationSlotAlter(cmd->slotname, two_phase);
+}
+
/*
* Load previously initiated logical slot and prepare for sending data (via
* WalSndLoop).
@@ -1820,6 +1860,13 @@ exec_replication_command(const char *cmd_string)
EndReplicationCommand(cmdtag);
break;
+ case T_AlterReplicationSlotCmd:
+ cmdtag = "ALTER_REPLICATION_SLOT";
+ set_ps_display(cmdtag);
+ AlterReplicationSlot((AlterReplicationSlotCmd *) cmd_node);
+ EndReplicationCommand(cmdtag);
+ break;
+
case T_StartReplicationCmd:
{
StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index a1aa946b30..b689b6906a 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1927,7 +1927,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "disable_on_error", "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 21e2af7387..dac3f27bc3 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/nodes/replnodes.h b/src/include/nodes/replnodes.h
index 4321ba8f86..5819276d19 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -72,6 +72,18 @@ typedef struct DropReplicationSlotCmd
} DropReplicationSlotCmd;
+/* ----------------------
+ * ALTER_REPLICATION_SLOT command
+ * ----------------------
+ */
+typedef struct AlterReplicationSlotCmd
+{
+ NodeTag type;
+ char *slotname;
+ List *options;
+} AlterReplicationSlotCmd;
+
+
/* ----------------------
* START_REPLICATION command
* ----------------------
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index a8a89dc784..24ab5117bd 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -214,7 +214,7 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
bool two_phase);
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
-
+extern void ReplicationSlotAlter(const char *name, bool two_phase);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
extern void ReplicationSlotCleanup(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 281626fa6f..8ba1599fad 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -359,6 +359,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
+/*
+ * walrcv_alter_slot_fn
+ */
+typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
+ const char *slotname,
+ bool two_phase);
+
/*
* walrcv_get_backend_pid_fn
*
@@ -400,6 +407,7 @@ typedef struct WalReceiverFunctionsType
walrcv_receive_fn walrcv_receive;
walrcv_send_fn walrcv_send;
walrcv_create_slot_fn walrcv_create_slot;
+ walrcv_alter_slot_fn walrcv_alter_slot;
walrcv_get_backend_pid_fn walrcv_get_backend_pid;
walrcv_exec_fn walrcv_exec;
walrcv_disconnect_fn walrcv_disconnect;
@@ -431,6 +439,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
+#define walrcv_alter_slot(conn, slotname, two_phase) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, 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 343e781896..3f19ae4f5f 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -235,6 +235,7 @@ extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
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 b15eddbff3..51c466f42e 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 | 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 444e563ff3..b7764c1074 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 8ce4cfc983..b6ce59763a 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] REL_16_0002-Alter-slot-option-two_phase-only-when-altering-true-.patch (4.0K, ../../OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-REL_16_0002-Alter-slot-option-two_phase-only-when-altering-true-.patch)
download | inline diff:
From 9adc3d08178b89a95f9e87b76743a796814aaf8f Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH 2/4] Alter slot option two_phase only when altering true to
false
---
src/backend/commands/subscriptioncmds.c | 3 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 72 +++++++++++++++++++
3 files changed, 75 insertions(+), 1 deletion(-)
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 59852fd9af..955d5e4899 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1493,7 +1493,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
/*
* Try to acquire the connection necessary for altering slot.
*/
- if (replaces[Anum_pg_subscription_subtwophasestate - 1])
+ if (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase)
{
bool must_use_password;
char *err;
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index bd673a9d68..9e2a458202 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] REL_16_0003-Abort-prepared-transactions-while-altering-two_phase.patch (5.4K, ../../OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-REL_16_0003-Abort-prepared-transactions-while-altering-two_phase.patch)
download | inline diff:
From e4ec647fc5114440b1061d1376caca73c03f7936 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH 3/4] Abort prepared transactions while altering two_phase to
false
---
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 +++++++++++++++++++
4 files changed, 68 insertions(+), 16 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 2148daba3c..e18e80cf4e 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2678,13 +2678,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++)
@@ -2693,11 +2693,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 955d5e4899..b1c00e36db 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1142,6 +1142,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
@@ -1158,13 +1160,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 dac3f27bc3..8eebfa7267 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] REL_16_0004-Add-force_alter-option.patch (7.4K, ../../OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-REL_16_0004-Add-force_alter-option.patch)
download | inline diff:
From 7588a8ddfdce6193c6bfaf43ab5a09c18ed4bc8a Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH 4/4] Add force_alter option
---
src/backend/commands/subscriptioncmds.c | 37 ++++++++++++++++++-
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, 59 insertions(+), 7 deletions(-)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index b1c00e36db..e38d808d32 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -72,6 +72,7 @@
#define SUBOPT_RUN_AS_OWNER 0x00001000
#define SUBOPT_LSN 0x00002000
#define SUBOPT_ORIGIN 0x00004000
+#define SUBOPT_FORCE_ALTER 0x00008000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -98,6 +99,7 @@ typedef struct SubOpts
bool runasowner;
char *origin;
XLogRecPtr lsn;
+ bool twophase_force;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -158,6 +160,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->runasowner = 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)
@@ -354,6 +358,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),
@@ -1134,7 +1147,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN |
+ SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1170,6 +1184,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 = ...)");
@@ -1179,8 +1203,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 */
@@ -1272,6 +1297,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 51c466f42e..946f3f6721 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -370,6 +370,9 @@ ERROR: two_phase requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+ERROR: force_alter must be specified with two_phase
\dRs+
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index b7764c1074..2f04675980 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -255,6 +255,9 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+
\dRs+
-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index a8135b671c..7c73a58f2a 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -85,16 +85,29 @@ $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(1), "transaction has been prepared on subscriber");
-$node_subscriber->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION sub DISABLE;
- ALTER SUBSCRIPTION sub SET (two_phase = off);
- ALTER SUBSCRIPTION sub ENABLE;");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub DISABLE;");
+
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql(
+ 'postgres', "ALTER SUBSCRIPTION sub SET (two_phase = off);");
+ok($stderr =~ /cannot alter two_phase = false when there are prepared transactions/,
+ 'ALTER SUBSCRIPTION failed');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exits");
+
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter = on);");
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(0), "prepared transaction done by worker is aborted");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE;");
+
$node_publisher->safe_psql( 'postgres',
"COMMIT PREPARED 'test_prepared_tab_full';");
$node_publisher->wait_for_catchup('sub');
--
2.43.0
^ permalink raw reply [nested|flat] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-10 14:16 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-15 15:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-22 08:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-22 12:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
@ 2024-04-23 12:15 ` Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-23 12:15 UTC (permalink / raw)
To: [email protected] <[email protected]>; +Cc: Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; 'Давыдов Виталий' <[email protected]>
Dear hackers,
Per recent commit (b29cbd3da), our patch needed to be rebased.
Here is an updated version.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v6-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (23.0K, ../../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v6-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
download | inline diff:
From d4bb208e621c0f47500fd1b4542a5c10ebd2ec59 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v6 1/4] Allow altering of two_phase option of a SUBSCRIPTION
This patch allows user to alter two_phase option of a subscriber provided no uncommitted
prepared transactions are pending on that subscription.
Author: Cherian Ajin, Hayato Kuroda
---
doc/src/sgml/ref/alter_subscription.sgml | 11 +--
src/backend/access/transam/twophase.c | 43 ++++++++++++
src/backend/commands/subscriptioncmds.c | 62 +++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 7 +-
src/backend/replication/logical/launcher.c | 21 ++++++
src/backend/replication/logical/worker.c | 2 +-
src/backend/replication/slot.c | 19 +++++-
src/backend/replication/walsender.c | 20 ++++--
src/bin/psql/tab-complete.c | 2 +-
src/include/access/twophase.h | 3 +
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 5 +-
src/include/replication/worker_internal.h | 1 +
src/test/regress/expected/subscription.out | 5 +-
src/test/regress/sql/subscription.sql | 5 +-
src/test/subscription/t/021_twophase.pl | 67 ++++++++++++++++++-
16 files changed, 235 insertions(+), 41 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index a78c1c3a47..e69132c39d 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -68,8 +68,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>,
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
- with <literal>refresh</literal> option as <literal>true</literal> and
- <command>ALTER SUBSCRIPTION ... SET (failover = on|off)</command>
+ with <literal>refresh</literal> option as <literal>true</literal>,
+ <command>ALTER SUBSCRIPTION ... SET (failover = on|off)</command> and
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = on|off)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
@@ -228,9 +229,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
- <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+ <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
+ <literal>two_phase</literal> can be altered only for disabled subscription.
</para>
<para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 8090ac9fc1..495f99a357 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,46 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
LWLockRelease(TwoPhaseStateLock);
return found;
}
+
+/*
+ * checkGid
+ */
+static bool
+checkGid(char *gid, Oid subid)
+{
+ int ret;
+ Oid subid_written,
+ xid;
+
+ ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid);
+
+ if (ret != 2 || subid != subid_written)
+ return false;
+
+ return true;
+}
+
+/*
+ * LookupGXactBySubid
+ * Check if the prepared transaction done by apply worker exists.
+ */
+bool
+LookupGXactBySubid(Oid subid)
+{
+ bool found = false;
+
+ LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+ for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
+ {
+ GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+
+ /* Ignore not-yet-valid GIDs. */
+ if (gxact->valid && checkGid(gxact->gid, subid))
+ {
+ found = true;
+ break;
+ }
+ }
+ LWLockRelease(TwoPhaseStateLock);
+ return found;
+}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e407428dbc..aa8a8e1f84 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -16,6 +16,7 @@
#include "access/htup_details.h"
#include "access/table.h"
+#include "access/twophase.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
@@ -1143,7 +1144,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
+ SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+ SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
SUBOPT_ORIGIN);
@@ -1151,6 +1153,47 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ /* XXX */
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /*
+ * two_phase can be only changed for disabled
+ * subscriptions
+ */
+ if (form->subenabled)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for enabled subscription",
+ "two_phase")));
+
+ /*
+ * Stop all the subscription workers, just in case. Workers
+ * may still survive even if the subscription is disabled.
+ */
+ logicalrep_workers_stop(subid);
+
+ /* Check whether the number of prepared transactions */
+ if (!opts.twophase &&
+ form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ LookupGXactBySubid(subid))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot disable two_phase when uncommitted prepared transactions present")));
+
+ /*
+ * The changed failover option of the slot can't be rolled
+ * back.
+ */
+ PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+
+ /* Change system catalog acoordingly */
+ values[Anum_pg_subscription_subtwophasestate - 1] =
+ CharGetDatum(opts.twophase ?
+ LOGICALREP_TWOPHASE_STATE_PENDING :
+ LOGICALREP_TWOPHASE_STATE_DISABLED);
+ replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
{
/*
@@ -1505,7 +1548,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1])
+ if (replaces[Anum_pg_subscription_subtwophasestate - 1] ||
+ replaces[Anum_pg_subscription_subfailover - 1])
{
bool must_use_password;
char *err;
@@ -1525,7 +1569,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+ walrcv_alter_slot(wrconn, sub->slotname, opts.twophase, opts.failover);
}
PG_FINALLY();
{
@@ -1562,7 +1606,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1672,16 +1715,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
* New workers won't be started because we hold an exclusive lock on the
* subscription till the end of the transaction.
*/
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
- subworkers = logicalrep_workers_find(subid, false);
- LWLockRelease(LogicalRepWorkerLock);
- foreach(lc, subworkers)
- {
- LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
-
- logicalrep_worker_stop(w->subid, w->relid);
- }
- list_free(subworkers);
+ logicalrep_workers_stop(subid);
/*
* Remove the no-longer-useful entry in the launcher's table of apply
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 3c2b1bb496..baef3bdec0 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover);
+ bool two_phase, bool failover);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,14 +1121,15 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool two_phase, bool failover)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s, FAILOVER %s )",
quote_identifier(slotname),
+ two_phase ? "true" : "false",
failover ? "true" : "false");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 66070e9131..94b73f3085 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,27 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockRelease(LogicalRepWorkerLock);
}
+/*
+ * Stop all the subscription workers.
+ */
+void
+logicalrep_workers_stop(Oid subid)
+{
+ List *subworkers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ subworkers = logicalrep_workers_find(subid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+ foreach(lc, subworkers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+ logicalrep_worker_stop(w->subid, w->relid);
+ }
+ list_free(subworkers);
+}
+
/*
* Stop the given logical replication parallel apply worker.
*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..374aa22091 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3911,7 +3911,7 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
- /* two-phase should not be altered */
+ /* two-phase should not be altered while the worker exists */
Assert(newsub->twophasestate == MySubscription->twophasestate);
/*
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index cebf44bb0f..621f35ab1e 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -800,8 +800,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
* Change the definition of the slot identified by the specified name.
*/
void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool two_phase, bool failover)
{
+ bool update_slot = false;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -844,12 +846,27 @@ ReplicationSlotAlter(const char *name, bool failover)
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot enable failover for a temporary replication slot"));
+ if (MyReplicationSlot->data.two_phase != two_phase)
+ {
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.two_phase = two_phase;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ update_slot = true;
+ }
+
+
if (MyReplicationSlot->data.failover != failover)
{
SpinLockAcquire(&MyReplicationSlot->mutex);
MyReplicationSlot->data.failover = failover;
SpinLockRelease(&MyReplicationSlot->mutex);
+ update_slot = true;
+ }
+
+ if (update_slot)
+ {
ReplicationSlotMarkDirty();
ReplicationSlotSave();
}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 9bf7c67f37..c45881554b 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,14 +1411,25 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
+ bool *two_phase, bool *failover)
{
+ bool two_phase_given = false;
bool failover_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
{
- if (strcmp(defel->defname, "failover") == 0)
+ if (strcmp(defel->defname, "two_phase") == 0)
+ {
+ if (two_phase_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ two_phase_given = true;
+ *two_phase = defGetBoolean(defel);
+ }
+ else if (strcmp(defel->defname, "failover") == 0)
{
if (failover_given)
ereport(ERROR,
@@ -1438,10 +1449,11 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
+ bool two_phase = false;
bool failover = false;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &two_phase, &failover);
+ ReplicationSlotAlter(cmd->slotname, two_phase, failover);
}
/*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 6fee3160f0..5ff84301cd 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
"password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit");
+ "streaming", "synchronous_commit", "two_phase");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 56248c0006..d493ed24c5 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -62,4 +62,7 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
TimestampTz origin_prepare_timestamp);
+
+extern bool LookupGXactBySubid(Oid subid);
+
#endif /* TWOPHASE_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 7b937d1a0c..2fcb11418f 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool two_phase,
+ bool failover);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 12f71fa99b..a443f402f5 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -377,6 +377,7 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
+ bool two_phase,
bool failover);
/*
@@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
-#define walrcv_alter_slot(conn, slotname, failover) \
- WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
+#define walrcv_alter_slot(conn, slotname, two_phase, failover) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, two_phase, failover)
#define walrcv_get_backend_pid(conn) \
WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 515aefd519..d5428263c1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -246,6 +246,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
Oid userid, Oid relid,
dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_workers_stop(Oid subid);
extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 0f2a25cdc1..51fa4b9690 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -377,10 +377,7 @@ HINT: To initiate replication, you must manually create the replication slot, e
regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-ERROR: unrecognized subscription parameter: "two_phase"
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
List of subscriptions
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 3e5ba4cb8c..a3886d79ca 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -256,10 +256,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
\dRs+
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 9437cd4c3b..e710f3c4c0 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,71 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
is($result, qq(2), 'replicated data in subscriber table');
+# Disable the subscription and alter it to two_phase = false,
+# verify that the altered subscription reflects the two_phase option.
+
+# Alter subscription two_phase to false
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is disabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(d), 'two-phase is disabled');
+
+# Now do a prepare on publisher and make sure that it is not replicated.
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_publisher->safe_psql(
+ 'postgres', qq{
+ BEGIN;
+ INSERT INTO tab_copy VALUES (100);
+ PREPARE TRANSACTION 'newgid';
+ });
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure that there is 0 prepared transaction on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(0), 'transaction is prepared on subscriber');
+
+# Now commit the insert and verify that it IS replicated
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';");
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Made sure that the commited transaction is replicated.
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
+is($result, qq(3), 'replicated data in subscriber table');
+
+# Alter subscription two_phase to true
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is enabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(e), 'two-phase is disabled');
+
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;");
$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
@@ -374,8 +439,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
# check all the cleanup
###############################
-$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
-
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_subscription");
is($result, qq(0), 'check subscription was dropped on subscriber');
--
2.43.0
[application/octet-stream] v6-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch (9.0K, ../../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v6-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch)
download | inline diff:
From 458bdbfeabae2e22dae84386a8b8e54126d101a3 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v6 2/4] Alter slot option two_phase only when altering true to
false
---
doc/src/sgml/ref/alter_subscription.sgml | 2 +-
src/backend/commands/subscriptioncmds.c | 23 +++++-
.../libpqwalreceiver/libpqwalreceiver.c | 21 ++++--
src/include/replication/walreceiver.h | 8 +--
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 72 +++++++++++++++++++
6 files changed, 114 insertions(+), 13 deletions(-)
create mode 100644 src/test/subscription/t/099_twophase_added.pl
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index e69132c39d..e54aa1b128 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -70,7 +70,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
with <literal>refresh</literal> option as <literal>true</literal>,
<command>ALTER SUBSCRIPTION ... SET (failover = on|off)</command> and
- <command>ALTER SUBSCRIPTION ... SET (two_phase = on|off)</command>
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = off)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index aa8a8e1f84..b02e21f535 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1184,7 +1184,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* The changed failover option of the slot can't be rolled
* back.
*/
- PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+ if (!opts.twophase)
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = off)");
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
@@ -1554,6 +1556,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
bool must_use_password;
char *err;
WalReceiverConn *wrconn;
+ bool two_phase_needs_to_be_updated;
+ bool failover_needs_to_be_updated;
/* Load the library providing us libpq calls. */
load_file("libpqwalreceiver", false);
@@ -1567,9 +1571,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
(errcode(ERRCODE_CONNECTION_FAILURE),
errmsg("could not connect to the publisher: %s", err)));
+ /*
+ * Consider which slot option must be altered.
+ *
+ * We must alter the failover option whenever subfailover is updated.
+ * Two_phase, however, is altered only when changing true to false.
+ */
+ two_phase_needs_to_be_updated =
+ (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase);
+ failover_needs_to_be_updated =
+ replaces[Anum_pg_subscription_subfailover - 1];
+
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.twophase, opts.failover);
+ if (two_phase_needs_to_be_updated || failover_needs_to_be_updated)
+ walrcv_alter_slot(wrconn, sub->slotname,
+ two_phase_needs_to_be_updated ? &opts.twophase : NULL,
+ failover_needs_to_be_updated ? &opts.failover : NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index baef3bdec0..546b599848 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool two_phase, bool failover);
+ const bool *two_phase, const bool *failover);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,16 +1121,25 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool two_phase, bool failover)
+ const bool *two_phase, const bool *failover)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s, FAILOVER %s )",
- quote_identifier(slotname),
- two_phase ? "true" : "false",
- failover ? "true" : "false");
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
+ quote_identifier(slotname));
+
+ if (two_phase)
+ appendStringInfo(&cmd, "TWO_PHASE %s%s ",
+ (*two_phase) ? "true" : "false",
+ failover ? ", " : "");
+
+ if (failover)
+ appendStringInfo(&cmd, "FAILOVER %s ",
+ (*failover) ? "true" : "false");
+
+ appendStringInfoString(&cmd, ");");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index a443f402f5..f30637aa4a 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -372,13 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
/*
* walrcv_alter_slot_fn
*
- * Change the definition of a replication slot. Currently, it only supports
- * changing the failover property of the slot.
+ * Change the definition of a replication slot. Currently, it supports
+ * changing the two_phase and the failover property of the slot.
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool two_phase,
- bool failover);
+ const bool *two_phase,
+ const bool *failover);
/*
* walrcv_get_backend_pid_fn
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..b4bd522c3d 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
't/031_column_list.pl',
't/032_subscribe_use_index.pl',
't/033_run_as_table_owner.pl',
+ 't/099_twophase_added.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
new file mode 100644
index 0000000000..c13a37675a
--- /dev/null
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -0,0 +1,72 @@
+# Copyright (c) 2021-2024, PostgreSQL Global Development Group
+
+# Additional tests for altering two_phase option
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10));
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10));
+$node_subscriber->start;
+
+# Define pre-existing tables on both nodes
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication, with two_phase = off
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION pub FOR ALL TABLES");
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION sub
+ CONNECTION '$publisher_connstr' PUBLICATION pub
+ WITH (two_phase = off, copy_data = off)");
+
+######
+# Check the case that prepared transactions exist on publisher node
+######
+
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(1, 5));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('sub');
+
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "transaction is not prepared on subscriber");
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION sub DISABLE;
+ ALTER SUBSCRIPTION sub SET (two_phase = on);
+ ALTER SUBSCRIPTION sub ENABLE;");
+
+$node_publisher->safe_psql( 'postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM tab_full;");
+is($result, q(5),
+ "prepared transactions done before altering can be replicated");
+
+done_testing();
--
2.43.0
[application/octet-stream] v6-0003-Abort-prepared-transactions-while-altering-two_ph.patch (7.1K, ../../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v6-0003-Abort-prepared-transactions-while-altering-two_ph.patch)
download | inline diff:
From d0c8138ccdf19dd9d4395855e5482cce496bda22 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v6 3/4] Abort prepared transactions while altering two_phase
to false
---
doc/src/sgml/ref/alter_subscription.sgml | 8 ++++-
src/backend/access/transam/twophase.c | 19 +++++-----
src/backend/commands/subscriptioncmds.c | 33 +++++++++++------
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 35 +++++++++++++++++++
5 files changed, 76 insertions(+), 22 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index e54aa1b128..926f560566 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -233,7 +233,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
- <literal>two_phase</literal> can be altered only for disabled subscription.
</para>
<para>
@@ -255,6 +254,13 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>
option is enabled.
</para>
+
+ <para>
+ <literal>two_phase</literal> can be altered only for disabled
+ subscriptions. When altering the parameter from <literal>true</literal>
+ to <literal>false</literal>, the backend process checks prepared
+ transactions done by the logical replication worker and aborts them.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 495f99a357..9121195725 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2702,13 +2702,13 @@ checkGid(char *gid, Oid subid)
}
/*
- * LookupGXactBySubid
- * Check if the prepared transaction done by apply worker exists.
+ * GetGidListBySubid
+ * Get a list of GIDs which is PREPARE'd by the given subscription.
*/
-bool
-LookupGXactBySubid(Oid subid)
+List *
+GetGidListBySubid(Oid subid)
{
- bool found = false;
+ List *list = NIL;
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
@@ -2717,11 +2717,10 @@ LookupGXactBySubid(Oid subid)
/* Ignore not-yet-valid GIDs. */
if (gxact->valid && checkGid(gxact->gid, subid))
- {
- found = true;
- break;
- }
+ list = lappend(list, pstrdup(gxact->gid));
+
}
LWLockRelease(TwoPhaseStateLock);
- return found;
+
+ return list;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index b02e21f535..8a36558b2c 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1156,6 +1156,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
/* XXX */
if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
{
+ List *prepared_xacts = NIL;
+
/*
* two_phase can be only changed for disabled
* subscriptions
@@ -1172,22 +1174,33 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
*/
logicalrep_workers_stop(subid);
- /* Check whether the number of prepared transactions */
- if (!opts.twophase &&
- form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
- LookupGXactBySubid(subid))
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("cannot disable two_phase when uncommitted prepared transactions present")));
-
/*
- * The changed failover option of the slot can't be rolled
- * back.
+ * If two phase was enabled, there is a possibility the
+ * transactions has already been PREPARE'd.
*/
if (!opts.twophase)
+ {
+ /*
+ * The changed failover option of the slot can't be rolled
+ * back.
+ */
PreventInTransactionBlock(isTopLevel,
"ALTER SUBSCRIPTION ... SET (two_phase = off)");
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ ListCell *cell;
+
+ /* Abort all listed transactions */
+ foreach(cell, prepared_xacts)
+ FinishPreparedTransaction((char *) lfirst(cell),
+ false);
+
+ list_free(prepared_xacts);
+ }
+ }
+
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index d493ed24c5..95770bbd69 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"
+#include "nodes/pg_list.h"
/*
* GlobalTransactionData is defined in twophase.c; other places have no
@@ -63,6 +64,6 @@ extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
TimestampTz origin_prepare_timestamp);
-extern bool LookupGXactBySubid(Oid subid);
+extern List *GetGidListBySubid(Oid subid);
#endif /* TWOPHASE_H */
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index c13a37675a..a8135b671c 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -69,4 +69,39 @@ $result = $node_subscriber->safe_psql('postgres',
is($result, q(5),
"prepared transactions done before altering can be replicated");
+######
+# Check the case that prepared transactions exist on subscriber node
+######
+
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(6, 10));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "transaction has been prepared on subscriber");
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION sub DISABLE;
+ ALTER SUBSCRIPTION sub SET (two_phase = off);
+ ALTER SUBSCRIPTION sub ENABLE;");
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "prepared transaction done by worker is aborted");
+
+$node_publisher->safe_psql( 'postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(10) FROM tab_full;");
+is($result, q(10),
+ "prepared transactions on publisher can be replicated");
+
done_testing();
--
2.43.0
[application/octet-stream] v6-0004-Add-force_alter-option.patch (8.2K, ../../OSBPR01MB2552936A8C9F9790A4CFD389F5112@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v6-0004-Add-force_alter-option.patch)
download | inline diff:
From 79b7d98f71c6bf6676ce868d16f8965709a4251a Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v6 4/4] Add force_alter option
---
doc/src/sgml/ref/alter_subscription.sgml | 9 +++--
src/backend/commands/subscriptioncmds.c | 33 ++++++++++++++++++-
src/test/regress/expected/subscription.out | 3 ++
src/test/regress/sql/subscription.sql | 3 ++
src/test/subscription/t/099_twophase_added.pl | 23 ++++++++++---
5 files changed, 62 insertions(+), 9 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 926f560566..e6228490a8 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -257,9 +257,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
<literal>two_phase</literal> can be altered only for disabled
- subscriptions. When altering the parameter from <literal>true</literal>
- to <literal>false</literal>, the backend process checks prepared
- transactions done by the logical replication worker and aborts them.
+ subscriptions. Altering the parameter from <literal>true</literal>
+ to <literal>false</literal> will be failed when there are prepared
+ transactions done by the logical replication worker. If you want to alter
+ the parameter forcibly in this case, <literal>force_alter</literal>
+ option must be set to <literal>true</literal> at the same time. If
+ specified, the backend process aborts prepared transactions.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 8a36558b2c..7ea4e85595 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -73,6 +73,7 @@
#define SUBOPT_FAILOVER 0x00002000
#define SUBOPT_LSN 0x00004000
#define SUBOPT_ORIGIN 0x00008000
+#define SUBOPT_FORCE_ALTER 0x00010000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -100,6 +101,7 @@ typedef struct SubOpts
bool failover;
char *origin;
XLogRecPtr lsn;
+ bool twophase_force;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -162,6 +164,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->failover = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_FORCE_ALTER))
+ opts->twophase_force = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -367,6 +371,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_FORCE_ALTER) &&
+ strcmp(defel->defname, "force_alter") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_FORCE_ALTER))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_FORCE_ALTER;
+ opts->twophase_force = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -1148,7 +1161,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1192,6 +1205,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
ListCell *cell;
+ /*
+ * Abort prepared transactions if force option is also
+ * specified. Otherwise raise an ERROR.
+ */
+ if (!opts.twophase_force)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot alter %s when there are prepared transactions",
+ "two_phase = false")));
+
/* Abort all listed transactions */
foreach(cell, prepared_xacts)
FinishPreparedTransaction((char *) lfirst(cell),
@@ -1321,6 +1344,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_suborigin - 1] = true;
}
+ /* force_alter cannot be used standalone */
+ if (IsSet(opts.specified_opts, SUBOPT_FORCE_ALTER) &&
+ !IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("%s must be specified with %s",
+ "force_alter", "two_phase")));
+
update_tuple = true;
break;
}
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 51fa4b9690..f607045b28 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -370,6 +370,9 @@ ERROR: two_phase requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+ERROR: force_alter must be specified with two_phase
\dRs+
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index a3886d79ca..80ab4dd9bc 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -255,6 +255,9 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
+-- fail - force_alter cannot be set alone
+ALTER SUBSCRIPTION regress_testsub SET (force_alter = true);
+
\dRs+
-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index a8135b671c..7c73a58f2a 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -85,16 +85,29 @@ $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(1), "transaction has been prepared on subscriber");
-$node_subscriber->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION sub DISABLE;
- ALTER SUBSCRIPTION sub SET (two_phase = off);
- ALTER SUBSCRIPTION sub ENABLE;");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub DISABLE;");
+
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql(
+ 'postgres', "ALTER SUBSCRIPTION sub SET (two_phase = off);");
+ok($stderr =~ /cannot alter two_phase = false when there are prepared transactions/,
+ 'ALTER SUBSCRIPTION failed');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exits");
+
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub SET (two_phase = off, force_alter = on);");
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(0), "prepared transaction done by worker is aborted");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE;");
+
$node_publisher->safe_psql( 'postgres',
"COMMIT PREPARED 'test_prepared_tab_full';");
$node_publisher->wait_for_catchup('sub');
--
2.43.0
^ permalink raw reply [nested|flat] 9+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
@ 2024-04-11 02:07 ` Hayato Kuroda (Fujitsu) <[email protected]>
1 sibling, 0 replies; 9+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-04-11 02:07 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,
> 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?
Alternative idea is that worker pass a list of prepared transaction as new
option in START_REPLICATION. This can reduce the number of inter-node
communications, but sometimes the list may be huge.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2024-04-23 12:15 UTC | newest]
Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-03-30 23:59 [PATCH v17 02/10] pg_stat_file and pg_ls_dir_* to use lstat().. Justin Pryzby <[email protected]>
2024-04-05 11:29 Re: Slow catchup of 2PC (twophase) transactions on replica in LR Ajin Cherian <[email protected]>
2024-04-10 11:18 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Amit Kapila <[email protected]>
2024-04-10 14:16 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-15 15:31 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Давыдов Виталий <[email protected]>
2024-04-22 08:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-22 12:54 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-23 12:15 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-04-11 02:07 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[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