Re: Perform streaming logical transactions by background workers and parallel apply

On Fri, Jul 22, 2022 at 8:26 AM wangw.fnst@fujitsu.com
<wangw.fnst@fujitsu.com> wrote:
>
> On Tues, Jul 19, 2022 at 10:29 AM I wrote:
> > Attach the news patches.
>
> Not able to apply patches cleanly because the change in HEAD (366283961a).
> Therefore, I rebased the patch based on the changes in HEAD.
>
> Attach the new patches.
>

Few comments on 0001:
======================
1.
-       <structfield>substream</structfield> <type>bool</type>
+       <structfield>substream</structfield> <type>char</type>
       </para>
       <para>
-       If true, the subscription will allow streaming of in-progress
-       transactions
+       Controls how to handle the streaming of in-progress transactions:
+       <literal>f</literal> = disallow streaming of in-progress transactions,
+       <literal>t</literal> = spill the changes of in-progress transactions to
+       disk and apply at once after the transaction is committed on the
+       publisher,
+       <literal>p</literal> = apply changes directly using a background worker

Shouldn't the description of 'p' be something like: apply changes
directly using a background worker, if available, otherwise, it
behaves the same as 't'

2.
Note that if an error happens when
+          applying changes in a background worker, the finish LSN of the
+          remote transaction might not be reported in the server log.

Is there any case where finish LSN can be reported when applying via
background worker, if not, then we should use 'won't' instead of
'might not'?

3.
+#define PG_LOGICAL_APPLY_SHM_MAGIC 0x79fb2447 // TODO Consider change

It is better to change this as the same magic number is used by
PG_TEST_SHM_MQ_MAGIC

4.
+ /* Ignore statistics fields that have been updated. */
+ s.cursor += IGNORE_SIZE_IN_MESSAGE;

Can we change the comment to: "Ignore statistics fields that have been
updated by the main apply worker."? Will it be better to name the
define as "SIZE_STATS_MESSAGE"?

5.
+/* Apply Background Worker main loop */
+static void
+LogicalApplyBgwLoop(shm_mq_handle *mqh, volatile ApplyBgworkerShared *shared)
{
...
...

+ apply_dispatch(&s);
+
+ if (ConfigReloadPending)
+ {
+ ConfigReloadPending = false;
+ ProcessConfigFile(PGC_SIGHUP);
+ }
+
+ MemoryContextSwitchTo(oldctx);
+ MemoryContextReset(ApplyMessageContext);

We should not process the config file under ApplyMessageContext. You
should switch context before processing the config file. See other
similar usages in the code.

6.
+/* Apply Background Worker main loop */
+static void
+LogicalApplyBgwLoop(shm_mq_handle *mqh, volatile ApplyBgworkerShared *shared)
{
...
...
+ MemoryContextSwitchTo(oldctx);
+ MemoryContextReset(ApplyMessageContext);
+ }
+
+ MemoryContextSwitchTo(TopMemoryContext);
+ MemoryContextReset(ApplyContext);
...
}

I don't see the need to reset ApplyContext here as we don't do
anything in that context here.

-- 
With Regards,
Amit Kapila.