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..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); } @@ -1999,6 +2000,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 @@ -3739,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/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; /* 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) \