v46-0007-Fix-apply-worker-empty-prepare.patch

From 842f12a3293ebfb33b961dede5505d827b32e1a9 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Mon, 1 Mar 2021 15:04:18 +1100
Subject: [PATCH v46] 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    | 668 +++++++++++++++++++++++++++-
 src/include/replication/worker_internal.h   |   3 +
 src/tools/pgindent/typedefs.list            |   1 +
 4 files changed, 815 insertions(+), 35 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index feb634e..5f897d3 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 it 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..fcf7bc2 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -209,6 +209,45 @@ 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 contest for the prepare spooling
+ */
+static MemoryContext PsfContext = NULL;
+
+/*
+ * 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 */
+	SharedFileSet *fileset;		/* shared file set for prepare spoolfile */
+}			PsfHashEntry;
+
+/*
+ * Hash table for storing the Prepared spoolfile info along with shared fileset.
+ */
+static HTAB *psf_hash = NULL;
+
+/*
+ * The spoolfile handle is only valid between begin_prepare and prepare.
+ */
+static BufFile *psf_fd = NULL;
+
+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_cleanup(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);
+
+/*
  * Serialize and deserialize changes for a toplevel transaction.
  */
 static void stream_cleanup_files(Oid subid, TransactionId xid);
@@ -730,6 +769,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 +783,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;
+
+			/* The begin_prepare's LSN has been overtaken. */
+
+			/*
+			 * We need a transaction for handling the buffile, used for
+			 * serializing prepared messages. This transaction lasts until the
+			 * commit_prepared/ rollback_prepared.
+			 */
+			ensure_transaction();
+
+			prepare_spoolfile_name(psfpath, sizeof(psfpath),
+								   MyLogicalRepWorker->subid, begin_data.gid);
+			prepare_spoolfile_create(psfpath);
+
+			/*
+			 * 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 +873,38 @@ 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_fd)
+	{
+		/* Write the PREPARE info to the psf file. */
+		Assert(prepare_spoolfile_handler(LOGICAL_REP_MSG_PREPARE, s));
+
+		/*
+		 * TODO - Flush the spoolfile so changes can survive a restart.
+		 */
+
+		/*
+		 * The psf_fd is meaningful only between begin_prepare and prepared.
+		 * So close it now. If we had been writing any messages to the psf_fd
+		 * (the spoolfile) then those will be applied later during
+		 * handle_commit_prepared.
+		 */
+		prepare_spoolfile_close();
+
+		/*
+		 * And end the transaction that was created by begin_prepare for
+		 * working with the psf buffiles.
+		 */
+		Assert(IsTransactionState());
+		CommitTransactionCommand();
+
+		in_remote_transaction = false;
+		return;
+	}
+
 	logicalrep_read_prepare(s, &prepare_data);
 
 	Assert(prepare_data.prepare_lsn == remote_final_lsn);
@@ -805,9 +954,61 @@ 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);
+
+		prepare_spoolfile_cleanup(psfpath);
+
+		/*
+		 * 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);
+	}
+
 	/* there is no transaction when COMMIT PREPARED is called */
 	ensure_transaction();
 
@@ -838,15 +1039,53 @@ 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_fd)
+		{
+			/*
+			 * 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_fd and the transaction are still lingering so
+			 * they need to be cleaned up now.
+			 */
+			prepare_spoolfile_close();
+
+			/*
+			 * And end the transaction that was created by the begin_prepare
+			 * for working with psf buffiles.
+			 */
+			Assert(IsTransactionState());
+			AbortCurrentTransaction();
+		}
+
+		prepare_spoolfile_cleanup(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 +1128,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 +1221,7 @@ apply_handle_origin(StringInfo s)
 	 * remote transaction and before any actual writes.
 	 */
 	if (!in_streamed_transaction &&
+		psf_fd == NULL &&
 		(!in_remote_transaction ||
 		 (IsTransactionState() && !am_tablesync_worker())))
 		ereport(ERROR,
@@ -1393,6 +1656,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 +1780,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 +1941,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 +2313,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 +2609,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.
 	 */
@@ -2349,6 +2641,14 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 													"LogicalStreamingContext",
 													ALLOCSET_DEFAULT_SIZES);
 
+	/*
+	 * This memory context is used when the prepare spooling is used. It is
+	 * reset at prepare commit/rollback time.
+	 */
+	PsfContext = AllocSetContextCreate(ApplyContext,
+									   "PsfContext",
+									   ALLOCSET_DEFAULT_SIZES);
+
 	/* mark as idle, before starting to loop */
 	pgstat_report_activity(STATE_IDLE, NULL);
 
@@ -2453,7 +2753,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_fd == NULL)
 		{
 			/*
 			 * If we didn't get any transactions for a while there might be
@@ -3378,3 +3678,367 @@ 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_fd ? "Do" : "Don't");
+
+	if (psf_fd == NULL)
+		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)
+{
+	MemoryContext oldctx;
+	bool		found;
+	PsfHashEntry *hentry;
+
+	elog(DEBUG1, "creating file \"%s\" for prepare changes", path);
+
+	Assert(psf_fd == NULL);
+
+	/* create or find the prepare spoolfile entry in the psf_hash */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_ENTER | HASH_FIND,
+										  &found);
+
+	/*
+	 * Create/open the bufFiles under the Prepare Spoolfile Context so that we
+	 * have those files until prepare commit/rollback.
+	 */
+	oldctx = MemoryContextSwitchTo(PsfContext);
+
+	if (!found)
+	{
+		MemoryContext savectx;
+		SharedFileSet *fileset;
+
+		elog(DEBUG1, "Not found file \"%s\". Create it.", path);
+		savectx = MemoryContextSwitchTo(ApplyContext);
+		fileset = palloc(sizeof(SharedFileSet));
+
+		SharedFileSetInit(fileset, NULL);
+		MemoryContextSwitchTo(savectx);
+
+		psf_fd = BufFileCreateShared(fileset, path);
+
+		/* Remember this path's fileset for the next time */
+		memcpy(hentry->name, path, MAXPGPATH);
+		hentry->fileset = fileset;
+	}
+	else
+	{
+		/*
+		 * Open the file and seek to the end of the file because we always
+		 * append the changes file.
+		 */
+		elog(DEBUG1, "Found file \"%s\". Open for append.", path);
+		psf_fd = BufFileOpenShared(hentry->fileset, path, O_RDWR);
+		BufFileSeek(psf_fd, 0, 0, SEEK_END);
+	}
+
+	MemoryContextSwitchTo(oldctx);
+
+	/* Sanity check */
+	Assert(prepare_spoolfile_exists(path));
+}
+
+/*
+ * Close the "current" spoolfile and unset the fd.
+ */
+static void
+prepare_spoolfile_close()
+{
+	if (psf_fd)
+		BufFileClose(psf_fd);
+	psf_fd = NULL;
+}
+
+/*
+ * Delete the specified psf spoolfile.
+ */
+static void
+prepare_spoolfile_cleanup(char *path)
+{
+	PsfHashEntry *hentry;
+
+	/* The current psf should be closed already, but make sure anyway. */
+	prepare_spoolfile_close();
+
+	/* And remove the path entry from the psf_hash */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_REMOVE,
+										  NULL);
+
+	/* By this time we must have created the entry */
+	Assert(hentry != NULL);
+
+	/* Delete the file and release the fileset memory */
+	SharedFileSetDeleteAll(hentry->fileset);
+	pfree(hentry->fileset);
+	hentry->fileset = NULL;
+
+	/* Reset the memory context used during the file creation */
+	MemoryContextReset(PsfContext);
+}
+
+/*
+ * 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;
+
+	Assert(psf_fd != NULL);
+
+	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 */
+	BufFileWrite(psf_fd, &len, sizeof(len));
+
+	/* then the action */
+	BufFileWrite(psf_fd, &action, sizeof(action));
+
+	/* and finally the remaining part of the buffer (after the XID) */
+	len = (s->len - s->cursor);
+
+	BufFileWrite(psf_fd, &s->data[s->cursor], len);
+}
+
+/*
+ * Is there a prepare spoolfile for the specified gid?
+ */
+static bool
+prepare_spoolfile_exists(char *path)
+{
+	bool		found;
+
+	/* Find the prepare spoolfile entry in the psf_hash */
+	hash_search(psf_hash,
+				path,
+				HASH_FIND,
+				&found);
+
+	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;
+	bool		found = false;
+	PsfHashEntry *hentry;
+	BufFile    *fd;
+
+	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);
+
+	/* Open the spool file */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_FIND,
+										  &found);
+	Assert(found);
+	fd = BufFileOpenShared(hentry->fileset, path, O_RDONLY);
+
+	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 = BufFileRead(fd, &len, sizeof(len));
+
+		/* 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 */
+		if (BufFileRead(fd, buffer, len) != 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 PREARE)
+		 * 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);
+		}
+	}
+
+	BufFileClose(fd);
+
+	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, "%u-%s.prep_changes", subid, gid);
+}
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 048681c..b7da03b 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1958,6 +1958,7 @@ ProtocolVersion
 PrsStorage
 PruneState
 PruneStepResult
+PsfContext
 PsqlScanCallbacks
 PsqlScanQuoteType
 PsqlScanResult
-- 
1.8.3.1