poc_add_running_catchanges_xacts_to_serialized_snapshot.patch

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 8da5f9089c..19123cbfa3 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -4821,6 +4821,45 @@ ReorderBufferToastReset(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	txn->toast_hash = NULL;
 }
 
+/*
+ * Return palloc'ed array of the transactions that have changed catalogs.
+ * The returned array is sorted in xidComparator order.
+ *
+ * The caller must free the returned array when done with it.
+ */
+TransactionId *
+ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb, size_t *xcnt_p)
+{
+	HASH_SEQ_STATUS hash_seq;
+	ReorderBufferTXNByIdEnt *ent;
+	TransactionId *xids;
+	size_t	xcnt = 0;
+	size_t	xcnt_space = 64; /* arbitrary number */
+
+	xids = (TransactionId *) palloc(sizeof(TransactionId) * xcnt_space);
+
+	hash_seq_init(&hash_seq, rb->by_txn);
+	while ((ent = hash_seq_search(&hash_seq)) != NULL)
+	{
+		ReorderBufferTXN *txn = ent->txn;
+
+		if (!rbtxn_has_catalog_changes(txn))
+			continue;
+
+		if (xcnt >= xcnt_space)
+		{
+			xcnt_space *= 2;
+			xids = repalloc(xids, sizeof(TransactionId) * xcnt_space);
+		}
+
+		xids[xcnt++] = txn->xid;
+	}
+
+	qsort(xids, xcnt, sizeof(TransactionId), xidComparator);
+
+	*xcnt_p = xcnt;
+	return xids;
+}
 
 /* ---------------------------------------
  * Visibility support for logical decoding
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 1119a12db9..c57d5f91d9 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -241,6 +241,26 @@ struct SnapBuild
 		 */
 		TransactionId *xip;
 	}			committed;
+
+	/*
+	 * Array of transactions that were running when the snapshot serialization
+	 * and changed system catalogs, but that are not committed.
+	 *
+	 * We normally rely on HEAP2_NEW_CID records and XLOG_XACT_INVALIDATIONS to
+	 * know if the transaction has changed the catalog. But it could happen that
+	 * the logical decoding decodes only the commit record of the transaction.
+	 * This array keeps track of the transactions that were running but not
+	 * committed when serializing and restoring a snapshot, and is used to add
+	 * such transactions to the snapshot.
+	 */
+	struct
+	{
+		/* number of transactions */
+		size_t		xcnt;
+
+		/* This array must be sorted in xidComparator order */
+		TransactionId *xip;
+	}			catchanges;
 };
 
 /*
@@ -306,6 +326,9 @@ AllocateSnapshotBuilder(ReorderBuffer *reorder,
 		palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
 	builder->committed.includes_all_transactions = true;
 
+	builder->catchanges.xcnt = 0;
+	builder->catchanges.xip = NULL;
+
 	builder->initial_xmin_horizon = xmin_horizon;
 	builder->start_decoding_at = start_lsn;
 	builder->building_full_snapshot = need_full_snapshot;
@@ -983,7 +1006,9 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 		 * Add subtransaction to base snapshot if catalog modifying, we don't
 		 * distinguish to toplevel transactions there.
 		 */
-		if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid))
+		if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid) ||
+			bsearch(&xid, builder->catchanges.xip, builder->catchanges.xcnt,
+					sizeof(TransactionId), xidComparator) != NULL)
 		{
 			sub_needs_timetravel = true;
 			needs_snapshot = true;
@@ -1012,7 +1037,9 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 	}
 
 	/* if top-level modified catalog, it'll need a snapshot */
-	if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
+	if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid) ||
+		bsearch(&xid, builder->catchanges.xip, builder->catchanges.xcnt,
+				sizeof(TransactionId), xidComparator) != NULL)
 	{
 		elog(DEBUG2, "found top level transaction %u, with catalog changes",
 			 xid);
@@ -1438,6 +1465,7 @@ SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
  *
  * struct SnapBuildOnDisk;
  * TransactionId * committed.xcnt; (*not xcnt_space*)
+ * TransactionId * catchanges.xcnt;
  *
  */
 typedef struct SnapBuildOnDisk
@@ -1578,8 +1606,17 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
 				(errcode_for_file_access(),
 				 errmsg("could not remove file \"%s\": %m", tmppath)));
 
+	/*
+	 * Update the transactions that are running and changes catalogs that are
+	 * not committed.
+	 */
+	if (builder->catchanges.xip)
+		pfree(builder->catchanges.xip);
+	builder->catchanges.xip = ReorderBufferGetCatalogChangesXacts(builder->reorder,
+																  &builder->catchanges.xcnt);
+
 	needed_length = sizeof(SnapBuildOnDisk) +
-		sizeof(TransactionId) * builder->committed.xcnt;
+		sizeof(TransactionId) * (builder->committed.xcnt + builder->catchanges.xcnt);
 
 	ondisk_c = MemoryContextAllocZero(builder->context, needed_length);
 	ondisk = (SnapBuildOnDisk *) ondisk_c;
@@ -1598,6 +1635,7 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
 	ondisk->builder.snapshot = NULL;
 	ondisk->builder.reorder = NULL;
 	ondisk->builder.committed.xip = NULL;
+	ondisk->builder.catchanges.xip = NULL;
 
 	COMP_CRC32C(ondisk->checksum,
 				&ondisk->builder,
@@ -1609,6 +1647,12 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
 	COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
 	ondisk_c += sz;
 
+	/* copy catalog-changes xacts */
+	sz = sizeof(TransactionId) * builder->catchanges.xcnt;
+	memcpy(ondisk_c, builder->catchanges.xip, sz);
+	COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
+	ondisk_c += sz;
+
 	FIN_CRC32C(ondisk->checksum);
 
 	/* we have valid data now, open tempfile and write it there */
@@ -1832,6 +1876,33 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
 	}
 	COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
 
+	/* restore catalog-changes xacts information */
+	sz = sizeof(TransactionId) * ondisk.builder.catchanges.xcnt;
+	ondisk.builder.catchanges.xip = MemoryContextAllocZero(builder->context, sz);
+	pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
+	readBytes = read(fd, ondisk.builder.catchanges.xip, sz);
+	pgstat_report_wait_end();
+	if (readBytes != sz)
+	{
+		int			save_errno = errno;
+
+		CloseTransientFile(fd);
+
+		if (readBytes < 0)
+		{
+			errno = save_errno;
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read file \"%s\": %m", path)));
+		}
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_DATA_CORRUPTED),
+					 errmsg("could not read file \"%s\": read %d of %zu",
+							path, readBytes, sz)));
+	}
+	COMP_CRC32C(checksum, ondisk.builder.catchanges.xip, sz);
+
 	if (CloseTransientFile(fd) != 0)
 		ereport(ERROR,
 				(errcode_for_file_access(),
@@ -1885,6 +1956,14 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
 	}
 	ondisk.builder.committed.xip = NULL;
 
+	builder->catchanges.xcnt = ondisk.builder.catchanges.xcnt;
+	if (builder->catchanges.xcnt > 0)
+	{
+		pfree(builder->committed.xip);
+		builder->catchanges.xip = ondisk.builder.catchanges.xip;
+	}
+	ondisk.builder.catchanges.xip = NULL;
+
 	/* our snapshot is not interesting anymore, build a new one */
 	if (builder->snapshot != NULL)
 	{
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 4a01f877e5..07e378d3ef 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -677,6 +677,7 @@ extern void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid);
 extern void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid);
 extern ReorderBufferTXN *ReorderBufferGetOldestTXN(ReorderBuffer *);
 extern TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb);
+extern TransactionId *ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb, size_t *xcnt_p);
 
 extern void ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);