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

On Thursday, May 5, 2022 1:46 PM Peter Smith <smithpb2250@gmail.com> wrote:

> Here are my review comments for v5-0001.
> I will take a look at the v5-0002 (TAP) patch another time.

Thanks for the comments !

> 4. Commit message
> 
> User can set the streaming option to 'on/off', 'apply'. For now,
> 'apply' means the streaming will be applied via a apply background if
> available. 'on' means the streaming transaction will be spilled to
> disk.
> 
> 
> I think "apply" might not be the best choice of values for this
> meaning, but I think Hou-san already said [1] that this was being
> reconsidered.

Yes, I am thinking over this along with some other related stuff[1] posted by Amit
and sawada. Will change this in next version.

[1] https://www.postgresql.org/message-id/flat/CAA4eK1%2B7D4qAQUQEE8zzQ0fGCqeBWd3rzTaY5N0jVs-VXFc_Xw%40mail.gmail.com

> 7. src/backend/commands/subscriptioncmds.c - defGetStreamingMode
> 
> +static char
> +defGetStreamingMode(DefElem *def)
> +{
> + /*
> + * If no parameter given, assume "true" is meant.
> + */
> + if (def->arg == NULL)
> + return SUBSTREAM_ON;
> 
> But is that right? IIUC all the docs said that the default is OFF.

I think it's right. "arg == NULL" means user specify the streaming option
without the value. Like CREATE SUBSCRIPTION xxx WITH(streaming). The value should
be 'on' in this case.


> 12. src/backend/replication/logical/origin.c - replorigin_session_setup
> 
> @@ -1110,7 +1110,11 @@ replorigin_session_setup(RepOriginId node)
>   if (curstate->roident != node)
>   continue;
> 
> - else if (curstate->acquired_by != 0)
> + /*
> + * We allow the apply worker to get the slot which is acquired by its
> + * leader process.
> + */
> + else if (curstate->acquired_by != 0 && acquire)
> 
> I still feel this is overly-cofusing. Shouldn't comment say "Allow the
> apply bgworker to get the slot...".
> 
> Also the parameter name 'acquire' is hard to reconcile with the
> comment. E.g. I feel all this would be easier to understand if the
> param was  was refactored with a name like 'bgworker' and the code was
> changed to:
> else if (curstate->acquired_by != 0 && !bgworker)
> 
> Of course, the value true/false would need to be flipped on calls too.
> This is the same as my previous comment [PSv4] #26.

I feel it's not good idea to mention bgworker in origin.c. I have remove this
comment and add some other comments in worker.c.

> 26. src/backend/replication/logical/worker.c - apply_handle_stream_abort
> 
> + if (found)
> + {
> + elog(LOG, "rolled back to savepoint %s", spname);
> + RollbackToSavepoint(spname);
> + CommitTransactionCommand();
> + subxactlist = list_truncate(subxactlist, i + 1);
> + }
> 
> Should that elog use the "[Apply BGW #%u]" format like the others for BGW?

I feel the "[Apply BGW #%u]" is a bit hacky and some of them comes from the old
patchset. I will recheck these logs and adjust them and change some log
level in next version.

> 27. src/backend/replication/logical/worker.c - apply_handle_stream_abort
> 
> Should this function be setting stream_apply_worker = NULL somewhere
> when all is done?
> 29. src/backend/replication/logical/worker.c - apply_handle_stream_commit
> 
> I am unsure, but should something be setting the stream_apply_worker =
> NULL somewhere when all is done?

I think the worker already be set to NULL in apply_handle_stream_stop.


> 32. src/backend/replication/logical/worker.c - ApplyBgwShutdown
> 
> +/*
> + * Set the failed flag so that the main apply worker can realize we have
> + * shutdown.
> + */
> +static void
> +ApplyBgwShutdown(int code, Datum arg)
> 
> If the 'code' param is deliberately unused it might be better to say
> so in the comment...

Not sure about this. After searching the codes, I think most of the callback
functions doesn't use and add comments for the 'code' param.


> 45. src/backend/utils/activity/wait_event.c
> 
> @@ -388,6 +388,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
>   case WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT:
>   event_name = "HashGrowBucketsReinsert";
>   break;
> + case WAIT_EVENT_LOGICAL_APPLY_WORKER_READY:
> + event_name = "LogicalApplyWorkerReady";
> + break;
> 
> I am not sure this is the best name for this event since the only
> place it is used (in apply_bgworker_wait_for) is not only waiting for
> READY state. Maybe a name like WAIT_EVENT_LOGICAL_APPLY_BGWORKER or
> WAIT_EVENT_LOGICAL_APPLY_WORKER_SYNC would be more appropriate? Need
> to change the wait_event.h also.

I noticed a similar named "WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE", so I changed
this to WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE.

> 47. src/test/regress/expected/subscription.out - missting test
> 
> Missing some test cases for all new option values? E.g. Where is the
> test using streaming value is set to 'apply'. Same comment as [PSv4]
> #81

The new option is tested in the second patch posted by Shi yu.

I addressed other comments from Peter and the 2PC related comment from Shi.
Here is the version patch.

Best regards,
Hou zj