Re: Parallel Apply

On Mon, Mar 30, 2026 at 3:50 PM Hayato Kuroda (Fujitsu)
<kuroda(dot)hayato(at)fujitsu(dot)com> wrote:
>
> The patch set needs to be rebased due to 06d859a, and here is a fixed version.
>
> Now we are busy for other projects but planning to work on this, after the
> feature freeze.
>

Few comments:
=============
1.
index 058a955e20c..9042470f500 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -75,6 +75,8 @@ typedef enum LogicalRepMsgType
LOGICAL_REP_MSG_STREAM_COMMIT = 'c',
LOGICAL_REP_MSG_STREAM_ABORT = 'A',
LOGICAL_REP_MSG_STREAM_PREPARE = 'p',
+ LOGICAL_REP_MSG_INTERNAL_DEPENDENCY = 'd',
+ LOGICAL_REP_MSG_INTERNAL_RELATION = 'i',
} LogicalRepMsgType;

I don't think it is good to classify these as logical_rep messages as
they would be probably used to communicate between leader and parallel
workers.

2.
+/*
+ * Record in-progress transactions from the given list that are being depended
+ * on into the shared hash table.
+ */
+void
+pa_record_dependency_on_transactions(List *depends_on_xids)
+{
+ foreach_xid(xid, depends_on_xids)
+ {

Neither the comments atop this function nor in the function itself
make the functionality clear. At the end, either we are removing the
txn entry or releasing the lock which is not matching with the
function name.

3.
@@ -307,6 +307,7 @@ typedef struct FlushPosition
dlist_node node;
XLogRecPtr local_end;
XLogRecPtr remote_end;
+ TransactionId pa_remote_xid;
} FlushPosition;

I think it would be better to add a few comments explaining each field
and or how this structure is used. I understand we are adding only one
new field but still it would help understanding the new addition to
it.

4.
+ /* Compute dependency only for non-streaming transaction */
+ if (in_streamed_transaction || (winfo && winfo->stream_txn))
+ return;
+

Either explain here or somewhere else how this feature interacts with
streaming transactions. If it is already explained elsewhere then it
would be better to add the reference here.

5.
+ if (!replica_identity_table)
+ replica_identity_table = replica_identity_create(ApplyContext,
+ REPLICA_IDENTITY_INITIAL_SIZE,
+ NULL);

We can add some comments why we create this in ApplyContext. BTW, can
we consider a separate ParallelApplyContext for this though I am not
sure if that will be really helpful. Do you have any thoughts on the
same?

6.
Commit message of 0004:
--
commit order
--

There is a case where columns have no foreign or primary keys, and integrity is
maintained at the application layer. In this case, the above RI mechanism cannot
detect any dependencies. For safety reasons, parallel apply workers preserve the
commit ordering done on the publisher side.

Here, we should say that as currently we don't have a way to track
replication progress for out-of-order commit transactions, parallel
apply workers preserve the commit ordering done on the publisher side.
Am I missing something?

7. Shouldn't we explain a bit about commit order dependency in the
--dedendency waiting-- section in commit message?

--
With Regards,
Amit Kapila.