Horton

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

Amit Kapila <amit.kapila16@gmail.com>

View thread View on postgr.es
From: Amit Kapila <amit.kapila16@gmail.com>
To: Peter Smith <smithpb2250@gmail.com>
Cc: Tom Lane <tgl@sss.pgh.pa.us>, vignesh C <vignesh21@gmail.com>, "tanghy.fnst@fujitsu.com" <tanghy.fnst@fujitsu.com>, Ajin Cherian <itsajin@gmail.com>, Greg Nancarrow <gregn4422@gmail.com>, PostgreSQL Hackers <pgsql-hackers@postgresql.org>
Date: 2021-07-23T10:08:05Z
Lists: pgsql-hackers 
On Tue, Jul 20, 2021 at 9:24 AM Peter Smith <smithpb2250@gmail.com> wrote:
>
> Please find attached the latest patch set v98*
>

Review comments:
================
1.
/*
- * Handle STREAM COMMIT message.
+ * Common spoolfile processing.
+ * Returns how many changes were applied.
  */
-static void
-apply_handle_stream_commit(StringInfo s)
+static int
+apply_spooled_messages(TransactionId xid, XLogRecPtr lsn)

Let's extract this common functionality (common to current code and
the patch) as a separate patch? I think we can commit this as a
separate patch.

2.
apply_spooled_messages()
{
..
  elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
  nchanges, path);
..
}

You have this DEBUG1 message in apply_spooled_messages and its
callers. You can remove it from callers as the patch already has
another debug message to indicate whether it is stream prepare or
stream commit. Also, if this is the only reason to return nchanges
from apply_spooled_messages() then we can get rid of that as well.

3.
+ /*
+ * 2. Mark the transaction as prepared. - Similar code as for
+ * apply_handle_prepare (i.e. two-phase non-streamed prepare)
+ */
+
+ /*
+ * BeginTransactionBlock is necessary to balance the EndTransactionBlock
+ * called within the PrepareTransactionBlock below.
+ */
+ BeginTransactionBlock();
+ CommitTransactionCommand(); /* Completes the preceding Begin command. */
+
+ /*
+ * Update origin state so we can restart streaming from correct position
+ * in case of crash.
+ */
+ replorigin_session_origin_lsn = prepare_data.end_lsn;
+ replorigin_session_origin_timestamp = prepare_data.prepare_time;
+
+ PrepareTransactionBlock(gid);

I think you can move this part into a common function
apply_handle_prepare_internal. If that is possible then you might want
to move this part into a common functionality patch as mentioned in
point-1.

4.
+ xid = logicalrep_read_stream_prepare(s, &prepare_data);
+ elog(DEBUG1, "received prepare for streamed transaction %u", xid);

It is better to have an empty line between the above code lines for
the sake of clarity.

5.
+/* Commit (and abort) information */
 typedef struct LogicalRepCommitData

How is this structure related to abort? Even if it is, why this
comment belongs to this patch?

6. Most of the code in logicalrep_write_stream_prepare() and
logicalrep_write_prepare() is same except for message. I think if we
want we can handle both of them with a single message by setting some
flag for stream case but probably there will be some additional
checking required on the worker-side. What do you think? I think if we
want to keep them separate then at least we should keep the common
functionality in logicalrep_write_*/logicalrep_read_* in separate
functions. This way we will avoid minor inconsistencies in-stream and
non-stream functions.

7.
+++ b/doc/src/sgml/protocol.sgml
@@ -2881,7 +2881,7 @@ The commands accepted in replication mode are:
    Begin Prepare and Prepare messages belong to the same transaction.
    It also sends changes of large in-progress transactions between a pair of
    Stream Start and Stream Stop messages. The last stream of such a transaction
-   contains a Stream Commit or Stream Abort message.
+   contains a Stream Prepare, Stream Commit or Stream Abort message.

I am not sure if it is correct to mention Stream Prepare here because
after that we will send commit prepared as well for such a
transaction. So, I think we should remove this change.

8.
-ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
-
 \dRs+

+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);

Is there a reason for this change in the tests?

9.
I think this contains a lot of streaming tests in 023_twophase_stream.
Let's keep just one test for crash-restart scenario (+# Check that 2PC
COMMIT PREPARED is decoded properly on crash restart.) where both
publisher and subscriber get restarted. I think others are covered in
one or another way by other existing tests. Apart from that, I also
don't see the need for the below tests:
# Do DELETE after PREPARE but before COMMIT PREPARED.
This is mostly the same as the previous test where the patch is testing Insert
# Try 2PC transaction works using an empty GID literal
This is covered in 021_twophase.

10.
+++ b/src/test/subscription/t/024_twophase_cascade_stream.pl
@@ -0,0 +1,271 @@
+
+# Copyright (c) 2021, PostgreSQL Global Development Group
+
+# Test cascading logical replication of 2PC.

In the above comment, you might want to say something about streaming.
In general, I am not sure if it is really adding value to have these
many streaming tests for cascaded setup and doing the whole setup
again after we have done in tests 022_twophase_cascade. I think it is
sufficient to do just one or two streaming tests by enhancing
022_twophase_cascade, you can alter subscription to enable streaming
after doing non-streaming tests.

11. Have you verified that all these tests went through the streaming
code path? If not, you can once enable DEBUG message in
apply_handle_stream_prepare() and see if all tests hit that.

-- 
With Regards,
Amit Kapila.

Commits

  1. Add prepare API support for streaming transactions in logical replication.

  2. Doc: minor improvements for logical replication protocol documentation.

  3. Fix test failure in 021_twophase.pl.

  4. Refactor to make common functions in proto.c and worker.c.

  5. Unify PostgresNode's new() and get_new_node() methods

  6. Fix potential buffer overruns in proto.c.

  7. Add support for prepared transactions to built-in logical replication.

  8. Refactor function parse_subscription_options.

  9. Allow enabling two-phase option via replication protocol.

  10. Don't use Asserts to check for violations of replication protocol.

  11. Improve psql tab completion for options of subcriptions and publications

  12. Rearrange logrep worker's snapshot handling some more.

  13. doc: Update information of new messages for logical replication.

  14. ALTER SUBSCRIPTION ... ADD/DROP PUBLICATION

  15. Allow pgoutput to send logical decoding messages.

  16. Refactor function parse_output_parameters.

  17. Avoid repeated decoding of prepared transactions after a restart.

  18. Fix an oversight in ReorderBufferFinishPrepared.

  19. Allow multiple xacts during table sync in logical replication.

  20. Fix replication of in-progress transactions in tablesync worker.

  21. Fix 'skip-empty-xacts' option in test_decoding for streaming mode.

  22. Use Enum for top level logical replication message types.

  23. Add support for streaming to built-in logical replication.

  24. Fix the logical streaming test.

  25. Implement streaming mode in ReorderBuffer.

  26. Extend the logical decoding output plugin API with stream methods.

  27. Store 2PC GID in commit/abort WAL recs for logical decoding