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

On Tue, Oct 27, 2020 at 3:25 PM Ajin Cherian <itsajin(at)gmail(dot)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.

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.

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..

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.

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.

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.

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?

--
With Regards,
Amit Kapila.