Add decoding_xid_status_hook and RBTXN_NO_STREAMING for storage-engine logical xids

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),

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;
 
 /*

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;

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;
 
 /*

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) \