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

contrib/test_decoding/Makefile

@@ -9,7 +9,8 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \
 ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \
 	oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \
 	twophase_snapshot catalog_change_snapshot skip_snapshot_restore \
-	invalidation_distribution
+	invalidation_distribution \
+	tuplecid_restart tuplecid_nested
 
 REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf
 ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf

contrib/test_decoding/expected/tuplecid_nested.out

@@ -0,0 +1,32 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s1_init s0_begin s0_topwrite s0_outer s0_inner s0_catinsert1 s0_release_inner s0_rollback_outer s2_vacuum s0_catinsert2 s0_commit s1_get_changes
+step s1_init: SELECT 'init' FROM pg_create_logical_replication_slot('tuplecid_nested_slot', 'test_decoding');
+?column?
+--------
+init    
+(1 row)
+
+step s0_begin: BEGIN;
+step s0_topwrite: INSERT INTO tbl1 VALUES (0);
+step s0_outer: SAVEPOINT outer_sp;
+step s0_inner: SAVEPOINT inner_sp;
+step s0_catinsert1: INSERT INTO user_cat VALUES (5, 'e');
+step s0_release_inner: RELEASE SAVEPOINT inner_sp;
+step s0_rollback_outer: ROLLBACK TO SAVEPOINT outer_sp;
+step s2_vacuum: VACUUM user_cat;
+step s0_catinsert2: INSERT INTO user_cat VALUES (6, 'f');
+step s0_commit: COMMIT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('tuplecid_nested_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%';
+data                                      
+------------------------------------------
+BEGIN                                     
+table public.tbl1: INSERT: val1[integer]:0
+COMMIT                                    
+(3 rows)
+
+?column?
+--------
+stop    
+(1 row)
+

contrib/test_decoding/expected/tuplecid_restart.out

@@ -0,0 +1,68 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s0_init s0_begin s0_savepoint s0_insert s1_checkpoint s1_get_changes s0_catinsert1 s0_rollback s2_vacuum s0_catinsert2 s1_checkpoint s1_get_changes s0_commit s1_get_changes
+step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding');
+?column?
+--------
+init    
+(1 row)
+
+step s0_begin: BEGIN;
+step s0_savepoint: SAVEPOINT sp1;
+step s0_insert: INSERT INTO tbl1 VALUES (1);
+step s1_checkpoint: CHECKPOINT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%';
+data
+----
+(0 rows)
+
+step s0_catinsert1: INSERT INTO user_cat VALUES (5, 'e');
+step s0_rollback: ROLLBACK TO SAVEPOINT sp1;
+step s2_vacuum: VACUUM user_cat;
+step s0_catinsert2: INSERT INTO user_cat VALUES (6, 'f');
+step s1_checkpoint: CHECKPOINT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%';
+data
+----
+(0 rows)
+
+step s0_commit: COMMIT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%';
+data  
+------
+BEGIN 
+COMMIT
+(2 rows)
+
+?column?
+--------
+stop    
+(1 row)
+
+
+starting permutation: s0_begin s0_savepoint s0_insert s1_init s0_catinsert1 s0_rollback s2_vacuum s0_catinsert2 s0_commit s1_get_changes
+step s0_begin: BEGIN;
+step s0_savepoint: SAVEPOINT sp1;
+step s0_insert: INSERT INTO tbl1 VALUES (1);
+step s1_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); <waiting ...>
+step s0_catinsert1: INSERT INTO user_cat VALUES (5, 'e');
+step s0_rollback: ROLLBACK TO SAVEPOINT sp1;
+step s2_vacuum: VACUUM user_cat;
+step s0_catinsert2: INSERT INTO user_cat VALUES (6, 'f');
+step s0_commit: COMMIT;
+step s1_init: <... completed>
+?column?
+--------
+init    
+(1 row)
+
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%';
+data
+----
+(0 rows)
+
+?column?
+--------
+stop    
+(1 row)
+

contrib/test_decoding/specs/tuplecid_nested.spec

@@ -0,0 +1,77 @@
+# Test for BUG #19555.
+#
+# Test decoding of a partially rolled back transaction where the aborted part
+# is nested: the aborting outer subtransaction never produced any WAL of its
+# own, so decoding cannot know its toplevel transaction, while an inner
+# subtransaction it rolls back did produce WAL and its tuplecid records still
+# have to be removed
+#
+# The tuplecid records written by the aborted inner subtransaction are queued
+# on the toplevel transaction and must be removed when the outer
+# subtransaction's abort is decoded; otherwise they collide with the records
+# of the line pointer that gets reused by the toplevel transaction's later
+# catalog insert.
+#
+#   - s0_topwrite makes the toplevel transaction produce its first WAL record
+#     before any subtransaction exists.
+#   - s0_outer establishes the outer savepoint, which never gets a WAL record
+#     of its own.
+#   - s0_catinsert1 generates an xl_heap_new_cid record in the inner
+#     subtransaction, which is then released.
+#   - s0_rollback_outer rolls the outer savepoint back.  Its abort record is
+#     written once the backend is already in TRANS_ABORT, so it carries no
+#     toplevel xid in its header and the outer subtransaction's association
+#     with the toplevel can never be established.  The released inner
+#     subtransaction is listed in the same abort record, though, and its
+#     tuplecid records still must be removed from the toplevel's list.
+#   - s2_vacuum reclaims the aborted row's line pointer (dead inserts are
+#     removed regardless of the xid horizon, so this is deterministic even
+#     while the toplevel transaction is still open); s0_catinsert2 then reuses
+#     the same tid with a different cmin.
+#   - s1_get_changes then decodes the toplevel commit and builds the tuplecid
+#     hash.
+setup
+{
+    DROP TABLE IF EXISTS tbl1;
+    DROP TABLE IF EXISTS user_cat;
+    CREATE TABLE tbl1 (val1 integer);
+    -- Four rows of ~1.5kB fill the first heap page to within a few hundred
+    -- bytes, so that vacuuming away the aborted fifth row leaves exactly one
+    -- line pointer for the next insert to reuse.
+    CREATE TABLE user_cat (c1 int, filler char(1500)) WITH (user_catalog_table = true);
+    INSERT INTO user_cat VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd');
+}
+
+teardown
+{
+    DROP TABLE tbl1;
+    DROP TABLE user_cat;
+    SELECT 'stop' FROM pg_drop_replication_slot('tuplecid_nested_slot');
+}
+
+session "s0"
+setup { SET synchronous_commit=on; }
+step "s0_begin" { BEGIN; }
+step "s0_topwrite" { INSERT INTO tbl1 VALUES (0); }
+step "s0_outer" { SAVEPOINT outer_sp; }
+step "s0_inner" { SAVEPOINT inner_sp; }
+step "s0_catinsert1" { INSERT INTO user_cat VALUES (5, 'e'); }
+step "s0_release_inner" { RELEASE SAVEPOINT inner_sp; }
+step "s0_rollback_outer" { ROLLBACK TO SAVEPOINT outer_sp; }
+step "s0_catinsert2" { INSERT INTO user_cat VALUES (6, 'f'); }
+step "s0_commit" { COMMIT; }
+
+session "s1"
+setup { SET synchronous_commit=on; }
+step "s1_init" { SELECT 'init' FROM pg_create_logical_replication_slot('tuplecid_nested_slot', 'test_decoding'); }
+# The user_cat rows carry ~1.5kB filler values which would flood the expected
+# output; filter them out (decoding still processes them server-side).
+step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('tuplecid_nested_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; }
+
+session "s2"
+setup { SET synchronous_commit=on; }
+step "s2_vacuum" { VACUUM user_cat; }
+
+# The slot is created before the transaction begins, so the final
+# s1_get_changes decodes the whole transaction through its commit.
+permutation "s1_init" "s0_begin" "s0_topwrite" "s0_outer" "s0_inner" "s0_catinsert1" "s0_release_inner" "s0_rollback_outer" "s2_vacuum" "s0_catinsert2" "s0_commit" "s1_get_changes"

contrib/test_decoding/specs/tuplecid_restart.spec

@@ -0,0 +1,74 @@
+# Test for BUG #19555.
+#
+# Test decoding of a transaction in which a catalog-modifying subtransaction
+# was rolled back, where the transaction spans a logical decoding restart
+# point.
+#
+# The tuplecid records written by the aborted subtransaction are queued on the
+# toplevel transaction and must be removed when the subtransaction's abort is
+# decoded; otherwise they collide with the records of the line pointer that
+# gets reused by the toplevel transaction's later catalog insert.
+#
+#   - s0_insert runs in a subtransaction and produces its first WAL record.
+#   - s0_catinsert1 generates an xl_heap_new_cid record in that subtransaction.
+#   - s0_rollback aborts the subtransaction.
+#   - s2_vacuum reclaims the aborted row's line pointer (dead inserts are
+#     removed regardless of the xid horizon, so this is deterministic even
+#     while the toplevel transaction is still open); s0_catinsert2 then reuses
+#     the same tid with a different cmin.
+#   - The transaction stays open across both checkpoints, so the final
+#     s1_get_changes replays the transaction's records from the first
+#     checkpoint and decodes its commit for output, building the tuplecid
+#     hash.
+setup
+{
+    DROP TABLE IF EXISTS tbl1;
+    DROP TABLE IF EXISTS user_cat;
+    CREATE TABLE tbl1 (val1 integer);
+    -- Four rows of ~1.5kB fill the first heap page to within a few hundred
+    -- bytes, so that vacuuming away the aborted fifth row leaves exactly one
+    -- line pointer for the next insert to reuse.
+    CREATE TABLE user_cat (c1 int, filler char(1500)) WITH (user_catalog_table = true);
+    INSERT INTO user_cat VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd');
+}
+
+teardown
+{
+    DROP TABLE tbl1;
+    DROP TABLE user_cat;
+    SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot');
+}
+
+session "s0"
+setup { SET synchronous_commit=on; }
+step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); }
+step "s0_begin" { BEGIN; }
+step "s0_savepoint" { SAVEPOINT sp1; }
+step "s0_insert" { INSERT INTO tbl1 VALUES (1); }
+step "s0_catinsert1" { INSERT INTO user_cat VALUES (5, 'e'); }
+step "s0_rollback" { ROLLBACK TO SAVEPOINT sp1; }
+step "s0_catinsert2" { INSERT INTO user_cat VALUES (6, 'f'); }
+step "s0_commit" { COMMIT; }
+
+session "s1"
+setup { SET synchronous_commit=on; }
+step "s1_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); }
+step "s1_checkpoint" { CHECKPOINT; }
+# The user_cat rows carry ~1.5kB filler values which would flood the expected
+# output; filter them out (decoding still processes them server-side).
+step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; }
+
+session "s2"
+setup { SET synchronous_commit=on; }
+step "s2_vacuum" { VACUUM user_cat; }
+
+# The transaction commits only after the second s1_get_changes, so the last
+# s1_get_changes both replays the transaction from the first checkpoint and
+# decodes its commit for output.
+permutation "s0_init" "s0_begin" "s0_savepoint" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_catinsert1" "s0_rollback" "s2_vacuum" "s0_catinsert2" "s1_checkpoint" "s1_get_changes" "s0_commit" "s1_get_changes"
+
+# Variant with the slot created while the transaction is already open: slot
+# creation waits for a consistent point past the open transaction, so the
+# transaction is never decoded by this slot at all (no output).  This locks in
+# the behavior that decoding cannot start in the middle of a transaction.
+permutation "s0_begin" "s0_savepoint" "s0_insert" "s1_init" "s0_catinsert1" "s0_rollback" "s2_vacuum" "s0_catinsert2" "s0_commit" "s1_get_changes"

src/backend/replication/logical/decode.c

@@ -880,7 +880,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);
@@ -912,7 +911,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);

src/backend/replication/logical/reorderbuffer.c

@@ -234,6 +234,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
@@ -3036,6 +3038,110 @@ 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.
+ *
+ * If the xid being aborted is not known to this reorderbuffer to be a
+ * subxact, we return without doing anything, which is correct because there
+ * cannot possibly be TUPLECIDs to clean up; the topxid would've been created
+ * by ReorderBufferAddNewTupleCids(), and the association between the two is
+ * known because we're in SNAPBUILD_FULL_SNAPSHOT mode (otherwise the NEW_CID
+ * WAL records aren't decoded.)
+ *
+ * (This cleanup is pointless when xid is a top-level transaction.  That case
+ * is handled instead by ReorderBufferCleanupTXN() clearing the whole list.)
+ *
+ * XXX this whole exercise is only needed because of keeping the TUPLECID list
+ * assigned to the toplevel transaction. Maybe it would be wiser to have a
+ * per-subxact list instead.  That would make ReorderBufferBuildTupleCidHash
+ * more expensive though.
+ */
+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, or that we entered FULL_SNAPSHOT after the subxid was assigned
+	 * as a child.  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 = txn->toptxn;
+
+	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);
+			ReorderBufferReturnChange(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.
@@ -3251,15 +3357,21 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
 /*
  * Add new (relfilenode, 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, RelFileNode node,
-							 ItemPointerData tid, CommandId cmin,
+							 ItemPointerData tid,
-							 CommandId cmax, CommandId combocid)
+							 CommandId cmin, CommandId cmax,
+							 CommandId combocid)
 {
 	ReorderBufferChange *change = ReorderBufferGetChange(rb);
 	ReorderBufferTXN *txn;
@@ -3271,6 +3383,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

@@ -864,7 +864,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_node, xlrec->target_tid,
 								 xlrec->cmin, xlrec->cmax,
 								 xlrec->combocid);

src/include/replication/reorderbuffer.h

@@ -140,8 +140,12 @@ typedef struct ReorderBufferChange
 		CommandId	command_id;
 
 		/*
-		 * New cid mapping for catalog changing transaction, set when action
+		 * tuplecid changes are used to follow catalog tuple movements (i.e.,
-		 * == *_INTERNAL_TUPLECID.
+		 * 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
 		{
@@ -150,6 +154,7 @@ typedef struct ReorderBufferChange
 			CommandId	cmin;
 			CommandId	cmax;
 			CommandId	combocid;
+			TransactionId subxid;
 		}			tuplecid;
 
 		/* Invalidation. */
@@ -659,15 +664,23 @@ void		ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId,
 void		ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
 void		ReorderBufferAbortOld(ReorderBuffer *, TransactionId xid);
 void		ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+void		ReorderBufferCleanupAbortedTupleCids(ReorderBuffer *rb,
+												 TransactionId xid,
+												 int nsubxacts,
+												 TransactionId *subxacts);
 void		ReorderBufferInvalidate(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
 
 void		ReorderBufferSetBaseSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
 void		ReorderBufferAddSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
 void		ReorderBufferAddNewCommandId(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
 										 CommandId cid);
-void		ReorderBufferAddNewTupleCids(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+void		ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
-										 RelFileNode node, ItemPointerData pt,
+										 TransactionId subxid,
-										 CommandId cmin, CommandId cmax, CommandId combocid);
+										 XLogRecPtr lsn, RelFileNode node,
+										 ItemPointerData tid,
+										 CommandId cmin, CommandId cmax,
+										 CommandId combocid);
+
 void		ReorderBufferAddInvalidations(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
 										  Size nmsgs, SharedInvalidationMessage *msgs);
 void		ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid,