Re: Skip collecting decoded changes of already-aborted transactions

On Tue, Nov 26, 2024 at 3:03 AM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
>
> I've attached a new version patch that incorporates all comments I got so far.
>

Review comments:
===============
1.
+ * The given transaction is marked as streamed if appropriate and the caller
+ * requested it by passing 'mark_txn_streaming' as true.
+ *
* 'txn_prepared' indicates that we have decoded the transaction at prepare
* time.
*/
static void
-ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
bool txn_prepared)
+ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
bool txn_prepared,
+ bool mark_txn_streaming)
{
...
}
+ else if (mark_txn_streaming && (rbtxn_is_toptxn(txn) ||
(txn->nentries_mem != 0)))
+ {
+ /*
+ * Mark the transaction as streamed, if appropriate.

The comments related to the above changes don't clarify in which cases
the 'mark_txn_streaming' should be set. Before this patch, it was
clear from the comments and code about the cases where we would decide
to mark it as streamed.

2.
+ /*
+ * Mark the transaction as aborted so we ignore future changes of this
+ * transaction.

/so we ignore/so we can ignore/

3.
* Helper function for ReorderBufferProcessTXN to handle the concurrent
- * abort of the streaming transaction. This resets the TXN such that it
- * can be used to stream the remaining data of transaction being processed.
- * This can happen when the subtransaction is aborted and we still want to
- * continue processing the main or other subtransactions data.
+ * abort of the streaming (prepared) transaction.
...

In the above comment, "... streaming (prepared)...", you added
prepared to imply that this function handles concurrent abort for both
in-progress and prepared transactions. Am I correct? If so, the
current change makes it less clear. If you see the comments at its
caller, they are clearer.

4.
+ /*
+ * Remember if the transaction is already aborted so we can detect when
+ * the transaction is concurrently aborted during the replay.
+ */
+ already_aborted = rbtxn_is_aborted(txn);
+
ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn,
txn->xact_time.prepare_time, txn->origin_id, txn->origin_lsn);

@@ -2832,10 +2918,10 @@ ReorderBufferPrepare(ReorderBuffer *rb,
TransactionId xid,
* when rollback prepared is decoded and sent, the downstream should be
* able to rollback such a xact. See comments atop DecodePrepare.
*
- * Note, for the concurrent_abort + streaming case a stream_prepare was
+ * Note, for the concurrent abort + streaming case a stream_prepare was
* already sent within the ReorderBufferReplay call above.
*/
- if (txn->concurrent_abort && !rbtxn_is_streamed(txn))
+ if (!already_aborted && rbtxn_is_aborted(txn) && !rbtxn_is_streamed(txn))
rb->prepare(rb, txn, txn->final_lsn);

It is not clear from the comments how the 'already_aborted' is
handled. I think after this patch we would have already truncated all
its changes. If so, why do we need to try to replay the changes of
such a xact?

5.
+/*
+ * Check the transaction status by looking CLOG and discard all changes if
+ * the transaction is aborted. The transaction status is cached in
+ * txn->txn_flags so we can skip future changes and avoid CLOG lookups on the
+ * next call. Return true if the transaction is aborted, otherwise return
+ * false.
+ *
+ * When the 'debug_logical_replication_streaming' is set to "immediate", we
+ * don't check the transaction status, meaning the caller will always process
+ * this transaction.
+ */
+static bool
+ReorderBufferTruncateTXNIfAborted(ReorderBuffer *rb, ReorderBufferTXN *txn)
+{

I think this function is being invoked to mark a sub-transaction as
aborted. It is better to explain in comments how it interacts with
sub-transactions, why it is okay to mark them as aborted, and how the
other parts of the system interact with it.

--
With Regards,
Amit Kapila.