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

Hayato Kuroda (Fujitsu) <kuroda.hayato@fujitsu.com>

From: "kuroda.hayato@fujitsu.com" <kuroda.hayato@fujitsu.com>
To: "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>
Cc: 'Amit Kapila' <amit.kapila16@gmail.com>, 'Dilip Kumar' <dilipbalaut@gmail.com>, 'Masahiko Sawada' <sawada.mshk@gmail.com>, "shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>, 'Peter Smith' <smithpb2250@gmail.com>, 'PostgreSQL Hackers' <pgsql-hackers@lists.postgresql.org>, "wangw.fnst@fujitsu.com" <wangw.fnst@fujitsu.com>
Date: 2022-10-06T12:39:35Z
Lists: pgsql-hackers
Dear Hou,

I put comments for v35-0001.

01. catalog.sgml

```
+       Controls how to handle the streaming of in-progress transactions:
+       <literal>f</literal> = disallow streaming of in-progress transactions,
+       <literal>t</literal> = spill the changes of in-progress transactions to
+       disk and apply at once after the transaction is committed on the
+       publisher,
+       <literal>p</literal> = apply changes directly using a parallel apply
+       worker if available (same as 't' if no worker is available)
```

I'm not sure why 't' means "spill the changes to file". Is it compatibility issue?

~~~
02. applyworker.c - parallel_apply_stream_abort

The argument abort_data is not modified in the function. Maybe "const" modifier should be added.
(Other functions should be also checked...)

~~~
03. applyparallelworker.c - parallel_apply_find_worker

```
+       ParallelApplyWorkerEntry *entry = NULL;
```

This may not have to be initialized here.

~~~
04. applyparallelworker.c - HandleParallelApplyMessages

```
+       static MemoryContext hpm_context = NULL;
```

I think "hpm" means "handle parallel message", so it should be "hpam".

~~~
05. launcher.c - logicalrep_worker_launch()

```
	if (is_subworker)
		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
	else
		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
```

I'm not sure why there are only bgw_type even if there are three types of apply workers. Is it for compatibility?

~~~
06. launcher.c - logicalrep_worker_stop_by_slot

An assertion like Assert(slot_no >=0 && slot_no < max_logical_replication_workers) should be added at the top of this function.

~~~
07. launcher.c - logicalrep_worker_stop_internal

```
+/*
+ * Workhorse for logicalrep_worker_stop(), logicalrep_worker_detach() and
+ * logicalrep_worker_stop_by_slot(). Stop the worker and wait for it to die.
+ */
+static void
+logicalrep_worker_stop_internal(LogicalRepWorker *worker)
```

I think logicalrep_worker_stop_internal() may be not "Workhorse" for logicalrep_worker_detach(). In the function internal function is called for parallel apply worker, and it does not main part of the detach function. 

~~~
08. worker.c - handle_streamed_transaction()

```
+       TransactionId current_xid = InvalidTransactionId;
```

This initialization is not needed. This is not used in non-streaming mode, otherwise it is substituted before used.

~~~
09. worker.c - handle_streamed_transaction()

```
+               case TRANS_PARALLEL_APPLY:
+                       /* Define a savepoint for a subxact if needed. */
+                       parallel_apply_start_subtrans(current_xid, stream_xid);
+                       return false;
```

Based on other case-block, Assert(am_parallel_apply_worker()) may be added at the top of this part.
This suggestion can be said for other swith-case statements.

~~~
10. worker.c - apply_handle_stream_start

```
+ *
+ * XXX We can avoid sending pair of the START/STOP messages to the parallel
+ * worker because unlike apply worker it will process only one
+ * transaction-at-a-time. However, it is not clear whether that is worth the
+ * effort because it is sent after logical_decoding_work_mem changes.
```

I can understand that START message is not needed, but is STOP really removable? If leader does not send STOP to its child, does it lose a chance to change the worker-state to IDLE_IN_TRANSACTION?  

~~~
11. worker.c - apply_handle_stream_start

Currently the number of received chunks have not counted, but it can do if a variable "nchunks" is defined and incremented in apply_handle_stream_start(). This this info may be useful to determine appropriate logical_decoding_work_mem for workloads. How do you think?

~~~
12. worker.c - get_transaction_apply_action

{} are not needed.


Best Regards,
Hayato Kuroda
FUJITSU LIMITED

Commits

Same data as JSON: GET /api/v1/messages/:b64id/commits the thread's linked commits as JSON, with link sources. API reference →
  1. Fix invalid memory access during the shutdown of the parallel apply worker.

  2. Fix assertion failure in apply worker.

  3. Use elog to report unexpected action in handle_streamed_transaction().

  4. Use appropriate wait event when sending data in the apply worker.

  5. Allow the logical_replication_mode to be used on the subscriber.

  6. Rename GUC logical_decoding_mode to logical_replication_mode.

  7. Display the leader apply worker's PID for parallel apply workers.

  8. Improve the code to decide and process the apply action.

  9. Document the newly added wait events added by commit 216a784829.

  10. Perform apply of large transactions by parallel workers.

  11. Wake up a subscription's replication worker processes after DDL.

  12. Add copyright notices to meson files

  13. Better document logical replication parameters

  14. Add a common function to generate the origin name.

  15. Harmonize parameter names in storage and AM code.

  16. Avoid using list_length() to test for empty list.

  17. Improve two comments related to a boolean DefElem's value

  18. Fix partition table's REPLICA IDENTITY checking on the subscriber.

  19. Fix data inconsistency between publisher and subscriber.

  20. Fix cache look-up failures while applying changes in logical replication.