From db576364abce33d53bfb21058cc5c7c1862186ef Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Tue, 11 Aug 2026 14:04:04 +0800 Subject: [PATCH v23 04/12] Prepare commit order preservation for parallel apply This patch adds infrastructure to preserve commit order between transactions applied in parallel and those applied by the leader, ensuring the subscriber matches the publisher's commit order. Previously, sequential application preserved order automatically, but the upcoming parallel apply does not. This patch introduces a dependency mechanism where each transaction waits for its immediate predecessor to commit, preserving the overall order. A new message type (PAWorkerMsgType.PA_MSG_XACT_DEPENDENCY) is introduced to carry transaction IDs that must be waited for. This is currently used for commit order dependencies, and will later be extended for row-level dependency tracking. The leader sends this message to a parallel apply worker before sending the COMMIT message, instructing the worker to wait for the preceding transaction. The leader itself also waits for the last parallelized transaction to finish before committing its own transaction. Author: Zhijie Hou Author: Hayato Kuroda --- .../replication/logical/applyparallelworker.c | 80 +++++++++++++++++++ src/backend/replication/logical/worker.c | 76 ++++++++++++++++++ src/include/replication/worker_internal.h | 2 + 3 files changed, 158 insertions(+) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 9f9bc78885b..57366b1fbfb 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -859,6 +859,24 @@ ProcessParallelApplyInterrupts(void) } } +/* + * Handle internal dependency information. + * + * Wait for all transactions listed in the message to commit. + */ +static void +apply_handle_internal_dependency(StringInfo s) +{ + int nxids = pq_getmsgint(s, 4); + + for (int i = 0; i < nxids; i++) + { + TransactionId xid = pq_getmsgint(s, 4); + + pa_wait_for_depended_transaction(xid); + } +} + /* * Handle internal relation information. * @@ -895,6 +913,9 @@ apply_handle_internal_message(StringInfo s) switch (action) { + case PA_MSG_XACT_DEPENDENCY: + apply_handle_internal_dependency(s); + break; case PA_MSG_RELMAP: apply_handle_internal_relation(s); break; @@ -2023,3 +2044,62 @@ pa_attach_parallelized_txn_hash(dsa_handle *pa_dsa_handle, MemoryContextSwitchTo(oldctx); } + +/* + * Wait for the given remote transaction to finish applying by a parallel apply + * worker. + * + * Both leader and parallel apply workers can call this function to wait for a + * parallelized transaction to finish. + */ +void +pa_wait_for_depended_transaction(TransactionId xid) +{ + /* + * Quick exit if parallelized_txns has not been initialized yet. This can + * happen when the leader worker calls this function before any parallel + * apply workers have been launched. + */ + if (!parallelized_txns) + return; + + elog(DEBUG1, "wait for depended xid %u", xid); + + for (;;) + { + ParallelizedTxnEntry *txn_entry; + + txn_entry = dshash_find(parallelized_txns, &xid, false); + + /* The entry is removed only if the transaction is committed */ + if (txn_entry == NULL) + break; + + dshash_release_lock(parallelized_txns, txn_entry); + + /* + * Wait for the parallel apply worker processing the given remote + * transaction to finish applying and release its lock. + */ + pa_lock_transaction(xid, AccessShareLock); + pa_unlock_transaction(xid, AccessShareLock); + + CHECK_FOR_INTERRUPTS(); + + /* + * Acquiring the lock successfully does not guarantee we can proceed. + * The worker may have errored out and released the lock while leaving + * its shared hash entry intact, or it may not have acquired the lock + * yet because it hasn't processed the BEGIN message. In either case, we + * must continue waiting in the loop until the parallel apply worker + * finishes applying the transaction, or until the leader notifies us of + * a failure and restarts all workers. + * + * The above race window is small and infrequent, so we avoid WaitLatch + * and just sleep briefly to prevent busy waiting. + */ + pg_usleep(1000L); + } + + elog(DEBUG1, "finish waiting for depended xid %u", xid); +} diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 45bdd881a69..77445115d11 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -663,6 +663,8 @@ static void set_wal_receiver_timeout(void); static void on_exit_clear_xact_state(int code, Datum arg); +static void maintain_commit_order_dependency(ParallelApplyWorkerInfo *winfo); + /* * Form the origin name for the subscription. * @@ -1306,6 +1308,8 @@ apply_handle_commit(StringInfo s) */ ProcessSyncingRelations(commit_data.end_lsn); + maintain_commit_order_dependency(NULL); + pgstat_report_activity(STATE_IDLE, NULL); reset_apply_error_context_info(); } @@ -6542,3 +6546,75 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo) return TRANS_LEADER_APPLY; } } + +/* + * Send an INTERNAL_DEPENDENCY message to a parallel apply worker. + */ +static void +send_internal_dependencies(ParallelApplyWorkerInfo *winfo, List *depends_on_xids) +{ + StringInfoData dependencies; + + initStringInfo(&dependencies); + + pq_sendbyte(&dependencies, LOGICAL_REP_MSG_INTERNAL_MESSAGE); + pq_sendbyte(&dependencies, PA_MSG_XACT_DEPENDENCY); + pq_sendint32(&dependencies, list_length(depends_on_xids)); + + foreach_xid(xid, depends_on_xids) + pq_sendint32(&dependencies, xid); + + if (!pa_send_data(winfo, dependencies.len, dependencies.data)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not send data to the logical replication parallel apply worker"))); + + pfree(dependencies.data); +} + +/* + * Ensure commit order is preserved for transactions applied by either a + * parallel worker or the leader. + * + * For parallelized transactions, the leader sends an internal dependency to the + * given parallel worker, instructing it to wait for the last parallelized + * transaction to finish before proceeding. + * + * For transactions applied directly by the leader (when winfo is NULL), the + * leader waits for the last parallelized transaction to commit here. + * + * Since each transaction waits for its immediate predecessor, waiting for the + * last parallelized transaction transitively means waiting for all preceding + * parallelized transactions, thereby preserving the publisher's commit order. + */ +static void +maintain_commit_order_dependency(ParallelApplyWorkerInfo *winfo) +{ + FlushPosition *last_txn_node; + TransactionId last_pa_remote_xid; + + /* Return early if there are no pending transactions */ + if (dlist_is_empty(&lsn_mapping)) + return; + + /* + * Get the last flush position entry, which could contain the XID of the + * most recent parallelized transaction. + */ + last_txn_node = dlist_tail_element(FlushPosition, node, &lsn_mapping); + last_pa_remote_xid = last_txn_node->pa_remote_xid; + + /* + * Nothing to do if the last transaction was applied by the leader, or if it + * is parallelized but has already committed. While on it, we also try to + * update the local_end of the last transaction node. + */ + if (!TransactionIdIsValid(last_pa_remote_xid) || + pa_get_last_commit_end(last_pa_remote_xid, &last_txn_node->local_end)) + return; + + if (winfo == NULL) + pa_wait_for_depended_transaction(last_pa_remote_xid); + else + send_internal_dependencies(winfo, list_make1_xid(last_pa_remote_xid)); +} diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index d4733946656..471e37f1382 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -247,6 +247,7 @@ typedef struct ParallelApplyWorkerInfo */ typedef enum PAWorkerMsgType { + PA_MSG_XACT_DEPENDENCY = 'd', PA_MSG_RELMAP = 'r', } PAWorkerMsgType; @@ -383,6 +384,7 @@ extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo, extern void pa_distribute_remote_rel_to_workers(LogicalRepRelation *rel); extern void pa_bind_flush_position(TransactionId xid, XLogRecPtr *local_end); extern bool pa_get_last_commit_end(TransactionId xid, XLogRecPtr *local_end); +extern void pa_wait_for_depended_transaction(TransactionId xid); #define isParallelApplyWorker(worker) ((worker)->in_use && \ (worker)->type == WORKERTYPE_PARALLEL_APPLY) -- 2.43.0