Re: logical replication empty transactions

On Thu, Jan 27, 2022 at 12:16 AM osumi.takamichi@fujitsu.com
<osumi.takamichi@fujitsu.com> wrote:
>
> On Tuesday, January 11, 2022 6:43 PM From: Ajin Cherian <itsajin@gmail.com> wrote:
> > Minor update to rebase the patch so that it applies clean on HEAD.
> Hi, let me share some additional comments on v16.
>
>
> (1) comment of pgoutput_change
>
> @@ -630,11 +688,15 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
>                                 Relation relation, ReorderBufferChange *change)
>  {
>         PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
> +       PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
>         MemoryContext old;
>         RelationSyncEntry *relentry;
>         TransactionId xid = InvalidTransactionId;
>         Relation        ancestor = NULL;
>
> +       /* If not streaming, should have setup txndata as part of BEGIN/BEGIN PREPARE */
> +       Assert(in_streaming || txndata);
> +
>
> In my humble opinion, the comment should not touch BEGIN PREPARE,
> because this patch's scope doesn't include two phase commit.
> (We could add this in another patch to extend the scope after the commit ?)
>

We have to include BEGIN PREPARE as well, as the txndata has to be
setup. Only difference is that we will not skip empty transaction in
BEGIN PREPARE

> This applies to pgoutput_truncate's comment.
>
> (2) "keep alive" should be "keepalive" in WalSndUpdateProgress
>
>         /*
> +        * When skipping empty transactions in synchronous replication, we need
> +        * to send a keep alive to keep the MyWalSnd locations updated.
> +        */
> +       force_keepalive_syncrep = send_keepalive && SyncRepEnabled();
> +
>
> Also, this applies to the comment for force_keepalive_syncrep.

Fixed.

>
> (3) Should finish the second sentence with period in the comment of pgoutput_message.
>
> @@ -845,6 +923,19 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
>         if (in_streaming)
>                 xid = txn->xid;
>
> +       /*
> +        * Output BEGIN if we haven't yet.
> +        * Avoid for streaming and non-transactional messages
>

Fixed.

> (4) "begin" can be changed to "BEGIN" in the comment of PGOutputTxnData definition.
>
> In the entire patch, when we express BEGIN message,
> we use capital letters "BEGIN" except for one place.
> We can apply the same to this place as well.
>
> +typedef struct PGOutputTxnData
> +{
> +       bool sent_begin_txn;    /* flag indicating whether begin has been sent */
> +} PGOutputTxnData;
> +
>

Fixed.

> (5) inconsistent way to write Assert statements with blank lines
>
> In the below case, it'd be better to insert one blank line
> after the Assert();
>
> +static void
> +pgoutput_begin(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
> +{
>         bool            send_replication_origin = txn->origin_id != InvalidRepOriginId;
> +       PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
>
> +       Assert(txndata);
>         OutputPluginPrepareWrite(ctx, !send_replication_origin);
>
>

Fixed.

> (6) new codes in the pgoutput_commit_txn looks messy slightly
>
> @@ -419,7 +455,25 @@ static void
>  pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
>                                         XLogRecPtr commit_lsn)
>  {
> -       OutputPluginUpdateProgress(ctx);
> +       PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
> +       bool            skip;
> +
> +       Assert(txndata);
> +
> +       /*
> +        * If a BEGIN message was not yet sent, then it means there were no relevant
> +        * changes encountered, so we can skip the COMMIT message too.
> +        */
> +       skip = !txndata->sent_begin_txn;
> +       pfree(txndata);
> +       txn->output_plugin_private = NULL;
> +       OutputPluginUpdateProgress(ctx, skip);
>
> Could we conduct a refactoring for this new part ?
> IMO, writing codes to free the data structure at the top
> of function seems weird.
>
> One idea is to export some part there
> and write a new function, something like below.
>
> static bool
> txn_sent_begin(ReorderBufferTXN *txn)
> {
>     PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
>     bool            needs_skip;
>
>     Assert(txndata);
>
>     needs_skip = !txndata->sent_begin_txn;
>
>     pfree(txndata);
>     txn->output_plugin_private = NULL;
>
>     return needs_skip;
> }
>
> FYI, I had a look at the v12-0002-Skip-empty-prepared-transactions-for-logical-rep.patch
> for reference of pgoutput_rollback_prepared_txn and pgoutput_commit_prepared_txn.
> Looks this kind of function might work for future extensions as well.
> What did you think ?

I changed a bit, but I'd hold a comprehensive rewrite when a future
patch supports skipping
empty transactions in two-phase transactions and streaming transactions.

regards,
Ajin Cherian