Re: Perform streaming logical transactions by background workers and parallel apply

Here are some review comments for v47-0001

(This review is a WIP - I will post more comments for this patch next week)

======

.../replication/logical/applyparallelworker.c

1.

+ * Copyright (c) 2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION src/backend/replication/logical/applyparallelworker.c
+ *

This IDENTIFICATION should be on 2 lines like it previously was
instead of wrapped into one line. For consistency with all other file
headers.

~~~

2. File header comment

+ * Since the database structure (schema of subscription tables, etc.) of
+ * publisher and subscriber may be different.

Incomplete sentence?

~~~

3.

+ * When the following two scenarios occur, a deadlock occurs.

Actually, you described three scenarios in this comment. Not two.

SUGGESTION
The following scenarios can cause a deadlock.

~~~

4.

+ * LA (waiting to acquire the local transaction lock) -> PA1 (waiting to
+ * acquire the lock on the unique index) -> PA2 (waiting to acquire the lock on
+ * the remote transaction) -> LA

"PA1" -> "PA-1"
"PA2" -> "PA-2"

~~~

5.

+ * To resolve this issue, we use non-blocking write and wait with a timeout. If
+ * timeout is exceeded, the LA report an error and restart logical replication.

"report" --> "reports"
"restart" -> "restarts"

OR

"LA report" -> "LA will report"

~~~

6. pa_wait_for_xact_state

+/*
+ * Wait until the parallel apply worker's transaction state reach or exceed the
+ * given xact_state.
+ */
+static void
+pa_wait_for_xact_state(ParallelApplyWorkerShared *wshared,
+ ParallelTransState xact_state)

"reach or exceed" -> "reaches or exceeds"

~~~

7. pa_stream_abort

+ /*
+ * Although the lock can be automatically released during transaction
+ * rollback, but we still release the lock here as we may not in a
+ * transaction.
+ */
+ pa_unlock_transaction(xid, AccessShareLock);

"but we still" -> "we still"
"we may not in a" -> "we may not be in a"

~~~

8.

+ pa_savepoint_name(MySubscription->oid, subxid, spname,
+ sizeof(spname));
+

Unnecessary wrapping

~~~

9.

+ for (i = list_length(subxactlist) - 1; i >= 0; i--)
+ {
+ TransactionId xid_tmp = lfirst_xid(list_nth_cell(subxactlist, i));
+
+ if (xid_tmp == subxid)
+ {
+ found = true;
+ break;
+ }
+ }
+
+ if (found)
+ {
+ RollbackToSavepoint(spname);
+ CommitTransactionCommand();
+ subxactlist = list_truncate(subxactlist, i + 1);
+ }

This code logic does not seem to require the 'found' flag. You can do
the RollbackToSavepoint/CommitTransactionCommand/list_truncate before
the break.

~~~

10. pa_lock/unlock _stream/_transaction

+/*
+ * Helper functions to acquire and release a lock for each stream block.
+ *
+ * Set locktag_field4 to 0 to indicate that it's a stream lock.
+ */

+/*
+ * Helper functions to acquire and release a lock for each local transaction.
+ *
+ * Set locktag_field4 to 1 to indicate that it's a transaction lock.

Should constants/defines/enums replace those magic numbers 0 and 1?

~~~

11. pa_lock_transaction

+ * Note that all the callers are passing remote transaction ID instead of local
+ * transaction ID as xid. This is because the local transaction ID will only be
+ * assigned while applying the first change in the parallel apply, but it's
+ * possible that the first change in parallel apply worker is blocked by a
+ * concurrently executing transaction in another parallel apply worker causing
+ * the leader cannot get local transaction ID.

"causing the leader cannot" -> "which means the leader cannot" (??)

======

src/backend/replication/logical/worker.c

12. TransApplyAction

+/*
+ * What action to take for the transaction.
+ *
+ * TRANS_LEADER_APPLY:
+ * The action means that we are in the leader apply worker and changes of the
+ * transaction are applied directly in the worker.
+ *
+ * TRANS_LEADER_SERIALIZE:
+ * It means that we are in the leader apply worker or table sync worker.
+ * Changes are written to temporary files and then applied when the final
+ * commit arrives.
+ *
+ * TRANS_LEADER_SEND_TO_PARALLEL:
+ * The action means that we are in the leader apply worker and need to send the
+ * changes to the parallel apply worker.
+ *
+ * TRANS_PARALLEL_APPLY:
+ * The action that we are in the parallel apply worker and changes of the
+ * transaction are applied directly in the worker.
+ */
+typedef enum

12a
Too many various ways of saying the same thing:

"The action means that we..."
"It means that we..."
"The action that we..." (typo?)

Please word all these comments consistently

~

12b.
"directly in the worker" -> "directly by the worker" (??) 2x

~~~

13. get_worker_name

+/*
+ * Return the name of the logical replication worker.
+ */
+static const char *
+get_worker_name(void)
+{
+ if (am_tablesync_worker())
+ return _("logical replication table synchronization worker");
+ else if (am_parallel_apply_worker())
+ return _("logical replication parallel apply worker");
+ else
+ return _("logical replication apply worker");
+}

This function belongs nearer the top of the module (above all the
error messages that are using it).

------
Kind Regards,
Peter Smith.
Fujitsu Australia