Re: [HACKERS] logical decoding of two-phase transactions

On Thu, Oct 29, 2020 at 11:19 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
>
> On Tue, Oct 27, 2020 at 3:25 PM Ajin Cherian <itsajin@gmail.com> wrote:
> >
> [v13 patch set]
> Few comments on v13-0001-Support-2PC-txn-base. I haven't checked v14
> version of patches so if you have fixed anything then ignore it.
>
> 1.
> --- a/src/include/replication/reorderbuffer.h
> +++ b/src/include/replication/reorderbuffer.h
> @@ -10,6 +10,7 @@
>  #define REORDERBUFFER_H
>
>  #include "access/htup_details.h"
> +#include "access/twophase.h"
>  #include "lib/ilist.h"
>  #include "storage/sinval.h"
>  #include "utils/hsearch.h"
> @@ -174,6 +175,9 @@ typedef struct ReorderBufferChange
>  #define RBTXN_IS_STREAMED         0x0010
>  #define RBTXN_HAS_TOAST_INSERT    0x0020
>  #define RBTXN_HAS_SPEC_INSERT     0x0040
> +#define RBTXN_PREPARE             0x0080
> +#define RBTXN_COMMIT_PREPARED     0x0100
> +#define RBTXN_ROLLBACK_PREPARED   0x0200
>
>  /* Does the transaction have catalog changes? */
>  #define rbtxn_has_catalog_changes(txn) \
> @@ -233,6 +237,24 @@ typedef struct ReorderBufferChange
>   ((txn)->txn_flags & RBTXN_IS_STREAMED) != 0 \
>  )
>
> +/* Has this transaction been prepared? */
> +#define rbtxn_prepared(txn) \
> +( \
> + ((txn)->txn_flags & RBTXN_PREPARE) != 0 \
> +)
> +
> +/* Has this prepared transaction been committed? */
> +#define rbtxn_commit_prepared(txn) \
> +( \
> + ((txn)->txn_flags & RBTXN_COMMIT_PREPARED) != 0 \
> +)
> +
> +/* Has this prepared transaction been rollbacked? */
> +#define rbtxn_rollback_prepared(txn) \
> +( \
> + ((txn)->txn_flags & RBTXN_ROLLBACK_PREPARED) != 0 \
> +)
> +
>
> I think the above changes should be moved to the second patch. There
> is no use of these macros in this patch and moreover they appear to be
> out-of-place.

Moved to second patch in the patchset.
>
> 2.
> @@ -127,6 +152,7 @@ pg_decode_startup(LogicalDecodingContext *ctx,
> OutputPluginOptions *opt,
>   ListCell   *option;
>   TestDecodingData *data;
>   bool enable_streaming = false;
> + bool enable_2pc = false;
>
> I think it is better to name this variable as enable_two_pc or enable_twopc.

Renamed it to enable_twophase so that it matches with the ctx member
ctx-twophase.
>
> 3.
> + xid = strtoul(strVal(elem->arg), NULL, 0);
> + if (xid == 0 || errno != 0)
> + data->check_xid_aborted = InvalidTransactionId;
> + else
> + data->check_xid_aborted = (TransactionId)xid;
> +
> + if (!TransactionIdIsValid(data->check_xid_aborted))
> + ereport(ERROR,
> + (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
> + errmsg("check-xid-aborted is not a valid xid: \"%s\"",
> + strVal(elem->arg))));
>
> Can't we write this as below and get rid of xid variable:
> data->check_xid_aborted= (TransactionId) strtoul(strVal(elem->arg), NULL, 0);
> if (!TransactionIdIsValid(data->check_xid_aborted) || errno)
>     ereport..

Updated. Small change so that errno is checked first.

>
> 4.
> + /* if check_xid_aborted is a valid xid, then it was passed in
> + * as an option to check if the transaction having this xid would be aborted.
> + * This is to test concurrent aborts.
> + */
>
> multi-line comments have the first line as empty.

Updated.

>
> 5.
> +     <para>
> +      The required <function>prepare_cb</function> callback is called whenever
> +      a transaction which is prepared for two-phase commit has been
> +      decoded. The <function>change_cb</function> callbacks for all modified
> +      rows will have been called before this, if there have been any modified
> +      rows.
> +<programlisting>
> +typedef void (*LogicalDecodePrepareCB) (struct LogicalDecodingContext *ctx,
> +                                        ReorderBufferTXN *txn,
> +                                        XLogRecPtr prepare_lsn);
> +</programlisting>
> +     </para>
> +    </sect3>
> +
> +    <sect3 id="logicaldecoding-output-plugin-commit-prepared">
> +     <title>Transaction Commit Prepared Callback</title>
> +
> +     <para>
> +      The required <function>commit_prepared_cb</function> callback
> is called whenever
> +      a transaction commit prepared has been decoded. The
> <parameter>gid</parameter> field,
> +      which is part of the <parameter>txn</parameter> parameter can
> be used in this
> +      callback.
>
> I think the last line "The <parameter>gid</parameter> field, which is
> part of the <parameter>txn</parameter> parameter can be used in this
> callback." in 'Transaction Commit Prepared Callback' should also be
> present in 'Transaction Prepare Callback' as we using the same in
> prepare API as well.

Updated.

>
> 6.
> +pg_decode_stream_prepare(LogicalDecodingContext *ctx,
> + ReorderBufferTXN *txn,
> + XLogRecPtr prepare_lsn)
> +{
> + TestDecodingData *data = ctx->output_plugin_private;
> +
> + if (data->skip_empty_xacts && !data->xact_wrote_changes)
> + return;
> +
> + OutputPluginPrepareWrite(ctx, true);
> +
> + if (data->include_xids)
> + appendStringInfo(ctx->out, "preparing streamed transaction TXN %u", txn->xid);
> + else
> + appendStringInfo(ctx->out, "preparing streamed transaction");
>
> I think we should include 'gid' as well in the above messages.

Updated.

>
> 7.
> @@ -221,12 +235,26 @@ StartupDecodingContext(List *output_plugin_options,
>   ctx->streaming = (ctx->callbacks.stream_start_cb != NULL) ||
>   (ctx->callbacks.stream_stop_cb != NULL) ||
>   (ctx->callbacks.stream_abort_cb != NULL) ||
> + (ctx->callbacks.stream_prepare_cb != NULL) ||
>   (ctx->callbacks.stream_commit_cb != NULL) ||
>   (ctx->callbacks.stream_change_cb != NULL) ||
>   (ctx->callbacks.stream_message_cb != NULL) ||
>   (ctx->callbacks.stream_truncate_cb != NULL);
>
>   /*
> + * To support two-phase logical decoding, we require
> prepare/commit-prepare/abort-prepare
> + * callbacks. The filter-prepare callback is optional. We however
> enable two-phase logical
> + * decoding when at least one of the methods is enabled so that we
> can easily identify
> + * missing methods.
> + *
> + * We decide it here, but only check it later in the wrappers.
> + */
> + ctx->twophase = (ctx->callbacks.prepare_cb != NULL) ||
> + (ctx->callbacks.commit_prepared_cb != NULL) ||
> + (ctx->callbacks.rollback_prepared_cb != NULL) ||
> + (ctx->callbacks.filter_prepare_cb != NULL);
> +
>
> I think stream_prepare_cb should be checked for the 'twophase' flag
> because we won't use this unless two-phase is enabled. Am I missing
> something?

Was fixed in v14.

regards,
Ajin Cherian
Fujitsu Australia