From 42a86ac7bd35c84a40de53212cc52c74076c8309 Mon Sep 17 00:00:00 2001 From: proto Date: Fri, 24 Jul 2026 17:08:17 +0000 Subject: [PATCH 1/2] Add decoding_xid_status_hook for storage-engine logical xids Storage engines whose transactions carry logical-only xids (no clog backing) cannot answer TransactionIdIsInProgress()/TransactionIdDidCommit() during streamed logical decoding: the clog lookup fails with 'could not access status of transaction N'. Let the engine supply the status via a hook, consulted by SetupCheckXidLive() and HandleConcurrentAbort() before falling back to the clog paths. --- src/backend/access/index/genam.c | 18 ++++++++++++++++-- src/backend/access/transam/xact.c | 2 ++ .../replication/logical/reorderbuffer.c | 14 ++++++++++++++ src/include/access/xact.h | 18 ++++++++++++++++++ 4 files changed, 50 insertions(+), 2 deletions(-) diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c index e172025fe04..85b15ee6835 100644 --- a/src/backend/access/index/genam.c +++ b/src/backend/access/index/genam.c @@ -483,8 +483,22 @@ systable_beginscan(Relation heapRelation, static inline void HandleConcurrentAbort() { - if (TransactionIdIsValid(CheckXidAlive) && - !TransactionIdIsInProgress(CheckXidAlive) && + if (!TransactionIdIsValid(CheckXidAlive)) + return; + + if (decoding_xid_status_hook) + { + DecodingXidStatus status = decoding_xid_status_hook(CheckXidAlive); + + if (status == DECODING_XID_ABORTED) + ereport(ERROR, + (errcode(ERRCODE_TRANSACTION_ROLLBACK), + errmsg("transaction aborted during system catalog scan"))); + if (status != DECODING_XID_NOT_HANDLED) + return; + } + + if (!TransactionIdIsInProgress(CheckXidAlive) && !TransactionIdDidCommit(CheckXidAlive)) ereport(ERROR, (errcode(ERRCODE_TRANSACTION_ROLLBACK), diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index bdd886ed31b..f3afc8f0fcd 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -95,6 +95,8 @@ int synchronous_commit = SYNCHRONOUS_COMMIT_ON; * concurrent aborts only in systable_* APIs. */ TransactionId CheckXidAlive = InvalidTransactionId; + +decoding_xid_status_hook_type decoding_xid_status_hook = NULL; bool bsysscan = false; /* diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index ce4282b728f..f80f9a68bf3 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -1999,6 +1999,20 @@ SetupCheckXidLive(TransactionId xid) * setup CheckXidAlive if it's not committed yet. We don't check if the * xid is aborted. That will happen during catalog access. */ + if (decoding_xid_status_hook) + { + DecodingXidStatus status = decoding_xid_status_hook(xid); + + if (status != DECODING_XID_NOT_HANDLED) + { + if (status == DECODING_XID_COMMITTED) + CheckXidAlive = InvalidTransactionId; + else + CheckXidAlive = xid; + return; + } + } + if (!TransactionIdDidCommit(xid)) CheckXidAlive = xid; else diff --git a/src/include/access/xact.h b/src/include/access/xact.h index 414f29267e7..59d7883bce3 100644 --- a/src/include/access/xact.h +++ b/src/include/access/xact.h @@ -84,6 +84,24 @@ extern PGDLLIMPORT int synchronous_commit; /* used during logical streaming of a transaction */ extern PGDLLIMPORT TransactionId CheckXidAlive; + +/* + * Status of a decoded transaction id as reported by a storage engine whose + * transactions carry logical-only xids (no clog backing). Consulted by the + * concurrent-abort detection during streamed logical decoding instead of + * TransactionIdIsInProgress()/TransactionIdDidCommit(), which would attempt + * a clog lookup for an xid that no transam machinery ever assigned. + */ +typedef enum DecodingXidStatus +{ + DECODING_XID_NOT_HANDLED, /* not a logical-only xid; use clog paths */ + DECODING_XID_IN_PROGRESS, + DECODING_XID_COMMITTED, + DECODING_XID_ABORTED +} DecodingXidStatus; + +typedef DecodingXidStatus (*decoding_xid_status_hook_type) (TransactionId xid); +extern PGDLLIMPORT decoding_xid_status_hook_type decoding_xid_status_hook; extern PGDLLIMPORT bool bsysscan; /* From a5a984d2f8cc232046c3fc2e5ae8f40f3483e7e8 Mon Sep 17 00:00:00 2001 From: proto Date: Fri, 24 Jul 2026 17:08:17 +0000 Subject: [PATCH 2/2] Add RBTXN_NO_STREAMING reorder buffer transaction flag Lets an rmgr decode callback exclude specific transactions from streaming selection while leaving streaming enabled for the rest of the decoding session. Checked in ReorderBufferLargestStreamableTopTXN() and in the post-serialize stream path. --- src/backend/replication/logical/reorderbuffer.c | 5 +++-- src/include/replication/reorderbuffer.h | 4 ++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index f80f9a68bf3..cc08dfbc317 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -794,7 +794,8 @@ ReorderBufferProcessPartialChange(ReorderBuffer *rb, ReorderBufferTXN *txn, if (ReorderBufferCanStartStreaming(rb) && !(rbtxn_has_partial_change(toptxn)) && rbtxn_is_serialized(txn) && - rbtxn_has_streamable_change(toptxn)) + rbtxn_has_streamable_change(toptxn) && + !rbtxn_no_streaming(toptxn)) ReorderBufferStreamTXN(rb, toptxn); } @@ -3753,7 +3754,7 @@ ReorderBufferLargestStreamableTopTXN(ReorderBuffer *rb) if ((largest == NULL || txn->total_size > largest_size) && (txn->total_size > 0) && !(rbtxn_has_partial_change(txn)) && - rbtxn_has_streamable_change(txn)) + rbtxn_has_streamable_change(txn) && !rbtxn_no_streaming(txn)) { largest = txn; largest_size = txn->total_size; diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 05fa54da999..88cee01e40a 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -170,6 +170,10 @@ typedef struct ReorderBufferChange #define RBTXN_HAS_STREAMABLE_CHANGE 0x0100 #define RBTXN_DISTR_INVAL_OVERFLOWED 0x0200 #define RBTXN_DISTR_SKIP_CLEANUP 0x0400 +#define RBTXN_NO_STREAMING 0x0800 + +/* Has a storage engine marked this transaction as not streamable? */ +#define rbtxn_no_streaming(txn) ( ((txn)->txn_flags & RBTXN_NO_STREAMING) != 0 ) /* Does the transaction have catalog changes? */ #define rbtxn_has_catalog_changes(txn) \