Re: Open a streamed block for transactional messages during decoding

On Mon, Oct 30, 2023 at 2:17 PM Zhijie Hou (Fujitsu)
<houzj(dot)fnst(at)fujitsu(dot)com> wrote:
>
> On Monday, October 30, 2023 12:20 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
> >
> > On Thu, Oct 26, 2023 at 2:01 PM Zhijie Hou (Fujitsu) <houzj(dot)fnst(at)fujitsu(dot)com>
> > wrote:
> > >
> > > On Thursday, October 26, 2023 12:42 PM Amit Kapila
> > <amit(dot)kapila16(at)gmail(dot)com> wrote:
> > > >
> > > > On Tue, Oct 24, 2023 at 5:27 PM Zhijie Hou (Fujitsu)
> > > > <houzj(dot)fnst(at)fujitsu(dot)com>
> > > > wrote:
> > > > >
> > > > > While reviewing the test_decoding code, I noticed that when
> > > > > skip_empty_xacts option is specified, it doesn't open the
> > > > > streaming
> > > > block( e.g.
> > > > > pg_output_stream_start) before streaming the transactional MESSAGE
> > > > > even if it's the first change in a streaming block.
> > > > >
> > > > > It looks inconsistent with what we do when streaming DML changes(e.g.
> > > > > pg_decode_stream_change()).
> > > > >
> > > > > Here is a small patch to open the stream block in this case.
> > > > >
> > > >
> > > > The change looks good to me though I haven't tested it yet. BTW, can
> > > > we change the comment: "Output stream start if we haven't yet, but
> > > > only for the transactional case." to "Output stream start if we
> > > > haven't yet for transactional messages"?
> > >
> > > Thanks for the review and I changed this as suggested.
> > >
> >
> > --- a/contrib/test_decoding/expected/stream.out
> > +++ b/contrib/test_decoding/expected/stream.out
> > @@ -29,7 +29,10 @@ COMMIT;
> > SELECT data FROM pg_logical_slot_get_changes('regression_slot',
> > NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
> > data
> > ----------------------------------------------------------
> > + opening a streamed block for transaction
> > streaming message: transactional: 1 prefix: test, sz: 50
> > + closing a streamed block for transaction aborting streamed
> > + (sub)transaction
> >
> > I was analyzing the reason for the additional message: "aborting streamed
> > (sub)transaction" in the above test and it seems to be due to the below check in
> > the function pg_decode_stream_abort():
> >
> > if (data->skip_empty_xacts && !xact_wrote_changes) return;
> >
> > Before the patch, we won't be setting the 'xact_wrote_changes' flag in txndata
> > which is fixed now. So, this looks okay to me. However, I have another
> > observation in this code which is that for aborts or subtransactions, we are not
> > checking the flag 'stream_wrote_changes', so we may end up emitting the
> > abort message even when no actual change has been streamed. I haven't tried
> > to generate a test to verify this observation, so I could be wrong as well but it is
> > worth analyzing such cases.
>
> I have confirmed that the mentioned case is possible(steps[1]): the
> sub-transaction doesn't output any data, but the stream abort for this
> sub-transaction will still be sent.
>
> But I think this may not be a problemic behavior, as even the pgoutput can
> behave similarly, e.g. If all the changes are filtered by row filter or table
> filter, then the stream abort will still be sent. The subscriber will skip
> handling the STREAM ABORT if the aborted txn was not applied.
>
> And if we want to fix this, in output plugin, we need to record if we have sent
> any changes for each sub-transaction so that we can decide whether to send the
> following stream abort or not. We cannot use 'stream_wrote_changes' because
> it's a per streamed block flag and there could be serval streamed blocks for one
> sub-txn. It looks a bit complicate to me.
>

I agree with your analysis. So, pushed the existing patch. BTW, sorry,
by mistake I used Peter's name as author.

--
With Regards,
Amit Kapila.