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

From: Masahiko Sawada <[email-masked]>
Date: Mon, 2 May 2022 15:16:55 +0900

On Fri, Apr 8, 2022 at 6:14 PM [email-masked]
<[email-masked]> wrote:
>
> On Wednesday, April 6, 2022 1:20 PM Amit Kapila <[email-masked]> wrote:
>
> > In this email, I would like to discuss allowing streaming logical
> > transactions (large in-progress transactions) by background workers
> > and parallel apply in general. The goal of this work is to improve the
> > performance of the apply work in logical replication.
> >
> > Currently, for large transactions, the publisher sends the data in
> > multiple streams (changes divided into chunks depending upon
> > logical_decoding_work_mem), and then on the subscriber-side, the apply
> > worker writes the changes into temporary files and once it receives
> > the commit, it read from the file and apply the entire transaction. To
> > improve the performance of such transactions, we can instead allow
> > them to be applied via background workers. There could be multiple
> > ways to achieve this:
> >
> > Approach-1: Assign a new bgworker (if available) as soon as the xact's
> > first stream came and the main apply worker will send changes to this
> > new worker via shared memory. We keep this worker assigned till the
> > transaction commit came and also wait for the worker to finish at
> > commit. This preserves commit ordering and avoid writing to and
> > reading from file in most cases. We still need to spill if there is no
> > worker available. We also need to allow stream_stop to complete by the
> > background worker to finish it to avoid deadlocks because T-1's
> > current stream of changes can update rows in conflicting order with
> > T-2's next stream of changes.
> >
>
> Attach the POC patch for the Approach-1 of "Perform streaming logical
> transactions by background workers". The patch is still a WIP patch as
> there are serval TODO items left, including:
>
> * error handling for bgworker
> * support for SKIP the transaction in bgworker
> * handle the case when there is no more worker available
>   (might need spill the data to the temp file in this case)
> * some potential bugs

Are you planning to support "Transaction dependency" Amit mentioned in
his first mail in this patch? IIUC since the background apply worker
applies the streamed changes as soon as receiving them from the main
apply worker, a conflict that doesn't happen in the current streaming
logical replication could happen.

Regards,

-- 
Masahiko Sawada
EDB:  https://www.enterprisedb.com/

---

From: Amit Kapila <[email-masked]>
Date: Mon, 2 May 2022 14:39:36 +0530

> logical replication could happen.

This patch seems to be waiting for stream_stop to finish, so I don't
see how the issues related to "Transaction dependency" can arise? What
type of conflict/issues you have in mind?

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 2 May 2022 20:35:47 +0900

> type of conflict/issues you have in mind?

Suppose we set both publisher and subscriber:

On publisher:
create table test (i int);
insert into test values (0);
create publication test_pub for table test;

On subscriber:
create table test (i int primary key);
create subscription test_sub connection '...' publication test_pub; --
value 0 is replicated via initial sync

Now, both 'test' tables have value 0.

And suppose two concurrent transactions are executed on the publisher
in following order:

TX-1:
begin;
insert into test select generate_series(0, 10000); -- changes will be streamed;

    TX-2:
    begin;
    delete from test where c = 0;
    commit;

TX-1:
commit;

With the current streaming logical replication, these changes will be
applied successfully since the deletion is applied before the
(streamed) insertion. Whereas with the apply bgworker, it fails due to
an unique constraint violation since the insertion is applied first.
I've confirmed that it happens with v5 patch.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Tue, 3 May 2022 09:45:14 +0530

> I've confirmed that it happens with v5 patch.

Good point but I am not completely sure if doing transaction
dependency tracking for such cases is really worth it. I feel for such
concurrent cases users can anyway now also get conflicts, it is just a
matter of timing. One more thing to check transaction dependency, we
might need to spill the data for streaming transactions in which case
we might lose all the benefits of doing it via a background worker. Do
we see any simple way to avoid this?

-- 
With Regards,
Amit Kapila.

---

From: Peter Smith <[email-masked]>
Date: Tue, 3 May 2022 17:16:41 +1000

> we see any simple way to avoid this?

Avoiding unexpected differences like this is why I suggested the
option should have to be explicitly enabled instead of being on by
default as it is in the current patch. See my review comment #14 [1].
It means the user won't have to change their existing code as a
workaround.

------
[1] https://www.postgresql.org/message-id/CAHut%2BPuqYP5eD5wcSCtk%3Da6KuMjat2UCzqyGoE7sieCaBsVskQ%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Peter Smith <[email-masked]>
Date: Wed, 4 May 2022 09:44:13 +1000

On Tue, May 3, 2022 at 5:16 PM Peter Smith <[email-masked]> wrote:
...

> [1] https://www.postgresql.org/message-id/CAHut%2BPuqYP5eD5wcSCtk%3Da6KuMjat2UCzqyGoE7sieCaBsVskQ%40mail.gmail.com

Sorry I was wrong above. It seems this behaviour was already changed
in the latest patch v5 so now the option value 'on' means what it
always did. Thanks!

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

---

From: Amit Kapila <[email-masked]>
Date: Wed, 4 May 2022 09:20:39 +0530

> we see any simple way to avoid this?

I think the other kind of problem that can happen here is delete
followed by an insert. If in the example provided by you, TX-1
performs delete (say it is large enough to cause streaming) and TX-2
performs insert then I think it will block the apply worker because
insert will start waiting infinitely. Currently, I think it will lead
to conflict due to insert but that is still solvable by allowing users
to remove conflicting rows.

It seems both these problems are due to the reason that the table on
publisher and subscriber has different constraints otherwise, we would
have seen the same behavior on the publisher as well.

There could be a few ways to avoid these and similar problems:
a. detect the difference in constraints between publisher and
subscribers like primary key and probably others (like whether there
is any volatile function present in index expression) when applying
the change and then we give ERROR to the user that she must change the
streaming mode to 'spill' instead of 'apply' (aka parallel apply).
b. Same as (a) but instead of ERROR just LOG this information and
change the mode to spill for the transactions that operate on that
particular relation.

I think we can cache this information in LogicalRepRelMapEntry.

Thoughts?

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 10 May 2022 14:04:42 +0900

> > we see any simple way to avoid this?

I agree that it is just a matter of timing. I think new issues that
haven't happened on the current streaming logical replication
depending on the timing could happen with this feature and vice versa.

> change the mode to spill for the transactions that operate on that

Given that it doesn't introduce a new kind of problem I don't think we
need special treatment for that at least in this feature. If we want
such modes we can discuss it separately.

Regards,

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 10 May 2022 14:09:14 +0900

> in the latest patch v5 so now the option value 'on' means what it

Having it optional seems a good idea. BTW can the user configure how
many apply bgworkers can be used per subscription or in the whole
system? Like max_sync_workers_per_subscription, is it better to have a
configuration parameter or a subscription option for that? If so,
setting it to 0 probably means to disable the parallel apply feature.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Tue, 10 May 2022 14:28:59 +0530

> setting it to 0 probably means to disable the parallel apply feature.

Yeah, that might be useful but we are already giving an option while
creating a subscription whether to allow parallelism, so will it be
useful to give one more way to disable this feature? OTOH, having
something like max_parallel_apply_workers/max_bg_apply_workers at the
system level can give better control for how much parallelism the user
wishes to allow for apply work. If we have such a new parameter then I
think max_logical_replication_workers should include apply workers,
parallel apply workers, and table synchronization? In such a case,
don't we need to think of increasing the default value of
max_logical_replication_workers?

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Tue, 10 May 2022 14:39:56 +0530

> depending on the timing could happen with this feature and vice versa.

Here by vice versa, do you mean some problems that can happen with
current code won't happen after new implementation? If so, can you
give one such example?

> need special treatment for that at least in this feature.
>

Isn't the problem related to infinite wait by insert as explained in
my previous email (in the above-quoted text) a new kind of problem
that won't exist in the current implementation?

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Wed, 11 May 2022 12:46:56 +0900

> that won't exist in the current implementation?

Sorry I had completely missed the point that the commit order won't be
changed. I agree that this new implementation would introduce a new
kind of issue as you mentioned above, and the opposite is not true.

Regarding the case you explained in the previous email I also think it
will happen with the parallel apply feature. The apply worker will be
blocked until the conflict is resolved. I'm not sure how to avoid
that. It would be not easy to compare constraints between publisher
and subscribers when replicating partitioning tables.

Regards,

---

From: Masahiko Sawada <[email-masked]>
Date: Wed, 11 May 2022 13:05:02 +0900

On Tue, May 10, 2022 at 5:59 PM Amit Kapila <[email-masked]> wrote:
> wishes to allow for apply work.

Or we can have something like
max_parallel_apply_workers_per_subscription that controls how many
parallel apply workers can launch per subscription. That also gives
better control for the number of parallel apply workers.

> If we have such a new parameter then I
> parallel apply workers, and table synchronization?

Agreed.

>  In such a case,

I think we would need to think about that if the parallel apply is
enabled by default but given that it's disabled by default I'm fine
with the current default value.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Wed, 11 May 2022 10:24:26 +0530

> and subscribers when replicating partitioning tables.

I agree that partitioned tables need some more thought but in some
simple cases where replication happens via individual partition tables
(default), we can detect as we do for normal tables. OTOH, when
replication happens via root (publish_via_partition_root) it could be
tricky as the partitions could be different on both sides. I think the
cases where we can't safely identify the constraint difference won't
be considered for apply via a new bg worker.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Wed, 11 May 2022 10:40:27 +0530

> better control for the number of parallel apply workers.

I think we can go either way in this matter as both have their pros
and cons. I feel limiting the parallel workers per subscription gives
better control but OTOH, it may not allow max usage of parallelism
because some quota from other subscriptions might remain unused. Let
us see what Hou-San or others think on this matter?

-- 
With Regards,
Amit Kapila.

---

From: "[email-masked]" <[email-masked]>
Date: Fri, 13 May 2022 08:52:32 +0000

On Wednesday, May 11, 2022 1:10 PM Amit Kapila <[email-masked]> wrote:
> > > > Having it optional seems a good idea. BTW can the user configure
> > > > how many apply bgworkers can be used per subscription or in the
> > > > whole system? Like max_sync_workers_per_subscription, is it better
> > > > to have a configuration parameter or a subscription option for
> > > > that? If so, setting it to 0 probably means to disable the parallel apply
> feature.
> > > >
> > >
> > > something like max_parallel_apply_workers/max_bg_apply_workers at
> > > the system level can give better control for how much parallelism
> > > the user wishes to allow for apply work.
> >
> I think we can go either way in this matter as both have their pros and cons. I
> feel limiting the parallel workers per subscription gives better control but
> OTOH, it may not allow max usage of parallelism because some quota from
> other subscriptions might remain unused. Let us see what Hou-San or others
> think on this matter?

Thanks for Amit and Sawada-san's comments !
I will think over these approaches and reply soon.

Best regards,
Hou zj

---

From: "[email-masked]" <[email-masked]>
Date: Wed, 25 May 2022 02:24:59 +0000

> I will think over these approaches and reply soon.
After reading the thread, I wrote two patches for these comments.

The first patch (see v6-0003):
Improve the feature as suggested in [1].
For the issue mentioned by Amit-san (there is a block problem in the case
mentioned by Sawada-san), after investigating, I think this issue is caused by
unique index. So I added a check to make sure the unique columns are the same
between publisher and subscriber.
For other cases, I added the check that if there is any non-immutable function
present in expression in subscriber's relation. Check from the following 3
items:
    a. The function in triggers;
    b. Column default value expressions and domain constraints;
    c. Constraint expressions.
BTW, I do not add partitioned table related code. I think this part needs other
additional modifications. I will add this later when these modifications are
finished.

The second patch (see v6-0004):
Improve the feature as suggested in [2].
Add a GUC "max_apply_bgworkers_per_subscription" to control parallelism. This
GUC controls how many apply background workers can be launched per
subscription. I set its default value to 3 and do not change the default value
of other GUCs.

[1] - https://www.postgresql.org/message-id/CAA4eK1JwahU_WuP3S%2B7POqta%3DPhm_3gxZeVmJuuoUq1NV%3DkrXA%40mail.gmail.com
[2] - https://www.postgresql.org/message-id/CAA4eK1%2B7D4qAQUQEE8zzQ0fGCqeBWd3rzTaY5N0jVs-VXFc_Xw%40mail.gmail.com

Attach the patches. (Did not change v6-0001 and v6-0002.)

Regards,
Wang wei

---

From: "[email-masked]" <[email-masked]>
Date: Sun, 29 May 2022 12:25:12 +0000

> Attach the patches. (Did not change v6-0001 and v6-0002.)
Hi,

Some review comments on the new patches from v6-0001 to v6-0004.

<v6-0001>

(1) create_subscription.sgml

+          the transaction is committed. Note that if an error happens when
+          applying changes in a background worker, it might not report the
+          finish LSN of the remote transaction in the server log.

I suggest to add a couple of sentences like below
to the section of logical-replication-conflicts in logical-replication.sgml.

"
Setting streaming mode to 'apply' can export invalid LSN as
finish LSN of failed transaction. Changing the streaming mode and
making the same conflict writes the finish LSN of the
failed transaction in the server log if required.
"

(2) ApplyBgworkerMain

+       PG_TRY();
+       {
+               LogicalApplyBgwLoop(mqh, pst);
+       }
+       PG_CATCH();
+       {

...

+               pgstat_report_subscription_error(MySubscription->oid, false);
+
+               PG_RE_THROW();
+       }
+       PG_END_TRY();

When I stream a transaction in-progress and it causes an error(duplication error),
seemingly the subscription stats (values in pg_stat_subscription_stats) don't
get updated properly. The 2nd argument should be true for apply error.

Also, I observe that both apply_error_count and sync_error_count
get updated together by error. I think we need to check this point as well.

<v6-0003>

(3) logicalrep_write_attrs

+       if (rel->rd_rel->relhasindex)
+       {
+               List       *indexoidlist = RelationGetIndexList(rel);
+               ListCell   *indexoidscan;
+               foreach(indexoidscan, indexoidlist)

and

+                       if (indexRel->rd_index->indisunique)
+                       {
+                               int             i;
+                               /* Add referenced attributes to idindexattrs */
+                               for (i = 0; i < indexRel->rd_index->indnatts; i++)

We don't have each blank line after variable declarations.
There might be some other codes where this point can be applied.
Please check.

(4)

+       /*
+        * If any unique index exist, check that they are same as remoterel.
+        */
+       if (!rel->sameunique)
+               ereport(ERROR,
+                               (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                                errmsg("cannot replicate relation with different unique index"),
+                                errhint("Please change the streaming option to 'on' instead of 'apply'.")));

When I create a logical replication setup with different constraints
and let streaming of in-progress transaction run,
I keep getting this error.

This should be documented as a restriction or something,
to let users know the replication progress can't go forward by
any differences written like in the commit-message in v6-0003.

Also, it would be preferable to test this as well, if we
don't dislike having TAP tests for this.

Best Regards,
	Takamichi Osumi

---

From: "[email-masked]" <[email-masked]>
Date: Thu, 2 Jun 2022 10:04:31 +0000

> Some review comments on the new patches from v6-0001 to v6-0004.
Thanks for your comments.

> failed transaction in the server log if required.
Add the sentences as suggested.

> get updated together by error. I think we need to check this point as well.
Yes, we should input "true" to 2nd argument here to log "apply error".
And after checking the second point you mentioned, I think it is caused by the
first point you mentioned and another reason:
With the patch v6 (or v7) and we specify option "apply", when a streamed
transaction causes an error (duplication error), the function
pgstat_report_subscription_error is invoke twice (in main apply worker and
apply background worker, see function ApplyWorkerMain()->start_apply() and
ApplyBgworkerMain). This means for one same error, we will send twice stats
message.
So to fix this, I removed the code that you mentioned and then, just invoke
function LogicalApplyBgwLoop here.

> There might be some other codes where this point can be applied.
Improve the formatting as you suggested. And I run pgindent for new patches.

> don't dislike having TAP tests for this.
Yes, you are right. Thank for your reminder.
I added this in the paragraph introducing value "apply" in
create_subscription.sgml:
```
To run in this mode, there are following two requirements. The first
is that the unique column should be the same between publisher and
subscriber; the second is that there should not be any non-immutable
function in subscriber-side replicated table.
```
Also added the related tests. (refer to 032_streaming_apply.pl in v8-0003)

I also made some other changes. The new patches and the modification details
were attached in [1].

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

Regards,
Wang wei