v48-0008-Fix-apply-worker-empty-prepare.patch

From 56963d06eff84220a560b2bdb362e9d94a5e012a Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 4 Mar 2021 20:01:59 +1100
Subject: [PATCH v48] Fix apply worker empty prepare.

By sad timing of apply/tablesync workers it is possible to have a "consistent snapshot" that spans prepare/commit in such a way that the tablesync did not do the prepare (because snapshot not consistent) and the apply worker does the prepare ('b') but it skips all the prepared operations [e.g. inserts] while the tablesync was still busy (see the condition of should_apply_changes_for_rel). Later, at the commit prepared time when the apply worker does the commit prepare ('K'), there is nothing committed (because the inserts were skipped earlier).

This patch implements a two-part fix as suggested [1] on hackers.

Part 1 - The begin_prepare handler of apply will always wait for any busy tablesync workers to acheive SYNCDONE/READY state.

Part 2 - If (after Part 1) the apply-worker's prepare is found to be lagging behind any of the sync-workers then the subsequent prepared operations will be spooled to a file to be replayed at commit_prepared time.

Discussion:
[1] https://www.postgresql.org/message-id/CAA4eK1L%3DdhuCRvyDvrXX5wZgc7s1hLRD29CKCK6oaHtVCPgiFA%40mail.gmail.com
---
 src/backend/replication/logical/tablesync.c | 178 +++++--
 src/backend/replication/logical/worker.c    | 731 +++++++++++++++++++++++++++-
 src/include/replication/worker_internal.h   |   3 +
 src/tools/pgindent/typedefs.list            |   2 +
 4 files changed, 879 insertions(+), 35 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 50c3ea7..97fc399 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -116,6 +116,9 @@
 #include "utils/snapmgr.h"
 
 static bool table_states_valid = false;
+static List *table_states_not_ready = NIL;
+static List *table_states_all = NIL;
+static void FetchTableStates(bool *started_tx);
 
 StringInfo	copybuf = NULL;
 
@@ -359,7 +362,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 		Oid			relid;
 		TimestampTz last_start_time;
 	};
-	static List *table_states = NIL;
 	static HTAB *last_start_times = NULL;
 	ListCell   *lc;
 	bool		started_tx = false;
@@ -367,42 +369,14 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	Assert(!IsTransactionState());
 
 	/* We need up-to-date sync state info for subscription tables here. */
-	if (!table_states_valid)
-	{
-		MemoryContext oldctx;
-		List	   *rstates;
-		ListCell   *lc;
-		SubscriptionRelState *rstate;
-
-		/* Clean the old list. */
-		list_free_deep(table_states);
-		table_states = NIL;
-
-		StartTransactionCommand();
-		started_tx = true;
-
-		/* Fetch all non-ready tables. */
-		rstates = GetSubscriptionNotReadyRelations(MySubscription->oid);
-
-		/* Allocate the tracking info in a permanent memory context. */
-		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
-		foreach(lc, rstates)
-		{
-			rstate = palloc(sizeof(SubscriptionRelState));
-			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
-			table_states = lappend(table_states, rstate);
-		}
-		MemoryContextSwitchTo(oldctx);
-
-		table_states_valid = true;
-	}
+	FetchTableStates(&started_tx);
 
 	/*
 	 * Prepare a hash table for tracking last start times of workers, to avoid
 	 * immediate restarts.  We don't need it if there are no tables that need
 	 * syncing.
 	 */
-	if (table_states && !last_start_times)
+	if (table_states_not_ready && !last_start_times)
 	{
 		HASHCTL		ctl;
 
@@ -416,7 +390,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	 * Clean up the hash table when we're done with all tables (just to
 	 * release the bit of memory).
 	 */
-	else if (!table_states && last_start_times)
+	else if (!table_states_not_ready && last_start_times)
 	{
 		hash_destroy(last_start_times);
 		last_start_times = NULL;
@@ -425,7 +399,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	/*
 	 * Process all tables that are being synchronized.
 	 */
-	foreach(lc, table_states)
+	foreach(lc, table_states_not_ready)
 	{
 		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 
@@ -1137,3 +1111,141 @@ copy_table_done:
 	wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
 	return slotname;
 }
+
+/*
+ * Common code to fetch the up-to-date sync state info into the static lists.
+ */
+static void
+FetchTableStates(bool *started_tx)
+{
+	*started_tx = false;
+
+	if (!table_states_valid)
+	{
+		MemoryContext oldctx;
+		List	   *rstates;
+		ListCell   *lc;
+		SubscriptionRelState *rstate;
+
+		/* Clean the old lists. */
+		list_free_deep(table_states_all);
+		table_states_all = NIL;
+		list_free_deep(table_states_not_ready);
+		table_states_not_ready = NIL;
+
+		StartTransactionCommand();
+		*started_tx = true;
+
+		/* Fetch all tables. */
+		rstates = GetSubscriptionRelations(MySubscription->oid);
+
+		/* Allocate the tracking info in a permanent memory context. */
+		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+		foreach(lc, rstates)
+		{
+			SubscriptionRelState *cur_rstate = (SubscriptionRelState *) lfirst(lc);
+
+			/* List of all states */
+			rstate = palloc(sizeof(SubscriptionRelState));
+			memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState));
+			table_states_all = lappend(table_states_all, rstate);
+
+			/* List of only not-ready states */
+			if (cur_rstate->state != SUBREL_STATE_READY)
+			{
+				rstate = palloc(sizeof(SubscriptionRelState));
+				memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState));
+				table_states_not_ready = lappend(table_states_not_ready, rstate);
+			}
+		}
+		MemoryContextSwitchTo(oldctx);
+
+		table_states_valid = true;
+	}
+}
+
+/*
+ * Are there any tablesyncs which have still not yet reached SYNCDONE/READY state?
+ */
+bool
+AnyTablesyncInProgress()
+{
+	bool		found_busy = false;
+	bool		started_tx = false;
+	int			count = 0;
+	ListCell   *lc;
+
+	/* We need up-to-date sync state info for subscription tables here. */
+	FetchTableStates(&started_tx);
+
+	/*
+	 * Process all not-READY tables to see if any are also not-SYNCDONE
+	 */
+	foreach(lc, table_states_not_ready)
+	{
+		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
+
+		count++;
+		elog(DEBUG1,
+			 "AnyTablesyncInProgress?: #%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 the list element of table_states_not_ready.
+		 *
+		 * So the "table_states_not_ready" list might end up having a READY
+		 * state in it even though there was none when it was initially
+		 * created. This is reason why we need to check for READY below.
+		 */
+		if (rstate->state != SUBREL_STATE_SYNCDONE &&
+			rstate->state != SUBREL_STATE_READY)
+		{
+			found_busy = true;
+			break;
+		}
+	}
+
+	if (started_tx)
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(false);
+	}
+
+	elog(DEBUG1,
+		 "AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s",
+		 count,
+		 found_busy ? "true" : "false");
+
+	return found_busy;
+}
+
+/*
+ * What is the biggest LSN from the all the known tablesyncs?
+ */
+XLogRecPtr
+BiggestTablesyncLSN()
+{
+	XLogRecPtr	biggest_lsn = InvalidXLogRecPtr;
+	ListCell   *lc;
+	int			count = 0;
+
+	foreach(lc, table_states_all)
+	{
+		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
+
+		count++;
+		if (rstate->lsn > biggest_lsn)
+			biggest_lsn = rstate->lsn;
+	}
+
+	elog(DEBUG1,
+		 "BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X",
+		 count,
+		 LSN_FORMAT_ARGS(biggest_lsn));
+
+	return biggest_lsn;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ad537f4..f753e33 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -209,6 +209,54 @@ static void subxact_info_add(TransactionId xid);
 static inline void cleanup_subxact_info(void);
 
 /*
+ * The following are for the support of a spoolfile for prepared messages.
+ */
+
+/*
+ * A Prepare spoolfile hash entry. We create this entry in the psf_hash. This is
+ * for maintaining a mapping between the name of the prepared spoolfile, and the
+ * corresponding fileset handles of same.
+ */
+typedef struct PsfHashEntry
+{
+	char		name[MAXPGPATH];	/* Hash key --- must be first */
+	bool allow_delete; /* ok to delete? */
+}			PsfHashEntry;
+
+/*
+ * Information about the "current" psf spoolfile.
+ */
+typedef struct PsfFile
+{
+	char	name[MAXPGPATH];/* psf name - same as the HTAB key. */
+	bool	is_spooling;	/* are we currently spooling to this file? */
+	File 	vfd;			/* -1 when the file is closed. */
+	off_t	cur_offset;		/* offset for the next write or read. Reset to 0
+							 * when file is opened. */
+} PsfFile;
+
+/*
+ * Hash table for storing the Prepared spoolfile info along with shared fileset.
+ */
+static HTAB *psf_hash = NULL;
+
+/*
+ * Information about the 'current' open spoolfile is only valid when spooling.
+ * This is flagged as 'is_spooling' only between begin_prepare and prepare.
+ */
+static PsfFile psf_cur = { .is_spooling = false, .vfd = -1, .cur_offset = 0 };
+
+static void prepare_spoolfile_create(char *path);
+static void prepare_spoolfile_write(char action, StringInfo s);
+static void prepare_spoolfile_close(void);
+static void prepare_spoolfile_delete(char *path);
+static bool prepare_spoolfile_exists(char *path);
+static void prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid);
+static int	prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata);
+static bool prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s);
+static void prepare_spoolfile_on_proc_exit(int status, Datum arg);
+
+/*
  * Serialize and deserialize changes for a toplevel transaction.
  */
 static void stream_cleanup_files(Oid subid, TransactionId xid);
@@ -730,6 +778,9 @@ apply_handle_begin_prepare(StringInfo s)
 {
 	LogicalRepBeginPrepareData begin_data;
 
+	/* Tablesync should never receive prepare. */
+	Assert(!am_tablesync_worker());
+
 	logicalrep_read_begin_prepare(s, &begin_data);
 
 	/*
@@ -741,6 +792,81 @@ apply_handle_begin_prepare(StringInfo s)
 				errmsg("transaction identifier \"%s\" is already in use",
 					   begin_data.gid)));
 
+	/*
+	 * A Problem:
+	 *
+	 * By sad timing of apply/tablesync workers it is possible to have a
+	 * “consistent snapshot” that spans prepare/commit in such a way that
+	 * the tablesync did not do the prepare (because snapshot not consistent)
+	 * and the apply worker does the begin prepare (‘b’) but it skips all
+	 * the prepared operations [e.g. inserts] while the tablesync was still
+	 * busy (see the condition of should_apply_changes_for_rel). Later at the
+	 * commit prepared time when the apply worker does the commit prepare
+	 * (‘K’), there is nothing in it (because the inserts were skipped
+	 * earlier).
+	 *
+	 * The following code has a 2-part workaround for that scenario.
+	 */
+	if (!am_tablesync_worker())
+	{
+		/*
+		 * Workaround Part 1 of 2:
+		 *
+		 * 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",
+			 LSN_FORMAT_ARGS(begin_data.end_lsn),
+			 LSN_FORMAT_ARGS(begin_data.final_lsn),
+			 LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn));
+
+		while (AnyTablesyncInProgress())
+		{
+			process_syncing_tables(begin_data.final_lsn);
+
+			/* This latch is to prevent 100% CPU looping. */
+			(void) WaitLatch(MyLatch,
+							 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+							 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+			ResetLatch(MyLatch);
+		}
+
+		/*
+		 * Workaround Part 2 of 2:
+		 *
+		 * If (when getting to SYNCDONE/READY state) some tablesync went
+		 * beyond this begin_prepare LSN then set all messages (until
+		 * prepared) will be saved to a spoolfile for replay later at
+		 * commit_prepared time.
+		 */
+		if (begin_data.final_lsn < BiggestTablesyncLSN())
+		{
+			char		psfpath[MAXPGPATH];
+			StringInfoData sid;
+
+			/*
+			 * Create the spoolfile.
+			 */
+			prepare_spoolfile_name(psfpath, sizeof(psfpath),
+								   MyLogicalRepWorker->subid, begin_data.gid);
+			prepare_spoolfile_create(psfpath);
+
+			/*
+			 * From now, until the handle_prepare we are spooling to the
+			 * current psf.
+			 */
+			psf_cur.is_spooling = true;
+
+			/*
+			 * Write BEGIN_PREPARE as the first message of the psf file.
+			 */
+			initStringInfo(&sid);
+			appendBinaryStringInfo(&sid, (char *)&begin_data, sizeof(begin_data));
+			Assert(prepare_spoolfile_handler(LOGICAL_REP_MSG_BEGIN_PREPARE, &sid));
+		}
+	}
+
 	remote_final_lsn = begin_data.final_lsn;
 
 	in_remote_transaction = true;
@@ -756,6 +882,50 @@ apply_handle_prepare(StringInfo s)
 {
 	LogicalRepPreparedTxnData prepare_data;
 
+	/*
+	 * If we were using a psf spoolfile, then write the PREPARE as the final
+	 * message. This prepare information will be used at commit_prepared time.
+	 */
+	if (psf_cur.is_spooling)
+	{
+		PsfHashEntry *hentry;
+
+		/* Write the PREPARE info to the psf file. */
+		prepare_spoolfile_handler(LOGICAL_REP_MSG_PREPARE, s);
+
+		/*
+		 * Flush the spoolfile, so changes can survive a restart.
+		 */
+		FileSync(psf_cur.vfd, WAIT_EVENT_DATA_FILE_SYNC);
+
+		/*
+		 * We are finished spooling to the current psf.
+		 */
+		psf_cur.is_spooling = false;
+
+		/*
+		 * The commit_prepare will need the spoolfile, so unregister it for
+		 * removal on proc-exit just in case there is an unexpected restart
+		 * between now and when commit_prepared happens.
+		 */
+		hentry = (PsfHashEntry *) hash_search(psf_hash,
+											  psf_cur.name,
+											  HASH_FIND,
+											  NULL);
+		Assert(hentry);
+		hentry->allow_delete = false;
+
+		/*
+		 * The psf_cur.vfd is meaningful only between begin_prepare and prepared.
+		 * So close it now. Any messages written to the psf will be applied
+		 * later during handle_commit_prepared.
+		 */
+		prepare_spoolfile_close();
+
+		in_remote_transaction = false;
+		return;
+	}
+
 	logicalrep_read_prepare(s, &prepare_data);
 
 	Assert(prepare_data.prepare_lsn == remote_final_lsn);
@@ -805,9 +975,63 @@ static void
 apply_handle_commit_prepared(StringInfo s)
 {
 	LogicalRepPreparedTxnData prepare_data;
+	char		psfpath[MAXPGPATH];
 
 	logicalrep_read_commit_prepared(s, &prepare_data);
 
+	/*
+	 * If this prepare's messages were being spooled to a file, then replay
+	 * them all now.
+	 */
+	prepare_spoolfile_name(psfpath, sizeof(psfpath),
+						   MyLogicalRepWorker->subid, prepare_data.gid);
+	if (prepare_spoolfile_exists(psfpath))
+	{
+		int			nchanges;
+		LogicalRepPreparedTxnData pdata;
+
+		/*
+		 * 1. replay the spooled messages
+		 */
+
+		ensure_transaction();
+
+		nchanges = prepare_spoolfile_replay_messages(psfpath, &pdata);
+		elog(DEBUG1,
+			 "apply_handle_commit_prepared: replayed %d (all) changes.",
+			 nchanges);
+
+		/*
+		 * 2. mark as PREPARED (use prepare_data info from the psf file)
+		 */
+
+		/*
+		 * BeginTransactionBlock is necessary to balance the
+		 * EndTransactionBlock called within the PrepareTransactionBlock
+		 * below.
+		 */
+		BeginTransactionBlock();
+		CommitTransactionCommand();
+
+		/*
+		 * Update origin state so we can restart streaming from correct
+		 * position in case of crash.
+		 */
+		replorigin_session_origin_lsn = pdata.end_lsn;
+		replorigin_session_origin_timestamp = pdata.preparetime;
+
+		PrepareTransactionBlock(pdata.gid);
+		CommitTransactionCommand();
+		pgstat_report_stat(false);
+
+		store_flush_position(pdata.end_lsn);
+
+		/*
+		 * Now that we replayed the psf it is no longer needed. Just delete it.
+		 */
+		prepare_spoolfile_delete(psfpath);
+	}
+
 	/* there is no transaction when COMMIT PREPARED is called */
 	ensure_transaction();
 
@@ -838,15 +1062,49 @@ static void
 apply_handle_rollback_prepared(StringInfo s)
 {
 	LogicalRepRollbackPreparedTxnData rollback_data;
+	bool		using_psf;
+	char		psfpath[MAXPGPATH];
 
 	logicalrep_read_rollback_prepared(s, &rollback_data);
 
 	/*
+	 * If this prepare's messages were being spooled to a file, then cleanup
+	 * the file.
+	 */
+	prepare_spoolfile_name(psfpath, sizeof(psfpath),
+						   MyLogicalRepWorker->subid, rollback_data.gid);
+	using_psf = prepare_spoolfile_exists(psfpath);
+	if (using_psf)
+	{
+		if (psf_cur.is_spooling)
+		{
+			/*
+			 * XXX - For some reason it is currently possible (due to bug?) it
+			 * is possibe to get here, after a restart, when there was a
+			 * begin_prepare but there was NO prepare. Since there was no
+			 * prepare, the psf_cur and the transaction are still lingering
+			 * so they need to be cleaned up now.
+			 */
+			prepare_spoolfile_close();
+		}
+
+		/*
+		 * We are finished with this spoolfile. Delete it.
+		 */
+		prepare_spoolfile_delete(psfpath);
+	}
+
+	/*
 	 * It is possible that we haven't received prepare because it occurred
 	 * before walsender reached a consistent point in which case we need to
 	 * skip rollback prepared.
+	 *
+	 * And we also skip the FinishPreparedTransaction if we're using the
+	 * Prepare Spoolfile (using_psf) because in that case there is no matching
+	 * PrepareTransactionBlock done yet.
 	 */
-	if (LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn,
+	if (!using_psf &&
+		LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn,
 					rollback_data.preparetime))
 	{
 		/*
@@ -889,10 +1147,33 @@ apply_handle_stream_prepare(StringInfo s)
 
 	Assert(!in_streamed_transaction);
 
+	/* Tablesync should never receive prepare. */
+	Assert(!am_tablesync_worker());
+
 	xid = logicalrep_read_stream_prepare(s, &prepare_data);
 	elog(DEBUG1, "received prepare for streamed transaction %u", xid);
 
 	/*
+	 * Wait for all the sync workers to reach the SYNCDONE/READY state.
+	 *
+	 * This is same waiting logic as in apply_handle_begin_prepare function
+	 * (see that function for more details about this).
+	 */
+	if (!am_tablesync_worker())
+	{
+		while (AnyTablesyncInProgress())
+		{
+			process_syncing_tables(prepare_data.end_lsn);
+
+			/* This latch is to prevent 100% CPU looping. */
+			(void) WaitLatch(MyLatch,
+							 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+							 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+			ResetLatch(MyLatch);
+		}
+	}
+
+	/*
 	 *
 	 * --------------------------------------------------------------------------
 	 * 1. Replay all the spooled operations - Similar code as for
@@ -959,6 +1240,7 @@ apply_handle_origin(StringInfo s)
 	 * remote transaction and before any actual writes.
 	 */
 	if (!in_streamed_transaction &&
+		!psf_cur.is_spooling &&
 		(!in_remote_transaction ||
 		 (IsTransactionState() && !am_tablesync_worker())))
 		ereport(ERROR,
@@ -1393,6 +1675,9 @@ apply_handle_insert(StringInfo s)
 	TupleTableSlot *remoteslot;
 	MemoryContext oldctx;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_INSERT, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s))
 		return;
 
@@ -1514,6 +1799,9 @@ apply_handle_update(StringInfo s)
 	RangeTblEntry *target_rte;
 	MemoryContext oldctx;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_UPDATE, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, s))
 		return;
 
@@ -1672,6 +1960,9 @@ apply_handle_delete(StringInfo s)
 	TupleTableSlot *remoteslot;
 	MemoryContext oldctx;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_DELETE, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, s))
 		return;
 
@@ -2041,6 +2332,9 @@ apply_handle_truncate(StringInfo s)
 	List	   *relids_logged = NIL;
 	ListCell   *lc;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_TRUNCATE, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, s))
 		return;
 
@@ -2334,6 +2628,23 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	}
 
 	/*
+	 * Initialize the psf_hash table if we haven't yet. This will be used for
+	 * the entire duration of the apply worker so create it in permanent
+	 * context.
+	 */
+	if (psf_hash == NULL)
+	{
+		HASHCTL		hash_ctl;
+		PsfHashEntry *hentry;
+
+		hash_ctl.keysize = sizeof(hentry->name);
+		hash_ctl.entrysize = sizeof(PsfHashEntry);
+		hash_ctl.hcxt = ApplyContext;
+		psf_hash = hash_create("PrepareSpoolfileHash", 1024, &hash_ctl,
+							   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+	}
+
+	/*
 	 * Init the ApplyMessageContext which we clean up after each replication
 	 * protocol message.
 	 */
@@ -2453,7 +2764,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		/* confirm all writes so far */
 		send_feedback(last_received, false, false);
 
-		if (!in_remote_transaction && !in_streamed_transaction)
+		if (!in_remote_transaction && !in_streamed_transaction && !psf_cur.is_spooling)
 		{
 			/*
 			 * If we didn't get any transactions for a while there might be
@@ -3201,6 +3512,9 @@ ApplyWorkerMain(Datum main_arg)
 	/* Attach to slot */
 	logicalrep_worker_attach(worker_slot);
 
+	/* Arrange to delete any unwanted psf file(s) at proc-exit */
+	on_proc_exit(prepare_spoolfile_on_proc_exit, 0);
+
 	/* Setup signal handling */
 	pqsignal(SIGHUP, SignalHandlerForConfigReload);
 	pqsignal(SIGTERM, die);
@@ -3378,3 +3692,416 @@ IsLogicalWorker(void)
 {
 	return MyLogicalRepWorker != NULL;
 }
+
+/*
+ * Handle the PREPARE spoolfile (if any)
+ *
+ * It can be necessary to redirect the PREPARE messages to a spoolfile (see
+ * apply_handle_begin_prepare) and then replay them back at the COMMIT PREPARED
+ * time. If needed, this is the common function to do that file redirection.
+ *
+ * Returns true if the message was redirected to the spoolfile, false
+ * otherwise (regular mode).
+ */
+static bool
+prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s)
+{
+	elog(DEBUG1,
+		 "prepare_spoolfile_handler for action '%c'. %s write to spool file",
+		 action,
+		 psf_cur.is_spooling ? "Do" : "Don't");
+
+	if (!psf_cur.is_spooling)
+		return false;
+
+	Assert(!in_streamed_transaction);
+
+	/* write the change to the current file */
+	prepare_spoolfile_write(action, s);
+
+	return true;
+}
+
+/*
+ * Create the spoolfile used to serialize the prepare messages.
+ */
+static void
+prepare_spoolfile_create(char *path)
+{
+	bool		found;
+	PsfHashEntry *hentry;
+
+	elog(DEBUG1, "creating file \"%s\" for prepare changes", path);
+
+	Assert(!psf_cur.is_spooling);
+
+	/* create or find the prepare spoolfile entry in the psf_hash */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_ENTER | HASH_FIND,
+										  &found);
+
+	if (!found)
+	{
+		elog(DEBUG1, "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)
+		{
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not create file \"%s\": %m", path)));
+		}
+		memcpy(psf_cur.name, path, sizeof(psf_cur.name));
+		psf_cur.cur_offset = 0;
+		hentry->allow_delete = true;
+	}
+	else
+	{
+		/*
+		 * 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);
+		psf_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY);
+		if (psf_cur.vfd < 0)
+		{
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not open file \"%s\": %m", path)));
+		}
+		memcpy(psf_cur.name, path, sizeof(psf_cur.name));
+		psf_cur.cur_offset = 0;
+		hentry->allow_delete = true;
+	}
+
+	/* Sanity checks */
+	Assert(psf_cur.vfd >= 0);
+	Assert(psf_cur.cur_offset == 0);
+	Assert(prepare_spoolfile_exists(path));
+}
+
+/*
+ * Close the "current" spoolfile and unset the fd.
+ */
+static void
+prepare_spoolfile_close()
+{
+	if (psf_cur.vfd >= 0)
+		FileClose(psf_cur.vfd);
+
+	/* Mark this fd as not valid to use anymore. */
+	psf_cur.is_spooling = false;
+	psf_cur.vfd = -1;
+	psf_cur.cur_offset = 0;
+}
+
+/*
+ * Delete the specified psf spoolfile, and any HTAB associated with it.
+ */
+static void
+prepare_spoolfile_delete(char *path)
+{
+	/* The current psf should be closed already, but make sure anyway. */
+	prepare_spoolfile_close();
+
+	/* Delete the file off the disk. */
+	unlink(path);
+
+	/* Remove any entry from the psf_hash, if present */
+	hash_search(psf_hash, path, HASH_REMOVE, NULL);
+}
+
+/*
+ * Serialize a change to the prepare spoolfile for the current toplevel transaction.
+ *
+ * The change is serialized in a simple format, with length (not including
+ * the length), action code (identifying the message type) and message
+ * contents (without the subxact TransactionId value).
+ */
+static void
+prepare_spoolfile_write(char action, StringInfo s)
+{
+	int			len;
+	int			bytes_written;
+
+	Assert(psf_cur.is_spooling);
+
+	elog(DEBUG1, "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 */
+	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 */
+	bytes_written = FileWrite(psf_cur.vfd, &action, sizeof(action),
+							  psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE);
+	Assert(bytes_written == sizeof(action));
+	psf_cur.cur_offset += bytes_written;
+
+	/* and finally the remaining part of the buffer (after the XID) */
+	len = (s->len - s->cursor);
+
+	bytes_written = FileWrite(psf_cur.vfd, &s->data[s->cursor], len,
+							  psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE);
+	Assert(bytes_written == len);
+	psf_cur.cur_offset += bytes_written;
+}
+
+/*
+ * Is there a prepare spoolfile for the specified path?
+ */
+static bool
+prepare_spoolfile_exists(char *path)
+{
+	bool		found;
+	PsfHashEntry *hentry;
+
+	/* Find the prepare spoolfile entry in the psf_hash */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_FIND,
+										  &found);
+
+	if (!found)
+	{
+		/*
+		 * Hash doesn't know about it, but perhaps the Hash was destroyed by a
+		 * restart, so let's check the file existence on disk.
+		 */
+		File fd = PathNameOpenFile(path, O_RDONLY | PG_BINARY);
+
+		found = fd >= 0;
+		if (fd >= 0)
+			FileClose(fd);
+
+		/*
+		 * And if it was found on disk then create the HTAB entry for it.
+		 */
+		if (found)
+		{
+			hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_ENTER,
+										  NULL);
+			hentry->allow_delete = false;
+		}
+	}
+
+	return found;
+}
+
+/*
+ * Replay (apply) all the prepared messages that are in the prepare spoolfile.
+ *
+ * [Note: this is similar to apply_spooled_messages function]
+ */
+static int
+prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata)
+{
+	StringInfoData s2;
+	int			nchanges = 0;
+	char	   *buffer = NULL;
+	MemoryContext oldctx,
+				oldctx2;
+	PsfFile		psf = { .is_spooling = false, .vfd = -1, .cur_offset = 0 };
+
+	elog(DEBUG1,
+		 "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"",
+		 path);
+
+	/*
+	 * Allocate memory required to process all the messages in
+	 * TopTransactionContext to avoid it getting reset after each message is
+	 * processed.
+	 */
+	oldctx = MemoryContextSwitchTo(TopTransactionContext);
+
+	psf.vfd = PathNameOpenFile(path, O_RDONLY | PG_BINARY);
+	if (psf.vfd < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from prepared spoolfile \"%s\": %m",
+						path)));
+
+	buffer = palloc(BLCKSZ);
+	initStringInfo(&s2);
+
+	MemoryContextSwitchTo(oldctx);
+
+	/*
+	 * Read the entries one by one and pass them through the same logic as in
+	 * apply_dispatch.
+	 */
+	while (true)
+	{
+		int			nbytes;
+		int			len;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* read length of the on-disk record */
+		nbytes = FileRead(psf.vfd, (char *) &len, sizeof(len),
+						  psf.cur_offset, WAIT_EVENT_DATA_FILE_READ);
+		psf.cur_offset += nbytes;
+
+		/* have we reached end of the file? */
+		if (nbytes == 0)
+			break;
+
+		/* do we have a correct length? */
+		if (nbytes != sizeof(len))
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read from prepared spoolfile \"%s\": %m",
+							path)));
+
+		Assert(len > 0);
+
+		/* make sure we have sufficiently large buffer */
+		buffer = repalloc(buffer, len);
+
+		/* and finally read the data into the buffer */
+		nbytes = FileRead(psf.vfd, buffer, len,
+						  psf.cur_offset, WAIT_EVENT_DATA_FILE_READ);
+		psf.cur_offset += nbytes;
+		if (nbytes != len)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read from prepared spoolfile \"%s\": %m",
+							path)));
+
+		/* copy the buffer to the stringinfo and call apply_dispatch */
+		resetStringInfo(&s2);
+		appendBinaryStringInfo(&s2, buffer, len);
+
+		/*
+		 * The psf spoolfile contents will have first and last messages as
+		 * BEGIN_PREPARE and PREPARE message respectively. These two are
+		 * processed specially within this function.
+		 *
+		 * BEGIN_PREPARE msg: This will be the first message of the psf file.
+		 * Use this begin_data information to set the remote_final_lsn.
+		 *
+		 * PREPARE msg: The prepare_data information is returned so that the
+		 * prepare lsn values are available to the caller (commit_prepared).
+		 * Unfortunately, just dispatching the PREPARE message is problematic
+		 * because its transaction commits have side effects on this replay
+		 * loop which is still running.
+		 *
+		 * All other message content (between the BEGIN_PREPARE and the PREPARE)
+		 * will be delivered to apply_dispatch as they normally would be.
+		 */
+		if (s2.data[0] == LOGICAL_REP_MSG_BEGIN_PREPARE)
+		{
+			LogicalRepBeginPrepareData bdata;
+
+			/* 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",
+				 bdata.gid,
+				 LSN_FORMAT_ARGS(bdata.final_lsn),
+				 LSN_FORMAT_ARGS(bdata.end_lsn));
+
+			/*
+			 * Make sure the handle apply_dispatch methods are aware we're in a remote
+			 * transaction.
+			 */
+			remote_final_lsn = bdata.final_lsn;
+			in_remote_transaction = true;
+			pgstat_report_activity(STATE_RUNNING, NULL);
+		}
+		else if (s2.data[0] == LOGICAL_REP_MSG_PREPARE)
+		{
+			/* 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",
+				 pdata->gid,
+				 LSN_FORMAT_ARGS(pdata->prepare_lsn),
+				 LSN_FORMAT_ARGS(pdata->end_lsn));
+		}
+		else
+		{
+			/* Ensure we are reading the data into our memory context. */
+			oldctx2 = MemoryContextSwitchTo(ApplyMessageContext);
+
+			apply_dispatch(&s2);
+
+			MemoryContextReset(ApplyMessageContext);
+
+			MemoryContextSwitchTo(oldctx2);
+
+			nchanges++;
+
+			if (nchanges % 1000 == 0)
+				elog(DEBUG1, "replayed %d changes from file '%s'",
+					 nchanges, path);
+		}
+	}
+
+	FileClose(psf.vfd);
+
+	pfree(buffer);
+	pfree(s2.data);
+
+	elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
+		 nchanges, path);
+
+	return nchanges;
+}
+
+/*
+ * Format the filename for the prepare spoolfile.
+ */
+static void
+prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid)
+{
+	PsfHashEntry *hentry;
+
+	/*
+	 * This name is used as the key in the psf_hash HTAB.
+	 *
+	 * Therefore, the name and the key must be exactly same lengths and padded
+	 * with '\0' so garbage does not impact the HTAB lookups.
+	 */
+	Assert(sizeof(hentry->name) == MAXPGPATH);
+	Assert(szpath == MAXPGPATH);
+	memset(path, '\0', MAXPGPATH);
+
+	snprintf(path, MAXPGPATH, "pg_twophase/%u-%s.prep_changes", subid, gid);
+}
+
+/*
+ * proc_exit callback to remove unwanted psf files.
+ */
+static void
+prepare_spoolfile_on_proc_exit(int status, Datum arg)
+{
+	HASH_SEQ_STATUS seq_status;
+	PsfHashEntry *hentry;
+
+	/* Iterate the HTAB looking for what file can be deleted. */
+	if (psf_hash)
+	{
+		hash_seq_init(&seq_status, psf_hash);
+		while ((hentry = (PsfHashEntry *) hash_seq_search(&seq_status)) != NULL)
+		{
+			char *path = hentry->name;
+
+			if (hentry->allow_delete)
+				prepare_spoolfile_delete(path);
+		}
+	}
+}
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 1cac75e..95d78e9 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -86,6 +86,9 @@ extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
 											  char *originname, int szorgname);
 extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);
 
+extern bool AnyTablesyncInProgress(void);
+extern XLogRecPtr BiggestTablesyncLSN(void);
+
 void		process_syncing_tables(XLogRecPtr current_lsn);
 void		invalidate_syncing_table_states(Datum arg, int cacheid,
 											uint32 hashvalue);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 745b51d..4ffcef5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1958,6 +1958,8 @@ ProtocolVersion
 PrsStorage
 PruneState
 PruneStepResult
+PsfFile
+PsfHashEntry
 PsqlScanCallbacks
 PsqlScanQuoteType
 PsqlScanResult
-- 
1.8.3.1