public inbox for [email protected]  
help / color / mirror / Atom feed
From: Ajin Cherian <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: Давыдов Виталий <[email protected]>
Cc: Heikki Linnakangas <[email protected]>
Cc: [email protected] <[email protected]>
Subject: Re: Slow catchup of 2PC (twophase) transactions on replica in LR
Date: Thu, 18 Apr 2024 16:26:22 +1000
Message-ID: <CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com> (raw)
In-Reply-To: <CAA4eK1KNrm8xE-ZOAS249m+DOk=YQ_YaoM-GsZEvC8rTMtdW8w@mail.gmail.com>
References: <8fab8-65d74c80-1-2f28e880@39088166>
	<CAA4eK1KOs3s6syZqUgrd2WvjTz64SGf0ToZcRoPMCKKH+M0YFQ@mail.gmail.com>
	<99df2-65d8d200-1-63c26880@172687752>
	<CAA4eK1KtjjPxe5tjWeGNf5CfF9qfceVQCCC9XGsKzR-ifQD9uA@mail.gmail.com>
	<ba4b6-65ddc580-9-5c3e3080@134362348>
	<CAA4eK1J6wOMkDDt3iOs+JqkvQvuwSORosTpqdejgvXqFmxex=A@mail.gmail.com>
	<badcf-65dde500-3-24ba3c80@28790566>
	<cedbe-65e0c000-1-6db17700@133269862>
	<[email protected]>
	<f9f1b-65e72c80-7-533dda00@76203995>
	<CAFPTHDbU4EkVmKL66i5BzrmQ6LFftQPx_TZ1euKt9fvVzQDyjA@mail.gmail.com>
	<CAA4eK1+n7M2S1OpoGWDd+YZkDCuURMdVRbvP0eELQUvWgmDneg@mail.gmail.com>
	<CAFPTHDa=pJSZ_4dV5DPAOapRSgPcyyUTP0WzGY2Rz_D3-gwraw@mail.gmail.com>
	<CAA4eK1K1fSkeK=kc26G5cq87vQG4=1qs_b+no4+ep654SeBy1w@mail.gmail.com>
	<OSBPR01MB2552707A847936E6803CFAA5F5092@OSBPR01MB2552.jpnprd01.prod.outlook.com>
	<CAA4eK1KY=uwXXuVMtuNTYHGFbbXgDveoFoP3UbxNXqxCAx8GBQ@mail.gmail.com>
	<OSBPR01MB25528F4B0B8178D3AA8DE2BFF5082@OSBPR01MB2552.jpnprd01.prod.outlook.com>
	<CAA4eK1KNrm8xE-ZOAS249m+DOk=YQ_YaoM-GsZEvC8rTMtdW8w@mail.gmail.com>

On Tue, Apr 16, 2024 at 4:25 PM Amit Kapila <[email protected]> wrote:

>
> >
>
> Can you please once consider the idea shared by me at [1] (One naive
> idea is that on the publisher .....) to solve this problem?
>
> [1] -
> https://www.postgresql.org/message-id/CAA4eK1K1fSkeK%3Dkc26G5cq87vQG4%3D1qs_b%2Bno4%2Bep654SeBy1w%40...
>
>
>
Expanding on Amit's idea, we found out that there is already a mechanism in
code to fully decode prepared transactions prior to a defined LSN where
two_phase is enabled using the "two_phase_at" LSN in the slot. Look at
ReorderBufferFinishPrepared() on how this is done. This code was not
working as expected in our patch because
we were setting two_phase on the slot to true as soon as the alter command
was received. This was not the correct way, initially when two_phase is
enabled, the two_phase changes to pending state and two_phase option on the
slot should only be set to true when two_phase moves from pending to
enabled. This will happen once the replication is restarted with two_phase
option. Look at code in  CreateDecodingContext() on how "two_phase_at" is
set in the slot when done this way. So we changed the code to not remotely
alter two_phase when toggling from false to true. With this change, now
even if there are pending transactions on the publisher when toggling
two_phase from false to true, these pending transactions will be fully
decoded and sent once the commit prepared is decoded as the pending
prepared transactions are prior to the "two_phase_at" LSN. With this patch,
now we are able to handle both pending prepared transactions when altering
two_phase from true to false as well as false to true.

Attaching the patch for your review and comments. Big thanks to Kuroda-san
for also working on the patch.

regards,
Ajin Cherian
Fujitsu Australia.


Attachments:

  [application/octet-stream] v4-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch (8.0K, ../CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com/3-v4-0002-Alter-slot-option-two_phase-only-when-altering-tr.patch)
  download | inline diff:
From 5e4e79a5056bae374911d96eef6c4d945e23903e Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v4 2/3] Alter slot option two_phase only when altering true to
 false

---
 src/backend/commands/subscriptioncmds.c       | 21 +++++-
 .../libpqwalreceiver/libpqwalreceiver.c       | 21 ++++--
 src/include/replication/walreceiver.h         |  8 +--
 src/test/subscription/meson.build             |  1 +
 src/test/subscription/t/099_twophase_added.pl | 72 +++++++++++++++++++
 5 files changed, 111 insertions(+), 12 deletions(-)
 create mode 100644 src/test/subscription/t/099_twophase_added.pl

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 3299a60fff..0d80d6e110 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -850,7 +850,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 			else if (opts.slot_name &&
 					 (opts.failover || walrcv_server_version(wrconn) >= 170000))
 			{
-				walrcv_alter_slot(wrconn, opts.slot_name, opts.twophase, opts.failover);
+				walrcv_alter_slot(wrconn, opts.slot_name, NULL, &opts.failover);
 			}
 		}
 		PG_FINALLY();
@@ -1564,6 +1564,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 		bool		must_use_password;
 		char	   *err;
 		WalReceiverConn *wrconn;
+		bool		two_phase_needs_to_be_updated;
+		bool		failover_needs_to_be_updated;
 
 		/* Load the library providing us libpq calls. */
 		load_file("libpqwalreceiver", false);
@@ -1577,9 +1579,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					(errcode(ERRCODE_CONNECTION_FAILURE),
 					 errmsg("could not connect to the publisher: %s", err)));
 
+		/*
+		 * Consider which slot option must be altered.
+		 *
+		 * We must alter the failover option whenever subfailover is updated.
+		 * Two_phase, however, is altered only when changing true to false.
+		 */
+		two_phase_needs_to_be_updated =
+						(replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+						 !opts.twophase);
+		failover_needs_to_be_updated =
+								replaces[Anum_pg_subscription_subfailover - 1];
+
 		PG_TRY();
 		{
-			walrcv_alter_slot(wrconn, sub->slotname, opts.twophase, opts.failover);
+			if (two_phase_needs_to_be_updated || failover_needs_to_be_updated)
+				walrcv_alter_slot(wrconn, sub->slotname,
+								  two_phase_needs_to_be_updated ? &opts.twophase : NULL,
+								  failover_needs_to_be_updated ? &opts.failover : NULL);
 		}
 		PG_FINALLY();
 		{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index baef3bdec0..546b599848 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
 								  CRSSnapshotAction snapshot_action,
 								  XLogRecPtr *lsn);
 static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
-								bool two_phase, bool failover);
+								const bool *two_phase, const bool *failover);
 static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
 static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
 									   const char *query,
@@ -1121,16 +1121,25 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
  */
 static void
 libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
-					bool two_phase, bool failover)
+					const bool *two_phase, const bool *failover)
 {
 	StringInfoData cmd;
 	PGresult   *res;
 
 	initStringInfo(&cmd);
-	appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s, FAILOVER %s )",
-					 quote_identifier(slotname),
-					 two_phase ? "true" : "false",
-					 failover ? "true" : "false");
+	appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
+					 quote_identifier(slotname));
+
+	if (two_phase)
+		appendStringInfo(&cmd, "TWO_PHASE %s%s ",
+						 (*two_phase) ? "true" : "false",
+						 failover ? ", " : "");
+
+	if (failover)
+		appendStringInfo(&cmd, "FAILOVER %s ",
+						 (*failover) ? "true" : "false");
+
+	appendStringInfoString(&cmd, ");");
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
 	pfree(cmd.data);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index a443f402f5..f30637aa4a 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -372,13 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
 /*
  * walrcv_alter_slot_fn
  *
- * Change the definition of a replication slot. Currently, it only supports
- * changing the failover property of the slot.
+ * Change the definition of a replication slot. Currently, it supports
+ * changing the two_phase and the failover property of the slot.
  */
 typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
 									  const char *slotname,
-									  bool two_phase,
-									  bool failover);
+									  const bool *two_phase,
+									  const bool *failover);
 
 /*
  * walrcv_get_backend_pid_fn
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..b4bd522c3d 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
       't/031_column_list.pl',
       't/032_subscribe_use_index.pl',
       't/033_run_as_table_owner.pl',
+      't/099_twophase_added.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
new file mode 100644
index 0000000000..c13a37675a
--- /dev/null
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -0,0 +1,72 @@
+# Copyright (c) 2021-2024, PostgreSQL Global Development Group
+
+# Additional tests for altering two_phase option
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+	qq(max_prepared_transactions = 10));
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+	qq(max_prepared_transactions = 10));
+$node_subscriber->start;
+
+# Define pre-existing tables on both nodes
+$node_publisher->safe_psql('postgres',
+    "CREATE TABLE tab_full (a int PRIMARY KEY);");
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication, with two_phase = off
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub FOR ALL TABLES");
+
+$node_subscriber->safe_psql(
+	'postgres', "
+	CREATE SUBSCRIPTION sub
+	CONNECTION '$publisher_connstr' PUBLICATION pub
+	WITH (two_phase = off, copy_data = off)");
+
+######
+# Check the case that prepared transactions exist on publisher node
+######
+
+$node_publisher->safe_psql(
+	'postgres', "
+	BEGIN;
+	INSERT INTO tab_full VALUES (generate_series(1, 5));
+	PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('sub');
+
+my $result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "transaction is not prepared on subscriber");
+
+$node_subscriber->safe_psql(
+    'postgres', "
+    ALTER SUBSCRIPTION sub DISABLE;
+    ALTER SUBSCRIPTION sub SET (two_phase = on);
+    ALTER SUBSCRIPTION sub ENABLE;");
+
+$node_publisher->safe_psql( 'postgres',
+    "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) FROM tab_full;");
+is($result, q(5),
+   "prepared transactions done before altering can be replicated");
+
+done_testing();
-- 
2.43.0



  [application/octet-stream] v4-0003-Abort-prepared-transactions-while-altering-two_ph.patch (7.0K, ../CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com/4-v4-0003-Abort-prepared-transactions-while-altering-two_ph.patch)
  download | inline diff:
From 35d6a2a4f3ffe42aeebb56f219cd7004777baf10 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v4 3/3] Abort prepared transactions while altering two_phase
 to false

---
 doc/src/sgml/ref/alter_subscription.sgml      | 10 +++++-
 src/backend/access/transam/twophase.c         | 19 +++++-----
 src/backend/commands/subscriptioncmds.c       | 27 +++++++++++---
 src/include/access/twophase.h                 |  3 +-
 src/test/subscription/t/099_twophase_added.pl | 35 +++++++++++++++++++
 5 files changed, 77 insertions(+), 17 deletions(-)

diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 20b45e36e0..cfd2a18e2c 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -231,7 +231,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
       <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
       Only a superuser can set <literal>password_required = false</literal>.
-      <literal>two_phase</literal> can be altered only for disabled subscription.
      </para>
 
      <para>
@@ -253,6 +252,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>
       option is enabled.
      </para>
+
+     <para>
+      <literal>two_phase</literal> can be altered only for disabled
+      subscriptions. When altering the parameter from <literal>true</literal>
+      to <literal>false</literal>,  the backend process checks prepared
+      transactions done by the logical replication worker and aborts them. If
+      prepared transactions are found, the parameter cannot be altered to
+      <literal>false</literal> inside a transaction block.
+     </para>
     </listitem>
    </varlistentry>
 
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 495f99a357..9121195725 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2702,13 +2702,13 @@ checkGid(char *gid, Oid subid)
 }
 
 /*
- * LookupGXactBySubid
- *		Check if the prepared transaction done by apply worker exists.
+ * GetGidListBySubid
+ *      Get a list of GIDs which is PREPARE'd by the given subscription.
  */
-bool
-LookupGXactBySubid(Oid subid)
+List *
+GetGidListBySubid(Oid subid)
 {
-	bool		found = false;
+	List *list = NIL;
 
 	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 	for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
@@ -2717,11 +2717,10 @@ LookupGXactBySubid(Oid subid)
 
 		/* Ignore not-yet-valid GIDs. */
 		if (gxact->valid && checkGid(gxact->gid, subid))
-		{
-			found = true;
-			break;
-		}
+			list = lappend(list, pstrdup(gxact->gid));
+
 	}
 	LWLockRelease(TwoPhaseStateLock);
-	return found;
+
+	return list;
 }
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 0d80d6e110..cf3e5c64b0 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1178,6 +1178,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				/* XXX */
 				if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
 				{
+					List *prepared_xacts = NIL;
+
 					/*
 					 * two_phase can be only changed for disabled
 					 * subscriptions
@@ -1194,13 +1196,28 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					 */
 					logicalrep_workers_stop(subid);
 
-					/* Check whether the number of prepared transactions */
+					/*
+					 * If two phase was enabled, there is a possibility the
+					 * transactions has already been PREPARE'd.
+					 */
 					if (!opts.twophase &&
 						form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
-						LookupGXactBySubid(subid))
-						ereport(ERROR,
-								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-								 errmsg("cannot disable two_phase when uncommitted prepared transactions present")));
+						(prepared_xacts = GetGidListBySubid(subid)) != NIL)
+					{
+						ListCell	*cell;
+
+						/* Must not be in the transaction */
+						PreventInTransactionBlock(isTopLevel,
+												  "ALTER SUBSCRIPTION ... SET (two_phase = ...)");
+
+						/* Abort all listed transactions */
+						foreach(cell, prepared_xacts)
+						{
+							FinishPreparedTransaction((char *) lfirst(cell),
+													  false);
+							prepared_xacts = list_delete_cell(prepared_xacts, cell);
+						}
+					}
 
 					/* Change system catalog acoordingly */
 					values[Anum_pg_subscription_subtwophasestate - 1] =
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index d493ed24c5..95770bbd69 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "storage/lock.h"
+#include "nodes/pg_list.h"
 
 /*
  * GlobalTransactionData is defined in twophase.c; other places have no
@@ -63,6 +64,6 @@ extern void restoreTwoPhaseData(void);
 extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
 						TimestampTz origin_prepare_timestamp);
 
-extern bool LookupGXactBySubid(Oid subid);
+extern List *GetGidListBySubid(Oid subid);
 
 #endif							/* TWOPHASE_H */
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index c13a37675a..a8135b671c 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -69,4 +69,39 @@ $result = $node_subscriber->safe_psql('postgres',
 is($result, q(5),
    "prepared transactions done before altering can be replicated");
 
+######
+# Check the case that prepared transactions exist on subscriber node
+######
+
+$node_publisher->safe_psql(
+	'postgres', "
+	BEGIN;
+	INSERT INTO tab_full VALUES (generate_series(6, 10));
+	PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "transaction has been prepared on subscriber");
+
+$node_subscriber->safe_psql(
+    'postgres', "
+    ALTER SUBSCRIPTION sub DISABLE;
+    ALTER SUBSCRIPTION sub SET (two_phase = off);
+    ALTER SUBSCRIPTION sub ENABLE;");
+
+$result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "prepared transaction done by worker is aborted");
+
+$node_publisher->safe_psql( 'postgres',
+    "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('sub');
+
+$result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(10) FROM tab_full;");
+is($result, q(10),
+   "prepared transactions on publisher can be replicated");
+
 done_testing();
-- 
2.43.0



  [application/octet-stream] v4-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch (22.5K, ../CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.com/5-v4-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIPT.patch)
  download | inline diff:
From 8295540696e9699ac8191f1c2b0e09b6b684cb50 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v4 1/3] Allow altering of two_phase option of a SUBSCRIPTION

This patch allows user to alter two_phase option of a subscriber provided no uncommitted
prepared transactions are pending on that subscription.

Author: Cherian Ajin, Hayato Kuroda
---
 doc/src/sgml/ref/alter_subscription.sgml      |  6 +-
 src/backend/access/transam/twophase.c         | 43 ++++++++++++
 src/backend/commands/subscriptioncmds.c       | 58 +++++++++++-----
 .../libpqwalreceiver/libpqwalreceiver.c       |  7 +-
 src/backend/replication/logical/launcher.c    | 21 ++++++
 src/backend/replication/logical/worker.c      |  2 +-
 src/backend/replication/slot.c                | 19 +++++-
 src/backend/replication/walsender.c           | 20 ++++--
 src/bin/psql/tab-complete.c                   |  2 +-
 src/include/access/twophase.h                 |  3 +
 src/include/replication/slot.h                |  3 +-
 src/include/replication/walreceiver.h         |  5 +-
 src/include/replication/worker_internal.h     |  1 +
 src/test/regress/expected/subscription.out    |  5 +-
 src/test/regress/sql/subscription.sql         |  5 +-
 src/test/subscription/t/021_twophase.pl       | 67 ++++++++++++++++++-
 16 files changed, 227 insertions(+), 40 deletions(-)

diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 413ce68ce2..20b45e36e0 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -227,9 +227,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
       <link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
       <link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
-      <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and
-      <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+      <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+      <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+      <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
       Only a superuser can set <literal>password_required = false</literal>.
+      <literal>two_phase</literal> can be altered only for disabled subscription.
      </para>
 
      <para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 8090ac9fc1..495f99a357 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,46 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
 	LWLockRelease(TwoPhaseStateLock);
 	return found;
 }
+
+/*
+ * checkGid
+ */
+static bool
+checkGid(char *gid, Oid subid)
+{
+	int			ret;
+	Oid			subid_written,
+				xid;
+
+	ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid);
+
+	if (ret != 2 || subid != subid_written)
+		return false;
+
+	return true;
+}
+
+/*
+ * LookupGXactBySubid
+ *		Check if the prepared transaction done by apply worker exists.
+ */
+bool
+LookupGXactBySubid(Oid subid)
+{
+	bool		found = false;
+
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+
+		/* Ignore not-yet-valid GIDs. */
+		if (gxact->valid && checkGid(gxact->gid, subid))
+		{
+			found = true;
+			break;
+		}
+	}
+	LWLockRelease(TwoPhaseStateLock);
+	return found;
+}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 5a47fa984d..3299a60fff 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -16,6 +16,7 @@
 
 #include "access/htup_details.h"
 #include "access/table.h"
+#include "access/twophase.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
@@ -849,7 +850,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 			else if (opts.slot_name &&
 					 (opts.failover || walrcv_server_version(wrconn) >= 170000))
 			{
-				walrcv_alter_slot(wrconn, opts.slot_name, opts.failover);
+				walrcv_alter_slot(wrconn, opts.slot_name, opts.twophase, opts.failover);
 			}
 		}
 		PG_FINALLY();
@@ -1165,7 +1166,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 			{
 				supported_opts = (SUBOPT_SLOT_NAME |
 								  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
-								  SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
+								  SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+								  SUBOPT_DISABLE_ON_ERR |
 								  SUBOPT_PASSWORD_REQUIRED |
 								  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
 								  SUBOPT_ORIGIN);
@@ -1173,6 +1175,41 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
 
+				/* XXX */
+				if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+				{
+					/*
+					 * two_phase can be only changed for disabled
+					 * subscriptions
+					 */
+					if (form->subenabled)
+						ereport(ERROR,
+								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+								 errmsg("cannot set %s for enabled subscription",
+										"two_phase")));
+
+					/*
+					 * Stop all the subscription workers, just in case. Workers
+					 * may still survive even if the subscription is disabled.
+					 */
+					logicalrep_workers_stop(subid);
+
+					/* Check whether the number of prepared transactions */
+					if (!opts.twophase &&
+						form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+						LookupGXactBySubid(subid))
+						ereport(ERROR,
+								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+								 errmsg("cannot disable two_phase when uncommitted prepared transactions present")));
+
+					/* Change system catalog acoordingly */
+					values[Anum_pg_subscription_subtwophasestate - 1] =
+						CharGetDatum(opts.twophase ?
+									 LOGICALREP_TWOPHASE_STATE_PENDING :
+									 LOGICALREP_TWOPHASE_STATE_DISABLED);
+					replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+				}
+
 				if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
 				{
 					/*
@@ -1521,7 +1558,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 	 * doing the database operations we won't be able to rollback altered
 	 * slot.
 	 */
-	if (replaces[Anum_pg_subscription_subfailover - 1])
+	if (replaces[Anum_pg_subscription_subtwophasestate - 1] ||
+		replaces[Anum_pg_subscription_subfailover - 1])
 	{
 		bool		must_use_password;
 		char	   *err;
@@ -1541,7 +1579,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 		PG_TRY();
 		{
-			walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+			walrcv_alter_slot(wrconn, sub->slotname, opts.twophase, opts.failover);
 		}
 		PG_FINALLY();
 		{
@@ -1578,7 +1616,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	char	   *subname;
 	char	   *conninfo;
 	char	   *slotname;
-	List	   *subworkers;
 	ListCell   *lc;
 	char		originname[NAMEDATALEN];
 	char	   *err = NULL;
@@ -1688,16 +1725,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	 * New workers won't be started because we hold an exclusive lock on the
 	 * subscription till the end of the transaction.
 	 */
-	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-	subworkers = logicalrep_workers_find(subid, false);
-	LWLockRelease(LogicalRepWorkerLock);
-	foreach(lc, subworkers)
-	{
-		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
-
-		logicalrep_worker_stop(w->subid, w->relid);
-	}
-	list_free(subworkers);
+	logicalrep_workers_stop(subid);
 
 	/*
 	 * Remove the no-longer-useful entry in the launcher's table of apply
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 3c2b1bb496..baef3bdec0 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
 								  CRSSnapshotAction snapshot_action,
 								  XLogRecPtr *lsn);
 static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
-								bool failover);
+								bool two_phase, bool failover);
 static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
 static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
 									   const char *query,
@@ -1121,14 +1121,15 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
  */
 static void
 libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
-					bool failover)
+					bool two_phase, bool failover)
 {
 	StringInfoData cmd;
 	PGresult   *res;
 
 	initStringInfo(&cmd);
-	appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+	appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( TWO_PHASE %s, FAILOVER %s )",
 					 quote_identifier(slotname),
+					 two_phase ? "true" : "false",
 					 failover ? "true" : "false");
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 66070e9131..94b73f3085 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,27 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 	LWLockRelease(LogicalRepWorkerLock);
 }
 
+/*
+ * Stop all the subscription workers.
+ */
+void
+logicalrep_workers_stop(Oid subid)
+{
+	List	   *subworkers;
+	ListCell   *lc;
+
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+	subworkers = logicalrep_workers_find(subid, false);
+	LWLockRelease(LogicalRepWorkerLock);
+	foreach(lc, subworkers)
+	{
+		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+		logicalrep_worker_stop(w->subid, w->relid);
+	}
+	list_free(subworkers);
+}
+
 /*
  * Stop the given logical replication parallel apply worker.
  *
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..374aa22091 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3911,7 +3911,7 @@ maybe_reread_subscription(void)
 	/* !slotname should never happen when enabled is true. */
 	Assert(newsub->slotname);
 
-	/* two-phase should not be altered */
+	/* two-phase should not be altered while the worker exists */
 	Assert(newsub->twophasestate == MySubscription->twophasestate);
 
 	/*
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index cebf44bb0f..621f35ab1e 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -800,8 +800,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
  * Change the definition of the slot identified by the specified name.
  */
 void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool two_phase, bool failover)
 {
+	bool		update_slot = false;
+
 	Assert(MyReplicationSlot == NULL);
 
 	ReplicationSlotAcquire(name, false);
@@ -844,12 +846,27 @@ ReplicationSlotAlter(const char *name, bool failover)
 				errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				errmsg("cannot enable failover for a temporary replication slot"));
 
+	if (MyReplicationSlot->data.two_phase != two_phase)
+	{
+		SpinLockAcquire(&MyReplicationSlot->mutex);
+		MyReplicationSlot->data.two_phase = two_phase;
+		SpinLockRelease(&MyReplicationSlot->mutex);
+
+		update_slot = true;
+	}
+
+
 	if (MyReplicationSlot->data.failover != failover)
 	{
 		SpinLockAcquire(&MyReplicationSlot->mutex);
 		MyReplicationSlot->data.failover = failover;
 		SpinLockRelease(&MyReplicationSlot->mutex);
 
+		update_slot = true;
+	}
+
+	if (update_slot)
+	{
 		ReplicationSlotMarkDirty();
 		ReplicationSlotSave();
 	}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index bc40c454de..be155067ce 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,14 +1411,25 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
  * Process extra options given to ALTER_REPLICATION_SLOT.
  */
 static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
+						  bool *two_phase, bool *failover)
 {
+	bool		two_phase_given = false;
 	bool		failover_given = false;
 
 	/* Parse options */
 	foreach_ptr(DefElem, defel, cmd->options)
 	{
-		if (strcmp(defel->defname, "failover") == 0)
+		if (strcmp(defel->defname, "two_phase") == 0)
+		{
+			if (two_phase_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			two_phase_given = true;
+			*two_phase = defGetBoolean(defel);
+		}
+		else if (strcmp(defel->defname, "failover") == 0)
 		{
 			if (failover_given)
 				ereport(ERROR,
@@ -1438,10 +1449,11 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
 static void
 AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
 {
+	bool		two_phase = false;
 	bool		failover = false;
 
-	ParseAlterReplSlotOptions(cmd, &failover);
-	ReplicationSlotAlter(cmd->slotname, failover);
+	ParseAlterReplSlotOptions(cmd, &two_phase, &failover);
+	ReplicationSlotAlter(cmd->slotname, two_phase, failover);
 }
 
 /*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 6fee3160f0..5ff84301cd 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end)
 	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
 		COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
 					  "password_required", "run_as_owner", "slot_name",
-					  "streaming", "synchronous_commit");
+					  "streaming", "synchronous_commit", "two_phase");
 	/* ALTER SUBSCRIPTION <name> SKIP ( */
 	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
 		COMPLETE_WITH("lsn");
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 56248c0006..d493ed24c5 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -62,4 +62,7 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
 extern void restoreTwoPhaseData(void);
 extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
 						TimestampTz origin_prepare_timestamp);
+
+extern bool LookupGXactBySubid(Oid subid);
+
 #endif							/* TWOPHASE_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 7b937d1a0c..2fcb11418f 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
 extern void ReplicationSlotPersist(void);
 extern void ReplicationSlotDrop(const char *name, bool nowait);
 extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool two_phase,
+								 bool failover);
 
 extern void ReplicationSlotAcquire(const char *name, bool nowait);
 extern void ReplicationSlotRelease(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 12f71fa99b..a443f402f5 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -377,6 +377,7 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
  */
 typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
 									  const char *slotname,
+									  bool two_phase,
 									  bool failover);
 
 /*
@@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 	WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
 #define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
 	WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
-#define walrcv_alter_slot(conn, slotname, failover) \
-	WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
+#define walrcv_alter_slot(conn, slotname, two_phase, failover) \
+	WalReceiverFunctions->walrcv_alter_slot(conn, slotname, two_phase, failover)
 #define walrcv_get_backend_pid(conn) \
 	WalReceiverFunctions->walrcv_get_backend_pid(conn)
 #define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 515aefd519..d5428263c1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -246,6 +246,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_workers_stop(Oid subid);
 extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 1eee6b17b8..9bba656e00 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -379,10 +379,7 @@ HINT:  To initiate replication, you must manually create the replication slot, e
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-ERROR:  unrecognized subscription parameter: "two_phase"
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 \dRs+
                                                                                                                 List of subscriptions
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 1b2a23ba7b..9ff151f806 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -257,10 +257,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
 
 \dRs+
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 
 \dRs+
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 9437cd4c3b..e710f3c4c0 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,71 @@ $result =
   $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
 is($result, qq(2), 'replicated data in subscriber table');
 
+# Disable the subscription and alter it to two_phase = false,
+# verify that the altered subscription reflects the two_phase option.
+
+# Alter subscription two_phase to false
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)");
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is disabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+       "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(d), 'two-phase is disabled');
+
+# Now do a prepare on publisher and make sure that it is not replicated.
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_publisher->safe_psql(
+       'postgres', qq{
+    BEGIN;
+    INSERT INTO tab_copy VALUES (100);
+    PREPARE TRANSACTION 'newgid';
+	});
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure that there is 0 prepared transaction on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(0), 'transaction is prepared on subscriber');
+
+# Now commit the insert and verify that it IS replicated
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';");
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Made sure that the commited transaction is replicated.
+$result =
+	$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
+is($result, qq(3), 'replicated data in subscriber table');
+
+# Alter subscription two_phase to true
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)");
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is enabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(e), 'two-phase is disabled');
+
 $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;");
 $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
 
@@ -374,8 +439,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
 # check all the cleanup
 ###############################
 
-$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
-
 $result = $node_subscriber->safe_psql('postgres',
 	"SELECT count(*) FROM pg_subscription");
 is($result, qq(0), 'check subscription was dropped on subscriber');
-- 
2.43.0



view thread (11+ messages)  latest in thread

reply

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Reply to all the recipients using the --to and --cc options:
  reply via email

  To: [email protected]
  Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
  Subject: Re: Slow catchup of 2PC (twophase) transactions on replica in LR
  In-Reply-To: <CAFPTHDaz4cyRN3NbjYtkqCKFj7Dv1u0Z8wviY8fnJW4+0CkF_Q@mail.gmail.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