RE: logical replication empty transactions

On Thursday, January 27, 2022 9:57 PM Ajin Cherian <itsajin@gmail.com> wrote:
Hi, thanks for your patch update.


> On Wed, Jan 26, 2022 at 8:33 PM osumi.takamichi@fujitsu.com
> <osumi.takamichi@fujitsu.com> wrote:
> >
> > On Tuesday, January 11, 2022 6:43 PM Ajin Cherian <itsajin@gmail.com>
> wrote:
> > (3) Is this patch's reponsibility to intialize the data in
> pgoutput_begin_prepare_txn ?
> >
> > @@ -433,6 +487,8 @@ static void
> >  pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx,
> > ReorderBufferTXN *txn)  {
> >         bool            send_replication_origin = txn->origin_id !=
> InvalidRepOriginId;
> > +       PGOutputTxnData    *txndata =
> MemoryContextAllocZero(ctx->context,
> > +
> > + sizeof(PGOutputTxnData));
> >
> >         OutputPluginPrepareWrite(ctx, !send_replication_origin);
> >         logicalrep_write_begin_prepare(ctx->out, txn);
> >
> >
> > Even if we need this initialization for either non streaming case or
> > non two_phase case, there can be another issue.
> > We don't free the allocated memory for this data, right ?
> > There's only one place to use free in the entire patch, which is in
> > the pgoutput_commit_txn(). So, corresponding free of memory looked
> > necessary in the two phase commit functions.
> >
> 
> Actually it is required for begin_prepare to set the data type, so that the checks
> in the pgoutput_change can make sure that the begin prepare is sent. I've also
> added a free in commit_prepared code.
Okay, but if we choose the design that this patch takes
care of the initialization in pgoutput_begin_prepare_txn(),
we need another free in pgoutput_rollback_prepared_txn().
Could you please add some codes similar to pgoutput_commit_prepared_txn() to the same ?
If we simply execute rollback prepared for non streaming transaction,
we don't free it.


Some other new minor comments.

(a) can be "synchronous replication", instead of "Synchronous Replication"

When we have a look at the syncrep.c, we use the former usually in
a normal comment.

 /*
+ * Check if Synchronous Replication is enabled
+ */

(b) move below pgoutput_truncate two codes to the case where if nrelids > 0.

@@ -770,6 +850,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                                  int nrelations, Relation relations[], ReorderBufferChange *change)
 {
        PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
+       PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
        MemoryContext old;
        RelationSyncEntry *relentry;
        int                     i;
@@ -777,6 +858,9 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
        Oid                *relids;
        TransactionId xid = InvalidTransactionId;

+       /* If not streaming, should have setup txndata as part of BEGIN/BEGIN PREPARE */
+       Assert(in_streaming || txndata);
+

(c) fix indent with spaces (for the one sentence of SyncRepEnabled)

@@ -539,6 +538,15 @@ SyncRepReleaseWaiters(void)
 }

 /*
+ * Check if Synchronous Replication is enabled
+ */
+bool
+SyncRepEnabled(void)
+{
+    return SyncRepRequested() && ((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined;
+}
+
+/*

This can be detected by git am.


Best Regards,
    Takamichi Osumi