From 6da5e4943ec95928a34502f76e9cdf698ef4b902 Mon Sep 17 00:00:00 2001 From: "yangboyu.yby" Date: Thu, 15 Oct 2026 19:05:19 +0800 Subject: [PATCH] logical decoding: 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. 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. --- contrib/test_decoding/meson.build | 1 + contrib/test_decoding/t/002_two_databases.pl | 164 +++++++++++++++++++ src/backend/replication/logical/decode.c | 79 ++++++++- src/backend/replication/logical/snapbuild.c | 24 ++- src/include/replication/snapbuild.h | 3 +- 5 files changed, 262 insertions(+), 9 deletions(-) create mode 100644 contrib/test_decoding/t/002_two_databases.pl diff --git a/contrib/test_decoding/meson.build b/contrib/test_decoding/meson.build index dbc0ef0f63e..12270efc1fb 100644 --- a/contrib/test_decoding/meson.build +++ b/contrib/test_decoding/meson.build @@ -84,6 +84,7 @@ tests += { 'tap': { 'tests': [ 't/001_repl_stats.pl', + 't/002_two_databases.pl', ], }, } diff --git a/contrib/test_decoding/t/002_two_databases.pl b/contrib/test_decoding/t/002_two_databases.pl new file mode 100644 index 00000000000..3894cda2328 --- /dev/null +++ b/contrib/test_decoding/t/002_two_databases.pl @@ -0,0 +1,164 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test that catalog commits in other databases do not disturb logical +# decoding: the snapshot rebuild and the distribution of the snapshot and +# the invalidation messages to the in-progress transactions are skipped for +# them unless shared catalogs were touched, and the decoding of the slot's +# own database produces correct results either way. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $node = PostgreSQL::Test::Cluster->new('test'); +$node->init(allows_streaming => 'logical'); +$node->start; + +# db1 holds the replication slot; db2 acts as the other database whose +# catalog commits must not disturb the decoding of db1. +$node->safe_psql('postgres', 'CREATE DATABASE db1'); +$node->safe_psql('postgres', 'CREATE DATABASE db2'); + +$node->safe_psql('db1', 'CREATE TABLE t1(a int)'); +$node->safe_psql('db1', + "SELECT pg_create_logical_replication_slot('slot_db1', 'test_decoding')"); +$node->safe_psql('db2', 'CREATE TABLE t2(a int)'); +$node->safe_psql('db2', 'CREATE ROLE regress_role1'); +$node->safe_psql('db2', 'CREATE PUBLICATION pub0 FOR ALL TABLES'); + +my $get_changes = qq( + SET log_min_messages = 'debug1'; + SELECT data FROM pg_logical_slot_get_changes('slot_db1', NULL, NULL, + 'include-xids', '0', 'skip-empty-xacts', '1')); + +# Shared-catalog DDL committed in db2 while a transaction is in progress in +# db1. The commit touches a shared catalog, so the snapshot and the +# invalidations are still distributed; the decoding of db1's transaction +# must produce correct results. +my $session1 = $node->background_psql('db1', on_error_stop => 1); +$session1->query_safe('BEGIN'); +$session1->query_safe('INSERT INTO t1 VALUES (1)'); + +my $session_role = $node->background_psql('db2', on_error_stop => 1); +$session_role->query_safe('BEGIN'); +$session_role->query_safe('ALTER ROLE regress_role1 CONNECTION LIMIT 10'); +my $xid_role = $session_role->query_safe('SELECT pg_current_xact_id()'); +chomp($xid_role); +$session_role->query_safe('COMMIT'); +$session_role->quit; + +$session1->query_safe('COMMIT'); +$session1->quit; + +my $result = $node->safe_psql('db1', $get_changes); +is($result, qq(BEGIN +table public.t1: INSERT: a[integer]:1 +COMMIT), + 'shared-catalog DDL in db2 does not disturb decoding of db1'); + +my $log = slurp_file($node->logfile); +like($log, + qr/transaction $xid_role committed in database \d+ has catalog changes relevant/, + 'shared-catalog DDL in db2 classified as relevant'); + +# A catalog DDL committed in db2, with a shared-catalog DDL rolled back to +# a savepoint, while a transaction is in progress in db1. The invalidation +# messages of the rolled back savepoint were WAL-logged before the rollback +# and cannot be retracted from the reorder buffer, but the commit record +# carries only the surviving messages, which touch no shared catalog, so +# the snapshot distribution is skipped; the decoding of db1's transaction +# must still produce correct results. +$session1 = $node->background_psql('db1', on_error_stop => 1); +$session1->query_safe('BEGIN'); +$session1->query_safe('INSERT INTO t1 VALUES (2)'); + +my $session2 = $node->background_psql('db2', on_error_stop => 1); +$session2->query_safe('BEGIN'); +$session2->query_safe('ALTER TABLE t2 RENAME TO t2_new'); +$session2->query_safe('SAVEPOINT s1'); +$session2->query_safe('ALTER ROLE regress_role1 RENAME TO regress_role2'); +$session2->query_safe('ROLLBACK TO SAVEPOINT s1'); +my $xid_savepoint = $session2->query_safe('SELECT pg_current_xact_id()'); +chomp($xid_savepoint); +$session2->query_safe('COMMIT'); +$session2->quit; + +$session1->query_safe('COMMIT'); +$session1->quit; + +$result = $node->safe_psql('db1', $get_changes); +is($result, qq(BEGIN +table public.t1: INSERT: a[integer]:2 +COMMIT), + 'rolled back shared-catalog DDL in db2 does not disturb decoding of db1'); + +$log = slurp_file($node->logfile); +like($log, + qr/transaction $xid_savepoint committed in database \d+ does not have catalog changes relevant/, + 'ddl+savepoint+ddl+rollback in db2 classified as irrelevant'); + +# A publication parameter change on a FOR ALL TABLES publication in db2 +# while a transaction is in progress in db1. The commit carries a +# whole-relcache invalidation broadcast (dbId and relId both 0), which is +# conservatively treated as touching a shared catalog, so the snapshot and +# the invalidations are still distributed; the decoding of db1's +# transaction must produce correct results. +$session1 = $node->background_psql('db1', on_error_stop => 1); +$session1->query_safe('BEGIN'); +$session1->query_safe('INSERT INTO t1 VALUES (3)'); + +my $session_pub = $node->background_psql('db2', on_error_stop => 1); +$session_pub->query_safe('BEGIN'); +$session_pub->query_safe( + "ALTER PUBLICATION pub0 SET (publish = 'insert, update, delete')"); +my $xid_pub = $session_pub->query_safe('SELECT pg_current_xact_id()'); +chomp($xid_pub); +$session_pub->query_safe('COMMIT'); +$session_pub->quit; + +$session1->query_safe('COMMIT'); +$session1->quit; + +$result = $node->safe_psql('db1', $get_changes); +is($result, qq(BEGIN +table public.t1: INSERT: a[integer]:3 +COMMIT), + 'whole-relcache broadcast in db2 does not disturb decoding of db1'); + +$log = slurp_file($node->logfile); +like($log, + qr/transaction $xid_pub committed in database \d+ has catalog changes relevant/, + 'whole-relcache broadcast in db2 classified as relevant'); + + +# Many catalog commits in db2 with the xmin of the historic snapshots pinned +# by an in-progress transaction. Each such commit is irrelevant to this slot, +# so no snapshot is distributed and the reorder buffer must not spill. +my $session3 = $node->background_psql('db2', on_error_stop => 1); +$session3->query_safe('BEGIN'); +$session3->query_safe('INSERT INTO t2_new VALUES (1)'); + +my $session4 = $node->background_psql('db2', on_error_stop => 1); +for my $i (1 .. 300) +{ + $session4->query_safe("CREATE TABLE spill_t_$i(a int)"); +} +$session4->quit; + +$session3->query_safe('COMMIT'); +$session3->quit; + +$node->safe_psql('db1', 'CHECKPOINT'); +$node->safe_psql( + 'db1', qq( + SET logical_decoding_work_mem = '64kB'; + SELECT count(*) FROM pg_logical_slot_get_changes('slot_db1', NULL, NULL, + 'include-xids', '0', 'skip-empty-xacts', '1'))); + +$result = $node->safe_psql('db1', + "SELECT spill_bytes FROM pg_stat_replication_slots WHERE slot_name = 'slot_db1'"); +is($result, '0', + 'catalog commits in db2 do not spill the reorder buffer'); + +done_testing(); diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 4a739230264..4b07b10bc4b 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 catalog_changes_relevant = true; int i; if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN) @@ -700,9 +703,25 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, commit_time = parsed->origin_timestamp; } + /* + * Catalog changes committed 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 != InvalidOid && + parsed->dbId != ctx->slot->data.database && + !InvalidationsTouchSharedCatalog(parsed->nmsgs, parsed->msgs)) + catalog_changes_relevant = false; + + elog(DEBUG1, + "transaction %u committed in database %u %s catalog changes relevant", + xid, parsed->dbId, + catalog_changes_relevant ? "has" : "does not have"); + SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid, parsed->nsubxacts, parsed->subxacts, - parsed->xinfo); + parsed->xinfo, catalog_changes_relevant); /* ---- * Check whether we are interested in this specific transaction, and tell @@ -1353,3 +1372,61 @@ DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, return false; } + +/* + * Check whether any of the given invalidation messages concerns 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 >= 0) + { + /* catcache message; dbId is 0 for a shared catalog */ + if (msg->cc.dbId == InvalidOid) + return true; + } + else if (msg->id == SHAREDINVALCATALOG_ID) + { + if (msg->cat.dbId == InvalidOid) + return true; + } + else if (msg->id == SHAREDINVALRELCACHE_ID) + { + /* + * Be conservative as shared, although this does not mean a + * shared catalog changed if relId is also InvalidOid, such as + * in publication DDL + */ + if (msg->rc.dbId == InvalidOid) + return true; + } + else if (msg->id == SHAREDINVALSNAPSHOT_ID) + { + if (msg->sn.dbId == InvalidOid) + return true; + } + else if (msg->id == SHAREDINVALRELSYNC_ID) + { + if (msg->rs.dbId == InvalidOid) + return true; + } + else + { + /* + * smgr and relmap invalidations are non-transactional, + * so they cannot show up in a commit record. + */ + elog(ERROR, "unrecognized invalidation message ID %d", + msg->id); + } + } + + return false; +} diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index 69caf5b02e7..a93a6a25351 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -980,14 +980,18 @@ SnapBuildPurgeOlderTxn(SnapBuild *builder) /* * Handle everything that needs to be done when a transaction commits + * + * catalog_changes_relevant indicates whether the committing transaction's + * catalog changes (if any) can affect the decoding of this session. */ void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, - int nsubxacts, TransactionId *subxacts, uint32 xinfo) + int nsubxacts, TransactionId *subxacts, uint32 xinfo, + bool catalog_changes_relevant) { int nxact; - bool needs_snapshot = false; + bool has_catalog_changes = false; bool needs_timetravel = false; bool sub_needs_timetravel = false; @@ -1034,7 +1038,7 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo)) { sub_needs_timetravel = true; - needs_snapshot = true; + has_catalog_changes = true; elog(DEBUG1, "found subtransaction %u:%u with catalog changes", xid, subxid); @@ -1064,7 +1068,7 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, { elog(DEBUG2, "found top level transaction %u, with catalog changes", xid); - needs_snapshot = true; + has_catalog_changes = true; needs_timetravel = true; SnapBuildAddCommittedTxn(builder, xid); } @@ -1089,7 +1093,7 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, builder->committed.includes_all_transactions = false; } - Assert(!needs_snapshot || needs_timetravel); + Assert(!has_catalog_changes || needs_timetravel); /* * Adjust xmax of the snapshot builder, we only do that for committed, @@ -1104,8 +1108,14 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, TransactionIdAdvance(builder->xmax); } - /* if there's any reason to build a historic snapshot, do so now */ - if (needs_snapshot) + /* + * If there's any reason to build a historic snapshot, do so now. + * When the catalog changes are irrelevant, neither a rebuild nor a + * distribution is needed, since the builder's snapshot still reflects + * every catalog visible to this session and invalidation messages + * will also be filtered out by database id. + */ + if (has_catalog_changes && catalog_changes_relevant) { /* * If we haven't built a complete snapshot yet there's no need to hand diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h index a22a83a2f23..c2d02619f45 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 catalog_changes_relevant); extern bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn); extern void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, -- 2.32.0.3.g01195cf9f