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

From: Amit Kapila <[email-masked]>
Date: Mon, 9 Jan 2023 14:21:03 +0530

On Sun, Jan 8, 2023 at 11:32 AM [email-masked]
<[email-masked]> wrote:
>
> On Sunday, January 8, 2023 11:59 AM [email-masked] <[email-masked]> wrote:
> > Attach the updated patch set.
>
> Sorry, the commit message of 0001 was accidentally deleted, just attach
> the same patch set again with commit message.
>

Pushed the first (0001) patch.

-- 
With Regards,
Amit Kapila.

---

From: "Shinoda, Noriyoshi (PN Japan FSIP)" <[email-masked]>
Date: Mon, 9 Jan 2023 09:32:01 +0000

Hi, Thanks for the great new feature.

Applied patches include adding wait events LogicalParallelApplyMain, LogicalParallelApplyStateChange. 
However, it seems that monitoring.sgml only contains descriptions for pg_locks. The attached patch adds relevant wait event information.
Please update if you have a better description.

Noriyoshi Shinoda
-----Original Message-----
From: Amit Kapila <[email-masked]> 
Sent: Monday, January 9, 2023 5:51 PM
To: [email-masked]
Cc: Masahiko Sawada <[email-masked]>; [email-masked]; Peter Smith <[email-masked]>; [email-masked]; PostgreSQL Hackers <[email-masked]>; Dilip Kumar <[email-masked]>
Subject: Re: Perform streaming logical transactions by background workers and parallel apply

On Sun, Jan 8, 2023 at 11:32 AM [email-masked] <[email-masked]> wrote:
> Sorry, the commit message of 0001 was accidentally deleted, just 
> attach the same patch set again with commit message.
>

--
With Regards,
Amit Kapila.

---

From: "[email-masked]" <[email-masked]>
Date: Mon, 9 Jan 2023 10:15:20 +0000

On Monday, January 9, 2023 5:32 PM Shinoda, Noriyoshi (PN Japan FSIP) <[email-masked]> wrote:
> However, it seems that monitoring.sgml only contains descriptions for
> pg_locks. The attached patch adds relevant wait event information.

Thanks for reporting. I think for LogicalParallelApplyStateChange we'd better
document it in a consistent style with LogicalSyncStateChange, so I have
slightly adjusted the patch for the same.

Best regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 10 Jan 2023 04:55:55 +0000

> Pushed the first (0001) patch.

Thanks for pushing, here are the remaining patches.
I reordered the patch number to put patches that are easier to
commit in the front of others.

Best regards,
Hou zj

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Tue, 10 Jan 2023 14:46:38 +0900 (JST)

Hello.

At Mon, 9 Jan 2023 14:21:03 +0530, Amit Kapila <[email-masked]> wrote in 
> Pushed the first (0001) patch.

It added the following error message.

+	seg = dsm_attach(handle);
+	if (!seg)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("unable to map dynamic shared memory segment")));

On the other hand we already have the following one in parallel.c
(another in pg_prewarm)

	seg = dsm_attach(DatumGetUInt32(main_arg));
	if (seg == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("could not map dynamic shared memory segment")));

Although I don't see a technical difference between the two, all the
other occurances including the just above (except test_shm_mq) use
"could not". A faint memory in my non-durable memory tells me that we
have a policy that we use "can/could not" than "unable".

(Mmm. I find ones in StartBackgroundWorker and sepgsql_client_auth.)

Shouldn't we use the latter than the former?  If that's true, it seems
to me that test_shm_mq also needs the same amendment to avoid the same
mistake in future.

=====
index 2e5914d5d9..a2d7474ed4 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -891,7 +891,7 @@ ParallelApplyWorkerMain(Datum main_arg)
        if (!seg)
                ereport(ERROR,
                                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-                                errmsg("unable to map dynamic shared memory segment")));
+                                errmsg("could not map dynamic shared memory segment")));
 
        toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
        if (!toc)
diff --git a/src/test/modules/test_shm_mq/worker.c b/src/test/modules/test_shm_mq/worker.c
index 8807727337..005b56023b 100644
--- a/src/test/modules/test_shm_mq/worker.c
+++ b/src/test/modules/test_shm_mq/worker.c
@@ -81,7 +81,7 @@ test_shm_mq_main(Datum main_arg)
        if (seg == NULL)
                ereport(ERROR,
                                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-                                errmsg("unable to map dynamic shared memory segment")));
+                                errmsg("could not map dynamic shared memory segment")));
        toc = shm_toc_attach(PG_TEST_SHM_MQ_MAGIC, dsm_segment_address(seg));
        if (toc == NULL)
                ereport(ERROR,
=====

regards.

-- 
Kyotaro Horiguchi
NTT Open Source Software Center

---

From: Amit Kapila <[email-masked]>
Date: Tue, 10 Jan 2023 12:01:43 +0530

> have a policy that we use "can/could not" than "unable".

Right, it is mentioned in docs [1] (see section "Tricky Words to Avoid").

>         toc = shm_toc_attach(PG_TEST_SHM_MQ_MAGIC, dsm_segment_address(seg));

Can you please start a new thread and post these changes as we are
proposing to change existing message as well?

[1] - https://www.postgresql.org/docs/devel/error-style-guide.html

-- 
With Regards,
Amit Kapila.

---

From: Dilip Kumar <[email-masked]>
Date: Tue, 10 Jan 2023 17:17:41 +0530

> commit in the front of others.

I was looking into 0001, IMHO the pid should continue to represent the
main apply worker. So the pid will always show the main apply worker
which is actually receiving all the changes for the subscription (in
short working as logical receiver) and if it is applying changes
through a parallel worker then it should put the parallel worker pid
in a new column called 'parallel_worker_pid' or
'parallel_apply_worker_pid' otherwise NULL.  Thoughts?

-- 
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

---

From: Peter Smith <[email-masked]>
Date: Thu, 12 Jan 2023 15:23:49 +1100

Hi, here are some review comments for patch v78-0001.

======

General

1. (terminology)

AFAIK everywhere until now we’ve been referring everywhere
(docs/comments/code) to the parent apply worker as the "leader apply
worker". Not the "main apply worker". Not the "apply leader worker".
Not any other variations...

>From this POV I think the worker member "apply_leader_pid" would be
better named "leader_apply_pid",  but I see that this was already
committed to HEAD differently.

Maybe it is not possible (or you don't want) to change that internal
member name but IMO at least all the new code and docs should try to
be using consistent terminology (e.g. leader_apply_XXX) where
possible.

======

Commit message

2.

main_worker_pid is Process ID of the leader apply worker, if this process is a
apply parallel worker. NULL if this process is a leader apply worker or a
synchronization worker.

IIUC, this text is just cut/paste from the monitoring.sgml. In a
review comment below I suggest some changes to that text, so then this
commit message should also change to be the same.

~~

3.

The new column can make it easier to distinguish leader apply worker and apply
parallel worker which is also similar to the 'leader_pid' column in
pg_stat_activity.

SUGGESTION
The new column makes it easier to distinguish parallel apply workers
from other kinds of workers. It is implemented this way to be similar
to the 'leader_pid' column in pg_stat_activity.

======

doc/src/sgml/logical-replication.sgml

4.

+   being synchronized. Moreover, if the streaming transaction is applied in
+   parallel, there will be additional workers.

SUGGESTION
there will be additional workers -> there may be additional parallel
apply workers

======

doc/src/sgml/monitoring.sgml

5. pg_stat_subscription

@@ -3198,11 +3198,22 @@ SELECT pid, wait_event_type, wait_event FROM
pg_stat_activity WHERE wait_event i

      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>apply_leader_pid</structfield> <type>integer</type>
+      </para>
+      <para>
+       Process ID of the leader apply worker, if this process is a apply
+       parallel worker. NULL if this process is a leader apply worker or a
+       synchronization worker.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>relid</structfield> <type>oid</type>
       </para>
       <para>
        OID of the relation that the worker is synchronizing; null for the
-       main apply worker
+       main apply worker and the parallel apply worker
       </para></entry>
      </row>

5a.

(Same as general comment #1 about terminology)

"apply_leader_pid" --> "leader_apply_pid"

~~

5b.

The current text feels awkward. I see it was copied from the similar
text of 'pg_stat_activity' but perhaps it can be simplified a bit.

SUGGESTION
Process ID of the leader apply worker if this process is a parallel
apply worker; otherwise NULL.

~~

5c.
BEFORE
null for the main apply worker and the parallel apply worker

AFTER
null for the leader apply worker and parallel apply workers

~~

5c.

        <structfield>relid</structfield> <type>oid</type>
       </para>
       <para>
        OID of the relation that the worker is synchronizing; null for the
-       main apply worker
+       main apply worker and the parallel apply worker
       </para></entry>

main apply worker -> leader apply worker

~~~

6.

@@ -3212,7 +3223,7 @@ SELECT pid, wait_event_type, wait_event FROM
pg_stat_activity WHERE wait_event i
       </para>
       <para>
        Last write-ahead log location received, the initial value of
-       this field being 0
+       this field being 0; null for the parallel apply worker
       </para></entry>
      </row>

BEFORE
null for the parallel apply worker

AFTER
null for parallel apply workers

~~~

7.

@@ -3221,7 +3232,8 @@ SELECT pid, wait_event_type, wait_event FROM
pg_stat_activity WHERE wait_event i
        <structfield>last_msg_send_time</structfield> <type>timestamp
with time zone</type>
       </para>
       <para>
-       Send time of last message received from origin WAL sender
+       Send time of last message received from origin WAL sender; null for the
+       parallel apply worker
       </para></entry>
      </row>

(same as #6)

BEFORE
null for the parallel apply worker

AFTER
null for parallel apply workers

~~~

8.

@@ -3230,7 +3242,8 @@ SELECT pid, wait_event_type, wait_event FROM
pg_stat_activity WHERE wait_event i
        <structfield>last_msg_receipt_time</structfield>
<type>timestamp with time zone</type>
       </para>
       <para>
-       Receipt time of last message received from origin WAL sender
+       Receipt time of last message received from origin WAL sender; null for
+       the parallel apply worker
       </para></entry>
      </row>

(same as #6)

BEFORE
null for the parallel apply worker

AFTER
null for parallel apply workers

~~~

9.

@@ -3239,7 +3252,8 @@ SELECT pid, wait_event_type, wait_event FROM
pg_stat_activity WHERE wait_event i
        <structfield>latest_end_lsn</structfield> <type>pg_lsn</type>
       </para>
       <para>
-       Last write-ahead log location reported to origin WAL sender
+       Last write-ahead log location reported to origin WAL sender; null for
+       the parallel apply worker
       </para></entry>
      </row>

(same as #6)

BEFORE
null for the parallel apply worker

AFTER
null for parallel apply workers

~~~

10.

@@ -3249,7 +3263,7 @@ SELECT pid, wait_event_type, wait_event FROM
pg_stat_activity WHERE wait_event i
       </para>
       <para>
        Time of last write-ahead log location reported to origin WAL
-       sender
+       sender; null for the parallel apply worker
       </para></entry>
      </row>
     </tbody>

(same as #6)

BEFORE
null for the parallel apply worker

AFTER
null for parallel apply workers

======

src/backend/catalog/system_views.sql

11.

@@ -949,6 +949,7 @@ CREATE VIEW pg_stat_subscription AS
             su.oid AS subid,
             su.subname,
             st.pid,
+            st.apply_leader_pid,
             st.relid,
             st.received_lsn,
             st.last_msg_send_time,

(Same as general comment #1 about terminology)

"apply_leader_pid" --> "leader_apply_pid"

======

src/backend/replication/logical/launcher.c

12.

+ if (worker.apply_leader_pid == InvalidPid)
  nulls[3] = true;
  else
- values[3] = LSNGetDatum(worker.last_lsn);
- if (worker.last_send_time == 0)
+ values[3] = Int32GetDatum(worker.apply_leader_pid);
+

12a.

(Same as general comment #1 about terminology)

"apply_leader_pid" --> "leader_apply_pid"

~~

12b.

I wondered if here the code should be using the
isParallelApplyWorker(worker) macro here for readability.

e.g.

if (isParallelApplyWorker(worker))
values[3] = Int32GetDatum(worker.apply_leader_pid);
else
  nulls[3] = true;

======

src/include/catalog/pg_proc.dat

13.

+  proallargtypes =>
'{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o}',
+  proargnames =>
'{subid,subid,relid,pid,apply_leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time}',

(Same as general comment #1 about terminology)

"apply_leader_pid" --> "leader_apply_pid"

======

src/test/regress/expected/rules.out

14.

@@ -2094,6 +2094,7 @@ pg_stat_ssl| SELECT s.pid,
 pg_stat_subscription| SELECT su.oid AS subid,
     su.subname,
     st.pid,
+    st.apply_leader_pid,
     st.relid,
     st.received_lsn,
     st.last_msg_send_time,
@@ -2101,7 +2102,7 @@ pg_stat_subscription| SELECT su.oid AS subid,
     st.latest_end_lsn,
     st.latest_end_time
    FROM (pg_subscription su
-     LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid,
pid, received_lsn, last_msg_send_time, last_msg_receipt_time,
latest_end_lsn, latest_end_time) ON ((st.subid = su.oid)));
+     LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid,
pid, apply_leader_pid, received_lsn, last_msg_send_time,
last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid
= su.oid)));
 pg_stat_subscription_stats| SELECT ss.subid,
     s.subname,
     ss.apply_error_count,

(Same comment as elsewhere)

"apply_leader_pid" --> "leader_apply_pid"

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Amit Kapila <[email-masked]>
Date: Thu, 12 Jan 2023 10:34:33 +0530

> "apply_leader_pid" --> "leader_apply_pid"

How about naming this as just leader_pid? I think it could be helpful
in the future if we decide to parallelize initial sync (aka parallel
copy) because then we could use this for the leader PID of parallel
sync workers as well.

-- 
With Regards,
Amit Kapila.

---

From: shveta malik <[email-masked]>
Date: Thu, 12 Jan 2023 16:21:09 +0530

> copy) because then we could use this for the leader PID of parallel

I still prefer leader_apply_pid.
leader_pid does not tell which 'operation' it belongs to. 'apply'
gives the clarity that it is apply related process.

The terms used in patch look very confusing. I had to read a few lines
multiple times to understand it.

1.
Summary says 'main_worker_pid' to be added but I do not see
'main_worker_pid' added in pg_stat_subscription, instead I see
'apply_leader_pid'. Am I missing something? Also, as stated above
'leader_apply_pid' makes more sense.
it is better to correct it everywhere (apply leader-->leader apply).
Once that is done, it can be reviewed again.

thanks
Shveta

---

From: Amit Kapila <[email-masked]>
Date: Thu, 12 Jan 2023 16:37:30 +0530

> gives the clarity that it is apply related process.

But then do you suggest that tomorrow if we allow parallel sync
workers then we have a separate column leader_sync_pid? I think that
doesn't sound like a good idea and moreover one can refer to docs for
clarification.

-- 
With Regards,
Amit Kapila.

---

From: "[email-masked]" <[email-masked]>
Date: Thu, 12 Jan 2023 12:34:05 +0000

On Thursday, January 12, 2023 7:08 PM Amit Kapila <[email-masked]> wrote:
> > > > @@ -3198,11 +3198,22 @@ SELECT pid, wait_event_type, wait_event
> > > > FROM pg_stat_activity WHERE wait_event i
> > > >
> > > >       <row>
> > > > +      <entry role="catalog_table_entry"><para
> > > > + role="column_definition">
> > > How about naming this as just leader_pid? I think it could be
> > > helpful in the future if we decide to parallelize initial sync (aka
> > > parallel
> But then do you suggest that tomorrow if we allow parallel sync workers then
> we have a separate column leader_sync_pid? I think that doesn't sound like a
> good idea and moreover one can refer to docs for clarification.

I agree that leader_pid would be better not only for future parallel copy sync feature,
but also it's more consistent with the leader_pid column in pg_stat_activity.

And here is the version patch which addressed Peter's comments and renamed all
the related stuff to leader_pid.

Best Regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Thu, 12 Jan 2023 12:34:08 +0000

> Hi, here are some review comments for patch v78-0001.

Thanks for your comments.

> From this POV I think the worker member "apply_leader_pid" would be better
> named "leader_apply_pid",  but I see that this was already committed to
> HEAD differently.
> 
> Maybe it is not possible (or you don't want) to change that internal member
> name but IMO at least all the new code and docs should try to be using
> consistent terminology (e.g. leader_apply_XXX) where possible.
> 
> ======
> 
> Commit message
> 
> 2.
> 
> IIUC, this text is just cut/paste from the monitoring.sgml. In a review comment
> below I suggest some changes to that text, so then this commit message
> should also change to be the same.

Changed.

> ~~
> 
> 3.
> 
> The new column can make it easier to distinguish leader apply worker and
> apply parallel worker which is also similar to the 'leader_pid' column in
> pg_stat_activity.
> 
> SUGGESTION
> The new column makes it easier to distinguish parallel apply workers from
> other kinds of workers. It is implemented this way to be similar to the
> 'leader_pid' column in pg_stat_activity.

Changed.

> there will be additional workers -> there may be additional parallel apply
> workers

Changed.

> "apply_leader_pid" --> "leader_apply_pid"

I changed this and all related stuff to "leader_pid" as I agree with Amit that
this might be useful for future features and is more consistent with the
leader_pid in pg_stat_activity.

> 
> ~~
> 
> 5b.
> 
> The current text feels awkward. I see it was copied from the similar text of
> 'pg_stat_activity' but perhaps it can be simplified a bit.
> 
> SUGGESTION
> Process ID of the leader apply worker if this process is a parallel apply worker;
> otherwise NULL.

I slightly adjusted this according Amit's suggestion which I think would provide
more information.

"Process ID of the leader apply worker, if this process is a parallel apply worker.
NULL if this process is a leader apply worker or does not participate in parallel apply, or a synchronization worker."
"

> null for the leader apply worker and parallel apply workers

Changed.

> main apply worker -> leader apply worker

Changed.

> null for parallel apply workers

Changed.

> null for parallel apply workers

Changed.

> null for parallel apply workers

Changed.

> null for parallel apply workers

Changed.

> null for parallel apply workers

Changed.

> values[3] = Int32GetDatum(worker.apply_leader_pid);

Changed.

Best Regards,
Hou Zhijie

---

From: Peter Smith <[email-masked]>
Date: Fri, 13 Jan 2023 13:25:43 +1100

Here are my review comments for v79-0001.

======

General

1.

When Amit suggested [1] changing the name just to "leader_pid" instead
of "leader_apply_pid" I thought he was only referring to changing the
view column name, not also the internal member names of the worker
structure. Maybe it is OK anyway, but please check if that was the
intention.

======

Commit message

2.

leader_pid is the process ID of the leader apply worker if this process is a
parallel apply worker. If this field is NULL, it indicates that the process is
a leader apply worker or does not participate in parallel apply, or a
synchronization worker.

~

This text is just cut/paste from the monitoring.sgml. In a review
comment below I suggest some changes to that text, so then this commit
message should also change to be the same.

======

3.

       <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>leader_pid</structfield> <type>integer</type>
+      </para>
+      <para>
+       Process ID of the leader apply worker if this process is a parallel
+       apply worker; NULL if this process is a leader apply worker or does not
+       participate in parallel apply, or a synchronization worker
+      </para></entry>

I felt this change is giving too many details and ended up just
muddying the water.

E.g. Now this says basically "NULL if AAA or BBB, or CCC" but that
makes it sounds like there are 3 other things the process could be
instead of a parallel worker. But that is not not really true unless
you are making some distinction between the main "apply worker" which
is a leader versus a main apply worker which is not a leader. IMO we
should not be making any distinction at all - the leader apply worker
and the main (not leader) apply worker are one-and-the-same process.

So, I still prefer my previous suggestion (see [2] #5b]

======

4.

+            st.leader_pid,
             st.relid,
             st.received_lsn,
             st.last_msg_send_time,

IMO it would be very useful to have an additional "kind" attribute for
this view. This will save the user from needing to do mental
gymnastics every time just to recognise what kind of process they are
looking at.

For example, I tried this:

CREATE VIEW pg_stat_subscription AS
    SELECT
            su.oid AS subid,
            su.subname,
            CASE
                WHEN st.relid IS NOT NULL THEN 'tablesync'
                WHEN st.leader_pid IS NOT NULL THEN 'parallel apply'
                ELSE 'leader apply'
            END AS kind,
            st.pid,
            st.leader_pid,
            st.relid,
            st.received_lsn,
            st.last_msg_send_time,
            st.last_msg_receipt_time,
            st.latest_end_lsn,
            st.latest_end_time
    FROM pg_subscription su
            LEFT JOIN pg_stat_get_subscription(NULL) st
                      ON (st.subid = su.oid);

and it results in much more readable output IMO:

test_sub=# select * from pg_stat_subscription;
 subid | subname |     kind     | pid  | leader_pid | relid |
received_lsn |      last_msg_send_time       |
last_msg_receipt_time     | lat
est_end_lsn |        latest_end_time
-------+---------+--------------+------+------------+-------+--------------+-------------------------------+-------------------------------+----
------------+-------------------------------
 16388 | sub1    | leader apply | 5281 |            |       |
0/1901378    | 2023-01-13 12:39:03.984249+11 | 2023-01-13
12:39:03.986157+11 | 0/1
901378      | 2023-01-13 12:39:03.984249+11
(1 row)

Thoughts?

------
[1] Amit - https://www.postgresql.org/message-id/CAA4eK1KYUbnthSPyo4VjnhMygB0c1DZtp0XC-V2-GSETQ743ww%40mail.gmail.com
[2] My v78-0001 review -
https://www.postgresql.org/message-id/CAHut%2BPvA10Bp9Jaw9OS2%2BpuKHr7ry_xB3Tf2-bbv5gyxD5E_gw%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: shveta malik <[email-masked]>
Date: Fri, 13 Jan 2023 08:22:22 +0530

> doesn't sound like a good idea and moreover one can refer to docs for
okay, leader_pid is fine I think.

thanks
Shveta

---

From: Amit Kapila <[email-masked]>
Date: Fri, 13 Jan 2023 09:06:55 +0530

> structure. Maybe it is OK anyway, but please check if that was the

Yes, that was the intention.

> I felt this change is giving too many details and ended up just

I see that we give a similar description for other parameters as well.
For example leader_pid in pg_stat_activity, see client_dn,
client_serial in pg_stat_ssl. It is better to be consistent here and
this gives the reader a bit more information when the value is NULL
for the new column.

> gymnastics every time just to recognise what kind of process they are

This could be a separate enhancement as the same should be true for
sync workers.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Fri, 13 Jan 2023 09:58:18 +0530

On Fri, Jan 13, 2023 at 9:06 AM Amit Kapila <[email-masked]> wrote:
> For example leader_pid in pg_stat_activity,
>

BTW, shouldn't we update leader_pid column in pg_stat_activity as well
to display apply leader PID for parallel apply workers? It will
currently display for other parallel operations like a parallel
vacuum, so I don't see a reason to not do the same for parallel apply
workers.

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Fri, 13 Jan 2023 14:02:13 +0900

> vacuum, so I don't see a reason to not do the same for parallel apply

+1

The parallel apply workers have different properties than the parallel
query workers since they execute different transactions and don't use
group locking but it would be a good hint for users to show the leader
and parallel apply worker processes are related. If users want to
check only parallel query workers they can use the backend_type
column.

Regards,

-- 
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com

---

From: Masahiko Sawada <[email-masked]>
Date: Fri, 13 Jan 2023 14:43:21 +0900

> the related stuff to leader_pid.

Here are two comments on v79-0003 patch.

+        /* Force to serialize messages if stream_serialize_threshold
is reached. */
+        if (stream_serialize_threshold != -1 &&
+                (stream_serialize_threshold == 0 ||
+                 stream_serialize_threshold < parallel_stream_nchunks))
+        {
+                parallel_stream_nchunks = 0;
+                return false;
+        }

I think it would be better if we show the log message ""logical
replication apply worker will serialize the remaining changes of
remote transaction %u to a file" even in stream_serialize_threshold
case.

IIUC parallel_stream_nchunks won't be reset if pa_send_data() failed
due to the timeout.

Regards,

---

From: Peter Smith <[email-masked]>
Date: Fri, 13 Jan 2023 17:19:58 +1100

Here are some review comments for patch v79-0002.

======

General

1.

I saw that earlier in this thread Hou-san [1] and Amit [2] also seemed
to say there is not much point for this patch.

So I wanted to +1 that same opinion.

I feel this patch just adds more complexity for almost no gain:
- reducing the 'max_apply_workers_per_suibscription' seems not very
common in the first place.
- even when the GUC is reduced, at that point in time all the workers
might be in use so there may be nothing that can be immediately done.
- IIUC the excess workers (for a reduced GUC) are going to get freed
naturally anyway over time as more transactions are completed so the
pool size will reduce accordingly.

~

OTOH some refactoring parts of this patch (e.g. the new pa_stop_worker
function) look better to me. I would keep those ones but remove all
the pa_stop_idle_workers function/call.

*** NOTE: The remainder of these review comments are maybe only
relevant if you are going to keep this pa_stop_idle_workers
behaviour...

======

Commit message

2.

If the max_parallel_apply_workers_per_subscription is changed to a
lower value, try to stop free workers in the pool to keep the number of
workers lower than half of the max_parallel_apply_workers_per_subscription

SUGGESTION

If the GUC max_parallel_apply_workers_per_subscription is changed to a
lower value, try to stop unused workers to keep the pool size lower
than half of max_parallel_apply_workers_per_subscription.

======

.../replication/logical/applyparallelworker.c

3. pa_free_worker

if (winfo->serialize_changes ||
list_length(ParallelApplyWorkerPool) >
(max_parallel_apply_workers_per_subscription / 2))
{
pa_stop_worker(winfo);
return;
}

winfo->in_use = false;
winfo->serialize_changes = false;

~

IMO the above code can be more neatly written using if/else because
then there is only one return point, and there is a place to write the
explanatory comment about the else.

SUGGESTION

if (winfo->serialize_changes ||
list_length(ParallelApplyWorkerPool) >
(max_parallel_apply_workers_per_subscription / 2))
{
pa_stop_worker(winfo);
}
else
{
/* Don't stop the worker. Only mark it available for re-use. */
winfo->in_use = false;
winfo->serialize_changes = false;
}

======

src/backend/replication/logical/worker.c

4. pa_stop_idle_workers

/*
 * Try to stop parallel apply workers that are not in use to keep the number of
 * workers lower than half of the max_parallel_apply_workers_per_subscription.
 */
void
pa_stop_idle_workers(void)
{
List    *active_workers;
ListCell   *lc;
int max_applyworkers = max_parallel_apply_workers_per_subscription / 2;

if (list_length(ParallelApplyWorkerPool) <= max_applyworkers)
return;

active_workers = list_copy(ParallelApplyWorkerPool);

foreach(lc, active_workers)
{
ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);

pa_stop_worker(winfo);

/* Recheck the number of workers. */
if (list_length(ParallelApplyWorkerPool) <= max_applyworkers)
break;
}

list_free(active_workers);
}

~

4a. function comment

SUGGESTION

Try to keep the worker pool size lower than half of the
max_parallel_apply_workers_per_subscription.

~

4b. function name

This is not stopping all idle workers, so maybe a more meaningful name
for this function is something more like "pa_reduce_workerpool"

~

4c.

IMO the "max_applyworkers" var is a misleading name. Maybe something
like "goal_poolsize" is better?

~

4d.

Maybe I misunderstand the logic for the pool, but shouldn't this be
checking the winfo->in_use flag before blindly stopping each worker?

======

src/backend/replication/logical/worker.c

5.

@@ -3630,6 +3630,13 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
  {
  ConfigReloadPending = false;
  ProcessConfigFile(PGC_SIGHUP);
+
+ /*
+ * Try to stop free workers in the pool in case the
+ * max_parallel_apply_workers_per_subscription is changed to a
+ * lower value.
+ */
+ pa_stop_idle_workers();
  }
5a.

SUGGESTED COMMENT
If max_parallel_apply_workers_per_subscription is changed to a lower
value, try to reduce the worker pool to match.

~

5b.

Instead of unconditionally calling pa_stop_idle_workers, shouldn't
this code compare the value of
max_parallel_apply_workers_per_subscription before/after the
ProcessConfigFile so it only calls if the GUC was lowered?

------
[1] Hou-san - https://www.postgresql.org/message-id/OS0PR01MB5716E527412A3481F90B4397941A9%40OS0PR01MB5716.jpnprd01.prod.outlook.com
[2] Amit - https://www.postgresql.org/message-id/CAA4eK1J%3D9m-VNRMHCqeG8jpX0CTn3Ciad2o4H-ogrZMDJ3tn4w%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: "[email-masked]" <[email-masked]>
Date: Fri, 13 Jan 2023 10:13:25 +0000

On Friday, January 13, 2023 1:02 PM Masahiko Sawada <[email-masked]> wrote:
> The parallel apply workers have different properties than the parallel query
> workers since they execute different transactions and don't use group locking
> but it would be a good hint for users to show the leader and parallel apply
> worker processes are related. If users want to check only parallel query workers
> they can use the backend_type column.

Agreed, and changed as suggested.

Attach the new version patch set which address the comments so far.

Best Regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Fri, 13 Jan 2023 10:13:31 +0000

On Friday, January 13, 2023 1:43 PM Masahiko Sawada <[email-masked]> wrote:
> > > > On Thu, Jan 12, 2023 at 10:34 AM Amit Kapila
> > > > <[email-masked]>
> > > wrote:
> > > > >
> > > > > On Thu, Jan 12, 2023 at 9:54 AM Peter Smith
> > > > > <[email-masked]>
> > > wrote:
> > > > > >
> > > > > >
> > > > > > @@ -3198,11 +3198,22 @@ SELECT pid, wait_event_type,
> > > > > > wait_event FROM pg_stat_activity WHERE wait_event i
> > > > > >
> > > > > >       <row>
> > > > > helpful in the future if we decide to parallelize initial sync
> > > > > (aka parallel
> > > > > copy) because then we could use this for the leader PID of
> > > > > parallel sync workers as well.
> > > > >
> > > > > --
> > > >
> > I agree that leader_pid would be better not only for future parallel
> > copy sync feature, but also it's more consistent with the leader_pid column in
> pg_stat_activity.
> >
> > And here is the version patch which addressed Peter's comments and
> > renamed all the related stuff to leader_pid.
> 

Thanks for the comments.

> I think it would be better if we show the log message ""logical replication apply
> worker will serialize the remaining changes of remote transaction %u to a file"
> even in stream_serialize_threshold case.

Agreed and changed.

> 
> IIUC parallel_stream_nchunks won't be reset if pa_send_data() failed due to the
> timeout.

Changed.

Best Regards,
Hou zj

---

From: Amit Kapila <[email-masked]>
Date: Sat, 14 Jan 2023 17:17:11 +0530

> > <[email-masked]> wrote:

In GetLogicalLeaderApplyWorker(), we can use shared lock instead
exclusive as we are just reading the workers array. Also, the function
name looks a bit odd to me, so I changed it to
GetLeaderApplyWorkerPid(). Also, it is better to use InvalidPid
instead of 0 when there is no valid value for leader_pid in
GetLeaderApplyWorkerPid(). Apart from that, I have made minor changes
in the comments, docs, and commit message. I am planning to push this
next week by Tuesday unless you or others have any major comments.

-- 
With Regards,
Amit Kapila.

---

From: Tomas Vondra <[email-masked]>
Date: Sun, 15 Jan 2023 18:09:00 +0100

Hi,

I think there's a bug in how get_transaction_apply_action() interacts
with handle_streamed_transaction() to decide whether the transaction is
streamed or not. Originally, the code was simply:

    /* not in streaming mode */
    if (!in_streamed_transaction)
        return false;

But now this decision was moved to get_transaction_apply_action(), which
does this:

    if (am_parallel_apply_worker())
    {
        return TRANS_PARALLEL_APPLY;
    }
    else if (in_remote_transaction)
    {
        return TRANS_LEADER_APPLY;
    }

and handle_streamed_transaction() then uses the result like this:

    /* not in streaming mode */
    if (apply_action == TRANS_LEADER_APPLY)
        return false;

Notice this is not equal to the original behavior, because the two flags
(in_remote_transaction and in_streamed_transaction) are not inverse.
That is,

   in_remote_transaction=false

does not imply we're processing streamed transaction. It's allowed both
flags are false, i.e. a change may be "non-transactional" and not
streamed, though the only example of such thing in the protocol are
logical messages. Which are however ignored in the apply worker, so I'm
not surprised no existing test failed on this.

So I think get_transaction_apply_action() should do this:

    if (am_parallel_apply_worker())
    {
        return TRANS_PARALLEL_APPLY;
    }
    else if (!in_streamed_transaction)
    {
        return TRANS_LEADER_APPLY;
    }

FWIW I've noticed this after rebasing the sequence decoding patch, which
adds another type of protocol message with the transactional vs.
non-transactional behavior, similar to "logical messages" except that in
this case the worker does not ignore that.

Also, I think get_transaction_apply_action() would deserve better
comments explaining how/why it makes the decisions.

regards

-- 
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

---

From: Peter Smith <[email-masked]>
Date: Mon, 16 Jan 2023 15:54:00 +1100

Here are some review comments for v81-0001.

======

Commit Message

1.

Additionally, update the leader_pid column in pg_stat_activity as well to
display the PID of the leader apply worker for parallel apply workers.

~

Probably it should not say both "Additionally" and "as well" in the
same sentence.

======

2.

 /*
+ * Return the pid of the leader apply worker if the given pid is the pid of a
+ * parallel apply worker, otherwise return InvalidPid.
+ */
+pid_t
+GetLeaderApplyWorkerPid(pid_t pid)
+{
+ int leader_pid = InvalidPid;
+ int i;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+ for (i = 0; i < max_logical_replication_workers; i++)
+ {
+ LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+ if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
+ {
+ leader_pid = w->leader_pid;
+ break;
+ }
+ }
+
+ LWLockRelease(LogicalRepWorkerLock);
+
+ return leader_pid;
+}

2a.
IIUC the IsParallelApplyWorker macro does nothing except check that
the leader_pid is not InvalidPid anyway, so AFAIK this algorithm does
not benefit from using this macro because we will want to return
InvalidPid anyway if the given pid matches.

So the inner condition can just say:

if (w->proc && w->proc->pid == pid)
{
leader_pid = w->leader_pid;
break;
}

~

2b.
A possible alternative comment.

BEFORE
Return the pid of the leader apply worker if the given pid is the pid
of a parallel apply worker, otherwise return InvalidPid.

AFTER
If the given pid has a leader apply worker then return the leader pid,
otherwise, return InvalidPid.

======

src/backend/utils/adt/pgstatfuncs.c

3.

@@ -434,6 +435,16 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
  values[28] = Int32GetDatum(leader->pid);
  nulls[28] = false;
  }
+ else
+ {
+ int leader_pid = GetLeaderApplyWorkerPid(beentry->st_procpid);
+
+ if (leader_pid != InvalidPid)
+ {
+ values[28] = Int32GetDatum(leader_pid);
+ nulls[28] = false;
+ }
+

3a.
There is an existing comment preceding this if/else but it refers only
to leaders of parallel groups. Should that comment be updated to
mention the leader apply worker too?

~

3b.
It may be unrelated to this patch, but it seems strange to me that the
nulls[28]/values[28] assignments are done where they are. Every other
nulls/values assignment of this function here is pretty much in the
correct numerical order except this one, so IMO this code ought to be
relocated to later in this same function.

------
Kind Regards,
Peter Smith.
Fujitsu Australia.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 16 Jan 2023 11:49:36 +0530

> not surprised no existing test failed on this.

Right, this is the reason we didn't catch it in our testing.

>         return TRANS_LEADER_APPLY;

Yeah, something like this would work but some of the callers other
than handle_streamed_transaction() also need to be changed. See
attached.

> comments explaining how/why it makes the decisions.

Okay, I have added the comments in get_transaction_apply_action() and
updated the comments to refer to the enum TransApplyAction where all
the actions are explained.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 16 Jan 2023 12:10:50 +0530

> leader_pid = w->leader_pid;

Yeah, this should also work but I feel the current one is explicit and
more clear.

> otherwise, return InvalidPid.

I don't think that is an improvement.

> mention the leader apply worker too?

Yeah, we can slightly adjust the comments. How about something like the below:
index 415e711729..7eb668634a 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -410,9 +410,9 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)

                        /*
                         * If a PGPROC entry was retrieved, display
wait events and lock
-                        * group leader information if any.  To avoid
extra overhead, no
-                        * extra lock is being held, so there is no guarantee of
-                        * consistency across multiple rows.
+                        * group leader or apply leader information if
any.  To avoid extra
+                        * overhead, no extra lock is being held, so
there is no guarantee
+                        * of consistency across multiple rows.
                         */
                        if (proc != NULL)
                        {
@@ -428,7 +428,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
                                /*
                                 * Show the leader only for active
parallel workers.  This
                                 * leaves the field as NULL for the
leader of a parallel
-                                * group.
+                                * group or the leader of a parallel apply.
                                 */
                                if (leader && leader->pid !=
beentry->st_procpid)

> relocated to later in this same function.

This is not related to the current patch but I see there is merit in
the current coding as it is better to retrieve all the fields of proc
together.

-- 
With Regards,
Amit Kapila.

---

From: Tomas Vondra <[email-masked]>
Date: Mon, 16 Jan 2023 17:33:32 +0100

Hi Amit,

Thanks for the patch, the changes seem reasonable to me and it does fix
the issue in the sequence decoding patch.

regards

---

From: Peter Smith <[email-masked]>
Date: Tue, 17 Jan 2023 08:42:57 +1100

> Yeah, this should also work but I feel the current one is explicit and

OK.

But, I have one last comment about this function -- I saw there are
already other functions that iterate max_logical_replication_workers
like this looking for things:
- logicalrep_worker_find
- logicalrep_workers_find
- logicalrep_worker_launch
- logicalrep_sync_worker_count

So I felt this new function (currently called GetLeaderApplyWorkerPid)
ought to be named similarly to those ones. e.g. call it something like
 "logicalrep_worker_find_pa_leader_pid".

>                                 if (leader && leader->pid !=

The updated comment LGTM.

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 17 Jan 2023 02:21:04 +0000

On Tuesday, January 17, 2023 5:43 AM Peter Smith <[email-masked]> wrote:
> > > + * Return the pid of the leader apply worker if the given pid is
> > > +the pid of a
> > > + for (i = 0; i < max_logical_replication_workers; i++) {
> > > + if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid) {
> > > + leader_pid = w->leader_pid; break; } }
> > > +
> > > the leader_pid is not InvalidPid anyway, so AFAIK this algorithm
> > > does not benefit from using this macro because we will want to
> > > return InvalidPid anyway if the given pid matches.
> > >
> But, I have one last comment about this function -- I saw there are already
> other functions that iterate max_logical_replication_workers like this looking
> for things:
> - logicalrep_worker_find
> So I felt this new function (currently called GetLeaderApplyWorkerPid) ought
> to be named similarly to those ones. e.g. call it something like

I am not sure we can use the name, because currently all the API name in launcher that
used by other module(not related to subscription) are like
AxxBxx style(see the functions in logicallauncher.h).
logicalrep_worker_xxx style functions are currently only declared in
worker_internal.h.

Best regards,
Hou zj

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 17 Jan 2023 12:05:09 +0900

> the actions are explained.

Thank you for the patch.

@@ -1710,6 +1712,7 @@ apply_handle_stream_stop(StringInfo s)
        }

        in_streamed_transaction = false;
+       stream_xid = InvalidTransactionId;

We reset stream_xid also in stream_close_file() but probably it's no
longer necessary?

How about adding an assertion in apply_handle_stream_start() to make
sure the stream_xid is invalid?

---
It's not related to this issue but I realized that if the action
returned by get_transaction_apply_action() is not handled in the
switch statement, we do only Assert(false). Is it better to raise an
error like "unexpected apply action %d" just in case in order to
detect failure cases also in the production environment?

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Tue, 17 Jan 2023 08:59:45 +0530

> We reset stream_xid also in stream_close_file() but probably it's no

I think so.

> sure the stream_xid is invalid?

I think it would be better to add such an assert in
apply_handle_begin/apply_handle_begin_prepare because there won't be a
problem if we start_stream message even when stream_xid is valid.
However, maybe it is better to add in all three functions
(apply_handle_begin/apply_handle_begin_prepare/apply_handle_stream_start).
What do you think?

> detect failure cases also in the production environment?

Yeah, that may be better. Shall we do that as part of this patch only
or as a separate patch?

-- 
With Regards,
Amit Kapila.

---

From: Peter Smith <[email-masked]>
Date: Tue, 17 Jan 2023 14:32:28 +1100

> logicalrep_worker_xxx style functions are currently only declared in

OK. I didn't know there was another header convention that you were
following.  In that case, it is fine to leave the name as-is.

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 17 Jan 2023 03:37:07 +0000

On Tuesday, January 17, 2023 11:32 AM Peter Smith <[email-masked]> wrote:
> > > On Mon, Jan 16, 2023 at 5:41 PM Amit Kapila
> > > > On Mon, Jan 16, 2023 at 10:24 AM Peter Smith
> > > > <[email-masked]>
> > > wrote:
> > > > >
> > > > > 2.
> > > > >
> > > > >  /*
> > > > > + * Return the pid of the leader apply worker if the given pid
> > > > > +is the pid of a
> > > > > +GetLeaderApplyWorkerPid(pid_t pid) {  int leader_pid =
> > > > > +InvalidPid;  int i;
> > > > > +
> > > > > + if (isParallelApplyWorker(w) && w->proc && pid ==
> > > > > + w->proc->pid) { leader_pid = w->leader_pid; break; } }
> > > > > +
> > > > > IIUC the IsParallelApplyWorker macro does nothing except check
> > > > > that the leader_pid is not InvalidPid anyway, so AFAIK this
> > > > > algorithm does not benefit from using this macro because we will
> > > > > want to return InvalidPid anyway if the given pid matches.
> > > > >
> > > > > if (w->proc && w->proc->pid == pid) { leader_pid =
> > > > > w->leader_pid; break; }
> > > > >
> > > >
> > > > Yeah, this should also work but I feel the current one is explicit
> > > > and more clear.
> > >
> > > OK.
> > >
> > > So I felt this new function (currently called
> > > GetLeaderApplyWorkerPid) ought to be named similarly to those ones.
> > > e.g. call it something like "logicalrep_worker_find_pa_leader_pid".
> > >
> >
> > I am not sure we can use the name, because currently all the API name
> > in launcher that used by other module(not related to subscription) are
> > like AxxBxx style(see the functions in logicallauncher.h).
> OK. I didn't know there was another header convention that you were following.
> In that case, it is fine to leave the name as-is.

Thanks for confirming!

Attach the new version 0001 patch which addressed all other comments.

Best regards,
Hou zj

---

From: Peter Smith <[email-masked]>
Date: Tue, 17 Jan 2023 14:48:34 +1100

> Attach the new version 0001 patch which addressed all other comments.

OK. I checked the differences between patches v81-0001/v82-0001 and
found everything I was expecting to see.

I have no more review comments for v82-0001.

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Amit Kapila <[email-masked]>
Date: Tue, 17 Jan 2023 10:25:22 +0530

> Yeah, that may be better. Shall we do that as part of this patch only

Please find attached the updated patches to address the above
comments. I think we can combine and commit them as one patch as both
are related.

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 17 Jan 2023 14:07:05 +0900

> comments. I think we can combine and commit them as one patch as both

Thank you for the patches! Looks good to me. And +1 to merge them.

Regards,

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 17 Jan 2023 06:05:53 +0000

On Tuesday, January 17, 2023 12:55 PM Amit Kapila <[email-masked]> wrote:
> > > > Okay, I have added the comments in get_transaction_apply_action()
> > > > and updated the comments to refer to the enum TransApplyAction
> > > > where all the actions are explained.
> > >
> > > Thank you for the patch.
> > >
> Please find attached the updated patches to address the above comments. I
> think we can combine and commit them as one patch as both are related.

Thanks for fixing these.
I have confirmed that all regression tests passed after applying the patches.
And the patches look good to me.

Best regards,
Hou zj

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 17 Jan 2023 15:46:14 +0900

> Attach the new version 0001 patch which addressed all other comments.

Thank you for updating the patch. Here is one comment:

@@ -426,14 +427,24 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)

                                /*
                                 * Show the leader only for active
parallel workers.  This
-                                * leaves the field as NULL for the
leader of a parallel
-                                * group.
+                                * leaves the field as NULL for the
leader of a parallel group
+                                * or the leader of parallel apply workers.
                                 */
                                if (leader && leader->pid !=
beentry->st_procpid)
                                {
                                        values[28] = Int32GetDatum(leader->pid);
                                        nulls[28] = false;
                                }
+                               else
+                               {
+                                       int
leader_pid = GetLeaderApplyWorkerPid(beentry->st_procpid);
+
+                                       if (leader_pid != InvalidPid)
+                                       {
+                                               values[28] =
Int32GetDatum(leader_pid);
+                                               nulls[28] = false;
+                                       }
+                               }
                        }

I'm slightly concerned that there could be overhead of executing
GetLeaderApplyWorkerPid () for every backend process except for
parallel query workers. The number of such backends could be large and
GetLeaderApplyWorkerPid() acquires the lwlock. For example, does it
make sense to check (st_backendType == B_BG_WORKER) before calling
GetLeaderApplyWorkerPid()? Or it might not be a problem since it's
LogicalRepWorkerLock which is not likely to be contended.

Regards,

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 17 Jan 2023 09:14:44 +0000

On Tuesday, January 17, 2023 2:46 PM Masahiko Sawada <[email-masked]> wrote:
>                                  * Show the leader only for active parallel
> workers.  This
> GetLeaderApplyWorkerPid () for every backend process except for parallel
> query workers. The number of such backends could be large and
> GetLeaderApplyWorkerPid() acquires the lwlock. For example, does it make
> sense to check (st_backendType == B_BG_WORKER) before calling

Thanks for the comment and I think your suggestion makes sense.
I have added the check before getting the leader pid. Here is the new version patch.

Best regards,
Hou zj

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 17 Jan 2023 23:37:02 +0900

> I have added the check before getting the leader pid. Here is the new version patch.

Thank you for updating the patch. Looks good to me.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Wed, 18 Jan 2023 10:05:58 +0530

> Thank you for updating the patch. Looks good to me.

Pushed.

-- 
With Regards,
Amit Kapila.

---

From: "[email-masked]" <[email-masked]>
Date: Wed, 18 Jan 2023 04:48:01 +0000

On Wed, Jan 18, 2023 12:36 PM Amit Kapila <[email-masked]> wrote:
> make

Rebased and attach remaining patches for reviewing.

Regards,
Wang Wei

---

From: Amit Kapila <[email-masked]>
Date: Wed, 18 Jan 2023 12:09:26 +0530

> Here are some review comments for patch v79-0002.

So, this is about the latest v84-0001-Stop-extra-worker-if-GUC-was-changed.

> pool size will reduce accordingly.

I am still not sure if it is worth pursuing this patch because of the
above reasons. I don't think it would be difficult to add this even at
a later point in time if we really see a use case for this.
Sawada-San, IIRC, you raised this point. What do you think?

The other point I am wondering is whether we can have a different way
to test partial serialization apart from introducing another developer
GUC (stream_serialize_threshold). One possibility could be that we can
have a subscription option (parallel_send_timeout or something like
that) with some default value (current_timeout used in the patch)
which will be used only when streaming = parallel. Users may want to
wait for more time before serialization starts depending on the
workload (say when resource usage is high on a subscriber-side
machine, or there are concurrent long-running transactions that can
block parallel apply for a bit longer time). I know with this as well
it may not be straightforward to test the functionality because we
can't be sure how many changes would be required for a timeout to
occur. This is just for brainstorming other options to test the
partial serialization functionality.

Thoughts?

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Thu, 19 Jan 2023 11:10:46 +0530

> partial serialization functionality.

Apart from the above, we can also have a subscription option to
specify parallel_shm_queue_size (queue_size used to determine the
queue between the leader and parallel worker) and that can be used for
this purpose. Basically, configuring it to a smaller value can help in
reducing the test time but still, it will not eliminate the need for
dependency on timing we have to wait before switching to partial
serialize mode. I think this can be used in production as well to tune
the performance depending on workload.

Yet another way is to use the existing parameter logical_decode_mode
[1]. If the value of logical_decoding_mode is 'immediate', then we can
immediately switch to partial serialize mode. This will eliminate the
dependency on timing. The one argument against using this is that it
won't be as clear as a separate parameter like
'stream_serialize_threshold' proposed by the patch but OTOH we already
have a few parameters that serve a different purpose when used on the
subscriber. For example, 'max_replication_slots' is used to define the
maximum number of replication slots on the publisher and the maximum
number of origins on subscribers. Similarly,
wal_retrieve_retry_interval' is used for different purposes on
subscriber and standby nodes.

[1] - https://www.postgresql.org/docs/devel/runtime-config-developer.html

-- 
With Regards,
Amit Kapila.

---

From: shveta malik <[email-masked]>
Date: Thu, 19 Jan 2023 15:44:08 +0530

Hi Amit,

On rethinking the complete model, what I feel is that the name
logical_decoding_mode is not something which defines modes of logical
decoding. We, I think, picked it based on logical_decoding_work_mem.
As per current implementation, the parameter 'logical_decoding_mode'
tells what happens when work-memory used by logical decoding reaches
its limit. So it is in-fact 'logicalrep_workmem_vacate_mode' or
'logicalrep_trans_eviction_mode'. So if it is set to immediate,
meaning vacate the work-memory immediately or evict transactions
immediately. Add buffered means the reverse (i.e. keep on buffering
transactions until we reach a limit). Now coming to subscribers, we
can reuse the same parameter. On subscriber as well, shared-memory
queue could be considered as its workmem and thus the name
'logicalrep_workmem_vacate_mode' might look more relevant.

On publisher:
logicalrep_workmem_vacate_mode=immediate, buffered.

On subscriber:
logicalrep_workmem_vacate_mode=stream_serialize  (or if we want to
keep immediate here too, that will also be fine).

Thoughts?
And I am assuming it is possible to change the GUC name before the
coming release. If not, please let me know, we can brainstorm other
ideas.

thanks
Shveta

---

From: shveta malik <[email-masked]>
Date: Thu, 19 Jan 2023 15:52:36 +0530

> its limit. So it is in-fact 'logicalrep_workmem_vacate_mode' or

Minor correction in what I said earlier:
As per current implementation, the parameter 'logical_decoding_mode'
more or less tells how to deal with workmem i.e. to keep it buffering
with txns until it reaches its limit or immediately vacate it.

thanks
Shveta

---

From: Masahiko Sawada <[email-masked]>
Date: Fri, 20 Jan 2023 15:17:51 +0900

> > partial serialization functionality.

I can see parallel_send_timeout idea could be useful somewhat but I'm
concerned users can tune this value properly. It's likely to indicate
something abnormal or locking issues if LA waits to write data to the
queue for more than 10s. Also, I think it doesn't make sense to allow
users to set this timeout to a very low value. If switching to partial
serialization mode early is useful in some cases, I think it's better
to provide it as a new mode, such as streaming = 'parallel-file' etc.

> the performance depending on workload.

A parameter for the queue size is interesting but I agree it will not
eliminate the need for dependency on timing.

> subscriber and standby nodes.

Using the existing parameter makes sense to me. But if we use
logical_decoding_mode also on the subscriber, as Shveta Malik also
suggested, probably it's better to rename it so as not to confuse. For
example, logical_replication_mode or something.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Mon, 23 Jan 2023 08:47:18 +0530

> example, logical_replication_mode or something.

+1. Among the options discussed, this sounds better.

-- 
With Regards,
Amit Kapila.

---

From: Dilip Kumar <[email-masked]>
Date: Mon, 23 Jan 2023 11:28:21 +0530

> +1. Among the options discussed, this sounds better.

Yeah, this looks better option with the parameter name
'logical_replication_mode'.

---

From: "[email-masked]" <[email-masked]>
Date: Mon, 23 Jan 2023 08:05:06 +0000

On Monday, January 23, 2023 11:17 AM Amit Kapila <[email-masked]> wrote:
> > > [1]. If the value of logical_decoding_mode is 'immediate', then we
> > > can immediately switch to partial serialize mode. This will
> > > eliminate the dependency on timing. The one argument against using
> > > this is that it won't be as clear as a separate parameter like
> > > 'stream_serialize_threshold' proposed by the patch but OTOH we
> > > already have a few parameters that serve a different purpose when
> > > used on the subscriber. For example, 'max_replication_slots' is used
> > > to define the maximum number of replication slots on the publisher
> > > and the maximum number of origins on subscribers. Similarly,

OK, here is patch set which does the same.
The first patch set only renames the GUC name, and the second patch uses
the GUC to test the partial serialization.

Best Regards,
Hou zj

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 23 Jan 2023 12:34:00 +0000

Dear Hou,

Thank you for updating the patch! Followings are my comments.

1. guc_tables.c

```
 static const struct config_enum_entry logical_decoding_mode_options[] = {
-       {"buffered", LOGICAL_DECODING_MODE_BUFFERED, false},
-       {"immediate", LOGICAL_DECODING_MODE_IMMEDIATE, false},
+       {"buffered", LOGICAL_REP_MODE_BUFFERED, false},
+       {"immediate", LOGICAL_REP_MODE_IMMEDIATE, false},
        {NULL, 0, false}
 };
```

This struct should be also modified.

2. guc_tables.c

```
-               {"logical_decoding_mode", PGC_USERSET, DEVELOPER_OPTIONS,
+               {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
                        gettext_noop("Allows streaming or serializing each change in logical decoding."),
                        NULL,
```

I felt the description seems not to be suitable for current behavior.
A short description should be like "Sets a behavior of logical replication", and
further descriptions can be added in lond description.

3. config.sgml

```
       <para>
        This parameter is intended to be used to test logical decoding and
        replication of large transactions for which otherwise we need to
        generate the changes till <varname>logical_decoding_work_mem</varname>
        is reached.
       </para>
```

I understood that this part described the usage of the parameter. How about adding
a statement like:

" Moreover, this can be also used to test the message passing between the leader
and parallel apply workers."

4. 015_stream.pl

```
+# Ensure that the messages are serialized.
```

In other parts "changes" are used instead of "messages". Can you change the word?

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Peter Smith <[email-masked]>
Date: Tue, 24 Jan 2023 14:43:08 +1100

Here are my review comments for patch v86-0001.

======
General

1.

IIUC the GUC name was made generic 'logical_replication_mode' so that
multiple developer GUCs are not needed later.

But IMO those current option values (buffered/immediate) for that GUC
are maybe a bit too generic. Perhaps in future, we might want more
granular control than that allows. e.g. I can imagine there might be
multiple different meanings for what "buffered" means. If there is any
chance of the generic values being problematic later then maybe they
should be made more specific up-front.

e.g. maybe like:
logical_replication_mode = buffered_decoding
logical_replication_mode = immediate_decoding

Thoughts?

======
Commit message

2.
Since we may extend the developer option logical_decoding_mode to to test the
parallel apply of large transaction on subscriber, rename this option to
logical_replication_mode to make it easier to understand.

~

2a
typo "to to"

typo "large transaction on subscriber" --> "large transactions on the
subscriber"

~

2b.
IMO better to rephrase the whole paragraph like shown below.

SUGGESTION

Rename the developer option 'logical_decoding_mode' to the more
flexible name 'logical_replication_mode' because doing so will make it
easier to extend this option in future to help test other areas of
logical replication.

======
doc/src/sgml/config.sgml

3.
Allows streaming or serializing changes immediately in logical
decoding. The allowed values of logical_replication_mode are buffered
and immediate. When set to immediate, stream each change if streaming
option (see optional parameters set by CREATE SUBSCRIPTION) is
enabled, otherwise, serialize each change. When set to buffered, which
is the default, decoding will stream or serialize changes when
logical_decoding_work_mem is reached.

~

IMO it's more clear to say the default when the options are first
mentioned. So I suggested removing the "which is the default" part,
and instead saying:

BEFORE
The allowed values of logical_replication_mode are buffered and immediate.

AFTER
The allowed values of logical_replication_mode are buffered and
immediate. The default is buffered.

======
src/backend/utils/misc/guc_tables.c

4.
@@ -396,8 +396,8 @@ static const struct config_enum_entry
ssl_protocol_versions_info[] = {
 };

I noticed this array is still called "logical_decoding_mode_options".
Was that deliberate?

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Amit Kapila <[email-masked]>
Date: Tue, 24 Jan 2023 10:12:37 +0530

> logical_replication_mode = immediate_decoding

For now, it seems the meaning of buffered/immediate suits our
debugging and test needs for publisher/subscriber. This is somewhat
explained in Shveta's email [1]. I also think in the future this
parameter could be extended for a different purpose but maybe it would
be better to invent some new values at that time as things would be
more clear. We could do what you are suggesting or in fact even use
different values for publisher and subscriber but not really sure
whether that will simplify the usage.

[1] - https://www.postgresql.org/message-id/CAJpy0uDzddK_ZUsB2qBJUbW_ZODYGoUHTaS5pVcYE2tzATCVXQ%40mail.gmail.com

-- 
With Regards,
Amit Kapila.

---

From: Peter Smith <[email-masked]>
Date: Tue, 24 Jan 2023 18:18:35 +1100

Here are some review comments for v86-0002

======
Commit message

1.
Use the use the existing developer option logical_replication_mode to test the
parallel apply of large transaction on subscriber.

~

Typo “Use the use the”

SUGGESTION (rewritten)
Give additional functionality to the existing developer option
'logical_replication_mode' to help test parallel apply of large
transactions on the subscriber.

~~~

2.
Maybe that commit message should also say extra TAP tests that have
been added to exercise the serialization part of the parallel apply?

BTW – I can see the TAP tests are testing full serialization (when the
GUC is 'immediate') but I not sure how is "partial" serialization
(when it has to switch halfway from shmem to files) being tested.

======
doc/src/sgml/config.sgml

3.
Allows streaming or serializing changes immediately in logical
decoding. The allowed values of logical_replication_mode are buffered
and immediate. When set to immediate, stream each change if streaming
option (see optional parameters set by CREATE SUBSCRIPTION) is
enabled, otherwise, serialize each change. When set to buffered, which
is the default, decoding will stream or serialize changes when
logical_decoding_work_mem is reached.
On the subscriber side, if streaming option is set to parallel, this
parameter also allows the leader apply worker to send changes to the
shared memory queue or to serialize changes. When set to buffered, the
leader sends changes to parallel apply workers via shared memory
queue. When set to immediate, the leader serializes all changes to
files and notifies the parallel apply workers to read and apply them
at the end of the transaction.

~

Because now this same developer GUC affects both the publisher side
and the subscriber side differently IMO this whole description should
be re-structured accordingly.

SUGGESTION (something like)

On the publisher side, ...

On the subscriber side, ...

~~~

4.
This parameter is intended to be used to test logical decoding and
replication of large transactions for which otherwise we need to
generate the changes till logical_decoding_work_mem is reached.

~

Maybe this paragraph needs rewording or moving. e.g. Isn't that
misleading now? Although this might be an explanation for the
publisher side, it does not seem relevant to the subscriber side's
behaviour.

======

5.
@ -1149,6 +1149,9 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size
nbytes, const void *data)
  Assert(!IsTransactionState());
  Assert(!winfo->serialize_changes);

+ if (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE)
+ return false;
+

I felt that code should have some comment, even if it is just
something quite basic like "/* For developer testing */"

======
.../t/018_stream_subxact_abort.pl

6.
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);

Is it necessary to TRUNCATE the table here? If everything is working
shouldn't the data be rolled back anyway?

~~~

7.
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 values(1);
+ SAVEPOINT sp;
+ INSERT INTO test_tab_2 values(1);
+ ROLLBACK TO sp;
+ COMMIT;
+ });

Perhaps this should insert 2 different values so then the verification
code can check the correct value remains instead of just checking
COUNT(*)?

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 24 Jan 2023 12:47:13 +0000

On Tuesday, January 24, 2023 3:19 PM Peter Smith <[email-masked]> wrote:
> 'logical_replication_mode' to help test parallel apply of large transactions on the
> subscriber.

Changed.

> ~~~
> 
> 2.
> Maybe that commit message should also say extra TAP tests that have been
> added to exercise the serialization part of the parallel apply?

Added.

> BTW – I can see the TAP tests are testing full serialization (when the GUC is
> 'immediate') but I not sure how is "partial" serialization (when it has to switch
> halfway from shmem to files) being tested.

The new tests are intended to test most of new code patch for partial
serialization by doing it from the beginning. Later, if required, we can add
different tests for it.

> 
> ======
> doc/src/sgml/config.sgml
> 
> 3.
> Allows streaming or serializing changes immediately in logical decoding. The
> allowed values of logical_replication_mode are buffered and immediate. When
> set to immediate, stream each change if streaming option (see optional
> parameters set by CREATE SUBSCRIPTION) is enabled, otherwise, serialize each
> change. When set to buffered, which is the default, decoding will stream or
> serialize changes when logical_decoding_work_mem is reached.
> On the subscriber side, if streaming option is set to parallel, this parameter also
> allows the leader apply worker to send changes to the shared memory queue or
> to serialize changes. When set to buffered, the leader sends changes to parallel
> apply workers via shared memory queue. When set to immediate, the leader
> serializes all changes to files and notifies the parallel apply workers to read and
> apply them at the end of the transaction.
> 
> ~
> 
> Because now this same developer GUC affects both the publisher side and the
> subscriber side differently IMO this whole description should be re-structured
> accordingly.
> 
> The allowed values of logical_replication_mode are buffered and immediate. The
> default is buffered.
> 

Changed.

> 
> ~~~
> 
> 4.
> This parameter is intended to be used to test logical decoding and replication of
> large transactions for which otherwise we need to generate the changes till
> Maybe this paragraph needs rewording or moving. e.g. Isn't that misleading
> now? Although this might be an explanation for the publisher side, it does not
> seem relevant to the subscriber side's behaviour.

Adjusted the description here.

> + if (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE) return
> + false;
> +
> 
> I felt that code should have some comment, even if it is just something quite
> basic like "/* For developer testing */"

Added.

> Is it necessary to TRUNCATE the table here? If everything is working shouldn't
> the data be rolled back anyway?

I think it's unnecessary, so removed.

> Perhaps this should insert 2 different values so then the verification code can
> check the correct value remains instead of just checking COUNT(*)?

I think testing the count should be ok as the nearby testcases are
also checking the count.

Best regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 24 Jan 2023 12:47:16 +0000

> typo "large transaction on subscriber" --> "large transactions on the subscriber"
> 
> ~
> 
> 2b.
> Rename the developer option 'logical_decoding_mode' to the more flexible
> name 'logical_replication_mode' because doing so will make it easier to extend
> this option in future to help test other areas of logical replication.

Changed.

> IMO it's more clear to say the default when the options are first mentioned. So I
> suggested removing the "which is the default" part, and instead saying:
> 
> BEFORE

I included this change in the 0002 patch.

> ssl_protocol_versions_info[] = {  };
> 
> + {"buffered", LOGICAL_REP_MODE_BUFFERED, false}, {"immediate",
> + LOGICAL_REP_MODE_IMMEDIATE, false},
>   {NULL, 0, false}
>  };
> 

No, I didn't notice this one. Changed.

Best Regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 24 Jan 2023 12:47:28 +0000

On Monday, January 23, 2023 8:34 PM Kuroda, Hayato wrote:
> 
> Followings are my comments.

> This struct should be also modified.

Modified.

> +               {"logical_replication_mode", PGC_USERSET,
> + DEVELOPER_OPTIONS,

I adjusted the description here.

>         generate the changes till
> <varname>logical_decoding_work_mem</varname>
>         is reached.
>        </para>
> ```
> 
> I understood that this part described the usage of the parameter. How about
> adding a statement like:
> 
> " Moreover, this can be also used to test the message passing between the
> leader and parallel apply workers."

Added.

Changed.

Best Regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 24 Jan 2023 12:49:43 +0000

> > Here are some review comments for v86-0002

Sorry, the patch set was somehow attached twice. Here is the correct new version
patch set which addressed all comments so far.

Best Regards,
Hou zj

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 24 Jan 2023 14:34:05 +0000

Dear Hou,

> patch set which addressed all comments so far.

Thank you for updating the patch! I confirmed that
All of my comments are addressed.

One comment:
In this test the rollback-prepared seems not to be executed.
This is because serializations are finished while handling PREPARE message
and the final state of transaction does not affect that, right?
I think it may be helpful to add a one line comment.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Peter Smith <[email-masked]>
Date: Wed, 25 Jan 2023 08:45:34 +1100

On Tue, Jan 24, 2023 at 11:49 PM [email-masked]
<[email-masked]> wrote:
...
> patch set which addressed all comments so far.

Here are my review comments for patch v87-0001.

======
src/backend/replication/logical/reorderbuffer.c

1.
@@ -210,7 +210,7 @@ int logical_decoding_work_mem;
 static const Size max_changes_in_memory = 4096; /* XXX for restore only */

 /* GUC variable */
-int logical_decoding_mode = LOGICAL_DECODING_MODE_BUFFERED;
+int logical_replication_mode = LOGICAL_REP_MODE_BUFFERED;

I noticed that the comment /* GUC variable */ is currently only above
the logical_replication_mode, but actually logical_decoding_work_mem
is a GUC variable too. Maybe this should be rearranged somehow then
change the comment "GUC variable" -> "GUC variables"?

======

@@ -4908,13 +4908,13 @@ struct config_enum ConfigureNamesEnum[] =
  },

  {
- {"logical_decoding_mode", PGC_USERSET, DEVELOPER_OPTIONS,
+ {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
  gettext_noop("Allows streaming or serializing each change in logical
decoding."),
  NULL,
  GUC_NOT_IN_SAMPLE
  },
- &logical_decoding_mode,
- LOGICAL_DECODING_MODE_BUFFERED, logical_decoding_mode_options,
+ &logical_replication_mode,
+ LOGICAL_REP_MODE_BUFFERED, logical_replication_mode_options,
  NULL, NULL, NULL
  },

That gettext_noop string seems incorrect. I think Kuroda-san
previously reported the same, but then you replied it has been fixed
already [1]

But this doesn't look fixed to me. (??)

======
src/include/replication/reorderbuffer.h

3.
@@ -18,14 +18,14 @@
 #include "utils/timestamp.h"

 extern PGDLLIMPORT int logical_decoding_work_mem;
-extern PGDLLIMPORT int logical_decoding_mode;
+extern PGDLLIMPORT int logical_replication_mode;

Probably here should also be a comment to say "/* GUC variables */"

------
[1] https://www.postgresql.org/message-id/OS0PR01MB5716AE9F095F9E7888987BC794C99%40OS0PR01MB5716.jpnprd01.prod.outlook.com

Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Peter Smith <[email-masked]>
Date: Wed, 25 Jan 2023 10:30:19 +1100

Here are my review comments for patch v87-0002.

======
doc/src/sgml/config.sgml

1.
        <para>
-        Allows streaming or serializing changes immediately in
logical decoding.
         The allowed values of <varname>logical_replication_mode</varname> are
-        <literal>buffered</literal> and <literal>immediate</literal>. When set
-        to <literal>immediate</literal>, stream each change if
+        <literal>buffered</literal> and <literal>immediate</literal>.
The default
+        is <literal>buffered</literal>.
+       </para>

I didn't think it was necessary to say “of logical_replication_mode”.
IMO that much is already obvious because this is the first sentence of
the description for logical_replication_mode.

(see also review comment #4)

~~~

2.
+       <para>
+        On the publisher side, it allows streaming or serializing changes
+        immediately in logical decoding.  When set to
+        <literal>immediate</literal>, stream each change if
         <literal>streaming</literal> option (see optional parameters set by
         <link linkend="sql-createsubscription"><command>CREATE
SUBSCRIPTION</command></link>)
         is enabled, otherwise, serialize each change.  When set to
-        <literal>buffered</literal>, which is the default, decoding will stream
-        or serialize changes when <varname>logical_decoding_work_mem</varname>
-        is reached.
+        <literal>buffered</literal>, decoding will stream or serialize changes
+        when <varname>logical_decoding_work_mem</varname> is reached.
        </para>

2a.
"it allows" --> "logical_replication_mode allows"

2b.
"decoding" --> "the decoding"

~~~

3.
+       <para>
+        On the subscriber side, if <literal>streaming</literal> option is set
+        to <literal>parallel</literal>, this parameter also allows the leader
+        apply worker to send changes to the shared memory queue or to serialize
+        changes.  When set to <literal>buffered</literal>, the leader sends
+        changes to parallel apply workers via shared memory queue.  When set to
+        <literal>immediate</literal>, the leader serializes all changes to
+        files and notifies the parallel apply workers to read and apply them at
+        the end of the transaction.
+       </para>

"this parameter also allows" --> "logical_replication_mode also allows"

~~~

4.
        <para>
         This parameter is intended to be used to test logical decoding and
         replication of large transactions for which otherwise we need to
         generate the changes till <varname>logical_decoding_work_mem</varname>
-        is reached.
+        is reached. Moreover, this can also be used to test the transmission of
+        changes between the leader and parallel apply workers.
        </para>

"Moreover, this can also" --> "It can also"

I am wondering would this sentence be better put at the top of the GUC
description. So then the first paragraph becomes like this:

SUGGESTION (I've also added another sentence "The effect of...")

The allowed values are buffered and immediate. The default is
buffered. This parameter is intended to be used to test logical
decoding and replication of large transactions for which otherwise we
need to generate the changes till logical_decoding_work_mem is
reached. It can also be used to test the transmission of changes
between the leader and parallel apply workers. The effect of
logical_replication_mode is different for the publisher and
subscriber:

On the publisher side...

On the subscriber side...
======
.../replication/logical/applyparallelworker.c

5.
+ /*
+ * In immeidate mode, directly return false so that we can switch to
+ * PARTIAL_SERIALIZE mode and serialize remaining changes to files.
+ */
+ if (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE)
+ return false;

Typo "immediate"

Also, I felt "directly" is not needed. "return false" and "directly
return false" is the same.

SUGGESTION
Using ‘immediate’ mode returns false to cause a switch to
PARTIAL_SERIALIZE mode so that the remaining changes will be
serialized.

======

6.
  {
  {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
- gettext_noop("Allows streaming or serializing each change in logical
decoding."),
- NULL,
+ gettext_noop("Controls the behavior of logical replication publisher
and subscriber"),
+ gettext_noop("If set to immediate, on the publisher side, it "
+ "allows streaming or serializing each change in "
+ "logical decoding. On the subscriber side, in "
+ "parallel streaming mode, it allows the leader apply "
+ "worker to serialize changes to files and notifies "
+ "the parallel apply workers to read and apply them at "
+ "the end of the transaction."),
  GUC_NOT_IN_SAMPLE
  },

6a. short description

User PoV behaviour should be the same. Instead, maybe say "controls
the internal behavior" or something like that?

~

6b. long description

IMO the long description shouldn’t mention ‘immediate’ mode first as it does.

BEFORE
If set to immediate, on the publisher side, ...

AFTER

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Amit Kapila <[email-masked]>
Date: Wed, 25 Jan 2023 10:05:01 +0530

> change the comment "GUC variable" -> "GUC variables"?

I think moving these variables together doesn't sound like a good idea
because logical_decoding_work_mem variable is defined with other
related variable. Also, if we are doing the last comment, I think that
will obviate the need for this.

> But this doesn't look fixed to me. (??)

Okay, so, how about the following for the 0001 patch:
short desc: Controls when to replicate each change.
long desc: On the publisher, it allows streaming or serializing each
change in logical decoding.

Then we can extend it as follows for the 0002 patch:
Controls when to replicate or apply each change
On the publisher, it allows streaming or serializing each change in
logical decoding. On the subscriber, it allows serialization of all
changes to files and notifies the parallel apply workers to read and
apply them at the end of the transaction.

> src/include/replication/reorderbuffer.h

Okay, we can do this.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Wed, 25 Jan 2023 11:57:41 +0530

> change in logical decoding.

I have updated the patch accordingly and it looks good to me. I'll
push this first patch early next week (Monday) unless there are more
comments.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 25 Jan 2023 06:51:26 +0000

Dear Amit,

> push this first patch early next week (Monday) unless there are more

Thanks for updating. I checked v88-0001 and I have no objection. LGTM.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: "[email-masked]" <[email-masked]>
Date: Wed, 25 Jan 2023 14:24:50 +0000

>          The allowed values of
> <varname>logical_replication_mode</varname> are
> IMO that much is already obvious because this is the first sentence of the
> description for logical_replication_mode.
> 

Changed.

> -        or serialize changes when

Changed.

Changed.

> The allowed values are buffered and immediate. The default is buffered. This
> parameter is intended to be used to test logical decoding and replication of large
> transactions for which otherwise we need to generate the changes till
> logical_decoding_work_mem is reached. It can also be used to test the
> transmission of changes between the leader and parallel apply workers. The
> effect of logical_replication_mode is different for the publisher and
> subscriber:
> 
> On the publisher side...
> 

I think your suggestion makes sense, so changed as suggested.

> Also, I felt "directly" is not needed. "return false" and "directly return false" is the
> same.
> 
> SUGGESTION
> Using ‘immediate’ mode returns false to cause a switch to PARTIAL_SERIALIZE
> mode so that the remaining changes will be serialized.

Changed.

> User PoV behaviour should be the same. Instead, maybe say "controls the
> internal behavior" or something like that?

Changed to "internal behavior xxx"

Changed.

Attach the new version patch set.
The 0001 patch is the same as the v88-0001 posted by Amit[1],
attach it here to make cfbot happy.

[1] https://www.postgresql.org/message-id/CAA4eK1JpWoaB63YULpQa1KDw_zBW-QoRMuNxuiP1KafPJzuVuw%40mail.gmail.com

Best Regards,
Hou zj

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Thu, 26 Jan 2023 03:36:57 +0000

Dear Hou,

Thank you for updating the patch! Followings are comments.

1. config.sgml

```
+        the changes till logical_decoding_work_mem is reached. It can also be
```

I think it should be sandwiched by <varname>.

2. config.sgml

```
+        On the publisher side, <varname>logical_replication_mode</varname> allows
+        allows streaming or serializing changes immediately in logical decoding.
```

Typo "allows allows" -> "allows"

3. test general

You confirmed that the leader started to serialize changes, but did not ensure the endpoint.
IIUC the parallel apply worker exits after applying serialized changes, and it is not tested yet.
Can we add polling the log somewhere?

4. 015_stream.pl

```
+is($result, qq(15000), 'all changes are replayed from file')
```

The statement may be unclear because changes can be also replicated when streaming = on.
How about: "parallel apply worker replayed all changes from file"?

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Masahiko Sawada <[email-masked]>
Date: Sat, 28 Jan 2023 01:36:26 +0900

> push this first patch early next week (Monday) unless there are more

The patch looks good to me too. Thank you for the patch.

Regards,

---

From: Peter Smith <[email-masked]>
Date: Mon, 30 Jan 2023 11:10:39 +1100

Patch v88-0001 LGTM.

Below are just some minor review comments about the commit message.

======
Commit message

1.
We have discussed having this parameter as a subscription option but
exposing a parameter that is primarily used for testing/debugging to users
didn't seem advisable and there is no other such parameter. The other
option we have discussed is to have a separate GUC for subscriber-side
testing but it appears that for the current testing existing parameter is
sufficient and avoids adding another GUC.

SUGGESTION
We discussed exposing this parameter as a subscription option, but it
did not seem advisable since it is primarily used for
testing/debugging and there is no other such developer option.

We also discussed having separate GUCs for publisher/subscriber-side,
but for current testing/debugging requirements, one GUC is sufficient.

~~

2.
Reviewed-by: Pater Smith, Kuroda Hayato, Amit Kapila

"Pater" --> "Peter"

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Peter Smith <[email-masked]>
Date: Mon, 30 Jan 2023 15:12:34 +1100

Here are my review comments for v88-0002.

======
General

1.
The test cases are checking the log content but they are not checking
for debug logs or untranslated elogs -- they are expecting a normal
ereport LOG that might be translated. I’m not sure if that is OK, or
if it is a potential problem.

======
doc/src/sgml/config.sgml

2.
On the publisher side, logical_replication_mode allows allows
streaming or serializing changes immediately in logical decoding. When
set to immediate, stream each change if streaming option (see optional
parameters set by CREATE SUBSCRIPTION) is enabled, otherwise,
serialize each change. When set to buffered, the decoding will stream
or serialize changes when logical_decoding_work_mem is reached.

2a.
typo "allows allows"  (Kuroda-san reported same)

2b.
"if streaming option" --> "if the streaming option"

~~~

3.
On the subscriber side, if streaming option is set to parallel,
logical_replication_mode also allows the leader apply worker to send
changes to the shared memory queue or to serialize changes.

SUGGESTION
On the subscriber side, if the streaming option is set to parallel,
logical_replication_mode can be used to direct the leader apply worker
to send changes to the shared memory queue or to serialize changes.

======

4.
  {
  {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
- gettext_noop("Controls when to replicate each change."),
- gettext_noop("On the publisher, it allows streaming or serializing
each change in logical decoding."),
+ gettext_noop("Controls the internal behavior of logical replication
publisher and subscriber"),
+ gettext_noop("On the publisher, it allows streaming or "
+ "serializing each change in logical decoding. On the "
+ "subscriber, in parallel streaming mode, it allows "
+ "the leader apply worker to serialize changes to "
+ "files and notifies the parallel apply workers to "
+ "read and apply them at the end of the transaction."),
  GUC_NOT_IN_SAMPLE
  },
Suggest re-wording the long description (subscriber part) to be more
like the documentation text.

BEFORE
On the subscriber, in parallel streaming mode, it allows the leader
apply worker to serialize changes to files and notifies the parallel
apply workers to read and apply them at the end of the transaction.

SUGGESTION
On the subscriber, if the streaming option is set to parallel, it
directs the leader apply worker to send changes to the shared memory
queue or to serialize changes and apply them at the end of the
transaction.

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Amit Kapila <[email-masked]>
Date: Mon, 30 Jan 2023 11:41:37 +0530

Pushed.

-- 
With Regards,
Amit Kapila.

---

From: "[email-masked]" <[email-masked]>
Date: Mon, 30 Jan 2023 06:23:17 +0000

> 
> ======
> General
> 
> 1.
> The test cases are checking the log content but they are not checking for
> debug logs or untranslated elogs -- they are expecting a normal ereport LOG
> that might be translated. I’m not sure if that is OK, or if it is a potential problem.

We have tests that check the ereport ERROR and ereport WARNING message(by
search for the ERROR or WARNING keyword for all the tap tests), so I think
checking the LOG should be fine.

> ======
> doc/src/sgml/config.sgml
> 
> 2.
> On the publisher side, logical_replication_mode allows allows streaming or
> serializing changes immediately in logical decoding. When set to immediate,
> stream each change if streaming option (see optional parameters set by
> CREATE SUBSCRIPTION) is enabled, otherwise, serialize each change. When set
> to buffered, the decoding will stream or serialize changes when

Changed.

> logical_replication_mode also allows the leader apply worker to send changes
> to the shared memory queue or to serialize changes.
> 
> SUGGESTION
> logical_replication_mode can be used to direct the leader apply worker to
> send changes to the shared memory queue or to serialize changes.

Changed.

> - gettext_noop("On the publisher, it allows streaming or serializing each
> change in logical decoding."),
> Suggest re-wording the long description (subscriber part) to be more like the
> documentation text.
> 
> BEFORE
> On the subscriber, in parallel streaming mode, it allows the leader apply worker
> to serialize changes to files and notifies the parallel apply workers to read and
> On the subscriber, if the streaming option is set to parallel, it directs the leader
> apply worker to send changes to the shared memory queue or to serialize
> changes and apply them at the end of the transaction.
> 

Changed.

Attach the new version patch which addressed all comments so far (the v88-0001
has been committed, so we only have one remaining patch this time).

Best Regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Mon, 30 Jan 2023 06:24:36 +0000

On Thursday, January 26, 2023 11:37 AM Kuroda, Hayato/黒田 隼人 <[email-masked]> wrote:
> Followings are comments.

Thanks for the comments.

> In this test the rollback-prepared seems not to be executed. This is because
> serializations are finished while handling PREPARE message and the final
> state of transaction does not affect that, right? I think it may be helpful
> to add a one line comment.

Yes, but I am slightly unsure if it would be helpful to add this as we only test basic
cases(mainly for code coverage) for partial serialization.

Added.

> 
> 2. config.sgml
> 
> ```
> +        On the publisher side,
> <varname>logical_replication_mode</varname> allows

Fixed.

> Can we add polling the log somewhere?

I checked other tests and didn't find some examples where we test the exit of
apply worker or table sync worker. And if the parallel apply worker doesn't stop in
this case, we will fail anyway when reusing this worker to handle the next
transaction because the queue is broken. So, I prefer to keep the tests short.

Changed.

Best regards,
Hou zj

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 30 Jan 2023 12:57:34 +0000

Dear Hou,

Thank you for updating the patch!
I checked your replies and new patch, and it seems good.
Currently I have no comments

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 30 Jan 2023 23:19:46 +0900

> has been committed, so we only have one remaining patch this time).

I have one comment on v89 patch:

+       /*
+        * Using 'immediate' mode returns false to cause a switch to
+        * PARTIAL_SERIALIZE mode so that the remaining changes will
be serialized.
+        */
+       if (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE)
+               return false;
+

Probably we might want to add unlikely() here since we could pass
through this path very frequently?

The rest looks good to me.

Regards,

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 31 Jan 2023 03:34:37 +0000

On Monday, January 30, 2023 10:20 PM Masahiko Sawada <[email-masked]> wrote:
> Probably we might want to add unlikely() here since we could pass through this
> path very frequently?

I think your comment makes sense, thanks.
I updated the patch for the same.

Best Regards,
Hou zj

---

From: Peter Smith <[email-masked]>
Date: Tue, 31 Jan 2023 16:40:30 +1100

Thanks for the updates to address all of my previous review comments.

Patch v90-0001 LGTM.

------
Kind Reagrds,
Peter Smith.
Fujitsu Australia

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 24 Apr 2023 10:55:44 +0900

> Pushed the first (0001) patch.

While looking at the worker.c, I realized that we have the following
code in handle_streamed_transaction():

        default:
            Assert(false);
            return false;       / silence compiler warning /

I think it's better to do elog(ERROR) instead of Assert() as it ends
up returning false in non-assertion builds, which might cause a
problem. And it's more consistent with other codes in worker.c. Please
find an attached patch.

Regards,

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 24 Apr 2023 11:50:37 +0900 (JST)

At Mon, 24 Apr 2023 10:55:44 +0900, Masahiko Sawada <[email-masked]> wrote in 
> problem. And it's more consistent with other codes in worker.c. Please

I concur that returning false is problematic.

For assertion builds, Assert typically provides more detailed
information than elog. However, in this case, it wouldn't matter much
since the worker would repeatedly restart even after a server-restart
for the same reason unless cosmic rays are involved. Moreover, the
situation doesn't justify server-restaring, as it would unnecessarily
involve other backends.

In my opinion, it is fine to replace the Assert with an ERROR.

regards.

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 24 Apr 2023 12:10:12 +0900 (JST)

At Mon, 24 Apr 2023 11:50:37 +0900 (JST), Kyotaro Horiguchi <[email-masked]> wrote in 
> In my opinion, it is fine to replace the Assert with an ERROR.

Sorry for posting multiple times in a row, but I'm a bit unceratin
whether we should use FATAL or ERROR for this situation. The stream is
not provided by user, and the session or process cannot continue.

regards.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 24 Apr 2023 08:59:07 +0530

> not provided by user, and the session or process cannot continue.

I think ERROR should be fine here similar to other cases in worker.c.

-- 
With Regards,
Amit Kapila.

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 24 Apr 2023 13:03:03 +0900 (JST)

At Mon, 24 Apr 2023 08:59:07 +0530, Amit Kapila <[email-masked]> wrote in 
> I think ERROR should be fine here similar to other cases in worker.c.

Sure, I don't have any issues with it.

regards.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 24 Apr 2023 10:54:11 +0530

> problem. And it's more consistent with other codes in worker.c. Please

I haven't tested it but otherwise, the changes look good to me.

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 24 Apr 2023 15:42:56 +0900

> I haven't tested it but otherwise, the changes look good to me.

Thanks for checking! Pushed.

Regards,

---

From: Alexander Lakhin <[email-masked]>
Date: Wed, 26 Apr 2023 12:00:02 +0300

Hello hackers,

Please look at a new anomaly that can be observed starting from 216a7848.

The following script:
echo "CREATE SUBSCRIPTION testsub CONNECTION 'dbname=nodb' PUBLICATION testpub WITH (connect = false);
ALTER SUBSCRIPTION testsub ENABLE;" | psql

sleep 1
rm $PGINST/lib/libpqwalreceiver.so
sleep 15
pg_ctl -D "$PGDB" stop -m immediate
grep 'TRAP:' server.log

Leads to multiple assertion failures:
CREATE SUBSCRIPTION
ALTER SUBSCRIPTION
waiting for server to shut down.... done
server stopped
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899323
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899416
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899427
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899439
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899538
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899547

server.log contains:
2023-04-26 11:00:58.797 MSK [2899300] LOG:  database system is ready to accept connections
2023-04-26 11:00:58.821 MSK [2899416] ERROR:  could not access file "libpqwalreceiver": No such file or directory
TRAP: failed Assert("MyProc->backendId != InvalidBackendId"), File: "lock.c", Line: 4439, PID: 2899416
postgres: logical replication apply worker for subscription 16385 (ExceptionalCondition+0x69)[0x558b2ac06d41]
postgres: logical replication apply worker for subscription 16385 (VirtualXactLockTableCleanup+0xa4)[0x558b2aa9fd74]
postgres: logical replication apply worker for subscription 16385 (LockReleaseAll+0xbb)[0x558b2aa9fe7d]
postgres: logical replication apply worker for subscription 16385 (+0x4588c6)[0x558b2aa2a8c6]
postgres: logical replication apply worker for subscription 16385 (shmem_exit+0x6c)[0x558b2aa87eb1]
postgres: logical replication apply worker for subscription 16385 (+0x4b5faa)[0x558b2aa87faa]
postgres: logical replication apply worker for subscription 16385 (proc_exit+0xc)[0x558b2aa88031]
postgres: logical replication apply worker for subscription 16385 (StartBackgroundWorker+0x147)[0x558b2aa0b4d9]
postgres: logical replication apply worker for subscription 16385 (+0x43fdc1)[0x558b2aa11dc1]
postgres: logical replication apply worker for subscription 16385 (+0x43ff3d)[0x558b2aa11f3d]
postgres: logical replication apply worker for subscription 16385 (+0x440866)[0x558b2aa12866]
postgres: logical replication apply worker for subscription 16385 (+0x440e12)[0x558b2aa12e12]
postgres: logical replication apply worker for subscription 16385 (BackgroundWorkerInitializeConnection+0x0)[0x558b2aa14396]
postgres: logical replication apply worker for subscription 16385 (main+0x21a)[0x558b2a932e21]

I understand, that removing libpqwalreceiver.so (or whole pginst/) is not
what happens in a production environment every day, but nonetheless it's a
new failure mode and it can produce many coredumps when testing.

IIUC, that assert will fail in case of any error raised between
ApplyWorkerMain()->logicalrep_worker_attach()->before_shmem_exit() and
ApplyWorkerMain()->InitializeApplyWorker()->BackgroundWorkerInitializeConnectionByOid()->InitPostgres().

Best regards,
Alexander

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Wed, 26 Apr 2023 10:41:22 +0000

On Wednesday, April 26, 2023 5:00 PM Alexander Lakhin <[email-masked]> wrote:
> echo "CREATE SUBSCRIPTION testsub CONNECTION 'dbname=nodb'
> PUBLICATION testpub WITH (connect = false);
> 2023-04-26 11:00:58.821 MSK [2899416] ERROR:  could not access file
> "libpqwalreceiver": No such file or directory
> postgres: logical replication apply worker for subscription 16385
> (ExceptionalCondition+0x69)[0x558b2ac06d41]
> postgres: logical replication apply worker for subscription 16385
> (VirtualXactLockTableCleanup+0xa4)[0x558b2aa9fd74]
> postgres: logical replication apply worker for subscription 16385
> (StartBackgroundWorker+0x147)[0x558b2aa0b4d9]
> postgres: logical replication apply worker for subscription 16385
> (BackgroundWorkerInitializeConnection+0x0)[0x558b2aa14396]

Thanks for reporting the issue.

I think the problem is that it tried to release locks in
logicalrep_worker_onexit() before the initialization of the process is complete
because this callback function was registered before the init phase. So I think we
can add a conditional statement before releasing locks. Please find an attached
patch.

Best Regards,
Hou zj

---

From: Amit Kapila <[email-masked]>
Date: Wed, 26 Apr 2023 16:51:11 +0530

> can add a conditional statement before releasing locks. Please find an attached

Yeah, this should work. Yet another possibility is to introduce a new
variable 'InitializingApplyWorker' similar to
'InitializingParallelWorker' and use that to prevent releasing locks.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Fri, 28 Apr 2023 08:21:22 +0530

> can add a conditional statement before releasing locks. Please find an attached

Alexander, does the proposed patch fix the problem you are facing?
Sawada-San, and others, do you see any better way to fix it than what
has been proposed?

-- 
With Regards,
Amit Kapila.

---

From: Alexander Lakhin <[email-masked]>
Date: Fri, 28 Apr 2023 08:00:01 +0300

Hello Amit and Zhijie,

> Sawada-San, and others, do you see any better way to fix it than what

Yes, the patch definitely fixes it.
Maybe some other onexit actions can be skipped in the non-normal mode,
but the assert-triggering LockReleaseAll() not called now.

Thank you!

Best regards,
Alexander

---

From: Masahiko Sawada <[email-masked]>
Date: Fri, 28 Apr 2023 15:18:01 +0900

> Sawada-San, and others, do you see any better way to fix it than what

I'm concerned that the idea of relying on IsNormalProcessingMode()
might not be robust since if we change the meaning of
IsNormalProcessingMode() some day it would silently break again. So I
prefer using something like InitializingApplyWorker, or another idea
would be to do cleanup work (e.g., fileset deletion and lock release)
in a separate callback that is registered after connecting to the
database.

While investigating this issue, I've reviewed the code around
callbacks and worker termination etc and I found a problem.

A parallel apply worker calls the before_shmem_exit callbacks in the
following order:

1. ShutdownPostgres()
2. logicalrep_worker_onexit()
3. pa_shutdown()

Since the worker is detached during logicalrep_worker_onexit(),
MyLogicalReplication->leader_pid is an invalid when we call
pa_shutdown():

static void
pa_shutdown(int code, Datum arg)
{
    Assert(MyLogicalRepWorker->leader_pid != InvalidPid);
    SendProcSignal(MyLogicalRepWorker->leader_pid,
                   PROCSIG_PARALLEL_APPLY_MESSAGE,
                   InvalidBackendId);

Also, if the parallel apply worker fails shm_toc_lookup() during the
initialization, it raises an error (because of noError = false) but
ends up a SEGV as MyLogicalRepWorker is still NULL.

I think that we should not use MyLogicalRepWorker->leader_pid in
pa_shutdown() but instead store the leader's pid to a static variable
before registering pa_shutdown() callback. And probably we can
remember the backend id of the leader apply worker to speed up
SendProcSignal().

FWIW, we might need to be careful about the timing when we call
logicalrep_worker_detach() in the worker's termination process. Since
we rely on IsLogicalParallelApplyWorker() for the parallel apply
worker to send ERROR messages to the leader apply worker, if an ERROR
happens after logicalrep_worker_detach(), we will end up with the
assertion failure.

            if (IsLogicalParallelApplyWorker())
                SendProcSignal(pq_mq_parallel_leader_pid,
                               PROCSIG_PARALLEL_APPLY_MESSAGE,
                               pq_mq_parallel_leader_backend_id);
            else
            {
                Assert(IsParallelWorker());

It normally would be a should-no-happen case, though.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Fri, 28 Apr 2023 14:31:11 +0530

On Fri, Apr 28, 2023 at 11:48 AM Masahiko Sawada <[email-masked]> wrote:
> prefer using something like InitializingApplyWorker,
>

I think if we change the meaning of IsNormalProcessingMode() then it
could also break the other places the similar check is being used.
However, I am fine with InitializingApplyWorker as that could be used
at other places as well. I just want to avoid adding another variable
by using IsNormalProcessingMode.

> or another idea

Yeah, but not sure if it's worth having multiple callbacks for cleanup work.

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 1 May 2023 12:52:06 +0900

> could also break the other places the similar check is being used.

Right, but I think it's unclear the relationship between the
processing modes and releasing session locks. If non-normal-processing
mode means we're still in the process initialization phase, why we
don't skip other cleanup works such as walrcv_disconnect() and
FileSetDeleteAll()?

> by using IsNormalProcessingMode.

I think it's less confusing.

> Yeah, but not sure if it's worth having multiple callbacks for cleanup work.

Fair point.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Tue, 2 May 2023 08:51:45 +0530

On Fri, Apr 28, 2023 at 11:48 AM Masahiko Sawada <[email-masked]> wrote:
> before registering pa_shutdown() callback.
>

Why not simply move the registration of pa_shutdown() to someplace
after logicalrep_worker_attach()? BTW, it seems we don't have access
to MyLogicalRepWorker->leader_pid till we attach to the worker slot
via logicalrep_worker_attach(), so we anyway need to do what you are
suggesting after attaching to the worker slot.

-- 
With Regards,
Amit Kapila.

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Tue, 2 May 2023 03:35:58 +0000

> in a separate callback that is registered after connecting to the

Thanks for the review. I agree that it’s better to use a new variable here.
Attach the patch for the same.

> It normally would be a should-no-happen case, though.

Yes, I think currently PA sends ERROR message before exiting,
so the callback functions are always fired after the above code which
looks fine to me.

Best Regards,
Hou zj

---

From: Amit Kapila <[email-masked]>
Date: Tue, 2 May 2023 09:46:21 +0530

> Attach the patch for the same.

+ *
+ * However, if the worker is being initialized, there is no need to release
+ * locks.
  */
- LockReleaseAll(DEFAULT_LOCKMETHOD, true);
+ if (!InitializingApplyWorker)
+ LockReleaseAll(DEFAULT_LOCKMETHOD, true);

Can we slightly reword this comment as: "The locks will be acquired
once the worker is initialized."?

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Wed, 3 May 2023 12:47:06 +0530

> once the worker is initialized."?

After making this modification, I pushed your patch. Thanks!

-- 
With Regards,
Amit Kapila.

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Fri, 5 May 2023 03:44:47 +0000

On Wednesday, May 3, 2023 3:17 PM Amit Kapila <[email-masked]> wrote:
> > > > > Sawada-San, and others, do you see any better way to fix it than
> > > > > what has been proposed?
> > > >
> > > > IsNormalProcessingMode() some day it would silently break again.
> > > > So I prefer using something like InitializingApplyWorker, or
> > > > another idea would be to do cleanup work (e.g., fileset deletion
> > > > and lock release) in a separate callback that is registered after
> > > > connecting to the database.
> > >
> > + * However, if the worker is being initialized, there is no need to
> > + release
> > + * locks.
> >   */

Thanks for pushing.

Attach another patch to fix the problem that pa_shutdown will access invalid
MyLogicalRepWorker. I personally want to avoid introducing new static variable,
so I only reorder the callback registration in this version.

When testing this, I notice a rare case that the leader is possible to receive
the worker termination message after the leader stops the parallel worker. This
is unnecessary and have a risk that the leader would try to access the detached
memory queue. This is more likely to happen and sometimes cause the failure in
regression tests after the registration reorder patch because the dsm is
detached earlier after applying the patch.

So, put the patch that detach the error queue before stopping worker as 0001
and the registration reorder patch as 0002.

Best Regards,
Hou zj

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 8 May 2023 12:07:51 +0900

On Tue, May 2, 2023 at 12:22 PM Amit Kapila <[email-masked]> wrote:
> after logicalrep_worker_attach()?

If we do that, the worker won't call dsm_detach() if it raises an
ERROR in logicalrep_worker_attach(), is that okay? It seems that it's
no practically problem since we call dsm_backend_shutdown() in
shmem_exit(), but if so why do we need to call it in pa_shutdown()?

Regards,

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Mon, 8 May 2023 03:52:44 +0000

On Monday, May 8, 2023 11:08 AM Masahiko Sawada <[email-masked]>

Hi,

> shmem_exit(), but if so why do we need to call it in pa_shutdown()?

I think the dsm_detach in pa_shutdown was intended to fire on_dsm_detach
callbacks to give callback a chance to report stat before the stat system is
shutdown, following what we do in ParallelWorkerShutdown() (e.g.
sharedfileset.c callbacks cause fd.c to do ReportTemporaryFileUsage(), so we
need to fire that earlier).

But for parallel apply, we currently only have one on_dsm_detach
callback(shm_mq_detach_callback) which doesn't report extra stats. So the
dsm_detach in pa_shutdown is only used to make it a bit future-proof in case
we add some other on_dsm_detach callbacks in the future which need to report
stats.

Best regards,
Hou zj

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 8 May 2023 14:38:19 +0900

> we add some other on_dsm_detach callbacks in the future which need to report

Make sense . Given that it's possible that we add other callbacks that
report stats in the future, I think it's better not to move the
position to register pa_shutdown() callback.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Mon, 8 May 2023 12:03:48 +0530

> position to register pa_shutdown() callback.

Hmm, what kind of stats do we expect to be collected before we
register pa_shutdown? I think if required we can register such a
callback after pa_shutdown. I feel without reordering the callbacks,
the fix would be a bit complicated as explained in my previous email,
so I don't think it is worth complicating this code unless really
required.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 8 May 2023 16:39:07 +0530

> detached earlier after applying the patch.

I think it is only possible for the leader apply can worker to try to
receive the error message from an error queue after your 0002 patch.
Because another place already detached from the queue before stopping
the parallel apply workers. So, I combined both the patches and
changed a few comments and a commit message. Let me know what you
think of the attached.

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 9 May 2023 11:19:20 +0900

> so I don't think it is worth complicating this code unless really

Fair point. I agree that the issue can be resolved by carefully
ordering the callback registration.

Another thing I'm concerned about is that since both the leader worker
and parallel worker detach DSM before logicalrep_worker_onexit(),
cleaning up work that touches DSM cannot be done in
logicalrep_worker_onexit(). If we need to do something in the future,
we would need to have another callback called before detaching DSM.
But I'm fine as it's not a problem for now.

Regards,

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 9 May 2023 11:19:47 +0900

> changed a few comments and a commit message. Let me know what you

I have one comment on the detaching error queue part:

+       /*
+        * Detach from the error_mq_handle for the parallel apply worker before
+        * stopping it. This prevents the leader apply worker from trying to
+        * receive the message from the error queue that might already
be detached
+        * by the parallel apply worker.
+        */
+       shm_mq_detach(winfo->error_mq_handle);
+       winfo->error_mq_handle = NULL;

In pa_detach_all_error_mq(), we try to detach error queues of all
workers in the pool. I think we should check if the queue is already
detached (i.e. is NULL) there. Otherwise, we will end up a SEGV if an
error happens after detaching the error queue and before removing the
worker from the pool.

Regards,