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

On Thur, Sep 22, 2022 at 16:08 PM Kuroda, Hayato/黒田 隼人 <kuroda.hayato@fujitsu.com> wrote:
> Dear Wang,
> 
> Thanks for updating the patch! Followings are comments for v33-0001.

Thanks for your comments.

> ===
> libpqwalreceiver.c
> 
> 01. inclusion
> 
> ```
> +#include "catalog/pg_subscription.h"
> ```
> 
> We don't have to include it because the analysis of parameters is done at caller.
> 
> ===
> launcher.c

Improved.

> 02. logicalrep_worker_launch()
> 
> ```
> +       /*
> +        * Return silently if the number of parallel apply workers reached the
> +        * limit per subscription.
> +        */
> +       if (is_subworker && nparallelapplyworkers >=
> max_parallel_apply_workers_per_subscription)
> ```
> 
> a.
> I felt that it might be kind if we output some debug messages.
> 
> b.
> The if statement seems to be more than 80 characters. You can move to new
> line around "nparallelapplyworkers >= ...".

Improved.

> ===
> applyparallelworker.c
> 
> 03. declaration
> 
> ```
> +/*
> + * Is there a message pending in parallel apply worker which we need to
> + * receive?
> + */
> +volatile bool ParallelApplyMessagePending = false;
> ```
> 
> I checked other flags that are set by signal handlers, their datatype seemed to
> be sig_atomic_t.
> Is there any reasons that you use normal bool? It should be changed if not.

Improved.

> 04. HandleParallelApplyMessages()
> 
> ```
> +               if (winfo->error_mq_handle == NULL)
> +                       continue;
> ```
> 
> a.
> I was not sure when the cell should be cleaned. Currently we clean up
> ParallelApplyWorkersList() only in the parallel_apply_start_worker(),
> but we have chances to remove such a cell like HandleParallelApplyMessages()
> or HandleParallelApplyMessage(). How do you think?
> 
> b.
> Comments should be added even if we keep this, like "exited worker, skipped".
> 
> ```
> +               else
> +                       ereport(ERROR,
> +
> (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> +                                        errmsg("lost connection to the leader apply worker")));
> ```
> 
> c.
> This function is called on the leader apply worker, so the hint should be "lost
> connection to the parallel apply worker".

=>b.
Added the following comment according to your suggestion.
`Skip if worker has exited`

=>c.
Fixed.

> ===
> worker.c
> 
> 07. handle_streamed_transaction()
> 
> ```
> + * For non-streamed transactions, returns false;
> ```
> 
> "returns false;" -> "returns false"

Improved. I changed the semicolon to a period

> apply_handle_commit_prepared(), apply_handle_abort_prepared()
> 
> These functions are not expected that parallel worker calls
> so I think Assert() should be added.

I am not sure if this modification is necessary since we do not modify the
non-streamed transaction related message like "COMMIT PREPARED" or "ROLLBACK
PREPARED".

> 08. UpdateWorkerStats()
> 
> ```
> -static void
> +void
>  UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
> ```
> 
> This function is called only in worker.c, should be static.
> 
> 09. subscription_change_cb()
> 
> ```
> -static void
> +void
>  subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
> ```
> 
> This function is called only in worker.c, should be static.

Improved.

> 10. InitializeApplyWorker()
> 
> ```
> +/*
> + * Initialize the database connection, in-memory subscription and necessary
> + * config options.
> + */
>  void
> -ApplyWorkerMain(Datum main_arg)
> +InitializeApplyWorker(void)
> ```
> 
> Some comments should be added about this is a common part of leader and
> parallel apply worker.

Added the following comment:
`The common initialization for leader apply worker and parallel apply worker.`

> ===
> logicalrepworker.h
> 
> 11. declaration
> 
> ```
> extern PGDLLIMPORT volatile bool ParallelApplyMessagePending;
> ```
> 
> Please refer above comment.
> 
> ===
> guc_tables.c

Improved.

Also rebased the patch set based on the changes in HEAD (26f7802).

Attach the new patch set.

Regards,
Wang wei