From f7c0476226447793353463b32286410f7b960027 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Fri, 5 Apr 2024 06:47:18 -0400 Subject: [PATCH v8 1/4] Allow altering of two_phase option of a SUBSCRIPTION This patch allows the user to alter the 'two_phase' option of a subscriber provided no uncommitted prepared transactions are pending on that subscription. Author: Cherian Ajin, Hayato Kuroda --- doc/src/sgml/ref/alter_subscription.sgml | 12 ++-- src/backend/access/transam/twophase.c | 62 ++++++++++++++++ src/backend/commands/subscriptioncmds.c | 67 +++++++++++++---- .../libpqwalreceiver/libpqwalreceiver.c | 9 +-- src/backend/replication/logical/launcher.c | 22 ++++++ src/backend/replication/logical/worker.c | 21 +----- src/backend/replication/slot.c | 18 ++++- src/backend/replication/walsender.c | 18 ++++- src/bin/psql/tab-complete.c | 2 +- src/include/access/twophase.h | 5 ++ src/include/replication/slot.h | 3 +- src/include/replication/walreceiver.h | 11 +-- src/include/replication/worker_internal.h | 1 + src/test/regress/expected/subscription.out | 5 +- src/test/regress/sql/subscription.sql | 5 +- src/test/subscription/t/021_twophase.pl | 71 ++++++++++++++++++- 16 files changed, 269 insertions(+), 63 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index a78c1c3a47..88e9a72147 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -68,8 +68,9 @@ ALTER SUBSCRIPTION name RENAME TO < Commands ALTER SUBSCRIPTION ... REFRESH PUBLICATION, ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ... - with refresh option as true and - ALTER SUBSCRIPTION ... SET (failover = on|off) + with refresh option as true, + ALTER SUBSCRIPTION ... SET (failover = on|off) and + ALTER SUBSCRIPTION ... SET (two_phase = on|off) cannot be executed inside a transaction block. These commands also cannot be executed when the subscription has @@ -228,9 +229,12 @@ ALTER SUBSCRIPTION name RENAME TO < disable_on_error, password_required, run_as_owner, - origin, and - failover. + origin, + failover, and + two_phase. Only a superuser can set password_required = false. + The two_phase parameter can only be altered when the + subscription is disabled. diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 8090ac9fc1..66fa591eb5 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2682,3 +2682,65 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, LWLockRelease(TwoPhaseStateLock); return found; } + +/* + * TwoPhaseTransactionGid + * Form the prepared transaction GID for two_phase transactions. + * + * Return the GID in the supplied buffer. + */ +void +TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) +{ + Assert(subid != InvalidRepOriginId); + + if (!TransactionIdIsValid(xid)) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg_internal("invalid two-phase transaction ID"))); + + snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); +} + +/* + * IsTwoPhaseTransactionGidForSubid + * Check whether the given GID (as formed by TwoPhaseTransactionGid) is + * for the specified 'subid'. + */ +static bool +IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid) +{ + int ret; + Oid subid_written; + TransactionId xid; + + ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid); + + return (ret == 2 && subid == subid_written); +} + +/* + * LookupGXactBySubid + * Check if the prepared transaction done by apply worker exists. + */ +bool +LookupGXactBySubid(Oid subid) +{ + bool found = false; + + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + for (int i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + + /* Ignore not-yet-valid GIDs. */ + if (gxact->valid && + IsTwoPhaseTransactionGidForSubid(subid, gxact->gid)) + { + found = true; + break; + } + } + LWLockRelease(TwoPhaseStateLock); + return found; +} diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index e407428dbc..90d967eb7c 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,52 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, parse_subscription_options(pstate, stmt->options, supported_opts, &opts); + if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT)) + { + /* + * Do not allow changing the two_phase option if the + * subscription is enabled. This is because the two_phase + * option of the slot on the publisher cannot be modified + * if the slot is currently acquired by the apply worker. + */ + if (form->subenabled) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot set %s for enabled subscription", + "two_phase"))); + + /* + * Stop all the subscription workers, just in case. Workers + * may still survive even if the subscription is disabled. + */ + logicalrep_workers_stop(subid); + + /* + * two_phase cannot be disabled if there are any + * uncommitted prepared transactions present. + */ + if (!opts.twophase && + form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && + LookupGXactBySubid(subid)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot disable two_phase when uncommitted prepared transactions present"), + errhint("Resolve these transactions and try again"))); + + /* + * The changed two_phase option of the slot can't be rolled + * back. + */ + PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)"); + + /* Change system catalog acoordingly */ + values[Anum_pg_subscription_subtwophasestate - 1] = + CharGetDatum(opts.twophase ? + LOGICALREP_TWOPHASE_STATE_PENDING : + LOGICALREP_TWOPHASE_STATE_DISABLED); + replaces[Anum_pg_subscription_subtwophasestate - 1] = true; + } + if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME)) { /* @@ -1505,7 +1553,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, * doing the database operations we won't be able to rollback altered * slot. */ - if (replaces[Anum_pg_subscription_subfailover - 1]) + if (replaces[Anum_pg_subscription_subfailover - 1] || + replaces[Anum_pg_subscription_subtwophasestate - 1]) { bool must_use_password; char *err; @@ -1525,7 +1574,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, PG_TRY(); { - walrcv_alter_slot(wrconn, sub->slotname, opts.failover); + walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase); } PG_FINALLY(); { @@ -1562,7 +1611,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) char *subname; char *conninfo; char *slotname; - List *subworkers; ListCell *lc; char originname[NAMEDATALEN]; char *err = NULL; @@ -1672,16 +1720,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..998bbd517a 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn); static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, - bool failover); + bool failover, bool two_phase); static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn); static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn, const char *query, @@ -1121,15 +1121,16 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, */ static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, - bool failover) + bool failover, bool two_phase) { StringInfoData cmd; PGresult *res; initStringInfo(&cmd); - appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )", + appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )", quote_identifier(slotname), - failover ? "true" : "false"); + failover ? "true" : "false", + two_phase ? "true" : "false"); res = libpqrcv_PQexec(conn->streamConn, cmd.data); pfree(cmd.data); diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 66070e9131..548f6e0edb 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -623,6 +623,28 @@ logicalrep_worker_stop(Oid subid, Oid relid) LWLockRelease(LogicalRepWorkerLock); } +/* + * Stop all the subscription workers. + */ +void +logicalrep_workers_stop(Oid subid) +{ + List *subworkers; + ListCell *lc; + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + /* XXX clarify the reason why not only running workers are listed. */ + subworkers = logicalrep_workers_find(subid, false); + LWLockRelease(LogicalRepWorkerLock); + foreach(lc, subworkers) + { + LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc); + + logicalrep_worker_stop(w->subid, w->relid); + } + list_free(subworkers); +} + /* * Stop the given logical replication parallel apply worker. * diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index b5a80fe3e8..dcf656fd45 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -402,7 +402,6 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata, CmdType operation); /* Compute GID for two_phase transactions */ -static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid); /* Functions for skipping changes */ static void maybe_start_skipping_changes(XLogRecPtr finish_lsn); @@ -3911,7 +3910,7 @@ maybe_reread_subscription(void) /* !slotname should never happen when enabled is true. */ Assert(newsub->slotname); - /* two-phase should not be altered */ + /* two-phase cannot be altered while the worker exists */ Assert(newsub->twophasestate == MySubscription->twophasestate); /* @@ -4396,24 +4395,6 @@ cleanup_subxact_info() subxact_data.nsubxacts_max = 0; } -/* - * Form the prepared transaction GID for two_phase transactions. - * - * Return the GID in the supplied buffer. - */ -static void -TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid) -{ - Assert(subid != InvalidRepOriginId); - - if (!TransactionIdIsValid(xid)) - ereport(ERROR, - (errcode(ERRCODE_PROTOCOL_VIOLATION), - errmsg_internal("invalid two-phase transaction ID"))); - - snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); -} - /* * Common function to run the apply loop with error handling. Disable the * subscription, if necessary. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index aa4ea387da..d0c8d5a4df 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -804,8 +804,10 @@ ReplicationSlotDrop(const char *name, bool nowait) * Change the definition of the slot identified by the specified name. */ void -ReplicationSlotAlter(const char *name, bool failover) +ReplicationSlotAlter(const char *name, bool failover, bool two_phase) { + bool update_slot = false; + Assert(MyReplicationSlot == NULL); ReplicationSlotAcquire(name, false); @@ -854,6 +856,20 @@ ReplicationSlotAlter(const char *name, bool failover) MyReplicationSlot->data.failover = failover; SpinLockRelease(&MyReplicationSlot->mutex); + update_slot = true; + } + + if (MyReplicationSlot->data.two_phase != two_phase) + { + SpinLockAcquire(&MyReplicationSlot->mutex); + MyReplicationSlot->data.two_phase = two_phase; + SpinLockRelease(&MyReplicationSlot->mutex); + + update_slot = true; + } + + if (update_slot) + { ReplicationSlotMarkDirty(); ReplicationSlotSave(); } diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index c623b07cf0..2e6ca35049 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1411,9 +1411,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd) * Process extra options given to ALTER_REPLICATION_SLOT. */ static void -ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) +ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, + bool *failover, bool *two_phase) { bool failover_given = false; + bool two_phase_given = false; /* Parse options */ foreach_ptr(DefElem, defel, cmd->options) @@ -1427,6 +1429,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover) failover_given = true; *failover = defGetBoolean(defel); } + else if (strcmp(defel->defname, "two_phase") == 0) + { + if (two_phase_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + two_phase_given = true; + *two_phase = defGetBoolean(defel); + } else elog(ERROR, "unrecognized option: %s", defel->defname); } @@ -1439,9 +1450,10 @@ static void AlterReplicationSlot(AlterReplicationSlotCmd *cmd) { bool failover = false; + bool two_phase = false; - ParseAlterReplSlotOptions(cmd, &failover); - ReplicationSlotAlter(cmd->slotname, failover); + ParseAlterReplSlotOptions(cmd, &failover, &two_phase); + ReplicationSlotAlter(cmd->slotname, failover, two_phase); } /* diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index d453e224d9..891face1b6 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end) else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "(")) COMPLETE_WITH("binary", "disable_on_error", "failover", "origin", "password_required", "run_as_owner", "slot_name", - "streaming", "synchronous_commit"); + "streaming", "synchronous_commit", "two_phase"); /* ALTER SUBSCRIPTION SKIP ( */ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "(")) COMPLETE_WITH("lsn"); diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index 56248c0006..d37e06fdee 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -62,4 +62,9 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning); extern void restoreTwoPhaseData(void); extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp); + +extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, + int szgid); +extern bool LookupGXactBySubid(Oid subid); + #endif /* TWOPHASE_H */ diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 1bc80960ef..014e216cf5 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific, extern void ReplicationSlotPersist(void); extern void ReplicationSlotDrop(const char *name, bool nowait); extern void ReplicationSlotDropAcquired(void); -extern void ReplicationSlotAlter(const char *name, bool failover); +extern void ReplicationSlotAlter(const char *name, bool failover, + bool two_phase); extern void ReplicationSlotAcquire(const char *name, bool nowait); extern void ReplicationSlotRelease(void); diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 12f71fa99b..31fa1257ec 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -372,12 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn, /* * walrcv_alter_slot_fn * - * Change the definition of a replication slot. Currently, it only supports - * changing the failover property of the slot. + * Change the definition of a replication slot. Currently, it supports + * changing the failover and the two_phase property of the slot. */ typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn, const char *slotname, - bool failover); + bool failover, + bool two_phase); /* * walrcv_get_backend_pid_fn @@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions; WalReceiverFunctions->walrcv_send(conn, buffer, nbytes) #define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \ WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) -#define walrcv_alter_slot(conn, slotname, failover) \ - WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover) +#define walrcv_alter_slot(conn, slotname, failover, two_phase) \ + WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase) #define walrcv_get_backend_pid(conn) \ WalReceiverFunctions->walrcv_get_backend_pid(conn) #define walrcv_exec(conn, exec, nRetTypes, retTypes) \ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 515aefd519..d5428263c1 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -246,6 +246,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype, Oid userid, Oid relid, dsm_handle subworker_dsm); extern void logicalrep_worker_stop(Oid subid, Oid relid); +extern void logicalrep_workers_stop(Oid subid); extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo); extern void logicalrep_worker_wakeup(Oid subid, Oid relid); extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 0f2a25cdc1..51fa4b9690 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -377,10 +377,7 @@ HINT: To initiate replication, you must manually create the replication slot, e regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 (1 row) ---fail - alter of two_phase option not supported. -ALTER SUBSCRIPTION regress_testsub SET (two_phase = false); -ERROR: unrecognized subscription parameter: "two_phase" --- but can alter streaming when two_phase enabled +-- We can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ List of subscriptions diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 3e5ba4cb8c..a3886d79ca 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -256,10 +256,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true); \dRs+ ---fail - alter of two_phase option not supported. -ALTER SUBSCRIPTION regress_testsub SET (two_phase = false); - --- but can alter streaming when two_phase enabled +-- We can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl index 9437cd4c3b..72df258000 100644 --- a/src/test/subscription/t/021_twophase.pl +++ b/src/test/subscription/t/021_twophase.pl @@ -367,6 +367,75 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); is($result, qq(2), 'replicated data in subscriber table'); +# Clean up +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); + +############################### +# Disable the subscription and alter it to two_phase = false, +# then verify that the altered subscription reflects the two_phase option. +############################### + +# Alter subscription two_phase to false +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy DISABLE"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy ENABLE"); + +# Wait for subscription startup +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy); + +# Make sure that the two-phase is disabled on the subscriber +$result = $node_subscriber->safe_psql('postgres', + "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';" +); +is($result, qq(d), 'two-phase should be disabled'); + +# Now do a prepare on the publisher and make sure that it is not replicated. +$node_publisher->safe_psql( + 'postgres', qq{ + BEGIN; + INSERT INTO tab_copy VALUES (100); + PREPARE TRANSACTION 'newgid'; + }); + +# Wait for the subscriber to catchup +$node_publisher->wait_for_catchup($appname_copy); + +# Make sure there are no prepared transactions on the subscriber +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*) FROM pg_prepared_xacts;"); +is($result, qq(0), 'should be no prepared transactions on subscriber'); + +# Now commit the insert and verify that it IS replicated +$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';"); + +# Wait for the subscriber to catchup +$node_publisher->wait_for_catchup($appname_copy); + +# Make sure that the committed transaction is replicated. +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;"); +is($result, qq(3), 'replicated data in subscriber table'); + +# Alter subscription two_phase to true +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy DISABLE"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_copy ENABLE"); + +# Wait for subscription startup +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy); + +# Make sure that the two-phase is enabled on the subscriber +$result = $node_subscriber->safe_psql('postgres', + "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';" +); +is($result, qq(e), 'two-phase should be enabled'); + $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;"); $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;"); @@ -374,8 +443,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;"); # check all the cleanup ############################### -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); - $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_subscription"); is($result, qq(0), 'check subscription was dropped on subscriber'); -- 2.43.0