From c1d8159ca8f489081171c73d80b158d2c9012c39 Mon Sep 17 00:00:00 2001 From: "yangboyu.yby" 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