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

On Tuesday, September 27, 2022 2:32 PM Kuroda, Hayato/黒田 隼人 <kuroda(dot)hayato(at)fujitsu(dot)com>
>
> Dear Wang,
>
> Followings are comments for your patchset.

Thanks for the comments.

> ====
> 0001
>
>
> 01. launcher.c - logicalrep_worker_stop_internal()
>
> ```
> +
> + Assert(LWLockHeldByMe(LogicalRepWorkerLock));
> +
> ```

Changed.

> I think it should be Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock,
> LW_SHARED)) because the lock is released one and acquired again as
> LW_SHARED.
> If newer function has been acquired lock as LW_EXCLUSIVE and call
> logicalrep_worker_stop_internal(),
> its lock may become weaker after calling it.
>
> 02. launcher.c - apply_handle_stream_start()
>
> ```
> + /*
> + * Notify handle methods we're processing a remote
> in-progress
> + * transaction.
> + */
> + in_streamed_transaction = true;
>
> - MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
> - FileSetInit(MyLogicalRepWorker->stream_fileset);
> + /*
> + * Start a transaction on stream start, this transaction
> will be
> + * committed on the stream stop unless it is a
> tablesync worker in
> + * which case it will be committed after processing all
> the
> + * messages. We need the transaction for handling the
> buffile,
> + * used for serializing the streaming data and subxact
> info.
> + */
> + begin_replication_step();
> ```
>
> Previously in_streamed_transaction was set after the begin_replication_step(),
> but the ordering is modified. Maybe we don't have to modify it if there is no
> particular reason.
>
> 03. launcher.c - apply_handle_stream_stop()
>
> ```
> + /* Commit the per-stream transaction */
> + CommitTransactionCommand();
> +
> + /* Reset per-stream context */
> + MemoryContextReset(LogicalStreamingContext);
> +
> + pgstat_report_activity(STATE_IDLE, NULL);
> +
> + in_streamed_transaction = false;
> ```
>
> Previously in_streamed_transaction was set after the MemoryContextReset(),
> but the ordering is modified.
> Maybe we don't have to modify it if there is no particular reason.

I adjusted the position of this due to some other improvements this time.

>
> 04. applyparallelworker.c - LogicalParallelApplyLoop()
>
> ```
> + shmq_res = shm_mq_receive(mqh, &len, &data, false);
> ...
> + if (ConfigReloadPending)
> + {
> + ConfigReloadPending = false;
> + ProcessConfigFile(PGC_SIGHUP);
> + }
> ```
>
>
> Here the parallel apply worker waits to receive messages and after dispatching
> it ProcessConfigFile() is called.
> It means that .conf will be not read until the parallel apply worker receives new
> messages and apply them.
>
> It may be problematic when users change log_min_message to debugXXX for
> debugging but the streamed transaction rarely come.
> They expected that detailed description appears on the log from next
> streaming chunk, but it does not.
>
> This does not occur in leader worker when it waits messages from publisher,
> because it uses libpqrcv_receive(), which works asynchronously.
>
> I 'm not sure whether it should be documented that the evaluation of GUCs may
> be delayed, how do you think?

I changed the shm_mq_receive to asynchronous mode which is also consistent with
what we did for Gather node when reading data from parallel query workers.

>
> ===
> 0004
>
> 05. logical-replication.sgml
>
> ```
> ...
> In that case, it may be necessary to change the streaming mode to on or off and
> cause the same conflicts again so the finish LSN of the failed transaction will be
> written to the server log.
> ...
> ```
>
> Above sentence is added by 0001, but it is not modified by 0004.
> Such transactions will be retried as streaming=on mode, so some descriptions
> related with it should be added.

Added.

Best regards,
Hou zj