v29-0005-Support-2PC-txn-spoolfile.patch

From d254d5526512d699c724c1a5af57145be8028ac3 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Wed, 2 Dec 2020 18:55:47 +1100
Subject: [PATCH v29] Support 2PC txn - spoolfile.

This patch only refactors to isolate the streaming spool-file processing to a separate function.
Later, two-phase commit logic will require this common processing to be called from multiple places.
---
 src/backend/replication/logical/worker.c | 48 ++++++++++++++++++++++----------
 1 file changed, 33 insertions(+), 15 deletions(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 8c7fad8..a4ec883 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -246,6 +246,8 @@ static void apply_handle_tuple_routing(ResultRelInfo *relinfo,
 									   LogicalRepRelMapEntry *relmapentry,
 									   CmdType operation);
 
+static int	apply_spooled_messages(TransactionId xid, XLogRecPtr lsn);
+
 /*
  * Should this worker apply changes for given relation.
  *
@@ -924,30 +926,21 @@ apply_handle_stream_abort(StringInfo s)
 }
 
 /*
- * Handle STREAM COMMIT message.
+ * Common spoolfile processing.
+ * Returns how many changes were applied.
  */
-static void
-apply_handle_stream_commit(StringInfo s)
+static int
+apply_spooled_messages(TransactionId xid, XLogRecPtr lsn)
 {
-	TransactionId xid;
 	StringInfoData s2;
 	int			nchanges;
 	char		path[MAXPGPATH];
 	char	   *buffer = NULL;
 	bool		found;
-	LogicalRepCommitData commit_data;
 	StreamXidHash *ent;
 	MemoryContext oldcxt;
 	BufFile    *fd;
 
-	Assert(!in_streamed_transaction);
-
-	xid = logicalrep_read_stream_commit(s, &commit_data);
-
-	elog(DEBUG1, "received commit for streamed transaction %u", xid);
-
-	ensure_transaction();
-
 	/*
 	 * Allocate file handle and memory required to process all the messages in
 	 * TopTransactionContext to avoid them getting reset after each message is
@@ -955,7 +948,7 @@ apply_handle_stream_commit(StringInfo s)
 	 */
 	oldcxt = MemoryContextSwitchTo(TopTransactionContext);
 
-	/* open the spool file for the committed transaction */
+	/* open the spool file for the committed/prepared transaction */
 	changes_filename(path, MyLogicalRepWorker->subid, xid);
 	elog(DEBUG1, "replaying changes from file \"%s\"", path);
 	ent = (StreamXidHash *) hash_search(xidhash,
@@ -970,7 +963,7 @@ apply_handle_stream_commit(StringInfo s)
 
 	MemoryContextSwitchTo(oldcxt);
 
-	remote_final_lsn = commit_data.commit_lsn;
+	remote_final_lsn = lsn;
 
 	/*
 	 * Make sure the handle apply_dispatch methods are aware we're in a remote
@@ -1045,6 +1038,31 @@ apply_handle_stream_commit(StringInfo s)
 	elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
 		 nchanges, path);
 
+	return nchanges;
+}
+
+/*
+ * Handle STREAM COMMIT message.
+ */
+static void
+apply_handle_stream_commit(StringInfo s)
+{
+	TransactionId xid;
+	LogicalRepCommitData commit_data;
+	int			nchanges = 0;
+
+	Assert(!in_streamed_transaction);
+
+	xid = logicalrep_read_stream_commit(s, &commit_data);
+
+	elog(DEBUG1, "received commit for streamed transaction %u", xid);
+
+	ensure_transaction();
+
+	nchanges = apply_spooled_messages(xid, commit_data.commit_lsn);
+
+	elog(DEBUG1, "apply_handle_stream_commit: replayed %d (all) changes.", nchanges);
+
 	apply_handle_commit_internal(s, &commit_data);
 
 	/* unlink the files with serialized changes and subxact info */
-- 
1.8.3.1