Reorderbuffer: handle catalog-change-bearing aborted subxacts better

When constructing a historic snapshot for logical decoding,
reorderbuffer needs to keep track of movement of tuples in catalog
tables.  For the most part this works correctly, except that when a
subtransaction that wrote a catalog tuple aborts, we failed to "forget"
about that change.  There's no known reports of anything actually
breaking except assertion failures, but it is likely possible to
construct cases where historical snapshots produce bogus visibility
answers, potentially causing incorrect decoding output.

To fix, have the decode of a subtransaction abort go over the TUPLECID
list that has been accumulated on the ReorderBufferTXN for its topmost
parent transaction, and remove all the entries that were marked as
originating from the aborted subtransaction or its children.  There are
some little details in which various devils lie, which the added code
comments explain (we hope).

Add a couple of isolation test cases that trip the assertions with the
original coding.  They're not ideal test design, because they pass in
non-assert builds without the fix.  But we'd rather have the fix be
verified in that limited scenario than nowhere at all, and they are
cheap enough.

This bug is sibling to the one fixed in commit 8c67d29fd5 and, like
that one, it goes all the way back to the introduction of
reorderbuffers, so backpatch to all live branches.

While reviewing this, I (Álvaro) was bothered by the snapmgr.c
tuplecid_data file-scoped variable, as it's easily confused with a
function parameter used widely in reorderbuffer.c.  In branches 19 and
master, rename that to avoid confusion.

Reported-by: Arseny Sher <a.sher@postgrespro.ru>
Reported-by: Mark Dilger <mark.dilger@enterprisedb.com>
Reported-by: Alexander Kozhemyakin <a.kozhemyakin@postgrespro.ru>
Author: Bingshuai Li <lucian1412@outlook.com>
Author: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Author: Zhijie Hou <houzhijie22@gmail.com>
Backpatch-through: 14
Bug: #19555
Discussion: https://postgr.es/m/87r2c8xocq.fsf@ars-thinkpad
Discussion: https://postgr.es/m/CAHgHdKu5e3XY5e90Tuaxq_R4WrKxSV734Q%2BLwo5y39Omp2A-Gg@mail.gmail.com
Discussion: https://postgr.es/m/19555-3c700f40a13045cd@postgresql.org

src/backend/replication/logical/decode.c

@@ -839,7 +839,6 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 			xl_xact_parsed_abort *parsed, TransactionId xid,
 			bool two_phase)
 {
-	int			i;
 	XLogRecPtr	origin_lsn = InvalidXLogRecPtr;
 	TimestampTz abort_time = parsed->xact_time;
 	RepOriginId origin_id = XLogRecGetOrigin(buf->record);
@@ -871,7 +870,15 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	}
 	else
 	{
-		for (i = 0; i < parsed->nsubxacts; i++)
+		/*
+		 * Remove tuplecid changes queued by the aborted subtransactions from
+		 * the toplevel's list.
+		 */
+		ReorderBufferCleanupAbortedTupleCids(ctx->reorder, xid,
+											 parsed->nsubxacts,
+											 parsed->subxacts);
+
+		for (int i = 0; i < parsed->nsubxacts; i++)
 		{
 			ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
 							   buf->record->EndRecPtr, abort_time);

src/backend/replication/logical/reorderbuffer.c

@@ -241,6 +241,8 @@ static void ReorderBufferTransferSnapToParent(ReorderBufferTXN *txn,
 											  ReorderBufferTXN *subtxn);
 
 static void AssertTXNLsnOrder(ReorderBuffer *rb);
+static bool TransactionIdInSubxactArray(TransactionId xid,
+										TransactionId *subxacts, int nsubxacts);
 
 /* ---------------------------------------
  * support functions for lsn-order iterating over the ->changes of a
@@ -3237,6 +3239,102 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
 	ReorderBufferCleanupTXN(rb, txn);
 }
 
+/*
+ * Remove TUPLECID changes from a top-level transaction's tuplecid list.
+ *
+ * 'xid' is a subtransaction being aborted, and 'subxacts' is the list of
+ * its children -- which are, obviously, also being aborted.
+ *
+ * Unlike other types of changes, tuplecid ones are always queued on the
+ * toplevel transaction (see ReorderBufferAddNewTupleCids), so we must search
+ * for these changes by subxid.  Because TIDs used by a catalog tuple written
+ * by a subtransaction that aborts can be reused later by a future command
+ * of the same transaction, we must remove these TUPLECIDs from the array,
+ * otherwise the tuplecid hash of the historical snapshot would be corrupt.
+ *
+ * We return without cleanup if the aborting transaction is not known to this
+ * reorderbuffer to be a subxact.  It may still have queued TUPLECIDs onto the
+ * toplevel xact, if this snapbuild missed the WAL record(s) carrying the
+ * association between the two, but that should not be a problem because in
+ * that case we would "skip" the transaction at DecodeCommit.
+ *
+ * (This cleanup is pointless when xid is a top-level transaction.  That case
+ * is handled instead by ReorderBufferCleanupTXN() clearing the whole list.)
+ */
+void
+ReorderBufferCleanupAbortedTupleCids(ReorderBuffer *rb,
+									 TransactionId xid,
+									 int nsubxacts,
+									 TransactionId *subxacts)
+{
+	ReorderBufferTXN *txn;
+	ReorderBufferTXN *toptxn;
+	dlist_mutable_iter it;
+
+	/*
+	 * Search the passed-in xids for the first known subtransaction, and use
+	 * it to find the toplevel transaction.
+	 */
+	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false);
+	for (int i = 0; txn == NULL && i < nsubxacts; i++)
+		txn = ReorderBufferTXNByXid(rb, subxacts[i], false, NULL,
+									InvalidXLogRecPtr, false);
+
+	/*
+	 * Skip when nothing usable was found.  No entry at all means no NEW_CID
+	 * record from these xids was decoded by this snapbuild, so nothing was
+	 * queued.  In the latter case, we won't find the TUPLECIDs, but we don't
+	 * care.  If this is a top-level xact, this cleanup is not necessary.
+	 */
+	if (txn == NULL || !rbtxn_is_known_subxact(txn))
+		return;
+
+	/* Get the top-level transaction whose tuplecid list we're to clean */
+	toptxn = rbtxn_get_toptxn(txn);
+
+	dlist_foreach_modify(it, &toptxn->tuplecids)
+	{
+		ReorderBufferChange *change;
+
+		change = dlist_container(ReorderBufferChange, node, it.cur);
+
+		Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
+
+		/*
+		 * Remove the entry if it was recorded on behalf of the aborting
+		 * subtransaction or one of its subcommitted children.  The subxact
+		 * array is known sorted in logical order because it comes directly
+		 * from the xl_xact_abort WAL record, which has them sorted that way
+		 * by construction (Xid assign order.)
+		 */
+		if (change->data.tuplecid.subxid == xid ||
+			TransactionIdInSubxactArray(change->data.tuplecid.subxid,
+										subxacts, nsubxacts))
+		{
+			dlist_delete(&change->node);
+			ReorderBufferFreeChange(rb, change, false);
+			Assert(toptxn->ntuplecids > 0);
+			toptxn->ntuplecids--;
+		}
+	}
+}
+
+/*
+ * Check whether the given xid exists in the subxacts array, which must be
+ * sorted in logical order.
+ */
+static bool
+TransactionIdInSubxactArray(TransactionId xid, TransactionId *subxacts,
+							int nsubxacts)
+{
+	/* if there are no subxacts, it's not there */
+	if (nsubxacts == 0)
+		return false;
+	/* otherwise bsearch the (logically ordered) given subxact array */
+	return bsearch(&xid, subxacts, nsubxacts,
+				   sizeof(TransactionId), xidLogicalComparator) != NULL;
+}
+
 /*
  * Invalidate cache for those transactions that need to be skipped just in case
  * catalogs were manipulated as part of the transaction.
@@ -3469,15 +3567,21 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
 /*
  * Add new (relfilelocator, tid) -> (cmin, cmax) mappings.
  *
+ * The xid is always the top-level xid, for easy hash construction in
+ * ReorderBufferBuildTupleCidHash.  We need the subxid separately for
+ * cleanup during subxact aborts; see ReorderBufferCleanupAbortedTupleCids.
+ *
  * We do not include this change type in memory accounting, because we
  * keep CIDs in a separate list and do not evict them when reaching
  * the memory limit.
  */
 void
-ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
+ReorderBufferAddNewTupleCids(ReorderBuffer *rb,
+							 TransactionId xid, TransactionId subxid,
 							 XLogRecPtr lsn, RelFileLocator locator,
-							 ItemPointerData tid, CommandId cmin,
-							 CommandId cmax, CommandId combocid)
+							 ItemPointerData tid,
+							 CommandId cmin, CommandId cmax,
+							 CommandId combocid)
 {
 	ReorderBufferChange *change = ReorderBufferAllocChange(rb);
 	ReorderBufferTXN *txn;
@@ -3489,6 +3593,7 @@ ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
 	change->data.tuplecid.cmin = cmin;
 	change->data.tuplecid.cmax = cmax;
 	change->data.tuplecid.combocid = combocid;
+	change->data.tuplecid.subxid = subxid;
 	change->lsn = lsn;
 	change->txn = txn;
 	change->action = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID;

src/backend/replication/logical/snapbuild.c

@@ -737,7 +737,7 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
 	 */
 	ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
 
-	ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
+	ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, xid, lsn,
 								 xlrec->target_locator, xlrec->target_tid,
 								 xlrec->cmin, xlrec->cmax,
 								 xlrec->combocid);

src/include/replication/reorderbuffer.h

@@ -136,8 +136,12 @@ typedef struct ReorderBufferChange
 		CommandId	command_id;
 
 		/*
-		 * New cid mapping for catalog changing transaction, set when action
-		 * == *_INTERNAL_TUPLECID.
+		 * tuplecid changes are used to follow catalog tuple movements (i.e.,
+		 * DDL changes); we need them to create the HistoricTupleCidHash table
+		 * used with historical snapshots.  For historical reasons we store
+		 * these with the toplevel txn, not each subtxn; and yet we need to
+		 * remove them on subxact aborts, so we record the subxid separately.
+		 * Set when action == *_INTERNAL_TUPLECID.
 		 */
 		struct
 		{
@@ -146,6 +150,7 @@ typedef struct ReorderBufferChange
 			CommandId	cmin;
 			CommandId	cmax;
 			CommandId	combocid;
+			TransactionId subxid;
 		}			tuplecid;
 
 		/* Invalidation. */
@@ -737,6 +742,10 @@ extern void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr
 							   TimestampTz abort_time);
 extern void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid);
 extern void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
+extern void ReorderBufferCleanupAbortedTupleCids(ReorderBuffer *rb,
+												 TransactionId xid,
+												 int nsubxacts,
+												 TransactionId *subxacts);
 extern void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
 
 extern void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid,
@@ -746,6 +755,7 @@ extern void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid,
 extern void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid,
 										 XLogRecPtr lsn, CommandId cid);
 extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
+										 TransactionId subxid,
 										 XLogRecPtr lsn, RelFileLocator locator,
 										 ItemPointerData tid,
 										 CommandId cmin, CommandId cmax, CommandId combocid);