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

On Wed, Sep 21, 2022 at 17:25 PM Peter Smith <smithpb2250@gmail.com> wrote:
> Here are some review comments for patch v30-0001.

Thanks for your comments.

> ======
> 
> 1. Commit message
> 
> In addition, the patch extends the logical replication STREAM_ABORT message
> so
> that abort_time and abort_lsn can also be sent which can be used to update the
> replication origin in parallel apply worker when the streaming transaction is
> aborted. Because this message extension is needed to support parallel
> streaming, meaning that parallel streaming is not supported for publications on
> servers < PG16.
> 
> "meaning that parallel streaming is not supported" -> "parallel
> streaming is not supported"

Improved as suggested.

> ======
> 
> 2. doc/src/sgml/logical-replication.sgml
> 
> @@ -1611,8 +1622,12 @@ CONTEXT:  processing remote data for
> replication origin "pg_16395" during "INSER
>     to the subscriber, plus some reserve for table synchronization.
>     <varname>max_logical_replication_workers</varname> must be set to at
> least
>     the number of subscriptions, again plus some reserve for the table
> -   synchronization.  Additionally the
> <varname>max_worker_processes</varname>
> -   may need to be adjusted to accommodate for replication workers, at least
> +   synchronization. In addition, if the subscription parameter
> +   <literal>streaming</literal> is set to <literal>parallel</literal>, please
> +   increase <literal>max_logical_replication_workers</literal> according to
> +   the desired number of parallel apply workers.  Additionally the
> +   <varname>max_worker_processes</varname> may need to be adjusted to
> +   accommodate for replication workers, at least
>     (<varname>max_logical_replication_workers</varname>
>     + <literal>1</literal>).  Note that some extensions and parallel queries
>     also take worker slots from <varname>max_worker_processes</varname>.
> 
> IMO it looks a bit strange to have "In addition" followed by "Additionally".
> 
> Also, "to accommodate for replication workers"? seems like a typo (but
> it is not caused by your patch)
> 
> BEFORE
> In addition, if the subscription parameter streaming is set to
> parallel, please increase max_logical_replication_workers according to
> the desired number of parallel apply workers.
> 
> AFTER (???)
> If the subscription parameter streaming is set to parallel,
> max_logical_replication_workers should be increased according to the
> desired number of parallel apply workers.

=> Reword
Improved as suggested.

=> typo?
Sorry, I am not sure. Do you mean
s/replication workers/workers for subscriptions/  or something else?
I think we should improve it in a new thread.

> ======
> 
> 4. .../replication/logical/applyparallelworker.c - parallel_apply_free_worker
> 
> + winfo->in_use = false;
> +
> + /* Are there enough workers in the pool? */
> + if (napplyworkers > (max_parallel_apply_workers_per_subscription / 2))
> + {
> 
> I felt the comment/logic about "enough" needs a bit more description.
> At least it should say to refer to the more detailed explanation atop
> worker.c

Added related comment atop this function.

> ======
> 
> 5. .../replication/logical/applyparallelworker.c - parallel_apply_setup_dsm
> 
> + /*
> + * Estimate how much shared memory we need.
> + *
> + * Because the TOC machinery may choose to insert padding of oddly-sized
> + * requests, we must estimate each chunk separately.
> + *
> + * We need one key to register the location of the header, and we need two
> + * other keys to track of the locations of the message queue and the error
> + * message queue.
> + */
> 
> "track of" -> "keep track of" ?

Improved.

> ======
> 
> 6. src/backend/replication/logical/launcher.c  - logicalrep_worker_detach
> 
>  logicalrep_worker_detach(void)
>  {
> + /* Stop the parallel apply workers. */
> + if (!am_parallel_apply_worker() && !am_tablesync_worker())
> + {
> + List    *workers;
> + ListCell   *lc;
> 
> The condition is not very obvious. This is why I previously suggested
> adding another macro/function like 'isLeaderApplyWorker'. In the
> absence of that, then I think the comment needs to be more
> descriptive.
> 
> SUGGESTION
> If this is the leader apply worker then stop the parallel apply workers.

Added the new function am_leader_apply_worker.

> ======
> 
> 7. src/backend/replication/logical/proto.c - logicalrep_read_stream_abort
> 
>  void
>  logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
> -   TransactionId subxid)
> +   TransactionId subxid, XLogRecPtr abort_lsn,
> +   TimestampTz abort_time, bool abort_info)
>  {
>   pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_ABORT);
> 
> @@ -1175,19 +1179,40 @@ logicalrep_write_stream_abort(StringInfo out,
> TransactionId xid,
>   /* transaction ID */
>   pq_sendint32(out, xid);
>   pq_sendint32(out, subxid);
> +
> + if (abort_info)
> + {
> + pq_sendint64(out, abort_lsn);
> + pq_sendint64(out, abort_time);
> + }
> 
> 
> The new param name 'abort_info' seems misleading.
> 
> Maybe a name like 'write_abort_info' is better?

Improved as suggested.

> ~~~
> 
> 8. src/backend/replication/logical/proto.c - logicalrep_read_stream_abort
> 
> +logicalrep_read_stream_abort(StringInfo in,
> + LogicalRepStreamAbortData *abort_data,
> + bool read_abort_lsn)
>  {
> - Assert(xid && subxid);
> + Assert(abort_data);
> +
> + abort_data->xid = pq_getmsgint(in, 4);
> + abort_data->subxid = pq_getmsgint(in, 4);
> 
> - *xid = pq_getmsgint(in, 4);
> - *subxid = pq_getmsgint(in, 4);
> + if (read_abort_lsn)
> + {
> + abort_data->abort_lsn = pq_getmsgint64(in);
> + abort_data->abort_time = pq_getmsgint64(in);
> + }
> 
> This name 'read_abort_lsn' is inconsistent with the 'abort_info' of
> the logicalrep_write_stream_abort.
> 
> I suggest change these to 'read_abort_info/write_abort_info'

Improved as suggested.

> ======
> 
> 9. src/backend/replication/logical/worker.c - file header comment
> 
> + * information is added to the ParallelApplyWorkersList. Once the worker
> + * finishes applying the transaction, we mark it available for use. Now,
> + * before starting a new worker to apply the streaming transaction, we check
> + * the list and use any worker, if available. Note that we maintain a maximum
> 
> 9a.
> "available for use." -> "available for re-use."
> 
> ~
> 
> 9b.
> "we check the list and use any worker, if available" -> "we check the
> list for any available worker"

Improved as suggested.

> ~~~
> 
> 10. src/backend/replication/logical/worker.c - handle_streamed_transaction
> 
> + /* write the change to the current file */
> + stream_write_change(action, s);
> + return true;
> 
> Uppercase the comment.

Improved as suggested.

> ~~~
> 
> 11. src/backend/replication/logical/worker.c - apply_handle_stream_abort
> 
> +static void
> +apply_handle_stream_abort(StringInfo s)
> +{
> + TransactionId xid;
> + TransactionId subxid;
> + LogicalRepStreamAbortData abort_data;
> + bool read_abort_lsn = false;
> + ParallelApplyWorkerInfo *winfo = NULL;
> + TransApplyAction apply_action;
> 
> The variable 'read_abort_lsn' name ought to be changed to match
> consistently the parameter name.

Improved as suggested.

> ======
> 
> 12. src/backend/replication/pgoutput/pgoutput.c - pgoutput_stream_abort
> 
> @@ -1843,6 +1850,8 @@ pgoutput_stream_abort(struct
> LogicalDecodingContext *ctx,
>     XLogRecPtr abort_lsn)
>  {
>   ReorderBufferTXN *toptxn;
> + PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
> + bool abort_info = (data->streaming == SUBSTREAM_PARALLEL);
> 
> The variable 'abort_info' name ought to be changed to be
> 'write_abort_info' (as suggested above) to match consistently the
> parameter name.

Improved as suggested.

Attach the new patch set.

Regards,
Wang wei