v48-0009-Fix-apply-worker-empty-prepare-dev-logs.patch

From 5d0c4bbe6459e13b6336d6934ba231b6e1bba928 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 4 Mar 2021 20:12:24 +1100
Subject: [PATCH v48] Fix apply worker empty prepare (dev logs).

NOT TO BE COMMITTED.

This patch is only for adding some developer logging which may help for debugging/testing the "Fix apply worker empty prepare" patch.
---
 src/backend/replication/logical/tablesync.c | 29 ++++++++---
 src/backend/replication/logical/worker.c    | 77 ++++++++++++++++++++++-------
 2 files changed, 83 insertions(+), 23 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 97fc399..f3984d4 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -282,6 +282,12 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 {
 	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
 
+	elog(LOG,
+		 "!!> process_syncing_tables_for_sync: state = '%c', current_lsn = %X/%X, relstate_lsn = %X/%X",
+		 MyLogicalRepWorker->relstate,
+		 LSN_FORMAT_ARGS(current_lsn),
+		 LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn));
+
 	if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
 		current_lsn >= MyLogicalRepWorker->relstate_lsn)
 	{
@@ -1127,6 +1133,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_all);
 		table_states_all = NIL;
@@ -1149,6 +1157,7 @@ FetchTableStates(bool *started_tx)
 			rstate = palloc(sizeof(SubscriptionRelState));
 			memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState));
 			table_states_all = lappend(table_states_all, rstate);
+			elog(LOG, "!!> FetchTableStates: table_states_all - added Table relid %u with state '%c'", rstate->relid, rstate->state);
 
 			/* List of only not-ready states */
 			if (cur_rstate->state != SUBREL_STATE_READY)
@@ -1156,12 +1165,17 @@ FetchTableStates(bool *started_tx)
 				rstate = palloc(sizeof(SubscriptionRelState));
 				memcpy(rstate, cur_rstate, 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");
+	}
 }
 
 /*
@@ -1175,6 +1189,8 @@ AnyTablesyncInProgress()
 	int			count = 0;
 	ListCell   *lc;
 
+	elog(LOG, "!!> AnyTablesyncInProgress?");
+
 	/* We need up-to-date sync state info for subscription tables here. */
 	FetchTableStates(&started_tx);
 
@@ -1186,8 +1202,8 @@ AnyTablesyncInProgress()
 		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 
 		count++;
-		elog(DEBUG1,
-			 "AnyTablesyncInProgress?: #%d. Table relid %u has state '%c'",
+		elog(LOG,
+			 "!!> AnyTablesyncInProgress?: #%d. Table relid %u has state '%c'",
 			 count,
 			 rstate->relid,
 			 rstate->state);
@@ -1204,6 +1220,7 @@ AnyTablesyncInProgress()
 		if (rstate->state != SUBREL_STATE_SYNCDONE &&
 			rstate->state != SUBREL_STATE_READY)
 		{
+			elog(LOG, "!!> AnyTablesyncInProgress?: Table relid %u is busy!", rstate->relid);
 			found_busy = true;
 			break;
 		}
@@ -1215,8 +1232,8 @@ AnyTablesyncInProgress()
 		pgstat_report_stat(false);
 	}
 
-	elog(DEBUG1,
-		 "AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s",
+	elog(LOG,
+		 "!!> AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s",
 		 count,
 		 found_busy ? "true" : "false");
 
@@ -1242,8 +1259,8 @@ BiggestTablesyncLSN()
 			biggest_lsn = rstate->lsn;
 	}
 
-	elog(DEBUG1,
-		 "BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X",
+	elog(LOG,
+		 "!!> BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X",
 		 count,
 		 LSN_FORMAT_ARGS(biggest_lsn));
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f753e33..3c71f54 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -815,14 +815,16 @@ apply_handle_begin_prepare(StringInfo s)
 		 * Make sure every tablesync has reached at least SYNCDONE state
 		 * before letting the apply worker proceed.
 		 */
-		elog(DEBUG1,
-			 "apply_handle_begin_prepare, end_lsn = %X/%X, final_lsn = %X/%X, lstate_lsn = %X/%X",
+		elog(LOG,
+			 "!!> apply_handle_begin_prepare, end_lsn = %X/%X, final_lsn = %X/%X, lstate_lsn = %X/%X",
 			 LSN_FORMAT_ARGS(begin_data.end_lsn),
 			 LSN_FORMAT_ARGS(begin_data.final_lsn),
 			 LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn));
 
 		while (AnyTablesyncInProgress())
 		{
+			elog(LOG, "!!> apply_handle_begin_prepare - waiting for all sync workers to be DONE/READY");
+
 			process_syncing_tables(begin_data.final_lsn);
 
 			/* This latch is to prevent 100% CPU looping. */
@@ -840,7 +842,12 @@ apply_handle_begin_prepare(StringInfo s)
 		 * prepared) will be saved to a spoolfile for replay later at
 		 * commit_prepared time.
 		 */
-		if (begin_data.final_lsn < BiggestTablesyncLSN())
+		if (begin_data.final_lsn < BiggestTablesyncLSN()
+#if 0
+			|| true				/* XXX - Add this line to force psf (for
+								 * easier debugging) */
+#endif
+			)
 		{
 			char		psfpath[MAXPGPATH];
 			StringInfoData sid;
@@ -908,6 +915,8 @@ apply_handle_prepare(StringInfo s)
 		 * removal on proc-exit just in case there is an unexpected restart
 		 * between now and when commit_prepared happens.
 		 */
+		elog(LOG,
+			"!!> apply_handle_prepare: Make sure the spoolfile is not removed on proc-exit");
 		hentry = (PsfHashEntry *) hash_search(psf_hash,
 											  psf_cur.name,
 											  HASH_FIND,
@@ -990,6 +999,8 @@ apply_handle_commit_prepared(StringInfo s)
 		int			nchanges;
 		LogicalRepPreparedTxnData pdata;
 
+		elog(LOG, "!!> apply_handle_commit_prepared: replaying the spooled messages");
+
 		/*
 		 * 1. replay the spooled messages
 		 */
@@ -997,8 +1008,8 @@ apply_handle_commit_prepared(StringInfo s)
 		ensure_transaction();
 
 		nchanges = prepare_spoolfile_replay_messages(psfpath, &pdata);
-		elog(DEBUG1,
-			 "apply_handle_commit_prepared: replayed %d (all) changes.",
+		elog(LOG,
+			 "!!> apply_handle_commit_prepared: replayed %d (all) changes.",
 			 nchanges);
 
 		/*
@@ -1103,6 +1114,7 @@ apply_handle_rollback_prepared(StringInfo s)
 	 * Prepare Spoolfile (using_psf) because in that case there is no matching
 	 * PrepareTransactionBlock done yet.
 	 */
+	elog(LOG, "!!> apply_handle_rollback_prepared: using_psf = %d", using_psf);
 	if (!using_psf &&
 		LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn,
 					rollback_data.preparetime))
@@ -2493,18 +2505,22 @@ apply_dispatch(StringInfo s)
 			return;
 
 		case LOGICAL_REP_MSG_BEGIN_PREPARE:
+			elog(LOG, "!!> ------ apply_handle_begin_prepare ------");
 			apply_handle_begin_prepare(s);
 			return;
 
 		case LOGICAL_REP_MSG_PREPARE:
+			elog(LOG, "!!> ------ apply_handle_prepare ------");
 			apply_handle_prepare(s);
 			return;
 
 		case LOGICAL_REP_MSG_COMMIT_PREPARED:
+			elog(LOG, "!!> ------ apply_handle_commit_prepared ------");
 			apply_handle_commit_prepared(s);
 			return;
 
 		case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
+			elog(LOG, "!!> ------ apply_handle_rollback_prepared ------");
 			apply_handle_rollback_prepared(s);
 			return;
 
@@ -3706,8 +3722,8 @@ IsLogicalWorker(void)
 static bool
 prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s)
 {
-	elog(DEBUG1,
-		 "prepare_spoolfile_handler for action '%c'. %s write to spool file",
+	elog(LOG,
+		 "!!> prepare_spoolfile_handler for action '%c'. %s write to spool file",
 		 action,
 		 psf_cur.is_spooling ? "Do" : "Don't");
 
@@ -3731,7 +3747,7 @@ prepare_spoolfile_create(char *path)
 	bool		found;
 	PsfHashEntry *hentry;
 
-	elog(DEBUG1, "creating file \"%s\" for prepare changes", path);
+	elog(LOG, "!!> creating file \"%s\" for prepare changes", path);
 
 	Assert(!psf_cur.is_spooling);
 
@@ -3743,7 +3759,7 @@ prepare_spoolfile_create(char *path)
 
 	if (!found)
 	{
-		elog(DEBUG1, "Not found file \"%s\". Create it.", path);
+		elog(LOG, "!!> Not found file \"%s\". Create it.", path);
 		psf_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY);
 		if (psf_cur.vfd < 0)
 		{
@@ -3761,7 +3777,7 @@ prepare_spoolfile_create(char *path)
 		 * Open the file and seek to the beginning because we always want to
 		 * create/overwrite this file.
 		 */
-		elog(DEBUG1, "Found file \"%s\". Overwrite it.", path);
+		elog(LOG, "!!> Found file \"%s\". Overwrite it.", path);
 		psf_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY);
 		if (psf_cur.vfd < 0)
 		{
@@ -3786,6 +3802,7 @@ prepare_spoolfile_create(char *path)
 static void
 prepare_spoolfile_close()
 {
+	elog(LOG, "!!> prepare_spoolfile_close");
 	if (psf_cur.vfd >= 0)
 		FileClose(psf_cur.vfd);
 
@@ -3801,6 +3818,8 @@ prepare_spoolfile_close()
 static void
 prepare_spoolfile_delete(char *path)
 {
+	elog(LOG, "!!> prepare_spoolfile_delete: \"%s\"", path);
+
 	/* The current psf should be closed already, but make sure anyway. */
 	prepare_spoolfile_close();
 
@@ -3826,18 +3845,20 @@ prepare_spoolfile_write(char action, StringInfo s)
 
 	Assert(psf_cur.is_spooling);
 
-	elog(DEBUG1, "prepare_spoolfile_write: writing action '%c'", action);
+	elog(LOG, "!!> prepare_spoolfile_write: writing action '%c'", action);
 
 	/* total on-disk size, including the action type character */
 	len = (s->len - s->cursor) + sizeof(char);
 
 	/* first write the size */
+	elog(LOG, "!!> prepare_spoolfile_write: A writing len bytes = %d", len);
 	bytes_written = FileWrite(psf_cur.vfd, (char *)&len, sizeof(len),
 							  psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE);
 	Assert(bytes_written == sizeof(len));
 	psf_cur.cur_offset += bytes_written;
 
 	/* then the action */
+	elog(LOG, "!!> prepare_spoolfile_write: B writing action = %c, %d bytes", action, (int)sizeof(action));
 	bytes_written = FileWrite(psf_cur.vfd, &action, sizeof(action),
 							  psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE);
 	Assert(bytes_written == sizeof(action));
@@ -3846,6 +3867,7 @@ prepare_spoolfile_write(char action, StringInfo s)
 	/* and finally the remaining part of the buffer (after the XID) */
 	len = (s->len - s->cursor);
 
+	elog(LOG, "!!> prepare_spoolfile_write: C writing len bytes = %d", len);
 	bytes_written = FileWrite(psf_cur.vfd, &s->data[s->cursor], len,
 							  psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE);
 	Assert(bytes_written == len);
@@ -3879,6 +3901,12 @@ prepare_spoolfile_exists(char *path)
 		if (fd >= 0)
 			FileClose(fd);
 
+		elog(LOG,
+			 "!!> prepare_spoolfile_exists: Prepared spoolfile \"%s\" was "
+			 "not found in the HTAB, but was %s on the disk.",
+			 path,
+			 found ? "found" : "not found");
+
 		/*
 		 * And if it was found on disk then create the HTAB entry for it.
 		 */
@@ -3888,10 +3916,16 @@ prepare_spoolfile_exists(char *path)
 										  path,
 										  HASH_ENTER,
 										  NULL);
+			elog(LOG, "!!> prepare_spoolfile_exists: Created new HTAB entry '%s'", hentry->name);
 			hentry->allow_delete = false;
 		}
 	}
 
+	elog(LOG,
+		 "!!> prepare_spoolfile_exists: Prepared spoolfile \"%s\" was %s",
+		 path,
+		 found ? "found" : "not found");
+
 	return found;
 }
 
@@ -3910,8 +3944,8 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 				oldctx2;
 	PsfFile		psf = { .is_spooling = false, .vfd = -1, .cur_offset = 0 };
 
-	elog(DEBUG1,
-		 "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"",
+	elog(LOG,
+		 "!!> prepare_spoolfile_replay_messages: replaying changes from file \"%s\"",
 		 path);
 
 	/*
@@ -3948,6 +3982,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 		nbytes = FileRead(psf.vfd, (char *) &len, sizeof(len),
 						  psf.cur_offset, WAIT_EVENT_DATA_FILE_READ);
 		psf.cur_offset += nbytes;
+		elog(LOG, "!!> prepare_spoolfile_replay_messages: nbytes = %d, len = %d", nbytes, len);
 
 		/* have we reached end of the file? */
 		if (nbytes == 0)
@@ -3966,6 +4001,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 		buffer = repalloc(buffer, len);
 
 		/* and finally read the data into the buffer */
+		elog(LOG, "!!> prepare_spoolfile_replay_messages: read %d bytes into buffer", len);
 		nbytes = FileRead(psf.vfd, buffer, len,
 						  psf.cur_offset, WAIT_EVENT_DATA_FILE_READ);
 		psf.cur_offset += nbytes;
@@ -4000,13 +4036,15 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 		{
 			LogicalRepBeginPrepareData bdata;
 
+			elog(LOG, "!!> prepare_spoolfile_replay_messages: Found the BEGIN_PREPARE info");
+
 			/* read/skip the action byte. */
 			Assert(pq_getmsgbyte(&s2) == LOGICAL_REP_MSG_BEGIN_PREPARE);
 
 			/* read the begin_data. */
 			logicalrep_read_begin_prepare(&s2, &bdata);
 
-			elog(DEBUG1, "BEGIN_PREPARE info: gid = '%s', final_lsn = %X/%X, end_lsn = %X/%X",
+			elog(LOG, "!!> BEGIN_PREPARE info: gid = '%s', final_lsn = %X/%X, end_lsn = %X/%X",
 				 bdata.gid,
 				 LSN_FORMAT_ARGS(bdata.final_lsn),
 				 LSN_FORMAT_ARGS(bdata.end_lsn));
@@ -4021,13 +4059,15 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 		}
 		else if (s2.data[0] == LOGICAL_REP_MSG_PREPARE)
 		{
+			elog(LOG, "!!> prepare_spoolfile_replay_messages: Found the PREPARE info");
+
 			/* read/skip the action byte. */
 			Assert(pq_getmsgbyte(&s2) == LOGICAL_REP_MSG_PREPARE);
 
 			/* read and return the prepare_data info to the caller */
 			logicalrep_read_prepare(&s2, pdata);
 
-			elog(DEBUG1, "PREPARE info: gid = '%s', prepare_lsn = %X/%X, end_lsn = %X/%X",
+			elog(LOG, "!!> PREPARE info: gid = '%s', prepare_lsn = %X/%X, end_lsn = %X/%X",
 				 pdata->gid,
 				 LSN_FORMAT_ARGS(pdata->prepare_lsn),
 				 LSN_FORMAT_ARGS(pdata->end_lsn));
@@ -4046,7 +4086,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 			nchanges++;
 
 			if (nchanges % 1000 == 0)
-				elog(DEBUG1, "replayed %d changes from file '%s'",
+				elog(LOG, "!!> replayed %d changes from file '%s'",
 					 nchanges, path);
 		}
 	}
@@ -4056,7 +4096,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
 	pfree(buffer);
 	pfree(s2.data);
 
-	elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
+	elog(LOG, "!!> replayed %d (all) changes from file \"%s\"",
 		 nchanges, path);
 
 	return nchanges;
@@ -4092,6 +4132,8 @@ prepare_spoolfile_on_proc_exit(int status, Datum arg)
 	HASH_SEQ_STATUS seq_status;
 	PsfHashEntry *hentry;
 
+	elog(LOG, "!!> prepare_spoolfile_on_proc_exit");
+
 	/* Iterate the HTAB looking for what file can be deleted. */
 	if (psf_hash)
 	{
@@ -4100,6 +4142,7 @@ prepare_spoolfile_on_proc_exit(int status, Datum arg)
 		{
 			char *path = hentry->name;
 
+			elog(LOG, "!!> prepare_spoolfile_proc_exit: found '%s'", path);
 			if (hentry->allow_delete)
 				prepare_spoolfile_delete(path);
 		}
-- 
1.8.3.1