diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 4f3c4efeff..e1d3dc99ca 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -716,8 +716,8 @@ HandleParallelApplyMessages(void) /* * Set up a dynamic shared memory segment. * - * We set up a control region that contains a ParallelApplyWorkerShared, plus - * one region to each of message queue and error queue. + * We set up a control region that contains a fixed worker info + * (ParallelApplyWorkerShared), a message queue, and an error queue. * * Returns true on success, false on failure. */ @@ -916,9 +916,9 @@ parallel_apply_get_in_xact(ParallelApplyWorkerShared *wshared) * Define a savepoint for a subxact in parallel apply worker if needed. * * The parallel apply worker can figure out if a new subtransaction was - * started by checking if the new change arrived with different xid. In that + * started by checking if the new change arrived with a different xid. In that * case define a named savepoint, so that we are able to rollback to it - * separately later. + * if required. */ void parallel_apply_subxact_info_add(TransactionId current_xid, TransactionId top_xid) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index d34de7f5ae..67ac2be818 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1346,6 +1346,11 @@ apply_handle_origin(StringInfo s) /* * Handle STREAM START message. + * + * XXX We can avoid sending pair of the START/STOP messages to the parallel + * worker because unlike apply worker it will process only one + * transaction-at-a-time. However, it is not clear whether that is worth the + * effort because it is sent after logical_decoding_work_mem changes. */ static void apply_handle_stream_start(StringInfo s) @@ -1442,7 +1447,7 @@ apply_handle_stream_start(StringInfo s) case TRANS_PARALLEL_APPLY: /* - * Make sure the handle apply_dispatch methods are aware we're in + * Make sure the apply_dispatch methods are aware that we're in * a remote transaction. */ in_remote_transaction = true; @@ -4556,16 +4561,29 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo) *winfo = NULL; if (am_parallel_apply_worker()) + { return TRANS_PARALLEL_APPLY; + } else if (in_remote_transaction) + { + /* + * The in_remote_transaction can be set both for leader and parallel + * apply worker but because of a prior check for the parallel worker, + * we will reach here only for leader apply worker. + */ return TRANS_LEADER_APPLY; + } /* * Check if we are processing this transaction using a parallel apply * worker and if so, send the changes to that worker. */ else if ((*winfo = parallel_apply_find_worker(xid))) + { return TRANS_LEADER_SEND_TO_PARALLEL; + } else + { return TRANS_LEADER_SERIALIZE; + } }