Re: Skip collecting decoded changes of already-aborted transactions

On Wed, Nov 13, 2024 at 8:23 PM Peter Smith <smithpb2250(at)gmail(dot)com> wrote:
>
> Hi Sawda-San,
>
> Here are some more review comments for the latest (accidentally called
> v6 again?) v6-0001 patch.

Thank you for reviewing the patch! Indeed, the previous version should
have been v7.

>
> ======
> contrib/test_decoding/sql/stats.sql
>
> 1.
> +-- Execute a transaction that is prepared and aborted. We detect that the
> +-- transaction is aborted before spilling changes, and then skip collecting
> +-- further changes.
>
> You had replied (referring to the above comment):
> I think we already mentioned the transaction is going to be spilled
> but actually not.
>
> ~
>
> Yes, spilling was already mentioned in the current comment but I felt
> it assumes the reader is expected to know details of why it was going
> to be spilled in the first place.

TBH we expect the reader, typically patch authors and reviewers, to
know it.ats1', NULL, NULL, 'skip-empty-xacts', '1');

> In other words, I thought the comment could include a bit more
> explanatory background info:
> (Also, it's not really "we detect" the abort -- it's the new postgres
> code of this patch that detects it.)
>
> SUGGESTION:
> Execute a transaction that is prepared but then aborted. The INSERT
> data exceeds the 'logical_decoding_work_mem limit' limit which
> normally would result in the transaction being spilled to disk, but
> now when Postgres detects the abort it skips the spilling and also
> skips collecting further changes.

But I'm concerned this explanation might be too detailed, and feel odd
to put this comment for the new added tests even though we're doing
similar tests in the same file. For instance, we have:

-- spilling the xact
BEGIN;
INSERT INTO stats_test SELECT 'serialize-topbig--1:'||g.i FROM
generate_series(1, 5000) g(i);
COMMIT;
SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot_st

How about rewording it to the following? I think it's better to
explain why we use a prepared transaction here:

+-- The INSERT changes are large enough to be spilled but not, because the
+-- transaction is aborted. The logical decoding skips collecting further
+-- changes too. The transaction is prepared to make sure the decoding processes
+-- the aborted transaction.

>
> ~~~
>
> 2.
> +-- Check if the transaction is not spilled as it's already aborted.
> +SELECT count(*) FROM
> pg_logical_slot_get_changes('regression_slot_stats4_twophase', NULL,
> NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
> +SELECT pg_stat_force_next_flush();
> +SELECT slot_name, spill_txns, spill_count FROM
> pg_stat_replication_slots WHERE slot_name =
> 'regression_slot_stats4_twophase';
> +
>
> /Check if the transaction is not spilled/Verify that the transaction
> was not spilled/

How about "Verify that the decoding doesn't spill already-aborted
transaction's changes."?

>
> ======
> .../replication/logical/reorderbuffer.c
>
> ReorderBufferResetTXN:
>
> 3.
> /* Discard the changes that we just streamed */
> - ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
> + ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn), true);
>
> Looking at the calling code for ReorderBufferResetTXN it seems this
> function can called for streaming OR prepared. So is it OK here to be
> passing hardwired 'true' as the txn_streaming parameter, or should
> that be passing rbtxn_is_streamed(txn)?

I think it should pass 'true' because otherwise the transaction won't
be marked as streamed.

After more thoughts, I think the name of txn_streaming is the source
of confusion. The flag is actually used to decide whether or not the
given transaction can be marked as streamed, but should not indicate
whether the transaction is being streamed because this function can be
called while streaming. So I renamed it to 'mark_txn_streaming' and
updated the comment.

>
> ~~~
>
> ReorderBufferLargestStreamableTopTXN:
>
> 4.
> 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_is_aborted(txn)))
> {
> largest = txn;
> largest_size = txn->total_size;
>
> I felt that this increasingly complicated code would be a lot easier
> to understand if you just separate the conditions into: (a) the ones
> that filter out transaction you don't care about; (b) the ones that
> check for the largest size. For example,
>
> SUGGESTION:
> dlist_foreach(...)
> {
> ...
>
> /* Don't consider these kinds of transactions for eviction. */
> if (rbtxn_has_partial_change(txn) ||
> !rbtxn_has_streamable_change(txn) || rbtxn_is_aborted(txn))
> continue;
>
> /* Find the largest of the eviction candidates. */
> if ((largest == NULL || txn->total_size > largest_size) &&
> (txn->total_size > 0))
> {
> largest = txn;
> largest_size = txn->total_size;
> }
> }

I like this idea.

>
> ~~~
>
> ReorderBufferCheckMemoryLimit:
>
> 5.
> + /* skip the transaction if already aborted */
> + if (ReorderBufferCheckTXNAbort(rb, txn))
> + {
> + /* All changes should be truncated */
> + Assert(txn->size == 0 && txn->total_size == 0);
> + continue;
> + }
>
> The "discard all changes accumulated so far" side-effect happening
> here is not very apparent from the function name. Maybe a better name
> for ReorderBufferCheckTXNAbort() would be something like
> 'ReorderBufferCleanupIfAbortedTXN()'.

Okay, since we use the term "Cleanup" for different meanings in
reorderbuffer.c (discarding all changes and deallocating the entry),
how about ReorderBufferTruncateTXNIfAborted()?

I've attached the updated patch (v8).

Regards,

--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com