Re: Parallel Apply

On Tue, 6 Oct 2026 at 12:07, Hayato Kuroda (Fujitsu)
<kuroda(dot)hayato(at)fujitsu(dot)com> wrote:
>
> Thanks for the contact. Here is an updated version.

Few comments on 0002, 0003 and 0004 patch:
1. Declaration without Implementation (Missing Code)
In src/include/replication/worker_internal.h, a new function
update_flush_position is declared:
+++ b/src/include/replication/worker_internal.h
@@ -334,7 +334,9 @@ extern void SetupApplyOrSyncWorker(int worker_slot);

extern void DisableSubscriptionAndExit(void);

-extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);
+extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn,
+
TransactionId pa_remote_xid);
+extern void update_flush_position(dlist_node *node, XLogRecPtr local_lsn);

However, the implementation for this function is missing from both
applyparallelworker.c and worker.c.

2. Misleading Comment in pa_get_last_commit_end
In src/backend/replication/logical/applyparallelworker.c, the comment
for pa_get_last_commit_end says:
+ /*
+ * Return InvalidXLogRecPtr if the transaction is still in
progress in the
+ * parallel apply worker.
+ */
+ if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
+ return false;

The internal comment incorrectly states that the function returns
InvalidXLogRecPtr, whereas it returns a boolean (false) to indicate
the transaction is
still in progress. The external comment correctly describes the return value.

3. In pa_setup_dsm (the leader path), the handles are declared on the
stack but not initialized:
@@ -402,6 +430,15 @@ pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
shm_mq *mq;
Size queue_size = DSM_QUEUE_SIZE;
Size error_queue_size = DSM_ERROR_QUEUE_SIZE;
+ dsa_handle parallel_apply_dsa_handle;
+ dshash_table_handle parallelized_txns_handle;
+
+ pa_attach_parallelized_txn_hash(&parallel_apply_dsa_handle,
+
&parallelized_txns_handle);
+
+ if (parallel_apply_dsa_handle == DSA_HANDLE_INVALID ||
+ parallelized_txns_handle == DSHASH_HANDLE_INVALID)
+ return false;

If pa_attach_parallelized_txn_hash were to return without assigning
these (which it does in the worker path, see below), the leader would
be checking
uninitialized memory against DSA_HANDLE_INVALID. While the leader
currently always hits a path that assigns them, initializing these to
INVALID at the
call site is safer.

4. We could set pa_dsa_handle and pa_dshash_handle outparams in
pa_attach_parallelized_txn_hash of the am_parallel_apply_worker()
branch to prevent check with uninitialized value in the caller:
+ parallelized_txns =
dshash_create(parallel_apply_dsa_area, &dsh_params, NULL);
+
+ /* Store handles in shared memory for other backends to use. */
+ *pa_dsa_handle = dsa_get_handle(parallel_apply_dsa_area);
+ *pa_dshash_handle =
dshash_get_hash_table_handle(parallelized_txns);
+ }
+ else if (am_parallel_apply_worker())
+ {
+ /* Attach to existing dynamic shared hash table. */
+ parallel_apply_dsa_area =
dsa_attach(MyParallelShared->parallel_apply_dsa_handle);
+ dsa_pin_mapping(parallel_apply_dsa_area);
+ parallelized_txns =
dshash_attach(parallel_apply_dsa_area, &dsh_params,
+
MyParallelShared->parallelized_txns_handle,
+
NULL);
+ }

5. Busy Waiting in pa_wait_for_depended_transaction, the dependency
wait loop uses a busy-wait strategy with a 1-millisecond sleep:

+ * The above race window is small and infrequent, so
we avoid WaitLatch
+ * and just sleep briefly to prevent busy waiting.
+ *
+ * XXX: might be a performance bottleneck due to busy waiting.
+ */
+ pg_usleep(1000L);

Have we tried to use a condition variable where the worker wakes up
any waiters instead of hard coding with 1 ms sleep.

Regards,
Vignesh