From 6a23469d7c32dec52609739437582b7f8f5a1661 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81lvaro=20Herrera?= Date: Wed, 7 Oct 2026 12:42:01 +0200 Subject: [PATCH v7 2/3] Refactor the code and improve some comments Author: Hayato Kuroda Author: Zhijie Hou --- src/backend/replication/logical/decode.c | 26 ++--- .../replication/logical/reorderbuffer.c | 103 +++++++++++------- src/include/replication/reorderbuffer.h | 6 +- 3 files changed, 75 insertions(+), 60 deletions(-) diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 0e7d4840a8e..2958183c298 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -893,32 +893,20 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, else { /* - * Remove tuplecid changes queued by the aborted (sub)transactions - * from the toplevel's list, before the transactions are torn down. - * The abort record's primary xid tells - * ReorderBufferCleanupSubTxnTupleCids() whether the whole toplevel - * is going away, in which case scanning the list is pointless. - * - * Note that we must not try to decide that from the primary xid's - * own association instead: an abort record is written once the - * subtransaction is already in TRANS_ABORT, so it never carries - * the toplevel xid in its header (IsSubxactTopXidLogPending() - * requires IsTransactionState()), and an outer subtransaction - * that never wrote WAL of its own -- e.g. a savepoint that only - * wraps other savepoints -- can therefore have no association at - * all, while the released inner subtransactions it rolls back do - * have one and their tuplecids still need to be removed. + * Remove tuplecid changes queued by the aborted subtransactions from + * the toplevel's list, before ReorderBufferAbort() tears the + * transactions down. */ + ReorderBufferCleanupAbortedSubTxnTupleCids(ctx->reorder, xid, + parsed->nsubxacts, + parsed->subxacts); + for (i = 0; i < parsed->nsubxacts; i++) { - ReorderBufferCleanupSubTxnTupleCids(ctx->reorder, - parsed->subxacts[i], - xid); ReorderBufferAbort(ctx->reorder, parsed->subxacts[i], buf->record->EndRecPtr, abort_time); } - ReorderBufferCleanupSubTxnTupleCids(ctx->reorder, xid, xid); ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr, abort_time); } diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index e1c6a7c4d32..dbc10dccc8a 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -240,6 +240,9 @@ static ReorderBufferTXN *ReorderBufferTXNByXid(ReorderBuffer *rb, XLogRecPtr lsn, bool create_as_top); static void ReorderBufferTransferSnapToParent(ReorderBufferTXN *txn, ReorderBufferTXN *subtxn); +static bool TransactionIdInSubxactArray(TransactionId xid, + TransactionId *subxacts, + int nsubxacts); static void AssertTXNLsnOrder(ReorderBuffer *rb); @@ -3158,13 +3161,13 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, } /* - * Remove tuplecid changes queued by subtransaction xid from its toplevel - * transaction's list. + * Remove tuplecid changes queued by aborted subtransactions from their + * toplevel transaction's list. * * Unlike regular changes, tuplecid changes are always queued on the toplevel * transaction (see ReorderBufferAddNewTupleCids), so they would otherwise - * survive the abort of the subtransaction that wrote them. That's wrong: - * they describe catalog tuple versions created or killed by the aborted + * survive the abort of the subtransaction that wrote them. That's wrong: they + * describe catalog tuple versions created or killed by the aborted * subtransaction, which never became visible, and keeping them can corrupt * ReorderBufferBuildTupleCidHash() at commit time -- both when the toplevel * transaction later reuses the same tid (the stale entry collides with the @@ -3172,56 +3175,56 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, * the tid again (a stale cmax would make a still-live tuple look deleted on * historic snapshots). * - * The cleanup is pointless when the whole toplevel transaction is - * being aborted, since ReorderBufferCleanupTXN() frees the whole list - * anyway; the caller passes the abort record's primary xid and we skip - * the scan when it is xid's toplevel. Note that the opposite decision - * -- cleaning only when the primary xid's own association is known -- - * would be wrong: abort records are written once the subtransaction is - * already in TRANS_ABORT, so they carry no toplevel xid in their - * header, and an outer subtransaction that never wrote WAL of its own - * never gets an association at all. When such an outer subtransaction - * rolls back, released inner subtransactions listed in its abort record - * do have known associations and are cleaned here. + * To reach the toplevel, we need one of the abort set's xids to be a known + * subtransaction. The primary xid may have no entry at all: an abort record + * never carries the toplevel xid, so an outer subtransaction that wrote no WAL + * of its own (a savepoint that only wraps other savepoints) is never assigned + * to its toplevel, while a subcommitted child that did write WAL is. Hence we + * search the subxacts too, rather than testing the primary xid alone. * - * If this pass does not know the association between the aborting - * subtransaction and its toplevel, there is nothing we can clean up here, and - * that's fine: such a pass must have started after the subtransaction's first - * (toplevel-xid-bearing) WAL record. A pass that will output-decode the - * toplevel commit cannot have started that late: a slot's restart point - * cannot advance past the oldest in-progress transaction - * (SnapBuildProcessRunningXacts()), so the commit-decoding pass has replayed - * the record that established the association. Conversely, once the restart - * point has advanced past that record, the commit has already been consumed - * by an earlier pass and is skipped here via SnapBuildXactNeedsSkip(), so - * stale tuplecid entries left behind in that case are dropped along with the - * transaction state without ever reaching ReorderBufferBuildTupleCidHash(). + * The cleanup is pointless either when the whole toplevel transaction is + * being aborted, since ReorderBufferCleanupTXN() frees the whole list + * anyway, or when the aborting transaction is a subtransaction that has no + * association with its toplevel. The latter can happen if the transaction + * was already consumed by an earlier decoding pass, so the restart point + * has advanced past the subtransaction's first (toplevel-xid-bearing) WAL + * record. Since the transaction is skipped, its tuplecid entries will + * never be referenced again, so no cleanup is needed. */ void -ReorderBufferCleanupSubTxnTupleCids(ReorderBuffer *rb, TransactionId xid, - TransactionId primary_xid) +ReorderBufferCleanupAbortedSubTxnTupleCids(ReorderBuffer *rb, + TransactionId xid, + int nsubxacts, + TransactionId *subxacts) { ReorderBufferTXN *txn; ReorderBufferTXN *toptxn; dlist_mutable_iter it; + /* + * Search the passed-in xids for a known subtransaction, and use it to + * find the toplevel transaction. + */ txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false); - /* unknown transaction, or unknown association: nothing to remove */ - if (txn == NULL || !rbtxn_is_known_subxact(txn)) - return; - - toptxn = rbtxn_get_toptxn(txn); + for (int i = 0; txn == NULL && i < nsubxacts; i++) + txn = ReorderBufferTXNByXid(rb, subxacts[i], false, NULL, + InvalidXLogRecPtr, false); /* - * The whole toplevel transaction is being aborted (the abort record's - * primary xid is xid's toplevel): ReorderBufferCleanupTXN() will free - * the whole list shortly, don't scan it once per aborted subxid. + * Skip when nothing usable was found. No entry at all means no NEW_CID + * record from these xids was decoded in this pass, so nothing was queued. + * An entry that is not a known subtransaction means either the toplevel + * is being aborted or this pass started after the assigning WAL record, + * so the commit is skipped here anyway. */ - if (toptxn->xid == primary_xid) + if (txn == NULL || !rbtxn_is_known_subxact(txn)) return; + /* Get the top-level transaction to clean up its tuplecid list */ + toptxn = rbtxn_get_toptxn(txn); + dlist_foreach_modify(it, &toptxn->tuplecids) { ReorderBufferChange *change; @@ -3230,15 +3233,37 @@ ReorderBufferCleanupSubTxnTupleCids(ReorderBuffer *rb, TransactionId xid, Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID); - if (change->data.tuplecid.subxid == xid) + /* + * Remove the entry if it's part of the aborting subtransaction or one + * of its children. + * + * The abort record's subxacts are preserved in logical order (see + * AtSubCommit_childXids), so we can use binary search to look up the + * xid. + */ + 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 xid is in an array sorted in logical XID order. + */ +static bool +TransactionIdInSubxactArray(TransactionId xid, TransactionId *subxacts, + int nsubxacts) +{ + return bsearch(&xid, subxacts, nsubxacts, + sizeof(TransactionId), xidLogicalComparator) != NULL; +} + /* * Abort all transactions that aren't actually running anymore because the * server restarted. diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 23c447c4953..7cfa43ad163 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -743,8 +743,10 @@ extern void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, XLogRecPtr end_lsn); extern void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time); -extern void ReorderBufferCleanupSubTxnTupleCids(ReorderBuffer *rb, TransactionId xid, - TransactionId primary_xid); +extern void ReorderBufferCleanupAbortedSubTxnTupleCids(ReorderBuffer *rb, + TransactionId xid, + int nsubxacts, + TransactionId *subxacts); extern void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid); extern void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); extern void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); -- 2.34.1