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

On Thu, 19 Jan 2023 at 12:06, Takamichi Osumi (Fujitsu)
<osumi.takamichi@fujitsu.com> wrote:
>
> Updated the comment and the function call.
>
> Kindly have a look at the updated patch v17.

Thanks for the updated patch, few comments:
1) min_apply_delay was accepting values like '600 m s h', I was not
sure if we should allow this:
alter subscription sub1 set (min_apply_delay = ' 600 m s h');

+                       /*
+                        * If no unit was specified, then explicitly
add 'ms' otherwise
+                        * the interval_in function would assume 'seconds'.
+                        */
+                       if (strspn(tmp, "-0123456789 ") == strlen(tmp))
+                               val = psprintf("%sms", tmp);
+                       else
+                               val = tmp;
+
+                       interval =
DatumGetIntervalP(DirectFunctionCall3(interval_in,
+

CStringGetDatum(val),
+

ObjectIdGetDatum(InvalidOid),
+
                                                  Int32GetDatum(-1)));

2) How about adding current_txn_wait_time in
pg_stat_subscription_stats, we can update the current_txn_wait_time
periodically, this will help the user to check approximately how much
time is left(min_apply_delay - stat value) before this transaction
will be applied in the subscription. If you agree this can be 0002
patch.

3) There is one check at parse_subscription_options and another check
in AlterSubscription, this looks like a redundant check in case of
alter subscription, can we try to merge and keep in one place:
/*
* The combination of parallel streaming mode and min_apply_delay is not
* allowed.
*/
if (IsSet(supported_opts, SUBOPT_MIN_APPLY_DELAY) &&
opts->min_apply_delay > 0)
{
if (opts->streaming == LOGICALREP_STREAM_PARALLEL)
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("%s and %s are mutually exclusive options",
   "min_apply_delay > 0", "streaming = parallel"));
}

if (IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY))
{
/*
* The combination of parallel streaming mode and
* min_apply_delay is not allowed.
*/
if (opts.min_apply_delay > 0)
if ((IsSet(opts.specified_opts, SUBOPT_STREAMING) && opts.streaming ==
LOGICALREP_STREAM_PARALLEL) ||
(!IsSet(opts.specified_opts, SUBOPT_STREAMING) && sub->stream ==
LOGICALREP_STREAM_PARALLEL))
ereport(ERROR,
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("cannot enable %s for subscription in %s mode",
   "min_apply_delay", "streaming = parallel"));

values[Anum_pg_subscription_subminapplydelay - 1] =
Int64GetDatum(opts.min_apply_delay);
replaces[Anum_pg_subscription_subminapplydelay - 1] = true;
}

4) typo "execeeds" should be "exceeds"

+          time on the subscriber. Any overhead of time spent in
logical decoding
+          and in transferring the transaction may reduce the actual wait time.
+          It is also possible that the overhead already execeeds the requested
+          <literal>min_apply_delay</literal> value, in which case no additional
+          wait is necessary. If the system clocks on publisher and subscriber
+          are not synchronized, this may lead to apply changes earlier than

Regards,
Vignesh