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

Wei Wang (Fujitsu) <wangw.fnst@fujitsu.com>

From: "wangw.fnst@fujitsu.com" <wangw.fnst@fujitsu.com>
To: Peter Smith <smithpb2250@gmail.com>
Cc: Amit Kapila <amit.kapila16@gmail.com>, "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>, Dilip Kumar <dilipbalaut@gmail.com>, Masahiko Sawada <sawada.mshk@gmail.com>, "shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>, PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Date: 2022-08-16T07:33:04Z
Lists: pgsql-hackers

Attachments

On Fri, August 12, 2022 12:46 PM Peter Smith <smithpb2250@gmail.com> wrote:
> Here are some review comments for v20-0003:
> 
> (Sorry - the reviews are time consuming, so I am lagging slightly
> behind the latest posted version)

Thanks for your comments.

> 1. <General>
> 
> 1a.
> There are a few comment modifications in this patch (e.g. changing
> FROM "in an apply background worker" TO "using an apply background
> worker"). e.g. I noticed lots of these in worker.c but they might be
> in other files too.
> 
> Although these are good changes, these are just tweaks to new comments
> introduced by patch 0001, so IMO such changes belong in that patch,
> not in this one.
> 
> 1b.
> Actually, there are still some comments says "by an apply background
> worker///" and some saying "using an apply background worker..." and
> some saying "in the apply background worker...". Maybe they are all
> OK, but it will be better if all such can be searched and made to have
> consistent wording

Improved.

> 2. Commit message
> 
> 2a.
> 
> Without these restrictions, the following scenario may occur:
> The apply background worker lock a row when processing a streaming
> transaction,
> after that the main apply worker tries to lock the same row when processing
> another transaction. At this time, the main apply worker waits for the
> streaming transaction to complete and the lock to be released, it won't send
> subsequent data of the streaming transaction to the apply background worker;
> the apply background worker waits to receive the rest of streaming transaction
> and can't finish this transaction. Then the main apply worker will wait
> indefinitely.
> 
> "background worker lock a row" -> "background worker locks a row"
> 
> "Then the main apply worker will wait indefinitely." -> really, you
> already said the main apply worker is waiting, so I think this
> sentence only needs to say: "Now a deadlock has occurred, so both
> workers will wait indefinitely."
> 
> 2b.
> 
> Text fragments are all common between:
> 
> i.   This commit message
> ii.  Text in pgdocs CREATE SUBSCRIPTION
> iii. Function comment for 'logicalrep_rel_mark_parallel_apply' in relation.c
> 
> After addressing other review comments please make sure all those 3
> parts are worded same.

Improved.

> 3. doc/src/sgml/ref/create_subscription.sgml
> 
> +          There are two requirements for using <literal>parallel</literal>
> +          mode: 1) the unique column in the table on the subscriber-side should
> +          also be the unique column on the publisher-side; 2) there cannot be
> +          any non-immutable functions used by the subscriber-side replicated
> +          table.
> 
> 3a.
> I am not sure – is "requirements" the correct word here, or maybe it
> should be "prerequisites".
> 
> 3b.
> Is it correct to say "should also be", or should that say "must also be"?

Improved.

> 4. src/backend/replication/logical/applybgworker.c -
> apply_bgworker_relation_check
> 
> + /*
> + * Skip check if not using apply background workers.
> + *
> + * If any worker is handling the streaming transaction, this check needs to
> + * be performed not only in the apply background worker, but also in the
> + * main apply worker. This is because without these restrictions, main
> + * apply worker may block apply background worker, which will cause
> + * infinite waits.
> + */
> + if (!am_apply_bgworker() &&
> + (list_length(ApplyBgworkersFreeList) == list_length(ApplyBgworkersList)))
> + return;
> 
> I struggled a bit to reconcile the comment with the condition. Is the
> !am_apply_bgworker() part of this even needed – isn't the
> list_length() check enough?

We need to check this for apply bgworker. (Both lists are "NIL" in apply
bgworker.)

> 5.
> 
> + /* We are in error mode and should give user correct error. */
> 
> I still [1, #3.4a] don't see the value in saying "should give correct
> error" (e.g. what's the alternative?).
> 
> Maybe instead of that comment it can just say:
> rel->parallel_apply = PARALLEL_APPLY_UNSAFE;

I changed if-statement to report the error:
If 'parallel_apply' isn't 'PARALLEL_APPLY_SAFE', then report the error.

> 6. src/backend/replication/logical/proto.c - RelationGetUniqueKeyBitmap
> 
> + /* Add referenced attributes to idindexattrs */
> + for (i = 0; i < indexRel->rd_index->indnatts; i++)
> + {
> + int attrnum = indexRel->rd_index->indkey.values[i];
> +
> + /*
> + * We don't include non-key columns into idindexattrs
> + * bitmaps. See RelationGetIndexAttrBitmap.
> + */
> + if (attrnum != 0)
> + {
> + if (i < indexRel->rd_index->indnkeyatts &&
> + !bms_is_member(attrnum - FirstLowInvalidHeapAttributeNumber, attunique))
> + attunique = bms_add_member(attunique,
> +    attrnum - FirstLowInvalidHeapAttributeNumber);
> + }
> + }
> 
> There are 2x comments in that code that are referring to
> 'idindexattrs' but I think it is a cut/paste problem because that
> variable name does not even exist in this copied function.

Fixed the comments.

> 7. src/backend/replication/logical/relation.c -
> logicalrep_rel_mark_parallel_apply
> 
> + /* Initialize the flag. */
> + entry->parallel_apply = PARALLEL_APPLY_SAFE;
> 
> I have unsuccessfully repeated the same review comment several times
> [1 #3.8] suggesting that this flag should not be initialized to SAFE.
> IMO the state should remain as UNKNOWN until you are either sure it is
> SAFE, or sure it is UNSAFE. Anyway, I'll give up on this point now;
> let's see what other people think.

Okay, I will follow the relevant comments later.

> 8. src/include/replication/logicalrelation.h
> 
> +/*
> + * States to determine if changes on one relation can be applied using an
> + * apply background worker.
> + */
> +typedef enum ParallelApplySafety
> +{
> + PARALLEL_APPLY_UNKNOWN = 0, /* unknown  */
> + PARALLEL_APPLY_SAFE, /* Can apply changes using an apply background
> +    worker */
> + PARALLEL_APPLY_UNSAFE /* Can not apply changes using an apply
> +    background worker */
> +} ParallelApplySafety;
> +
> 
> I think the values are self-explanatory so the comments for every
> value add nothing here, particularly since the enum itself has a
> comment saying the same thing. I'm not sure if you accidentally missed
> my previous comment [1, #3.12b] about this, or just did not agree with
> it.

Changed.

> 9. .../subscription/t/015_stream.pl
> 
> +# "streaming = parallel" does not support non-immutable functions, so change
> +# the function in the defult expression of column "c".
> +$node_subscriber->safe_psql(
> + 'postgres', qq{
> +ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT
> to_timestamp(1284352323);
> +ALTER SUBSCRIPTION tap_sub SET(streaming = parallel, binary = off);
> +});
> 
> 9a.
> typo "defult"
> 
> 9b.
> The problem with to_timestamp(1284352323) is that it looks like it
> must be some special value, but in fact AFAIK you don't care at all
> what value timestamp this is. I think it would be better here to just
> use to_timestamp(0) or to_timestamp(999) or similar so the number is
> obviously not something of importance.
> 
> ======
> 
> 10. .../subscription/t/016_stream.pl
> 
> +# "streaming = parallel" does not support non-immutable functions, so change
> +# the function in the defult expression of column "c".
> +$node_subscriber->safe_psql(
> + 'postgres', qq{
> +ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT
> to_timestamp(1284352323);
> +ALTER SUBSCRIPTION tap_sub SET(streaming = parallel);
> +});
> 
> 10a. Ditto 9a.
> 10b. Ditto 9b.
> 
> ======
> 
> 11. .../subscription/t/022_twophase_cascade.pl
> 
> +# "streaming = parallel" does not support non-immutable functions, so change
> +# the function in the defult expression of column "c".
> +$node_B->safe_psql(
> + 'postgres', "ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT
> to_timestamp(1284352323);");
> +$node_C->safe_psql(
> + 'postgres', "ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT
> to_timestamp(1284352323);");
> +
> 
> 11a. Ditto 9a.
> 11b. Ditto 9b.
> 
> ======
> 
> 12. .../subscription/t/023_twophase_stream.pl
> 
> +# "streaming = parallel" does not support non-immutable functions, so change
> +# the function in the defult expression of column "c".
> +$node_subscriber->safe_psql(
> + 'postgres', qq{
> +ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT
> to_timestamp(1284352323);
> +ALTER SUBSCRIPTION tap_sub SET(streaming = parallel);
> +});
> 
> 12a. Ditto 9a.
> 12b. Ditto 9b.

Improved.

> 13. .../subscription/t/032_streaming_apply.pl
> 
> +# Drop default value on the subscriber, now it works.
> +$node_subscriber->safe_psql('postgres',
> + "ALTER TABLE test_tab1 ALTER COLUMN b DROP DEFAULT");
> 
> Maybe for these tests like this it would be better to test if it works
> OK using an immutable DEFAULT function instead of just completely
> removing the bad function to make it work.
> 
> I think maybe the same was done for TRIGGER tests. There was a test
> for a trigger with a bad function, and then the trigger was removed.
> What about including a test for the trigger with a good function?

Improved.

Attach the new patches.

Regards,
Wang wei

Commits

Same data as JSON: GET /api/v1/messages/:b64id/commits the thread's linked commits as JSON, with link sources. API reference →
  1. Fix invalid memory access during the shutdown of the parallel apply worker.

  2. Fix assertion failure in apply worker.

  3. Use elog to report unexpected action in handle_streamed_transaction().

  4. Use appropriate wait event when sending data in the apply worker.

  5. Allow the logical_replication_mode to be used on the subscriber.

  6. Rename GUC logical_decoding_mode to logical_replication_mode.

  7. Display the leader apply worker's PID for parallel apply workers.

  8. Improve the code to decide and process the apply action.

  9. Document the newly added wait events added by commit 216a784829.

  10. Perform apply of large transactions by parallel workers.

  11. Wake up a subscription's replication worker processes after DDL.

  12. Add copyright notices to meson files

  13. Better document logical replication parameters

  14. Add a common function to generate the origin name.

  15. Harmonize parameter names in storage and AM code.

  16. Avoid using list_length() to test for empty list.

  17. Improve two comments related to a boolean DefElem's value

  18. Fix partition table's REPLICA IDENTITY checking on the subscriber.

  19. Fix data inconsistency between publisher and subscriber.

  20. Fix cache look-up failures while applying changes in logical replication.