Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1oWY2A-0008Cx-Vx for pgsql-hackers@arkaria.postgresql.org; Fri, 09 Sep 2022 07:02:39 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.92) (envelope-from ) id 1oWY29-0003Sx-R4 for pgsql-hackers@arkaria.postgresql.org; Fri, 09 Sep 2022 07:02:37 +0000 Received: from magus.postgresql.org ([2a02:c0:301:0:ffff::29]) by malur.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1oWY29-0003So-F3 for pgsql-hackers@lists.postgresql.org; Fri, 09 Sep 2022 07:02:37 +0000 Received: from mail-oa1-x29.google.com ([2001:4860:4864:20::29]) by magus.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_128_GCM_SHA256:128) (Exim 4.92) (envelope-from ) id 1oWY26-0005Q1-Om for pgsql-hackers@lists.postgresql.org; Fri, 09 Sep 2022 07:02:37 +0000 Received: by mail-oa1-x29.google.com with SMTP id 586e51a60fabf-127ba06d03fso1724529fac.3 for ; Fri, 09 Sep 2022 00:02:34 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20210112; h=content-transfer-encoding:cc:to:subject:message-id:date:from :in-reply-to:references:mime-version:from:to:cc:subject:date; bh=vvJUET2Dvyj8fKsZSn/dRg1qGgjvQLzWvWL3ySZEwGU=; b=FDz69xfHOp9oBH5NRYDzCkJ8a1Db9V7mfTiWDaZXIZZS1o3E8Ty48W8i8uIS/XjHFb pESKr3VDXaBeEpGtfrED/5QE2ophByXpSivn/vztVgmvMPUhRz+R1HoF+iGNCYG1KJgr vEBwPay9oKXqfn+TvP6d1GZ1W60Hz5/GndRfXqokg/d/KplZC3dwM15fG+vinSaAQ6Xd OBCk/kuCNED+ojPNvan+/GtkmgswhhvT5nsTnvK/7diSfJfPGdx5Ksjps5xyfropHR/7 4n6HDyu7+61BPIToBFmWj+aZgjGD48uyPCO6ke3/fD1KCZcDiF4uU0QgdsTbapzXUOV7 Wotg== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20210112; h=content-transfer-encoding:cc:to:subject:message-id:date:from :in-reply-to:references:mime-version:x-gm-message-state:from:to:cc :subject:date; bh=vvJUET2Dvyj8fKsZSn/dRg1qGgjvQLzWvWL3ySZEwGU=; b=Z6rRdv2egPNg7q9hUCpwSzOkE3bwoaKNbyXoG7I7LOO/wCZIhV6iBEfPqagaDNIW7O PbWKlxCf9RLm9C+4EBb6T28rpBbTBRZRgaNlB2zzaM965GPkeQtwBvBLvcFmzQVdQOGp /2SovfHF413WpSPdlQT7dA5oA82V9v1iXD9bVIZrl3glVl/ntoJ1J/Afrzb23o4cvd6O 94YbZrY8ED/qhbeziAeJDtto+i3eM16BjJOHkPTVYjayQ8I9UfdhRTM5VCKfY8ScSsHV 7gdYbAwtQ6U7N5fvrSrYrj6kFT9JWRaYr1lsoYFs9m/7N+te47a0sd2zzmBOXcTIBpH7 AyBw== X-Gm-Message-State: ACgBeo0UOlHhKqy+rLGh92gX97NTbDx7Uvfg8ed5yIhM37diT+/nIZiV QhAs0Y0hz8aKmv44/n8fGuIhSg9zXXsbR1Ix6ujv8REM3U8= X-Google-Smtp-Source: AA6agR6e3iwR/JGrnve5snY0roZu099H51HgIAx/y39xG5ouJdwgawINwyK27nAP+rAl5mqO/Iev8rIrXOs3orS5+mw= X-Received: by 2002:a05:6808:1995:b0:343:4888:fb7b with SMTP id bj21-20020a056808199500b003434888fb7bmr3120253oib.163.1662706952893; Fri, 09 Sep 2022 00:02:32 -0700 (PDT) MIME-Version: 1.0 References: In-Reply-To: From: Peter Smith Date: Fri, 9 Sep 2022 17:02:16 +1000 Message-ID: Subject: Re: Perform streaming logical transactions by background workers and parallel apply To: "houzj.fnst@fujitsu.com" Cc: Amit Kapila , "wangw.fnst@fujitsu.com" , Dilip Kumar , Masahiko Sawada , "shiy.fnst@fujitsu.com" , PostgreSQL Hackers Content-Type: text/plain; charset="UTF-8" Content-Transfer-Encoding: quoted-printable List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Archived-At: Precedence: bulk Here are my review comments for the v28-0001 patch: (There may be some overlap with other people's review comments and/or some fixes already made). =3D=3D=3D=3D=3D=3D 1. Commit Message In addition, the patch extends the logical replication STREAM_ABORT message= so that abort_time and abort_lsn can also be sent which can be used to update = the replication origin in parallel apply worker when the streaming transaction = is aborted. ~ Should this also mention that because this message extension is needed to support parallel streaming, meaning that parallel streaming is not supported for publications on servers < PG16? =3D=3D=3D=3D=3D=3D 2. doc/src/sgml/config.sgml Specifies maximum number of logical replication workers. This incl= udes - both apply workers and table synchronization workers. + apply leader workers, parallel apply workers, and table synchroniz= ation + workers. "apply leader workers" -> "leader apply workers" ~~~ 3. max_logical_replication_workers (integer) Specifies maximum number of logical replication workers. This includes apply leader workers, parallel apply workers, and table synchronization workers. Logical replication workers are taken from the pool defined by max_worker_processes. The default value is 4. This parameter can only be set at server start. ~ I did not really understand why the default is 4. Because the default tablesync workers is 2, and the default parallel workers is 2, but what about accounting for the apply worker? Therefore, shouldn't max_logical_replication_workers default be 5 instead of 4? =3D=3D=3D=3D=3D=3D 4. src/backend/commands/subscriptioncmds.c - defGetStreamingMode + } + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("%s requires a Boolean value or \"parallel\"", + def->defname))); + return SUBSTREAM_OFF; /* keep compiler quiet */ +} Some whitespace before the ereport and the return might be tidier. =3D=3D=3D=3D=3D=3D 5. src/backend/libpq/pqmq.c + { + if (IsParallelWorker()) + SendProcSignal(pq_mq_parallel_leader_pid, + PROCSIG_PARALLEL_MESSAGE, + pq_mq_parallel_leader_backend_id); + else + { + Assert(IsLogicalParallelApplyWorker()); + SendProcSignal(pq_mq_parallel_leader_pid, + PROCSIG_PARALLEL_APPLY_MESSAGE, + pq_mq_parallel_leader_backend_id); + } + } This code can be simplified if you want to. For example, { ProcSignalReason reason; Assert(IsParallelWorker() || IsLogicalParallelApplyWorker()); reason =3D IsParallelWorker() ? PROCSIG_PARALLEL_MESSAGE : PROCSIG_PARALLEL_APPLY_MESSAGE; SendProcSignal(pq_mq_parallel_leader_pid, reason, pq_mq_parallel_leader_backend_id); } =3D=3D=3D=3D=3D=3D 6. src/backend/replication/logical/applyparallelworker.c Is there a reason why this file is called applyparallelworker.c instead of parallelapplyworker.c? Now this name is out of step with names of all the new typedefs etc. ~~~ 7. +/* + * There are three fields in each message received by parallel apply worke= r: + * start_lsn, end_lsn and send_time. Because we have updated these statist= ics + * in leader apply worker, we could ignore these fields in parallel apply + * worker (see function LogicalRepApplyLoop). + */ +#define SIZE_STATS_MESSAGE (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz)) SUGGESTION (Just dded word "the" and change "could" -> "can") There are three fields in each message received by the parallel apply worker: start_lsn, end_lsn and send_time. Because we have updated these statistics in the leader apply worker, we can ignore these fields in the parallel apply worker (see function LogicalRepApplyLoop). ~~~ 8. +/* + * List that stores the information of parallel apply workers that were + * started. Newly added worker information will be removed from the list a= t the + * end of the transaction when there are enough workers in the pool. Besid= es, + * exited workers will be removed from the list after being detected. + */ +static List *ParallelApplyWorkersList =3D NIL; Perhaps this comment can give more explanation of what is meant by the part that says "when there are enough workers in the pool". ~~~ 9. src/backend/replication/logical/applyparallelworker.c - parallel_apply_can_start + /* + * Don't start a new parallel worker if not in streaming parallel mode. + */ + if (MySubscription->stream !=3D SUBSTREAM_PARALLEL) + return false; "streaming parallel mode." -> "parallel streaming mode." ~~~ 10. + /* + * For streaming transactions that are being applied using parallel apply + * worker, we cannot decide whether to apply the change for a relation tha= t + * is not in the READY state (see should_apply_changes_for_rel) as we won'= t + * know remote_final_lsn by that time. So, we don't start the new parallel + * apply worker in this case. + */ + if (!AllTablesyncsReady()) + return false; "using parallel apply worker" -> "using a parallel apply worker" ~~~ 11. + /* + * Do not allow parallel apply worker to be started in the parallel apply + * worker. + */ + if (am_parallel_apply_worker()) + return false; I guess the comment is valid but it sounds strange. SUGGESTION Only leader apply workers can start parallel apply workers. ~~~ 12. + if (am_parallel_apply_worker()) + return false; Maybe this code should be earlier in this function, because surely this is a less costly test than the test for !AllTablesyncsReady()? ~~~ 13. src/backend/replication/logical/applyparallelworker.c - parallel_apply_start_worker +/* + * Start a parallel apply worker that will be used for the specified xid. + * + * If a parallel apply worker is not in use then re-use it, otherwise star= t a + * fresh one. Cache the worker information in ParallelApplyWorkersHash key= ed by + * the specified xid. + */ "is not in use" -> "is found but not in use" ? ~~~ 14. + /* Failed to start a new parallel apply worker. */ + if (winfo =3D=3D NULL) + return; There seem to be quite a lot of places (like this example) where something may go wrong and the behaviour apparently will just silently fall-back to using the non-parallel streaming. Maybe that is OK, but I am just wondering how can the user ever know this has happened? Maybe the docs can mention that this could happen and give some description of what processes users can look for (or some other strategy) so they can just confirm that the parallel streaming is really working like they assume it to be? ~~~ 15. + * Set this flag in the leader instead of the parallel apply worker to + * avoid the race condition where the leader has already started waiting + * for the parallel apply worker to finish processing the transaction(set + * the in_parallel_apply_xact to false) while the child process has not ye= t + * processed the first STREAM_START and has not set the + * in_parallel_apply_xact to true. Missing whitespace before "(" ~~~ 16. src/backend/replication/logical/applyparallelworker.c - parallel_apply_find_worker + /* Return the cached parallel apply worker if valid. */ + if (stream_apply_worker !=3D NULL) + return stream_apply_worker; Perhaps 'cur_stream_parallel_apply_winfo' is a better name for this var? ~~~ 17. src/backend/replication/logical/applyparallelworker.c - parallel_apply_free_worker +/* + * Remove the parallel apply worker entry from the hash table. And stop th= e + * worker if there are enough workers in the pool. + */ +void +parallel_apply_free_worker(ParallelApplyWorkerInfo *winfo, TransactionId x= id) I think the reason for doing the "enough workers in the pool" logic needs some more explanation. ~~~ 18. + if (napplyworkers > (max_parallel_apply_workers_per_subscription / 2)) + { + logicalrep_worker_stop_by_slot(winfo->shared->logicalrep_worker_slot_no, + winfo->shared->logicalrep_worker_generation); + + ParallelApplyWorkersList =3D list_delete_ptr(ParallelApplyWorkersList, wi= nfo); + + shm_mq_detach(winfo->mq_handle); + shm_mq_detach(winfo->error_mq_handle); + dsm_detach(winfo->dsm_seg); + pfree(winfo); + } + else + winfo->in_use =3D false; Maybe it is easier to remove this "else" and just unconditionally set winfo->in_use =3D false BEFORE the check to free the entire winfo. ~~~ 19. src/backend/replication/logical/applyparallelworker.c - LogicalParallelApplyLoop + ApplyMessageContext =3D AllocSetContextCreate(ApplyContext, + "ApplyMessageContext", + ALLOCSET_DEFAULT_SIZES); Should the name of this context be "ParallelApplyMessageContext"? ~~~ 20. src/backend/replication/logical/applyparallelworker.c - HandleParallelApplyMessage + default: + { + elog(ERROR, "unrecognized message type received from parallel apply worker: %c (message length %d bytes)", + msgtype, msg->len); + } "received from" -> "received by" ~~~ 21. src/backend/replication/logical/applyparallelworker.c - HandleParallelApplyMessages +/* + * Handle any queued protocol messages received from parallel apply worker= s. + */ +void +HandleParallelApplyMessages(void) 21a. "received from" -> "received by" ~ 21b. I wonder if this comment should give some credit to the function in parallel.c - because this seems almost a copy of all that code. ~~~ 22. src/backend/replication/logical/applyparallelworker.c - parallel_apply_set_xact_finish +/* + * Set the in_parallel_apply_xact flag for the current parallel apply work= er. + */ +void +parallel_apply_set_xact_finish(void) Should that "Set" really be saying "Reset" or "Clear"? =3D=3D=3D=3D=3D=3D 23. src/backend/replication/logical/launcher.c - logicalrep_worker_launch + nparallelapplyworkers =3D logicalrep_parallel_apply_worker_count(subid); + + /* + * Return silently if the number of parallel apply workers reached the + * limit per subscription. + */ + if (is_subworker && nparallelapplyworkers >=3D max_parallel_apply_workers_per_subscription) + { + LWLockRelease(LogicalRepWorkerLock); + return false; } I=E2=80=99m not sure if this is a good idea to be so silent. How will the u= ser know if they should increase the GUC parameter or not if it never tells them that the value is too low? ~~~ 24. /* Now wait until it attaches. */ - WaitForReplicationWorkerAttach(worker, generation, bgw_handle); + return WaitForReplicationWorkerAttach(worker, generation, bgw_handle); The comment feels a tiny bit misleading, because there is a chance that this might not attach at all and return false if something goes wrong. ~~~ 25. src/backend/replication/logical/launcher.c - logicalrep_worker_stop +void +logicalrep_worker_stop_by_slot(int slot_no, uint16 generation) +{ + LogicalRepWorker *worker =3D &LogicalRepCtx->workers[slot_no]; + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + + /* Return if the generation doesn't match or the worker is not alive. */ + if (worker->generation !=3D generation || + worker->proc =3D=3D NULL) + return; + + logicalrep_worker_stop_internal(worker); + + LWLockRelease(LogicalRepWorkerLock); +} I think this condition should be changed and reversed, otherwise you might return before releasing the lock (??) SUGGESTION { LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); /* Stop only if the worker is alive and the generation matches. */ if (worker && worker->proc && worker->generation =3D=3D generation) logicalrep_worker_stop_internal(worker); LWLockRelease(LogicalRepWorkerLock); } ~~~ 26 src/backend/replication/logical/launcher.c - logicalrep_worker_stop_inte= rnal +/* + * Workhorse for logicalrep_worker_stop() and logicalrep_worker_detach(). = Stop + * the worker and wait for it to die. + */ ... and logicalrep_worker_stop_by_slot() ~~~ 27. src/backend/replication/logical/launcher.c - logicalrep_worker_detach + /* + * This is the leader apply worker; stop all the parallel apply workers + * previously started from here. + */ + if (!isParallelApplyWorker(MyLogicalRepWorker)) 27a. The comment does not match the code. If this *is* the leader apply worker then why do we have the condition to check that? Maybe only needs a comment update like SUGGESTION If this is the leader apply worker then stop all the parallel... ~ 27b. Code seems also assuming it cannot be a tablesync worker but it is not checking that. I am wondering if it will be better to have yet another macro/inline to do isLeaderApplyWorker() that will make sure this really is the leader apply worker. (This review comment suggestion is repeated later below). =3D=3D=3D=3D=3D=3D 28. src/backend/replication/logical/worker.c - STREAMED TRANSACTIONS commen= t + * If no worker is available to handle the streamed transaction, the data = is + * written to temporary files and then applied at once when the final comm= it + * arrives. SUGGESTION If streaming =3D true, or if streaming =3D parallel but there are not parallel apply workers available to handle the streamed transaction, the data is written to... ~~~ 29. src/backend/replication/logical/worker.c - TransactionApplyAction /* * What action to take for the transaction. * * TA_APPLY_IN_LEADER_WORKER means that we are in the leader apply worker a= nd * changes of the transaction are applied directly in the worker. * * TA_SERIALIZE_TO_FILE means that we are in leader apply worker and change= s * are written to temporary files and then applied when the final commit * arrives. * * TA_APPLY_IN_PARALLEL_WORKER means that we are in the parallel apply work= er * and changes of the transaction are applied directly in the worker. * * TA_SEND_TO_PARALLEL_WORKER means that we are in the leader apply worker = and * need to send the changes to the parallel apply worker. */ typedef enum { /* The action for non-streaming transactions. */ TA_APPLY_IN_LEADER_WORKER, /* Actions for streaming transactions. */ TA_SERIALIZE_TO_FILE, TA_APPLY_IN_PARALLEL_WORKER, TA_SEND_TO_PARALLEL_WORKER } TransactionApplyAction; ~ 29a. I think if you change all those enum names slightly (e.g. like below) then they can be more self-explanatory: TA_NOT_STREAMING_LEADER_APPLY TA_STREAMING_LEADER_SERIALIZE TA_STREAMING_LEADER_SEND_TO_PARALLEL TA_STREAMING_PARALLEL_APPLY ~ 29b. * TA_APPLY_IN_LEADER_WORKER means that we are in the leader apply worker a= nd * changes of the transaction are applied directly in the worker. Maybe that should mention this is for the non-streaming case, or if you change all the enums names like in 29a. then there is no need because it is more self-explanatory. ~~~ 30. src/backend/replication/logical/worker.c - should_apply_changes_for_rel * Note that for streaming transactions that are being applied in parallel + * apply worker, we disallow applying changes on a table that is not in + * the READY state, because we cannot decide whether to apply the change a= s we + * won't know remote_final_lsn by that time. "applied in parallel apply worker" -> "applied in the parallel apply worker= " ~~~ 31. + errdetail("Cannot handle streamed replication transaction by parallel " + "apply workers until all tables are synchronized."))); "by parallel apply workers" -> "using parallel apply workers" (?) ~~~ 32. src/backend/replication/logical/worker.c - handle_streamed_transaction Now that there is an apply_action enum I felt it is better for this code to be using a switch instead of all the if/else. Furthermore, it might be better to put the switch case in a logical order (e.g. same as the suggested enums value order of #29a). ~~~ 33. src/backend/replication/logical/worker.c - apply_handle_stream_prepare (same as comment #32) Now that there is an apply_action enum I felt it is better for this code to be using a switch instead of all the if/else. Furthermore, it might be better to put the switch case in a logical order (e.g. same as the suggested enums value order of #29a). ~~~ 34. src/backend/replication/logical/worker.c - apply_handle_stream_start (same as comment #32) Now that there is an apply_action enum I felt it is better for this code to be using a switch instead of all the if/else. Furthermore, it might be better to put the switch case in a logical order (e.g. same as the suggested enums value order of #29a). ~~~ 35. + else if (apply_action =3D=3D TA_SERIALIZE_TO_FILE) + { + /* + * Notify handle methods we're processing a remote in-progress + * transaction. + */ + in_streamed_transaction =3D true; + + /* + * Since no parallel apply worker is available for the first + * stream start, serialize all the changes of the transaction. + * "Since no parallel apply worker is available". I don't think the comment is quite correct. Maybe it is doing the serialization because the user simply did not request to use the parallel mode at all? ~~~ 36. src/backend/replication/logical/worker.c - apply_handle_stream_stop (same as comment #32) Now that there is an apply_action enum I felt it is better for this code to be using a switch instead of all the if/else. Furthermore, it might be better to put the switch case in a logical order (e.g. same as the suggested enums value order of #29a). ~~~ 37. src/backend/replication/logical/worker.c - apply_handle_stream_abort + /* + * Check whether the publisher sends abort_lsn and abort_time. + * + * Note that the paralle apply worker is only started when the publisher + * sends abort_lsn and abort_time. + */ typo "paralle" ~~~ 38. (same as comment #32) Now that there is an apply_action enum I felt it is better for this code to be using a switch instead of all the if/else. Furthermore, it might be better to put the switch case in a logical order (e.g. same as the suggested enums value order of #29a). ~~~ 39. + /* + * Set in_parallel_apply_xact to true again as we only aborted the + * subtransaction and the top transaction is still in progress. No + * need to lock here because currently only the apply leader are + * accessing this flag. + */ "are accessing" -> "is accessing" ~~~ 40. src/backend/replication/logical/worker.c - apply_handle_stream_commit (same as comment #32) Now that there is an apply_action enum I felt it is better for this code to be using a switch instead of all the if/else. Furthermore, it might be better to put the switch case in a logical order (e.g. same as the suggested enums value order of #29a). ~~~ 41. src/backend/replication/logical/worker.c - store_flush_position + /* Skip if not the leader apply worker */ + if (am_parallel_apply_worker()) + return; + Code might be better to implement/use a new function so it can check something like !am_leader_apply_worker() ~~~ 42. src/backend/replication/logical/worker.c - InitializeApplyWorker +/* + * Initialize the database connection, in-memory subscription and necessar= y + * config options. + */ I still think this should mention that this is common initialization code for "both leader apply workers, and parallel apply workers" ~~~ 43. src/backend/replication/logical/worker.c - ApplyWorkerMain - /* This is main apply worker */ + /* This is leader apply worker */ "is leader" -> "is the leader" ~~~ 44. src/backend/replication/logical/worker.c - IsLogicalParallelApplyWorker +/* + * Is current process a logical replication parallel apply worker? + */ +bool +IsLogicalParallelApplyWorker(void) +{ + return am_parallel_apply_worker(); +} + It seems a bit strange to have this function IsLogicalParallelApplyWorker, and also am_parallel_apply_worker() which are basically identical except one of them is static and one is not. I wonder if there should be just one function. And if you really do need 2 names for consistency then you can just define a synonym like #define am_parallel_apply_worker IsLogicalParallelApplyWorker ~~~ 45. src/backend/replication/logical/worker.c - get_transaction_apply_action +/* + * Return the action to take for the given transaction. Also return the + * parallel apply worker information if the action is + * TA_SEND_TO_PARALLEL_WORKER. + */ +static TransactionApplyAction +get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo) I think this should be slightly more clear to say that *winfo is assigned to the destination parallel worker info (if the action is TA_SEND_TO_PARALLEL_WORKER), otherwise *winfo is assigned NULL (see also #46 below) ~~~ 46. +static TransactionApplyAction +get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo) +{ + if (am_parallel_apply_worker()) + return TA_APPLY_IN_PARALLEL_WORKER; + else if (in_remote_transaction) + return TA_APPLY_IN_LEADER_WORKER; + + /* + * Check if we are processing this transaction using a parallel apply + * worker and if so, send the changes to that worker. + */ + else if ((*winfo =3D parallel_apply_find_worker(xid))) + return TA_SEND_TO_PARALLEL_WORKER; + else + return TA_SERIALIZE_TO_FILE; +} The code is a bit quirky at the moment because sometimes the *winfo will be assigned NULL and sometimes it will be assigned valid value, and sometimes it will still be unassigned. I suggest always assigning it either NULL or valid. SUGGESTIONS static TransactionApplyAction get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **w= info) { *winfo =3D NULL; <=3D=3D add this default assignment ... =3D=3D=3D=3D=3D=3D 47. src/backend/storage/ipc/procsignal.c - procsignal_sigusr1_handler @@ -657,6 +658,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS) if (CheckProcSignal(PROCSIG_LOG_MEMORY_CONTEXT)) HandleLogMemoryContextInterrupt(); + if (CheckProcSignal(PROCSIG_PARALLEL_APPLY_MESSAGE)) + HandleParallelApplyMessageInterrupt(); + I wasn=E2=80=99t sure about the placement of this new code because those CheckProcSignal don=E2=80=99t seem to have any particular order. I think th= is belongs adjacent to the PROCSIG_PARALLEL_MESSAGE since it has the most in common with that one. =3D=3D=3D=3D=3D=3D 48. src/backend/tcop/postgres.c @@ -3377,6 +3377,9 @@ ProcessInterrupts(void) if (LogMemoryContextPending) ProcessLogMemoryContextInterrupt(); + + if (ParallelApplyMessagePending) + HandleParallelApplyMessages(); (like #47) I think this belongs adjacent to the ParallelMessagePending check since it has most in common with that one. =3D=3D=3D=3D=3D=3D 49. src/include/replication/worker_internal.h @@ -60,6 +64,12 @@ typedef struct LogicalRepWorker */ FileSet *stream_fileset; + /* + * PID of leader apply worker if this slot is used for a parallel apply + * worker, InvalidPid otherwise. + */ + pid_t apply_leader_pid; + /* Stats. */ XLogRecPtr last_lsn; TimestampTz last_send_time; Whitespace indent of the new member ok? ~~~ 50. +typedef struct ParallelApplyWorkerShared +{ + slock_t mutex; + + /* + * Flag used to ensure commit ordering. + * + * The parallel apply worker will set it to false after handling the + * transaction finish commands while the apply leader will wait for it to + * become false before proceeding in transaction finish commands (e.g. + * STREAM_COMMIT/STREAM_ABORT/STREAM_PREPARE). + */ + bool in_parallel_apply_xact; + + /* Information from the corresponding LogicalRepWorker slot. */ + uint16 logicalrep_worker_generation; + + int logicalrep_worker_slot_no; +} ParallelApplyWorkerShared; Whitespace indents of the new members ok? ~~~ 51. /* Main memory context for apply worker. Permanent during worker lifetime.= */ extern PGDLLIMPORT MemoryContext ApplyContext; +extern PGDLLIMPORT MemoryContext ApplyMessageContext; Maybe there should be a blank line between those externs, because the comment applies only to the first one, right? Alternatively modify the comment. ~~~ 52. src/include/replication/worker_internal.h - am_parallel_apply_worker I thought it might be worthwhile to also add another function like am_leader_apply_worker(). I noticed at least one place in this patch where it could have been called. SUGGESTION static inline bool am_parallel_apply_worker(void) { return !isParallelApplyWorker(MyLogicalRepWorker) && !am_tablesync_worker()= ; } =3D=3D=3D=3D=3D=3D 53. src/include/storage/procsignal.h @@ -35,6 +35,7 @@ typedef enum PROCSIG_WALSND_INIT_STOPPING, /* ask walsenders to prepare for shutdown = */ PROCSIG_BARRIER, /* global barrier interrupt */ PROCSIG_LOG_MEMORY_CONTEXT, /* ask backend to log the memory contexts */ + PROCSIG_PARALLEL_APPLY_MESSAGE, /* Message from parallel apply workers */ (like #47) I think this new enum belongs adjacent to the PROCSIG_PARALLEL_MESSAGE since it has most in common with that one =3D=3D=3D=3D=3D=3D 54. src/tools/pgindent/typedefs.list Missing TransactionApplyAction? ------ Kind Regards, Peter Smith. Fujitsu Australia