Re: Parallel Apply

On Tue, Apr 21, 2026 at 4:49 PM Zhijie Hou (Fujitsu)
<houzj(dot)fnst(at)fujitsu(dot)com> wrote:
>
> On Monday, April 20, 2026 7:06 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
> >
> > On Fri, Apr 17, 2026 at 12:57 PM shveta malik <shveta(dot)malik(at)gmail(dot)com>
> > wrote:
> > >
> > > 2)
> > > Calling handle_dependency_on_change() from
> > > handle_streamed_transaction() is misleading, since the former is
> > > intended for non-streaming transactions, while the latter handles
> > > streaming ones.
> > >
> >
> > Can you first explain in which case, do we need to handle dependency
> > for streamed transactions? IIUC, it is done in later patches, so we
> > can move this part of code to later patches such that these should be
> > able to handle stream transactions along with parallel-non-stream
> > transactions.
>
> Yes, it's done in later patches, I moved the corresponding logic
> to that patch in the updated version.
>
> >
> > >
> > > 5)
> > > I started reading 002's commit message, I think it is not that clear.
> > > I was trying to find if we have actual value for remote-xid which is
> > > key to hash tbale. But I think it is hash-table for only xid as key
> > > for faster access may be? If so, can we please improve commit messagee
> > > little bit?
> > >
> >
> > Right, and it is better to clarify if the transaction to wait is local
> > or remote?
>
> Improved the commit message.
>
> >
> >
> > Few other comments:
> > ===================
> > 1.
> > @@ -1916,7 +2015,106 @@ apply_handle_commit(StringInfo s)
> > {
> > ...
> > + case TRANS_LEADER_PARTIAL_SERIALIZE:
> > + Assert(winfo);
> > +
> > + /*
> > + * Build a dependency with the last committed transaction if not
> > + * already done.
> > + */
> > + if (apply_action != TRANS_LEADER_SEND_TO_PARALLEL)
> > + build_dependency_with_last_committed_txn(winfo);
> > +
> > + stream_open_and_write_change(remote_xid,
> > LOGICAL_REP_MSG_COMMIT,
> > + &original_msg);
> > +
> > + pa_set_fileset_state(winfo->shared, FS_SERIALIZE_DONE);
> > +
> > + /* Finish processing the transaction. */
> > + pa_xact_finish(winfo, commit_data.end_lsn);
> >
> > Can we move the serialize_to_file case handling as a separate patch,
> > probably at the end, if possible? It will simplify the base patches
> > and make them easier to review.
>
> Done as suggested.
>
> >
> > 2.
> > + /*
> > + * The last remote transaction that modified the relation's schema or
> > + * truncated the relation.
> > + */
> > + TransactionId last_depended_xid;
> >
> > It will be better to explain a bit on how it is used?
>
> Added.
>
> Here is updated patch set which addressed all comments so far.
>
> For 0001, I refactored the INTERNAL_MESSAGE handling to centralize the
> processing of both internal and logical replication messages. We now add one bit
> LOGICAL_REP_MSG_INTERNAL_MESSAGE to LogicalRepMsgType to indicate internal
> messages. In apply_dispatch, the worker can then check the sub-internal
> message type after reading LOGICAL_REP_MSG_INTERNAL_MESSAGE. This avoids
> the maintenance burden of ensuring that sub-internal message types do not
> conflict with LogicalRepMsgType values.
>
> Additionally, I've improved a few comments and the commit message based on
> internal feedback from Peter Smith.
>

Thank You for the patch.

Regarding 0001, I did not understand the need of having 2 separate messages:

+#define PARALLEL_APPLY_INTERNAL_MESSAGE 'i'
+ LOGICAL_REP_MSG_INTERNAL_MESSAGE = 'i',

And the need of sending both together in 0003:

+send_internal_dependencies(ParallelApplyWorkerInfo *winfo, List
*depends_on_xids)
+{
+ pq_sendbyte(&dependencies, PARALLEL_APPLY_INTERNAL_MESSAGE);
+ pq_sendbyte(&dependencies, LOGICAL_REP_MSG_INTERNAL_MESSAGE);

Also, it is confusing that above 2 are 'i' and
WORKER_INTERNAL_MSG_RELATION is also 'i'. Code has become very tricky
to understand now.

Reviewing everything, I feel having 'i' outside of LogicalRepMsgType
was better. I think it will eb better to retain
PARALLEL_APPLY_INTERNAL_MESSAGE and getting rid of
LOGICAL_REP_MSG_INTERNAL_MESSAGE. And when any worker intercepts
PARALLEL_APPLY_INTERNAL_MESSAGE, it need not dispatch
(apply_dispatch), instead it can handle it using
apply_handle_internal_message()

Goign above way:
--Messaged received from pub can be handled using apply_dispatch.
--Messages generated from leader to be handled separately/internally
using apply_handle_internal_message().

That way we have clear-cut boundary between the two types and less confusion.

Also, can we use 'R' for WORKER_INTERNAL_MSG_RELATION similar to
LOGICAL_REP_MSG_RELATION i.e. if 'i' is followed by 'R', then it means
it is internal relation-msg instead of pub's one? 'R' seems a better
choice over 'i' here.

Thoughts?

thanks
Shveta