public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: Hayato Kuroda (Fujitsu) <[email protected]>
To: 'Давыдов Виталий' <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Ajin Cherian <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
Date: Mon, 22 Apr 2024 12:54:10 +0000
Message-ID: <OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OSBPR01MB2552F73968D5A460157B57B9F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
References: <8fab8-65d74c80-1-2f28e880@39088166>
<CAA4eK1KOs3s6syZqUgrd2WvjTz64SGf0ToZcRoPMCKKH+M0YFQ@mail.gmail.com>
<99df2-65d8d200-1-63c26880@172687752>
<CAA4eK1KtjjPxe5tjWeGNf5CfF9qfceVQCCC9XGsKzR-ifQD9uA@mail.gmail.com>
<ba4b6-65ddc580-9-5c3e3080@134362348>
<CAA4eK1J6wOMkDDt3iOs+JqkvQvuwSORosTpqdejgvXqFmxex=A@mail.gmail.com>
<badcf-65dde500-3-24ba3c80@28790566>
<cedbe-65e0c000-1-6db17700@133269862>
<[email protected]>
<f9f1b-65e72c80-7-533dda00@76203995>
<CAFPTHDbU4EkVmKL66i5BzrmQ6LFftQPx_TZ1euKt9fvVzQDyjA@mail.gmail.com>
<CAA4eK1+n7M2S1OpoGWDd+YZkDCuURMdVRbvP0eELQUvWgmDneg@mail.gmail.com>
<CAFPTHDa=pJSZ_4dV5DPAOapRSgPcyyUTP0WzGY2Rz_D3-gwraw@mail.gmail.com>
<CAA4eK1K1fSkeK=kc26G5cq87vQG4=1qs_b+no4+ep654SeBy1w@mail.gmail.com>
<23d1ca-66169f80-1f-5a90aa00@207285398>
<26b30e-661d4880-9-62d2df00@231402233>
<OSBPR01MB2552F73968D5A460157B57B9F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
> 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
view thread (9+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
In-Reply-To: <OSBPR01MB25521E993657E754B0C51A98F5122@OSBPR01MB2552.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox