diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 2e2cd0d535..158dbd93c9 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -707,10 +707,10 @@ pa_stop_idle_workers(void) } /* - * Check if the parallel apply worker is pending due to spooled messages. + * Returns true, if there is any pending spooled message, false otherwise. */ static bool -pa_has_spooled_message_pending() +pa_is_spooled_message_pending() { PartialFileSetState fileset_state; @@ -870,7 +870,7 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh) else if (shmq_res == SHM_MQ_WOULD_BLOCK) { /* Check if changes have been serialized to a file. */ - if (pa_has_spooled_message_pending()) + if (pa_is_spooled_message_pending()) { pa_spooled_messages(); } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index faebceb4c7..59a4366fec 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1558,15 +1558,17 @@ apply_handle_stream_start(StringInfo s) pa_send_data(winfo, s->len, s->data, !first_segment); + /* + * Once we start serializing the changes, the parallel apply worker + * will wait for the leader to release the stream lock until the + * end of the transaction. So, we don't need to release the lock or + * increment the stream count in that case. + */ if (!winfo->serialize_changes) { /* - * Unlock the shared object lock when no change was serialized - * so that the parallel apply worker can continue to receive - * changes. No need to take the lock if switched to serialize - * mode because the parallel apply worker should wait for the - * leader to release the stream lock until the end of the - * transaction. + * Unlock the shared object lock so that the parallel apply + * worker can continue to receive changes. */ if (!first_segment) pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);