Horton
Menu

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

Zhijie Hou (Fujitsu) <houzj.fnst@fujitsu.com>

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

Attachments

On Wednesday, December 7, 2022 6:49 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
> 
> On Wed, Dec 7, 2022 at 8:28 AM houzj.fnst@fujitsu.com
> <houzj.fnst@fujitsu.com> wrote:
> >
> > Besides, I fixed a bug where there could still be messages left in
> > memory queue and the PA has started to apply spooled message.
> >
> 
> Few comments on the recent changes in the patch:
> ========================================
> 1. It seems you need to set FS_SERIALIZE_DONE in
> stream_prepare/commit/abort. They are still directly setting the state as
> READY. Am, I missing something or you forgot to change it?

It's my miss, changed.

> 2.
>   case TRANS_PARALLEL_APPLY:
>   pa_stream_abort(&abort_data);
> 
> + /*
> + * Reset the stream_fd after aborting the toplevel transaction in
> + * case the parallel apply worker is applying spooled messages */ if
> + (toplevel_xact) stream_fd = NULL;
> 
> I think we can keep the handling of stream file the same in
> abort/commit/prepare code path.

Changed.

> 3. It is already pointed out by Peter that it is better to add some comments in
> pa_spooled_messages() function that we won't be immediately able to apply
> changes after the lock is released, it will be done in the next cycle.

Added.

> 4. Shall we rename FS_SERIALIZE as FS_SERIALIZE_IN_PROGRESS? That will
> appear consistent with FS_SERIALIZE_DONE.

Agreed, changed.

> 5. Comment improvements:
> diff --git a/src/backend/replication/logical/worker.c
> b/src/backend/replication/logical/worker.c
> index b26d587ae4..921d973863 100644
> --- a/src/backend/replication/logical/worker.c
> +++ b/src/backend/replication/logical/worker.c
> @@ -1934,8 +1934,7 @@ apply_handle_stream_abort(StringInfo s)  }
> 
>  /*
> - * Check if the passed fileno and offset are the last fileno and position of
> - * the fileset, and report an ERROR if not.
> + * Ensure that the passed location is fileset's end.
>   */
>  static void
>  ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno, @@
> -2084,9 +2083,9 @@ apply_spooled_messages(FileSet *stream_fileset,
> TransactionId xid,
>                 nchanges++;
> 
>                 /*
> -                * Break the loop if stream_fd is set to NULL which
> means the parallel
> -                * apply worker has finished applying the transaction.
> The parallel
> -                * apply worker should have closed the file before committing.
> +                * It is possible the file has been closed because we
> have processed
> +                * some transaction end message like stream_commit in
> which case that
> +                * must be the last message.
>                  */

Merged, thanks.

Attach the new version patch which addressed all above comments and part of
comments from[1] except some comments that are being discussed.

Apart from above, according to the comment from Amit and Sawada-san[2], the new
version patch won't stop the parallel worker due to subscription parameter
change, it will absorb the change instead, and the leader will anyway detect
the parameter change and stop all workers later.

Based on this, I also removed the maybe_reread_subscription() call in parallel
apply worker's main loop, because we need to make sure we won't update the local
subscription parameter in the middle of the transaction. And we will call
maybe_reread_subscription() before starting a transaction in parallel apply
worker anyway(in maybe_reread_subscription()), so remove that check is fine and
can save some codes.

[1] https://www.postgresql.org/message-id/CAD21AoCZ3i9w1Rz-81Lv1QB%2BJGP60Ypiom4%2BwM9eP3aQTx0STQ%40mail.gmail.com
[2] https://www.postgresql.org/message-id/CAD21AoAzYstJVM0nMVnXZoeYamqD2j92DkWVH%3DYbGtA4yzy19A%40mail.gmail.com

Best regards,
Hou zj

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.