From a6fd142f9d7b8331c93e1cfdc2ec3a1de4c1ad36 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Tue, 23 Mar 2021 20:27:19 +1100
Subject: [PATCH v66] Fix to allow REFRESH PUBLICATION for two_phase PENDING.

* pg_dump of subscription with two_phase will restore with subscription
  having tristate PENDING but still needed to allow REFRESH PUBLICATION
  because it gets restored with no tables.

* Also, if there are 0 tables in the subscription do not transition from
  PENDING to ENABLED because that will prevent using ALTER SUBSCRIPTION
  REFRESH PUBLICATION with no way to add tables.

* Includes minor updates to PG docs.
---
 doc/src/sgml/ref/alter_subscription.sgml    | 12 ++++++----
 src/backend/commands/subscriptioncmds.c     |  4 ++--
 src/backend/replication/logical/tablesync.c | 37 ++++++++++++++++++++++++-----
 src/backend/replication/logical/worker.c    | 24 ++++++++++++++++---
 4 files changed, 61 insertions(+), 16 deletions(-)

diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 85cc8bb..d5e891b 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -62,11 +62,13 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
   </para>
 
   <para>
-   Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command> and
-   <command>ALTER SUBSCRIPTION ... SET PUBLICATION ...</command> with refresh
-   option as true cannot be executed inside a transaction block. They also
-   cannot be executed with <literal>copy_data = true</literal> if the
-   subscription is using <literal>two_phase</literal> commit.
+   Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>, and
+   <command>ALTER SUBSCRIPTION ... SET PUBLICATION ...</command> with
+   <literal>refresh = true</literal> cannot be executed inside a transaction
+   block. They also cannot be executed with <literal>copy_data = true</literal>
+   when the subscription has <literal>two_phase</literal> commit enabled. See
+   column <literal>subtwophasestate</literal> of
+   <xref linkend="catalog-pg-subscription"/> to know the actual two-phase state.
   </para>
  </refsect1>
 
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 38863bd..89abc81 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1026,7 +1026,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt, bool isTopLevel)
 					 * See ALTER_SUBSCRIPTION_REFRESH for details why this is
 					 * not allowed.
 					 */
-					if (sub->twophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED && copy_data)
+					if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && copy_data)
 						ereport(ERROR,
 								(errcode(ERRCODE_SYNTAX_ERROR),
 								 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
@@ -1084,7 +1084,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt, bool isTopLevel)
 				 *
 				 * For more details see comments atop worker.c.
 				 */
-				if (sub->twophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED && copy_data)
+				if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && copy_data)
 					ereport(ERROR,
 							(errcode(ERRCODE_SYNTAX_ERROR),
 							 errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when two_phase is enabled"),
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index b4f454f..679b73f 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -409,11 +409,30 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	{
 		if (AllTablesyncsReady())
 		{
-			ereport(LOG,
-					(errmsg("logical replication apply worker for subscription \"%s\" will restart so two_phase can be enabled",
-							MySubscription->name)));
+			List *subrels = NIL;
+			bool become_two_phase_enabled = false;
 
-			proc_exit(0);
+			if (!started_tx)
+			{
+				StartTransactionCommand();
+				started_tx = true;
+			}
+			subrels = GetSubscriptionRelations(MySubscription->oid);
+
+			/*
+			 * If there are no tables then leave the state as PENDING, which
+			 * allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to work.
+			 */
+			become_two_phase_enabled = list_length(subrels) > 0;
+
+			if (become_two_phase_enabled)
+			{
+				ereport(LOG,
+						(errmsg("logical replication apply worker for subscription \"%s\" will restart so two_phase can be enabled",
+								MySubscription->name)));
+
+				proc_exit(0);
+			}
 		}
 	}
 
@@ -1159,8 +1178,11 @@ FetchTableStates(bool *started_tx)
 		list_free_deep(table_states_not_ready);
 		table_states_not_ready = NIL;
 
-		StartTransactionCommand();
-		*started_tx = true;
+		if (!IsTransactionState())
+		{
+			StartTransactionCommand();
+			*started_tx = true;
+		}
 
 		/* Fetch all non-ready tables. */
 		rstates = GetSubscriptionNotReadyRelations(MySubscription->oid);
@@ -1181,6 +1203,9 @@ FetchTableStates(bool *started_tx)
 
 /*
  * Are all tablesyncs READY?
+ *
+ * Note: This function is not suitable to be called from outside of apply or
+ * tablesync workers because MySubscription needs to be already initialized.
  */
 bool
 AllTablesyncsReady(void)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 43304e0..c27db32 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3377,15 +3377,33 @@ ApplyWorkerMain(Datum main_arg)
 
 	if (!am_tablesync_worker())
 	{
+		bool	become_two_phase_enabled = false;
+
 		/*
 		 * Even when the two_phase mode is requested by the user, it remains
 		 * as the tri-state PENDING until all tablesyncs have reached READY
 		 * state. Only then, can it become properly ENABLED.
 		 */
-		bool		all_tables_ready = AllTablesyncsReady();
+		if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
+		{
+			if (AllTablesyncsReady())
+			{
+				List *subrels = NIL;
+
+				StartTransactionCommand();
+				subrels = GetSubscriptionRelations(MySubscription->oid);
+				CommitTransactionCommand();
+
+				/*
+				 * If there are no tables then leave the state as PENDING,
+				 * otherwise ALTER SUBSCRIPTION ... REFRESH PUBLICATION is
+				 * not allowed.
+				 */
+				become_two_phase_enabled = list_length(subrels) > 0;
+			}
+		}
 
-		if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
-			all_tables_ready)
+		if (become_two_phase_enabled)
 		{
 			/* Start streaming with two_phase enabled */
 			options.proto.logical.twophase = true;
-- 
1.8.3.1

