From c1d8159ca8f489081171c73d80b158d2c9012c39 Mon Sep 17 00:00:00 2001
From: "yangboyu.yby" <yangboyu.yby@alibaba-inc.com>
Date: Tue, 25 Aug 2026 17:07:06 +0800
Subject: [PATCH] Skip snapshot distribution for catalog commits in other
 databases

A catalog-changing commit in any database made every logical decoding
session rebuild its historic snapshot and distribute it, together with
the invalidation messages, to all in-progress transactions, regardless
of which database the commit belonged to.  With a long-running
transaction pinning xmin, the distributed snapshot's xip array keeps
growing, so spill files in pg_replslot grew quadratically with the
number of catalog commits, multiplied by the number of slots.

A decoding session is connected to a single database and only ever
reads that database's catalogs plus shared catalogs.  Catalog changes
committed in another database that touch no shared catalog can never
affect it, so such commits now skip rebuilding the snapshot and
distributing it, while the snapshot builder state is still maintained.
Whether a commit touched a shared catalog is determined from the
invalidation messages in its commit record, which works because every
shared catalog is covered by a catcache or by
RelationInvalidatesSnapshotsOnly(), so any write to one of them
generates invalidation messages.
---
 src/backend/replication/logical/decode.c    | 47 ++++++++++++++++++++-
 src/backend/replication/logical/snapbuild.c | 36 ++++++++++------
 src/include/replication/snapbuild.h         |  3 +-
 3 files changed, 72 insertions(+), 14 deletions(-)

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index c944be4ac83..8810686db24 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -68,6 +68,8 @@ static inline bool FilterPrepare(LogicalDecodingContext *ctx,
 static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx,
 							  XLogRecordBuffer *buf, Oid txn_dbid,
 							  ReplOriginId origin_id);
+static bool InvalidationsTouchSharedCatalog(uint32 nmsgs,
+											SharedInvalidationMessage *msgs);

 /*
  * Take every XLogReadRecord()ed record and perform the actions required to
@@ -692,6 +694,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	XLogRecPtr	origin_lsn = InvalidXLogRecPtr;
 	TimestampTz commit_time = parsed->xact_time;
 	ReplOriginId origin_id = XLogRecGetOrigin(buf->record);
+	bool		distribute = true;
 	int			i;

 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
@@ -700,9 +703,19 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		commit_time = parsed->origin_timestamp;
 	}

+	/*
+	 * Catalog changes in another database cannot influence the decoding of
+	 * this slot, unless shared catalogs were touched.  All changes to shared
+	 * catalogs produce invalidation messages, so we can use the presence of
+	 * such messages to determine that.
+	 */
+	if (parsed->dbId != ctx->slot->data.database &&
+		!InvalidationsTouchSharedCatalog(parsed->nmsgs, parsed->msgs))
+		distribute = false;
+
 	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
 					   parsed->nsubxacts, parsed->subxacts,
-					   parsed->xinfo);
+					   parsed->xinfo, distribute);

 	/* ----
 	 * Check whether we are interested in this specific transaction, and tell
@@ -1344,3 +1357,35 @@ DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,

 	return false;
 }
+
+/*
+ * Check whether any of the given invalidation messages concern a shared
+ * catalog.
+ */
+static bool
+InvalidationsTouchSharedCatalog(uint32 nmsgs, SharedInvalidationMessage *msgs)
+{
+	uint32		i;
+
+	for (i = 0; i < nmsgs; i++)
+	{
+		SharedInvalidationMessage *msg = &msgs[i];
+
+		if (msg->id == SHAREDINVALSMGR_ID)
+		{
+			/* shared relations have a zero dbOid in their RelFileLocator */
+			if (msg->sm.rlocator.dbOid == InvalidOid)
+				return true;
+		}
+		else if (msg->cc.dbId == InvalidOid)
+		{
+			/*
+			 * All the other message types carry the database ID (0 for a
+			 * shared relation) right after the type field.
+			 */
+			return true;
+		}
+	}
+
+	return false;
+}
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index f60bcf09605..86796ff125b 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -940,10 +940,14 @@ SnapBuildPurgeOlderTxn(SnapBuild *builder)

 /*
  * Handle everything that needs to be done when a transaction commits
+ *
+ * distribute indicates whether the committing transaction's catalog changes
+ * are relevant to the database this decoding session is connected to.
  */
 void
 SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
-				   int nsubxacts, TransactionId *subxacts, uint32 xinfo)
+				   int nsubxacts, TransactionId *subxacts, uint32 xinfo,
+				   bool distribute)
 {
 	int			nxact;

@@ -1075,14 +1079,22 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 			return;

 		/*
-		 * Decrease the snapshot builder's refcount of the old snapshot, note
-		 * that it still will be used if it has been handed out to the
-		 * reorderbuffer earlier.
+		 * To rebuild the builder's snapshot, decrease the snapshot builder's
+		 * refcount of the old snapshot, note that it still will be used if
+		 * it has been handed out to the reorderbuffer earlier.  Skip the
+		 * rebuilding when the catalog changes are not relevant to our
+		 * database.
 		 */
-		if (builder->snapshot)
-			SnapBuildSnapDecRefcount(builder->snapshot);
+		if (distribute || builder->snapshot == NULL)
+		{
+			if (builder->snapshot)
+				SnapBuildSnapDecRefcount(builder->snapshot);

-		builder->snapshot = SnapBuildBuildSnapshot(builder);
+			builder->snapshot = SnapBuildBuildSnapshot(builder);
+
+			/* refcount of the snapshot builder for the new snapshot */
+			SnapBuildSnapIncRefcount(builder->snapshot);
+		}

 		/* we might need to execute invalidations, add snapshot */
 		if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
@@ -1092,14 +1104,14 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 										 builder->snapshot);
 		}

-		/* refcount of the snapshot builder for the new snapshot */
-		SnapBuildSnapIncRefcount(builder->snapshot);
-
 		/*
 		 * Add a new catalog snapshot and invalidations messages to all
-		 * currently running transactions.
+		 * currently running transactions, unless the committing
+		 * transaction's catalog changes cannot influence the decoding of
+		 * our database.
 		 */
-		SnapBuildDistributeSnapshotAndInval(builder, lsn, xid);
+		if (distribute)
+			SnapBuildDistributeSnapshotAndInval(builder, lsn, xid);
 	}
 }

diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index a22a83a2f23..bbbad7ff75d 100644
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -85,7 +85,8 @@ extern void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr);

 extern void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
 							   TransactionId xid, int nsubxacts,
-							   TransactionId *subxacts, uint32 xinfo);
+							   TransactionId *subxacts, uint32 xinfo,
+							   bool distribute);
 extern bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid,
 								   XLogRecPtr lsn);
 extern void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
--
2.43.7

