From 6f1bb6bd1512c119a02880eca0b3c3ee47e7847a Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Sat, 20 Mar 2021 12:18:15 +1100
Subject: [PATCH v65 3/3] Fix apply worker (dev logs)

NOT TO BE COMMITTED.

This patch is only for adding some developer logging which may help for
debugging/testing the patch.
---
 src/backend/replication/logical/tablesync.c | 38 +++++++++++++++++++++++++++++
 src/backend/replication/logical/worker.c    |  4 +--
 2 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 233649e..6858dc3 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -407,6 +407,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	 */
 	if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
 	{
+		elog(LOG, "!!> two_phase enable is still pending");
 		if (AllTablesyncsReady())
 		{
 			ereport(LOG,
@@ -1155,6 +1156,8 @@ FetchTableStates(bool *started_tx)
 		ListCell   *lc;
 		SubscriptionRelState *rstate;
 
+		elog(LOG, "!!> FetchTableStates: Re-fetching the state list caches");
+
 		/* Clean the old lists. */
 		list_free_deep(table_states_not_ready);
 		table_states_not_ready = NIL;
@@ -1172,11 +1175,16 @@ FetchTableStates(bool *started_tx)
 			rstate = palloc(sizeof(SubscriptionRelState));
 			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
 			table_states_not_ready = lappend(table_states_not_ready, rstate);
+			elog(LOG, "!!> FetchTableStates: table_states_not_ready - added Table relid %u with state '%c'", rstate->relid, rstate->state);
 		}
 		MemoryContextSwitchTo(oldctx);
 
 		table_states_valid = true;
 	}
+	else
+	{
+		elog(LOG, "!!> FetchTableStates: Already up-to-date");
+	}
 }
 
 /*
@@ -1190,6 +1198,8 @@ AllTablesyncsReady(void)
 	int			count = 0;
 	ListCell   *lc;
 
+	elog(LOG, "!!> AllTablesyncsReady");
+
 	/* We need up-to-date sync state info for subscription tables here. */
 	FetchTableStates(&started_tx);
 
@@ -1201,6 +1211,12 @@ AllTablesyncsReady(void)
 		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 
 		count++;
+		elog(LOG,
+			 "!!> AllTablesyncsReady: #%d. Table relid %u has state '%c'",
+			 count,
+			 rstate->relid,
+			 rstate->state);
+
 		/*
 		 * When the process_syncing_tables_for_apply changes the state from
 		 * SYNCDONE to READY, that change is actually written directly into
@@ -1221,6 +1237,7 @@ AllTablesyncsReady(void)
 		 */
 		if (rstate->state != SUBREL_STATE_READY)
 		{
+			elog(LOG, "!!> AllTablesyncsReady: Table relid %u is busy!", rstate->relid);
 			found_busy = true;
 			break;
 		}
@@ -1232,6 +1249,11 @@ AllTablesyncsReady(void)
 		pgstat_report_stat(false);
 	}
 
+	elog(LOG,
+		 "!!> AllTablesyncsReady: Scanned %d tables, and found busy = %s",
+		 count,
+		 found_busy ? "true" : "false");
+
 	/* When no tablesyncs are busy, then all are READY */
 	return !found_busy;
 }
@@ -1279,4 +1301,20 @@ UpdateTwoPhaseState(char new_state)
 	table_close(rel, RowExclusiveLock);
 
 	CommitTransactionCommand();
+
+#if 1
+	/* This is just debugging, for confirmation the update worked. */
+	{
+		Subscription *new_s;
+
+		StartTransactionCommand();
+		new_s = GetSubscription(MySubscription->oid, false);
+		elog(LOG,
+			 "!!> 2PC Tri-state for \"%s\": '%c' ==> '%c'",
+			 MySubscription->name,
+			 MySubscription->twophasestate,
+			 new_s->twophasestate);
+		CommitTransactionCommand();
+	}
+#endif
 }
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 1768268..5e6bbc7 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3399,8 +3399,8 @@ ApplyWorkerMain(Datum main_arg)
 			walrcv_startstreaming(wrconn, &options);
 		}
 
-		ereport(DEBUG1,
-			(errmsg("logical replication apply worker for subscription \"%s\" two_phase is %s.",
+		ereport(LOG,
+			(errmsg("!!> logical replication apply worker for subscription \"%s\" two_phase is %s.",
 			MySubscription->name,
 			MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
 			MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
-- 
1.8.3.1

