Re: Time delayed LR (WAS Re: logical replication restrictions)

On Tue, Feb 7, 2023 at 10:13 AM Kyotaro Horiguchi
<horikyota.ntt@gmail.com> wrote:
>
> At Mon, 6 Feb 2023 13:10:01 +0000, "Takamichi Osumi (Fujitsu)" <osumi.takamichi@fujitsu.com> wrote in
> > The attached patch v29 has included your changes.
>
> catalogs.sgml
>
> +      <para>
> +       The minimum delay (ms) for applying changes.
> +      </para></entry>
>
> I think we don't use unit symbols that way. Namely I think we would
> write it as "The minimum delay for applying changes in milliseconds"
>

Okay, if we prefer to use milliseconds, then how about: "The minimum
delay, in milliseconds, for applying changes"?

>
> alter_subscription.sgml
>
>        are <literal>slot_name</literal>,
>        <literal>synchronous_commit</literal>,
>        <literal>binary</literal>, <literal>streaming</literal>,
> -      <literal>disable_on_error</literal>, and
> -      <literal>origin</literal>.
> +      <literal>disable_on_error</literal>,
> +      <literal>origin</literal>, and
> +      <literal>min_apply_delay</literal>.
>       </para>
>
> By the way, is there any rule for the order among the words?
>

Currently, it is in the order in which the corresponding features are added.

> They
> don't seem in alphabetical order nor in the same order to the
> create_sbuscription page.
>

In create_subscription page also, it appears to be in the order in
which those are added with a difference that they are divided into two
categories (parameters that control what happens during subscription
creation and parameters that control the subscription's replication
behavior after it has been created)

>  (I seems like in the order of SUBOPT_*
> symbols, but I'm not sure it's a good idea..)
>
>
> subscriptioncmds.c
>
> +                                       if (opts.streaming == LOGICALREP_STREAM_PARALLEL &&
> +                                               !IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY) &&
sub->minapplydelay> 0)
 
> ..
> +                                       if (opts.min_apply_delay > 0 &&
> +                                               !IsSet(opts.specified_opts, SUBOPT_STREAMING) && sub->stream ==
LOGICALREP_STREAM_PARALLEL)
>
> Don't we wrap the lines?
>
>
> worker.c
>
> +               if (wal_receiver_status_interval > 0 &&
> +                       diffms > wal_receiver_status_interval * 1000L)
> +               {
> +                       WaitLatch(MyLatch,
> +                                         WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
> +                                         wal_receiver_status_interval * 1000L,
> +                                         WAIT_EVENT_RECOVERY_APPLY_DELAY);
> +                       send_feedback(last_received, true, false, true);
> +               }
> +               else
> +                       WaitLatch(MyLatch,
> +                                         WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
> +                                         diffms,
> +                                         WAIT_EVENT_RECOVERY_APPLY_DELAY);
>
> send_feedback always handles the case where
> wal_receiver_status_interval == 0.
>

It only handles when force is false but here we are using that as
true. So, not sure, if what you said would be an improvement.

>  thus we can simply wait for
> min(wal_receiver_status_interval, diffms) then call send_feedback()
> unconditionally.
>
>
> -start_apply(XLogRecPtr origin_startpos)
> +start_apply(void)
>
> -LogicalRepApplyLoop(XLogRecPtr last_received)
> +LogicalRepApplyLoop(void)
>
> Does this patch requires this change?
>

I think this is because the scope of last_received has been changed so
that it can be used to pass in send_feedback() during the delay.

-- 
With Regards,
Amit Kapila.