logical_decoding_xact_bookkeep.patch

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 5f59613..2512912 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -42,6 +42,7 @@
 #include "replication/reorderbuffer.h"
 #include "replication/snapbuild.h"
 #include "storage/standby.h"
+#include "utils/builtins.h"
 
 typedef struct XLogRecordBuffer
 {
@@ -85,6 +86,9 @@ static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx,
 							  XLogRecordBuffer *buf, Oid dbId,
 							  RepOriginId origin_id);
 
+/* record previous restart_lsn running xacts */
+xl_running_xacts *last_running = NULL;
+
 /*
  * Take every XLogReadRecord()ed record and perform the actions required to
  * decode it using the output plugin already setup in the logical decoding
@@ -402,6 +406,28 @@ DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 			{
 				xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);
 
+				/* record restart_lsn running xacts */
+				if (MyReplicationSlot && (buf->origptr == MyReplicationSlot->data.restart_lsn))
+				{
+					if (last_running)
+						free(last_running);
+
+					last_running = NULL;
+
+					/*
+					 * xl_running_xacts contains a xids Flexible Array
+					 * and its size is subxcnt + xcnt.
+					 * Take that into account while allocating
+					 * the memory for last_running.
+					 */
+					last_running = (xl_running_xacts *) malloc(sizeof(xl_running_xacts)
+																+ sizeof(TransactionId )
+																* (running->subxcnt + running->xcnt));
+					memcpy(last_running, running, sizeof(xl_running_xacts)
+														 + (sizeof(TransactionId)
+														 * (running->subxcnt + running->xcnt)));
+				}
+
 				SnapBuildProcessRunningXacts(builder, buf->origptr, running);
 
 				/*
@@ -678,6 +704,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	TimestampTz commit_time = parsed->xact_time;
 	RepOriginId origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
+	bool force_travel_and_snapshot = false;
 
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
@@ -685,8 +712,29 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		commit_time = parsed->origin_timestamp;
 	}
 
+	/*
+	 * Check if the commit contain catalog invalidations
+	 * and if the xid was part of the restart_lsn
+	 * running ones
+	 */
+	if ((parsed->xinfo & XACT_XINFO_HAS_INVALS) && last_running)
+	{
+		/* make last_running->xids bsearch()able */
+		qsort(last_running->xids,
+			  last_running->subxcnt + last_running->xcnt,
+			  sizeof(TransactionId), xidComparator);
+
+		/*
+		 * Is this xid part of the known running ones
+		 * in the restart_lsn RUNNING_XACT entry?
+		 */
+		force_travel_and_snapshot = TransactionIdInArray(xid, last_running->xids,
+														 last_running->subxcnt
+														 + last_running->xcnt);
+	}
+
 	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
-					   parsed->nsubxacts, parsed->subxacts);
+					   parsed->nsubxacts, parsed->subxacts, force_travel_and_snapshot);
 
 	/* ----
 	 * Check whether we are interested in this specific transaction, and tell
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 91600ac..4a5610b 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -4892,7 +4892,7 @@ ApplyLogicalMappingFile(HTAB *tuplecid_data, Oid relid, const char *fname)
 /*
  * Check whether the TransactionId 'xid' is in the pre-sorted array 'xip'.
  */
-static bool
+bool
 TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num)
 {
 	return bsearch(&xid, xip, num,
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index ed3acad..6be39d1 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -910,15 +910,19 @@ SnapBuildPurgeCommittedTxn(SnapBuild *builder)
  */
 void
 SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
-				   int nsubxacts, TransactionId *subxacts)
+				   int nsubxacts, TransactionId *subxacts, bool force_travel_and_snapshot)
 {
 	int			nxact;
-
-	bool		needs_snapshot = false;
-	bool		needs_timetravel = false;
-	bool		sub_needs_timetravel = false;
+	bool		needs_snapshot;
+	bool		needs_timetravel;
+	bool		sub_needs_timetravel;
 
 	TransactionId xmax = xid;
+	/*
+	 * if force_travel_and_snapshot is set to true
+	 * then proceed as if there are any catalog modifying subxacts.
+	 */
+	needs_snapshot = needs_timetravel = sub_needs_timetravel = force_travel_and_snapshot;
 
 	/*
 	 * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 565a961..7fc79f3 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -685,4 +685,6 @@ void		ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);
 
 void		StartupReorderBuffer(void);
 
+extern bool TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num);
+
 #endif
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index fbabce6..46f7df0 100644
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -80,7 +80,7 @@ extern XLogRecPtr SnapBuildInitialConsistentPoint(SnapBuild *builder);
 
 extern void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
 							   TransactionId xid, int nsubxacts,
-							   TransactionId *subxacts);
+							   TransactionId *subxacts, bool force_travel_and_snapshot);
 extern bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid,
 								   XLogRecPtr lsn);
 extern void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,