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

Here are my review comments for the patch v25-0001.

======
Commit Message

1.
The other possibility is to apply the delay at the end of the parallel
apply transaction but that would cause issues related to resource
bloat and locks being held for a long time.

~

SUGGESTION
We chose not to apply the delay at the end of the parallel apply
transaction because that would cause issues related to resource bloat
and locks being held for a long time.

======
doc/src/sgml/config.sgml

2.
+ <para>
+ For time-delayed logical replication, the apply worker sends a feedback
+ message to the publisher every
+ <varname>wal_receiver_status_interval</varname> milliseconds. Make sure
+ to set <varname>wal_receiver_status_interval</varname> less than the
+ <varname>wal_sender_timeout</varname> on the publisher, otherwise, the
+ <literal>walsender</literal> will repeatedly terminate due to timeout
+ error. Note that if <varname>wal_receiver_status_interval</varname> is
+ set to zero, the apply worker sends no feedback messages during the
+ <literal>min_apply_delay</literal> period.
+ </para>

2a.
"due to timeout error." --> "due to timeout errors."

~

2b.
Shouldn't this also cross-ref to CREATE SUBSCRIPTION docs? Because the
above mentions 'min_apply_delay' but that is not defined on this page.

======
doc/src/sgml/ref/create_subscription.sgml

3.
+ <para>
+ By default, the subscriber applies changes as soon as possible. This
+ parameter allows the user to delay the application of changes by a
+ given time period. If the value is specified without units, it is
+ taken as milliseconds. The default is zero (no delay). See
+ <xref linkend="config-setting-names-values"/> for details on the
+ available valid time unites.
+ </para>

Typo: "unites"

~~~

4.
+ <para>
+ Any delay becomes effective after all initial table synchronization
+ has finished and occurs before each transaction starts to get applied
+ on the subscriber. The delay is calculated as the difference between
+ the WAL timestamp as written on the publisher and the current 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 exceeds the requested
+ <literal>min_apply_delay</literal> value, in which case no delay is
+ applied. If the system clocks on publisher and subscriber are not
+ synchronized, this may lead to apply changes earlier than expected,
+ but this is not a major issue because this parameter is typically
+ much larger than the time deviations between servers. Note that if
+ this parameter is set to a long delay, the replication will stop if
+ the replication slot falls behind the current LSN by more than
+ <link
linkend="guc-max-slot-wal-keep-size"><literal>max_slot_wal_keep_size</literal></link>.
+ </para>

"Any delay becomes effective after all initial table
synchronization..." --> "Any delay becomes effective only after all
initial table synchronization..."

~~~

5.
+ <warning>
+ <para>
+ Delaying the replication means there is a much longer time between
+ making a change on the publisher, and that change being committed
+ on the subscriber. This can impact the performance of synchronous
+ replication. See <xref linkend="guc-synchronous-commit"/>
+ parameter.
+ </para>
+ </warning>

I'm not sure why this was text changed to say "means there is a much
longer time" instead of "can mean there is a much longer time".

IMO the previous wording was better because this current text makes an
assumption about what the user has configured -- e.g. if they
configured only 1ms delay then the warning text is not really
relevant.

~~~

6.
Why was the example (it existed when I last looked at patch v19)
removed? Personally, I found that example to be a useful reminder that
the min_apply_delay can specify units other than just 'ms'.

======
src/backend/commands/subscriptioncmds.c

7. parse_subscription_options

+ /*
+ * The combination of parallel streaming mode and min_apply_delay is not
+ * allowed. This is because we start applying the transaction stream as
+ * soon as the first change arrives without knowing the transaction's
+ * prepare/commit time. This means we cannot calculate the underlying
+ * network/decoding lag between publisher and subscriber, and so always
+ * waiting for the full 'min_apply_delay' period might include unnecessary
+ * delay.
+ *
+ * The other possibility is to apply the delay at the end of the parallel
+ * apply transaction but that would cause issues related to resource bloat
+ * and locks being held for a long time.
+ */

I think the 2nd paragraph should be changed slightly as follows (like
review comment #1)

SUGGESTION
Note - we chose not to apply the delay at the end of the parallel
apply transaction because that would cause issues related to resource
bloat and locks being held for a long time.

~~~

8.
+ if (IsSet(supported_opts, SUBOPT_MIN_APPLY_DELAY) &&
+ opts->min_apply_delay > 0 && opts->streaming == LOGICALREP_STREAM_PARALLEL)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),

Saying "> 0" (in the condition) is not strictly necessary here, since
it is never < 0.

~~~

9. AlterSubscription

+ /*
+ * The combination of parallel streaming mode and
+ * min_apply_delay is not allowed. See
+ * parse_subscription_options for details of the reason.
+ */
+ if (opts.streaming == LOGICALREP_STREAM_PARALLEL)
+ if ((IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY) &&
opts.min_apply_delay > 0) ||
+ (!IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY) &&
sub->minapplydelay > 0))

Saying "> 0" (in the condition) is not strictly necessary here, since
it is never < 0.

~~~

10.
+ 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)

Saying "> 0" (in the condition) is not strictly necessary here, since
it is never < 0.

~~~

11. defGetMinApplyDelay

+ /*
+ * Check lower bound. parse_int() has already been confirmed that result
+ * is less than or equal to INT_MAX.
+ */

The parse_int already checks < INT_MAX. But on return from that
function, don’t you need to check again that it is < PG_INT32_MAX (in
case those are different)

(I think Kuroda-san already suggested same as this)

======
src/backend/replication/logical/worker.c

12.
+/*
+ * In order to avoid walsender timeout for time-delayed logical replication the
+ * apply worker keeps sending feedback messages during the delay period.
+ * Meanwhile, the feature delays the apply before the start of the
+ * transaction and thus we don't write WAL records for the suspended changes
+ * during the wait. When the apply worker sends a feedback message during the
+ * delay, we should not make positions of the flushed and apply LSN overwritten
+ * by the last received latest LSN. See send_feedback() for details.
+ */

"we should not make positions of the flushed and apply LSN
overwritten" --> "we should overwrite positions of the flushed and
apply LSN"

~~~

14. send_feedback

@@ -3738,8 +3867,15 @@ send_feedback(XLogRecPtr recvpos, bool force,
bool requestReply)
/*
* No outstanding transactions to flush, we can report the latest received
* position. This is important for synchronous replication.
+ *
+ * If the logical replication subscription has unprocessed changes then do
+ * not inform the publisher that the received latest LSN is already
+ * applied and flushed, otherwise, the publisher will make a wrong
+ * assumption about the logical replication progress. Instead, it just
+ * sends a feedback message to avoid a replication timeout during the
+ * delay.
*/

"Instead, it just sends" --> "Instead, just send"

======
src/bin/pg_dump/pg_dump.h

15. SubscriptionInfo

@@ -661,6 +661,7 @@ typedef struct _SubscriptionInfo
char *subdisableonerr;
char *suborigin;
char *subsynccommit;
+ int subminapplydelay;
char *subpublications;
} SubscriptionInfo;

Should this also be "int32" to match the other member type changes?

======
src/test/subscription/t/032_apply_delay.pl

16.
+# Make sure the apply worker knows to wait for more than 500ms
+check_apply_delay_log($node_subscriber, $offset, "0.5");

"knows to wait for more than" --> "waits for more than"

(this occurs in a couple of places)

------
Kind Regards,
Peter Smith.
Fujitsu Australia