poc_remember_last_running_xacts_v2.patch

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index aa2427ba73..13d0c16541 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -343,6 +343,26 @@ standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 			{
 				xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);
 
+				/*
+				 * We rely on HEAP2_NEW_CID records and XACT_INVALIDATIONS to know
+				 * if the transaction has changed the catalog, and that information
+				 * is not serialized to SnapBuilder.  Therefore, if the logical
+				 * decoding decodes the commit record of the transaction that actually
+				 * has done catalog changes without these records, we miss to add
+				 * the xid to the snapshot so up creating the wrong snapshot. To
+				 * avoid such a problem, if the COMMIT record of the xid listed in
+				 * the RUNNING_XACTS record read at the start of logical decoding
+				 * has XACT_XINFO_HAS_INVALS flag, we mark both the top transaction
+				 * and its substransactions as containing catalog changes (see also
+				 * ReorderBufferSetLastRunningXactsCatalogChanges()). Since we cannot
+				 * know which transactions actually have done catalog changes only
+				 * by reading the COMMIT record we do that for both.  So we might
+				 * mark an xid that actually has not done that but it’s not a
+				 * problem since we use historic snapshot only for reading system
+				 * catalogs.
+				 */
+				ReorderBufferProcessLastRunningXacts(ctx->reorder, running);
+
 				SnapBuildProcessRunningXacts(builder, buf->origptr, running);
 
 				/*
@@ -627,6 +647,15 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		commit_time = parsed->origin_timestamp;
 	}
 
+	/*
+	 * Set the last running xacts as containing catalog change if necessary.
+	 * This must be done before SnapBuildCommitTxn() so that we include catalog
+	 * change transactions to the historic snapshot.
+	 */
+	ReorderBufferSetLastRunningXactsCatalogChanges(ctx->reorder, xid, parsed->xinfo,
+												   parsed->nsubxacts, parsed->subxacts,
+												   buf->origptr);
+
 	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
 					   parsed->nsubxacts, parsed->subxacts);
 
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 8da5f9089c..fa5785e679 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -353,6 +353,9 @@ ReorderBufferAllocate(void)
 	buffer->outbufsize = 0;
 	buffer->size = 0;
 
+	buffer->last_running_xacts = NULL;
+	buffer->n_last_running_xacts = -1;
+
 	buffer->spillTxns = 0;
 	buffer->spillCount = 0;
 	buffer->spillBytes = 0;
@@ -5161,3 +5164,103 @@ restart:
 		*cmax = ent->cmax;
 	return true;
 }
+
+void
+ReorderBufferProcessLastRunningXacts(ReorderBuffer *rb, xl_running_xacts *running)
+{
+	/* Quick exit if there is no longer last running xacts */
+	if (likely(rb->n_last_running_xacts == 0))
+		return;
+
+	/* First call, build the last running xact list */
+	if (rb->n_last_running_xacts == -1)
+	{
+		int nxacts = running->subxcnt + running->xcnt;
+		Size sz = sizeof(TransactionId) * nxacts;;
+
+		rb->last_running_xacts = MemoryContextAlloc(rb->context, sz);
+		memcpy(rb->last_running_xacts, running->xids, sz);
+		qsort(rb->last_running_xacts, nxacts, sizeof(TransactionId), xidComparator);
+
+		rb->n_last_running_xacts = nxacts;
+
+		return;
+	}
+
+	/*
+	 * Purge xids in the last running xacts list if we can do that for at least
+	 * one xid.
+	 */
+	if (NormalTransactionIdPrecedes(rb->last_running_xacts[0],
+									running->oldestRunningXid))
+	{
+		TransactionId *workspace;
+		int nxids = 0;
+
+		workspace = MemoryContextAlloc(rb->context, rb->n_last_running_xacts);
+		for (int i = 0; i < rb->n_last_running_xacts; i++)
+		{
+			if (NormalTransactionIdPrecedes(rb->last_running_xacts[i],
+											running->oldestRunningXid))
+				;	/* remove */
+			else
+				workspace[nxids++] = rb->last_running_xacts[i];
+		}
+
+		if (nxids > 0)
+			memcpy(rb->last_running_xacts, workspace, sizeof(TransactionId) * nxids);
+		else
+		{
+			pfree(rb->last_running_xacts);
+			rb->last_running_xacts = NULL;
+		}
+
+		rb->n_last_running_xacts = nxids;
+	}
+}
+
+void
+ReorderBufferSetLastRunningXactsCatalogChanges(ReorderBuffer *rb, TransactionId xid,
+											   uint32 xinfo, int subxcnt,
+											   TransactionId *subxacts, XLogRecPtr lsn)
+{
+	void *test;
+
+	/*
+	 * Skip if there is no longer last running xacts information or the COMMIT record
+	 * doesn't have invalidation message, which is a common case.
+	 */
+	if (likely(rb->n_last_running_xacts == 0 || !(xinfo & XACT_XINFO_HAS_INVALS)))
+		return;
+
+	test = bsearch(&xid, rb->last_running_xacts, rb->n_last_running_xacts,
+				   sizeof(TransactionId), xidComparator);
+
+	if (test == NULL)
+	{
+		for (int i = 0; i < subxcnt; i++)
+		{
+			test = bsearch(&subxacts[i], rb->last_running_xacts, rb->n_last_running_xacts,
+						   sizeof(TransactionId), xidComparator);
+
+			if (test != NULL)
+				break;
+		}
+	}
+
+	/*
+	 * If this committed transaction is the one that was running at the time when
+	 * decoding the first RUNNING_XACTS record and have done catalog changes, we
+	 * can mark the top transaction and its subtransactions as catalog-changes.
+	 */
+	if (test != NULL)
+	{
+		ReorderBufferXidSetCatalogChanges(rb, xid, lsn);
+
+		for (int i = 0; i < subxcnt; i++)
+		{
+			ReorderBufferAssignChild(rb, xid, subxacts[i], lsn);
+			ReorderBufferXidSetCatalogChanges(rb, subxacts[i], lsn);
+		}
+	}
+}
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 4a01f877e5..c06df80e08 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -12,6 +12,7 @@
 #include "access/htup_details.h"
 #include "lib/ilist.h"
 #include "storage/sinval.h"
+#include "storage/standby.h"
 #include "utils/hsearch.h"
 #include "utils/relcache.h"
 #include "utils/snapshot.h"
@@ -593,6 +594,9 @@ struct ReorderBuffer
 	/* memory accounting */
 	Size		size;
 
+	TransactionId *last_running_xacts;
+	int n_last_running_xacts;	/* -1 for initial value */
+
 	/*
 	 * Statistics about transactions spilled to disk.
 	 *
@@ -682,4 +686,9 @@ extern void ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);
 
 extern void StartupReorderBuffer(void);
 
+void		ReorderBufferProcessLastRunningXacts(ReorderBuffer *rb, xl_running_xacts *running);
+void		ReorderBufferSetLastRunningXactsCatalogChanges(ReorderBuffer *rb, TransactionId xid,
+														   uint32 xinfo, int subxcnt,
+														   TransactionId *subxacts, XLogRecPtr lsn);
+
 #endif