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

From: Kyotaro Horiguchi <[email-masked]>
Date: Thu, 12 Jan 2023 12:03:38 +0900 (JST)

At Wed, 11 Jan 2023 12:46:24 +0000, "Hayato Kuroda (Fujitsu)" <[email-masked]> wrote in 
> them. Which version is better?

Some comments by a quick loock, different from the above.

+         CONNECTION 'host=192.168.1.50 port=5432 user=foo dbname=foodb'

I understand that we (not PG people, but IT people) are supposed to
use in documents a certain set of special addresses that is guaranteed
not to be routed in the field.

> TEST-NET-1 : 192.0.2.0/24
> TEST-NET-2 : 198.51.100.0/24
> TEST-NET-3 : 203.0.113.0/24

(I found 192.83.123.89 in the postgres_fdw doc, but it'd be another issue..)

+			if (strspn(tmp, "-0123456789 ") == strlen(tmp))

Do we need to bother spending another memory block for apparent
non-digits here?

+						errmsg(INT64_FORMAT " ms is outside the valid range for parameter \"%s\"",

We don't use INT64_FORMAT in translatable message strings. Cast then
use %lld instead.

This message looks unfriendly as it doesn't suggest the valid range,
and it shows the input value in a different unit from what was in the
input. A I think we can spell it as "\"%s\" is outside the valid range
for subsciription parameter \"%s\" (0 .. <INT_MAX> in millisecond)"

+	int64		min_apply_delay;
..
+			if (ms < 0 || ms > INT_MAX)

Why is the variable wider than required?

+					errmsg("%s and %s are mutually exclusive options",
+						   "min_apply_delay > 0", "streaming = parallel"));

Mmm. Couldn't we refuse 0 as min_apply_delay?

+						sub->minapplydelay > 0)
...
+					if (opts.min_apply_delay > 0 &&

Is there any reason for the differenciation?

+								errmsg("cannot set %s for subscription with %s",
+									   "streaming = parallel", "min_apply_delay > 0"));

I think that this shoud be more like human-speking. Say, "cannot
enable min_apply_delay for subscription in parallel streaming mode" or
something..  The same is applicable to the nearby message.

+static void maybe_delay_apply(TimestampTz ts);

  apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
-					   XLogRecPtr lsn)
+					   XLogRecPtr lsn, TimestampTz ts)

"ts" looks too generic. Couldn't it be more specific?
We need a explanation for the parameter in the function comment.

+	if (!am_parallel_apply_worker())
+	{
+		Assert(ts > 0);
+		maybe_delay_apply(ts);

It seems to me better that the if condition and assertion are checked
inside maybe_delay_apply().

regards.

-- 
Kyotaro Horiguchi
NTT Open Source Software Center

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Thu, 12 Jan 2023 15:39:23 +0000

> Some comments by a quick loock, different from the above.
Horiguchi-san, thanks for your review !

> I understand that we (not PG people, but IT people) are supposed to use in
> documents a certain set of special addresses that is guaranteed not to be
> routed in the field.
> 
Fixed. If necessary we can create another thread for this.

> Do we need to bother spending another memory block for apparent non-digits
> here?
Yes. The characters are necessary to handle an issue reported in [1].
The issue happened if the user inputs a negative value,
then the length comparison became different between strspn and strlen
and the input value was recognized as seconds, when
the unit wasn't described. This led to a wrong error message for the user.

Those addition of such characters solve the issue.

> +						errmsg(INT64_FORMAT " ms
> is outside the valid range for parameter
> +\"%s\"",
> 
Thanks for teaching us. Fixed.

> This message looks unfriendly as it doesn't suggest the valid range, and it
> shows the input value in a different unit from what was in the input. A I think we
> can spell it as "\"%s\" is outside the valid range for subsciription parameter
> \"%s\" (0 .. <INT_MAX> in millisecond)"
Makes sense. I incorporated the valid range with the aligned format of recovery_min_apply_delay.
FYI, the physical replication's GUC doesn't write the unites for the range like below.
I followed and applied this style.

---
LOG:  -1 ms is outside the valid range for parameter "recovery_min_apply_delay" (0 .. 2147483647)
FATAL:  configuration file "/home/k5user/new/pg/l/make_v15/slave/postgresql.conf" contains errors
---

> Why is the variable wider than required?
You are right. Fixed.

> Mmm. Couldn't we refuse 0 as min_apply_delay?
Sorry, the previous patch's behavior wasn't consistent with this error message.

In the previous patch, if we conducted alter subscription
with stream = parallel and min_apply_delay = 0 (from a positive value) at the same time,
the alter command failed, although this should succeed by this time-delayed feature specification.
We fixed this part accordingly by some more tests in AlterSubscription().

By the way, we should allow users to change min_apply_dealy to 0
whenever they want from different value. Then, we didn't restrict
this kind of operation.

> Is there any reason for the differenciation?
Yes. The former is the object for an existing subscription configuration.
For example, if we alter subscription with setting streaming = 'parallel'
for a subscription created with min_apply_delay = '1 day', we
need to reject the alter command. The latter is new settings.

> +
> 	errmsg("cannot set %s for subscription with %s",
> +
> "streaming = parallel", "min_apply_delay > 0"));
> 
> I think that this shoud be more like human-speking. Say, "cannot enable
> min_apply_delay for subscription in parallel streaming mode" or something..
> The same is applicable to the nearby message.
Reworded the error messages. Please check.

> We need a explanation for the parameter in the function comment.
Changed it to finish_ts, since it indicates commit/prepare time.
This terminology should be aligned with finish lsn.

> It seems to me better that the if condition and assertion are checked inside
> maybe_delay_apply().
Fixed.

[1] - https://www.postgresql.org/message-id/CALDaNm3Bpzhh60nU-keuGxMPb-OhcqsfpCN3ysfCfCJ-2ShYPA%40mail.gmail.com

Best Regards,
	Takamichi Osumi

---

From: vignesh C <[email-masked]>
Date: Sat, 14 Jan 2023 11:57:29 +0530

> > It seems to me better that the if condition and assertion are checked inside

Thanks for the updated patch, Few comments:
1) Since the min_apply_delay = 3, but you have specified 2s, there
might be a possibility that it can log delay as 1000ms due to
pub/sub/network delay and the test can fail randomly, If we cannot
ensure this log file value, check_apply_delay_time verification alone
should be sufficient.
+is($result, qq(5|1|5), 'check if the new rows were applied to subscriber');
+
+check_apply_delay_log("logical replication apply delay", "2000");

2) I'm not sure if this will add any extra coverage as the altering
value of min_apply_delay is already tested in the regression, if so
this test can be removed:
+# Test ALTER SUBSCRIPTION. Delay 86460 seconds (1 day 1 minute).
+$node_subscriber->safe_psql('postgres',
+       "ALTER SUBSCRIPTION tap_sub SET (min_apply_delay = 86460000)"
+);
+
+# New row to trigger apply delay.
+$node_publisher->safe_psql('postgres',
+       "INSERT INTO test_tab VALUES (0, 'foobar')");
+
+check_apply_delay_log("logical replication apply delay", "80000000");

3) We generally keep the subroutines before the tests, it can be kept
accordingly:
3.a)
+sub check_apply_delay_log
+{
+       my ($message, $expected) = @_;
+       $expected = 0 unless defined $expected;
+
+       my $old_log_location = $log_location;

3.b)
+sub check_apply_delay_time
+{
+       my ($primary_key, $expected_diffs) = @_;
+
+       my $inserted_time_on_pub = $node_publisher->safe_psql('postgres', qq[
+               SELECT extract(epoch from c) FROM test_tab WHERE a =
$primary_key;
+       ]);
+

4) typo "more then once" should be "more than once"
+ regress_testsub | regress_subscription_user | f       |
{testpub,testpub1,testpub2} | f      | off       | d                |
f                | any    |                    0 | off
| dbname=regress_doesnotexist | 0/0
 (1 row)

 -- fail - publication used more then once
@@ -316,10 +316,10 @@ ERROR:  publication "testpub3" is not in
subscription "regress_testsub"
 -- ok - delete publications
 ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1,
testpub2 WITH (refresh = false);
 \dRs+

5) This can be changed to "Is it larger than expected?"
+               # Is it larger than expected ?
+               cmp_ok($logged_delay, '>', $expected,
+                       "The wait time of the apply worker is long
enough expectedly"
+               );

6) 2022 should be changed to 2023
+++ b/src/test/subscription/t/032_apply_delay.pl
@@ -0,0 +1,170 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+# Test replication apply delay

7) Termination full stop is not required for single line comments:
7.a)
+use Test::More;
+
+# Create publisher node.
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');

7.b) +
+# Create subscriber node.
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');

7.c) +
+# Create some preexisting content on publisher.
+$node_publisher->safe_psql('postgres',

7.d) similarly in rest of the files

8) Is it possible to add one test for spooling also?

Regards,
Vignesh

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Tue, 17 Jan 2023 11:00:11 +0000

Hi,

On Saturday, January 14, 2023 3:27 PM vignesh C <[email-masked]> wrote:
> 1) Since the min_apply_delay = 3, but you have specified 2s, there might be a
> possibility that it can log delay as 1000ms due to pub/sub/network delay and
> the test can fail randomly, If we cannot ensure this log file value,
> check_apply_delay_time verification alone should be sufficient.
> +is($result, qq(5|1|5), 'check if the new rows were applied to
> +subscriber');
> +
You are right. Removed the left-time check of the 1st call of check_apply_delay_log(). 

> 2) I'm not sure if this will add any extra coverage as the altering value of
> min_apply_delay is already tested in the regression, if so this test can be
> removed:
While addressing this point, I've noticed that there is a
behavior difference between physical replication's recovery_min_apply_delay
and this feature when stopping the replication during delays.

At present, in the latter case,
the apply worker exits without applying the suspended transaction
after ALTER SUBSCRIPTION DISABLE command for the subscription.
Meanwhile, there is no "disabling" command for physical replication,
but I checked the behavior about what happens for promoting a secondary
during the delay of recovery_min_apply_delay for physical replication as one example.
The transaction has become visible even in the promoting in the middle of delay.

I'm not sure if I should make the time-delayed LR aligned with this behavior.
Does someone has an opinion for this ?

By the way, the above test code can be used for the test case
when the apply worker is in a delay but the transaction has been canceled by
ALTER SUBSCRIPTION DISABLE command. So, I didn't remove it at this stage.
Fixed.

>  -- fail - publication used more then once @@ -316,10 +316,10 @@ ERROR:
> publication "testpub3" is not in subscription "regress_testsub"
This was an existing typo on HEAD. Addressed in other thread in [1].

Fixed.
 
Fixed.

> 7.d) similarly in rest of the files
Removed the periods for single line comments.

> 8) Is it possible to add one test for spooling also?
There is a streaming transaction case in the TAP test already.

I conducted some minor comment modifications along with above changes.
Kindly have a look at the v16.

[1] - https://www.postgresql.org/message-id/flat/TYCPR01MB83737EA140C79B7D099F65E8EDC69%40TYCPR01MB8373.jpnprd01.prod.outlook.com

Best Regards,
	Takamichi Osumi

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 17 Jan 2023 12:45:14 +0000

Dear hackers,

> Does someone has an opinion for this ?

I put my opinion here. The current specification is correct; we should not follow
a physical replication manner.
One motivation for this feature is to offer opportunities to correct data loss
errors. When accidental delete events occur, DBA can stop propagations on subscribers
by disabling the subscription, with the patch at present.
IIUC, when the subscription is disabled before transactions are started,
workers exit and stop applications. This feature delays starting txns, so we
should regard such an alternation as that is executed before the transaction.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Peter Smith <[email-masked]>
Date: Wed, 18 Jan 2023 18:06:17 +1100

Here are my review comments for the latest patch v16-0001. (excluding
the test code)

======

General

1.

Since the value of min_apply_delay cannot be < 0,  I was thinking
probably it should have been declared everywhere in this patch as a
uint64 instead of an int64, right?

======

Commit message

2.

If the subscription sets min_apply_delay parameter, the logical
replication worker will delay the transaction commit for min_apply_delay
milliseconds.

~

IMO there should be another sentence before this just to say that a
new parameter is being added:

e.g.
This patch implements a new subscription parameter called 'min_apply_delay'.

======

doc/src/sgml/config.sgml

3.

+      <para>
+       For time-delayed logical replication, the apply worker sends a Standby
+       Status Update message to the corresponding publisher per the indicated
+       time of this parameter. Therefore, if this parameter is longer than
+       <literal>wal_sender_timeout</literal> on the publisher, then the
+       walsender doesn't get any update message during the delay and repeatedly
+       terminates due to the timeout errors. Hence, make sure this parameter is
+       shorter than the <literal>wal_sender_timeout</literal> of the publisher.
+       If this parameter is set to zero with time-delayed replication, the
+       apply worker doesn't send any feedback messages during the
+       <literal>min_apply_delay</literal>.
+      </para>

This paragraph seemed confusing. I think it needs to be reworded to
change all of the "this parameter" references because there are at
least 3 different parameters mentioned in this paragraph. e.g. maybe
just change them to explicitly name the parameter you are talking
about.

I also think it needs to mention the ‘min_apply_delay’ subscription
parameter up-front and then refer to it appropriately.

The end result might be something like I wrote below (this is just my
guess – probably you can word it better).

SUGGESTION
For time-delayed logical replication (i.e. when the subscription is
created with parameter min_apply_delay > 0), the apply worker sends a
Standby Status Update message to the publisher with a period of
wal_receiver_status_interval . Make sure to set
wal_receiver_status_interval less than the wal_sender_timeout on the
publisher, otherwise, the walsender will repeatedly terminate due to
the timeout errors. If wal_receiver_status_interval is set to zero,
the apply worker doesn't send any feedback messages during the
subscriber’s min_apply_delay period.

======

doc/src/sgml/ref/create_subscription.sgml

4.

+         <para>
+          By default, the subscriber applies changes as soon as possible. As
+          with the physical replication feature
+          (<xref linkend="guc-recovery-min-apply-delay"/>), it can be useful to
+          have a time-delayed logical replica. This parameter lets the user to
+          delay the application of changes by a specified amount of
time. If this
+          value is specified without units, it is taken as milliseconds. The
+          default is zero(no delay).
+         </para>

4a.
As with the physical replication feature (recovery_min_apply_delay),
it can be useful to have a time-delayed logical replica.

IMO not sure that the above sentence is necessary. It seems only to be
saying that this parameter can be useful. Why do we need to say that?

~

4b.
"This parameter lets the user to delay" -> "This parameter lets the user delay"
OR
"This parameter lets the user to delay" -> "This parameter allows the
user to delay"

~

4c.
"If this value is specified without units" -> "If the value is
specified without units"

~

4d.
"zero(no delay)." -> "zero (no delay)."

----

5.

+         <para>
+          The delay occurs only on WAL records for transaction begins and after
+          the initial table synchronization. It is possible that the
+          replication delay between publisher and subscriber exceeds the value
+          of this parameter, in which case no delay is added. Note that the
+          delay is calculated between the WAL time stamp as written on
+          publisher and the current time on the subscriber. Time
spent in logical
+          decoding and in transferring the transaction may reduce the
actual wait
+          time. 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>

I think the first part can be reworded slightly. See what you think
about the suggestion below.

SUGGESTION
Any delay occurs only on WAL records for transaction begins after all
initial table synchronization has finished.  The delay is calculated
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
'min_apply_delay' value, in which case no additional wait is
necessary. If the system clocks...

----

6.

+  <para>
+   Setting streaming to <literal>parallel</literal> mode and
<literal>min_apply_delay</literal>
+   simultaneously is not supported.
+  </para>

SUGGESTION
A non-zero min_apply_delay parameter is not allowed when streaming in
parallel mode.

======

src/backend/commands/subscriptioncmds.c

7. parse_subscription_options

@@ -404,6 +445,17 @@ parse_subscription_options(ParseState *pstate,
List *stmt_options,
  "slot_name = NONE", "create_slot = false")));
  }
  }
+
+ /* Test the combination of streaming mode and min_apply_delay */
+ 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"));
+ }

SUGGESTION (comment)
The combination of parallel streaming mode and min_apply_delay is not allowed.

~~~

8. AlterSubscription (general)

I observed during testing there are 3 different errors….

At subscription CREATE time you can get this error:
ERROR:  min_apply_delay > 0 and streaming = parallel are mutually
exclusive options

If you try to ALTER the min_apply_delay when already streaming =
parallel you can get this error:
ERROR:  cannot enable min_apply_delay for subscription in streaming =
parallel mode

If you try to ALTER the streaming to be parallel if there is already a
min_apply_delay > 0 then you can get this error:
ERROR:  cannot enable streaming = parallel mode for subscription with
min_apply_delay

~

IMO there is no need to have 3 different error message texts.  I think
all these cases are explained by just the first text (ERROR:
min_apply_delay > 0 and streaming = parallel are mutually exclusive
options)

~~~

9. AlterSubscription

@@ -1098,6 +1152,18 @@ AlterSubscription(ParseState *pstate,
AlterSubscriptionStmt *stmt,

  if (IsSet(opts.specified_opts, SUBOPT_STREAMING))
  {
+ /*
+ * Test the combination of streaming mode and
+ * min_apply_delay
+ */
+ 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))
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot enable %s mode for subscription with %s",
+    "streaming = parallel", "min_apply_delay"));
+

9a.
SUGGESTION (comment)
The combination of parallel streaming mode and min_apply_delay is not allowed.

~

9b.
(see AlterSubscription general review comment #8 above)
Here you can use the same comment error message that says
min_apply_delay > 0 and streaming = parallel are mutually exclusive
options.

~~~

10. AlterSubscription

@@ -1111,6 +1177,25 @@ AlterSubscription(ParseState *pstate,
AlterSubscriptionStmt *stmt,
  = true;
  }

+ if (IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY))
+ {
+ /*
+ * Test the combination of streaming mode and
+ * min_apply_delay
+ */
+ 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;
+ }

10a.
SUGGESTION (comment)
The combination of parallel streaming mode and min_apply_delay is not allowed.

~

10b.
(see AlterSubscription general review comment #8 above)
Here you can use the same comment error message that says
min_apply_delay > 0 and streaming = parallel are mutually exclusive
options.

======

.../replication/logical/applyparallelworker.c

11.

@@ -704,7 +704,8 @@ pa_process_spooled_messages_if_required(void)
  {
  apply_spooled_messages(&MyParallelShared->fileset,
     MyParallelShared->xid,
-    InvalidXLogRecPtr);
+    InvalidXLogRecPtr,
+    0);

IMO this passing of 0 is a bit strange because it is currently acting
like a dummy value since the apply_spooled_messages will never make
use of the 'finish_ts' anyway (since this call is from a parallel
apply worker).

I think a better way to code this might be to pass the 0 (same as you
are doing here) but inside the apply_spooled_messages change the code:

FROM
if (!am_parallel_apply_worker())
maybe_delay_apply(finish_ts);

TO
if (finish_ts)
maybe_delay_apply(finish_ts);

That does 2 things.
- It makes the passed-in 0 have some meaning
- It simplifies the apply_spooled_messages code

======

src/backend/replication/logical/worker.c

12.

@@ -318,6 +318,17 @@ static List *on_commit_wakeup_workers_subids = NIL;
 bool in_remote_transaction = false;
 static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;

+/*
+ * In order to avoid walsender's timeout during time-delayed replication,
+ * it's necessary to keep sending feedback messages during the delay from the
+ * worker process. Meanwhile, the feature delays the apply before starting the
+ * transaction and thus we don't write WALs for the suspended changes during
+ * the wait. Hence, in the case the worker process 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.
+ */
+static XLogRecPtr last_received = InvalidXLogRecPtr;

12a.
Suggest a small change to the first sentence of the comment.

BEFORE
In order to avoid walsender's timeout during time-delayed replication,
it's necessary to keep sending feedback messages during the delay from
the worker process.

AFTER
In order to avoid walsender timeout for time-delayed replication the
worker process keeps sending feedback messages during the delay
period.

~

12b.
"Hence, in the case" -> "When"

~~~

13. forward declare

-static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
+static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply,
+   bool in_delaying_apply);

Change the param name:

"in_delaying_apply" -> "in_delayed_apply” (??)

~~~

14. maybe_delay_apply

+ /* Nothing to do if no delay set */
+ if (MySubscription->minapplydelay <= 0)
+ return;

IIUC min_apply_delay cannot be < 0 so this condition could simply be:

if (!MySubscription->minapplydelay)
return;

~~~

15. maybe_delay_apply

+ /*
+ * The min_apply_delay parameter is ignored until all tablesync workers
+ * have reached READY state. If we allow the delay during the catchup
+ * phase, once we reach the limit of tablesync workers, it will impose a
+ * delay for each subsequent worker. It means it will take a long time to
+ * finish the initial table synchronization.
+ */
+ if (!AllTablesyncsReady())
+ return;

SUGGESTION (slight rewording)
The min_apply_delay parameter is ignored until all tablesync workers
have reached READY state. This is because if we allowed the delay
during the catchup phase, then once we reached the limit of tablesync
workers it would impose a delay for each subsequent worker. That would
cause initial table synchronization completion to take a long time.

~~~

16. maybe_delay_apply

+ while (true)
+ {
+ long diffms;
+
+ ResetLatch(MyLatch);
+
+ CHECK_FOR_INTERRUPTS();

IMO there should be some small explanatory comment here at the top of
the while loop.

~~~

17. apply_spooled_messages

@@ -2024,6 +2141,21 @@ apply_spooled_messages(FileSet *stream_fileset,
TransactionId xid,
  int fileno;
  off_t offset;

+ /*
+ * Should we delay the current transaction?
+ *
+ * Unlike the regular (non-streamed) cases, the delay is applied in a
+ * STREAM COMMIT/STREAM PREPARE message for streamed transactions. The
+ * STREAM START message does not contain a commit/prepare time (it will be
+ * available when the in-progress transaction finishes). Hence, it's not
+ * appropriate to apply a delay at that time.
+ *
+ * It's not allowed to execute time-delayed replication with parallel
+ * apply feature.
+ */
+ if (!am_parallel_apply_worker())
+ maybe_delay_apply(finish_ts);

That whole comment part "Unlike the regular (non-streamed) cases"
seems misplaced here.  Perhaps this part of the comment is better put
into the function header where the meaning of 'finish_ts' is
explained?

~~~

18. apply_spooled_messages

+ * It's not allowed to execute time-delayed replication with parallel
+ * apply feature.
+ */
+ if (!am_parallel_apply_worker())
+ maybe_delay_apply(finish_ts);

As was mentioned in comment #11 above this code could be changed like

if (finish_ts)
maybe_delay_apply(finish_ts);
then you don't even need to make mention of "parallel apply" at all here.

OTOH if you want to still have the parallel apply comment then maybe
reword it like this:
"It is not allowed to combine time-delayed replication with the
parallel apply feature."

~~~

19. apply_spooled_messages

If you chose not to do my suggestion from comment #11, then there are
2 identical conditions (!am_parallel_apply_worker()); In this case, I
was wondering if it would be better to refactor to use a single
condition instead.

~~~

20. send_feedback
(same as comment #13)

Maybe change the new param name to “in_delayed_apply”?

~~~

21.

@@ -3737,8 +3869,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.
+ *
+ * During the delay of time-delayed replication, do not tell the publisher
+ * that the received latest LSN is already applied and flushed at this
+ * stage, since we don't apply the transaction yet. If we do so, it leads
+ * to a wrong assumption of logical replication progress on the publisher
+ * side. Here, we just send a feedback message to avoid publisher's
+ * timeout during the delay.
  */

Minor rewording of the comment

SUGGESTION
If the subscriber side apply is delayed (because of time-delayed
replication) then do not tell the publisher that the received latest
LSN is already applied and flushed, otherwise, it leads to the
publisher side making a wrong assumption of logical replication
progress. Instead, we just send a feedback message to avoid a
publisher timeout during the delay.

======

src/bin/pg_dump/pg_dump.c

22.

@@ -4546,9 +4547,14 @@ getSubscriptions(Archive *fout)
    LOGICALREP_TWOPHASE_STATE_DISABLED);

  if (fout->remoteVersion >= 160000)
- appendPQExpBufferStr(query, " s.suborigin\n");
+ appendPQExpBufferStr(query,
+ " s.suborigin,\n"
+ " s.subminapplydelay\n");
  else
- appendPQExpBuffer(query, " '%s' AS suborigin\n", LOGICALREP_ORIGIN_ANY);
+ {
+ appendPQExpBuffer(query, " '%s' AS suborigin,\n", LOGICALREP_ORIGIN_ANY);
+ appendPQExpBufferStr(query, " 0 AS subminapplydelay\n");
+ }

Can’t those appends in the else part can be combined to a single
appendPQExpBuffer

appendPQExpBuffer(query,
" '%s' AS suborigin,\n"
" 0 AS subminapplydelay\n"
LOGICALREP_ORIGIN_ANY);

======

src/include/catalog/pg_subscription.h

23.

@@ -70,6 +70,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId)
BKI_SHARED_RELATION BKI_ROW
  XLogRecPtr subskiplsn; /* All changes finished at this LSN are
  * skipped */

+ int64 subminapplydelay; /* Replication apply delay */
+
  NameData subname; /* Name of the subscription */

  Oid subowner BKI_LOOKUP(pg_authid); /* Owner of the subscription */

SUGGESTION (for comment)
Replication apply delay (ms)

~~

24.

@@ -120,6 +122,7 @@ typedef struct Subscription
  * in */
  XLogRecPtr skiplsn; /* All changes finished at this LSN are
  * skipped */
+ int64 minapplydelay; /* Replication apply delay */

SUGGESTION (for comment)
Replication apply delay (ms)

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

---

From: Peter Smith <[email-masked]>
Date: Thu, 19 Jan 2023 12:41:58 +1100

...
> min_apply_delay > 0 and streaming = parallel are mutually exclusive

After checking the regression test output I can see the merit of your
separate error messages like this, even if they are maybe not strictly
necessary. So feel free to ignore my previous review comment.

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

---

From: Peter Smith <[email-masked]>
Date: Thu, 19 Jan 2023 12:49:14 +1100

>  Here are my review comments for the latest patch v16-0001. (excluding

And here are some review comments for the v16-0001 test code.

======

src/test/regress/sql/subscription.sql

1. General
For all comments

"time delayed replication" -> "time-delayed replication" maybe is better?

~~~

2.
-- fail - utilizing streaming = parallel with time delayed replication
is not supported.

For readability please put a blank line before this test.

~~~

3.
-- success -- value without unit is taken as milliseconds

"value" -> "min_apply_delay value"

~~~

4.
-- success -- interval is converted into ms and stored as integer

"interval" -> "min_apply_delay interval"

"integer" -> "an integer"

~~~

5.
You could also add another test where min_apply_delay is 0

Then the following combination can be confirmed OK -- success create
subscription with (streaming=parallel, min_apply_delay=0)

~~

6.
-- fail - alter subscription with min_apply_delay should fail when
streaming = parallel is set.
CREATE SUBSCRIPTION regress_testsub CONNECTION
'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect =
false, streaming = parallel);

There is another way to do this test without creating a brand-new
subscription. You could just alter the existing subscription like:
ALTER ... SET (min_apply_delay = 0)
then ALTER ... SET (parallel = streaming)
then ALTER ... SET (min_apply_delay = 123)

======

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

7. sub check_apply_delay_log

    my ($node_subscriber, $message, $expected) = @_;

Why pass in the message text? I is always the same so can be hardwired
in this function, right?

~~~

8.
# Get the delay time in the server log

"int the server log" -> "from the server log" (?)

~~~

9.
        qr/$message: (\d+) ms/
        or die "could not get delayed time";
        my $logged_delay = $1;

        # Is it larger than expected?
        cmp_ok($logged_delay, '>', $expected,
            "The wait time of the apply worker is long enough expectedly"
        );

9a.
"could not get delayed time" -> "could not get the apply worker wait time"

9b.
"The wait time of the apply worker is long enough expectedly" -> "The
apply worker wait time has expected duration"

~~~

10.
sub check_apply_delay_time

Maybe a brief explanatory comment for this function is needed to
explain the unreplicated column c.

~~~

11.
$node_subscriber->safe_psql('postgres',
    "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr
application_name=$appname' PUBLICATION tap_pub WITH (streaming = on,
min_apply_delay = '3s')"

I think there should be a comment here highlighting that you are
setting up a subscriber time delay of 3 seconds, and then later you
can better describe the parameters for the checking functions...

e.g. (add this comment)
# verifies that the subscriber lags the publisher by at least 3 seconds
check_apply_delay_time($node_publisher, $node_subscriber, '5', '3');

e.g.
# verifies that the subscriber lags the publisher by at least 3 seconds
check_apply_delay_time($node_publisher, $node_subscriber, '8', '3');

~~~

12.
# Test whether ALTER SUBSCRIPTION changes the delayed time of the apply worker
# (1 day 1 minute).
$node_subscriber->safe_psql('postgres',
    "ALTER SUBSCRIPTION tap_sub SET (min_apply_delay = 86460000)"
);

Update the comment with another note.
# Note - The extra 1 min is to account for any decoding/network overhead.

~~~

13.
# Make sure we have long enough min_apply_delay after the ALTER command
check_apply_delay_log($node_subscriber, "logical replication apply
delay", "80000000");

IMO the expectation of 1 day (86460000 ms) wait time might be a better
number for your "expected" value.

So update the comment/call like this:

# Make sure the apply worker knows to wait for more than 1 day (86400000 ms)
check_apply_delay_log($node_subscriber, "logical replication apply
delay", "86400000");

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

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Thu, 19 Jan 2023 06:35:58 +0000

> And here are some review comments for the v16-0001 test code.
Hi, thanks for your review !

Fixed.

> ~~~
> 
> 2.
> -- fail - utilizing streaming = parallel with time delayed replication is not
> supported.
> 
Fixed.

Fixed.

> "integer" -> "an integer"
Both are fixed.

> subscription with (streaming=parallel, min_apply_delay=0)
This combination is added with the modification for #6.

> ~~
> 
> 6.
> -- fail - alter subscription with min_apply_delay should fail when streaming =
> parallel is set.
> 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false,
> streaming = parallel);
> 
> There is another way to do this test without creating a brand-new subscription.
> You could just alter the existing subscription like:
> then ALTER ... SET (parallel = streaming) then ALTER ... SET (min_apply_delay
> = 123)
Fixed.

> Why pass in the message text? I is always the same so can be hardwired in this
> function, right?
Fixed.

Fixed.

> "The wait time of the apply worker is long enough expectedly" -> "The apply
> worker wait time has expected duration"
Both are fixed.

> Maybe a brief explanatory comment for this function is needed to explain the
> unreplicated column c.
Added.

> I think there should be a comment here highlighting that you are setting up a
> subscriber time delay of 3 seconds, and then later you can better describe the
> parameters for the checking functions...
Added a comment for CREATE SUBSCRIPTION command.

> check_apply_delay_time($node_publisher, $node_subscriber, '8', '3');
Added.

> ~~~
> 
> 12.
> # Test whether ALTER SUBSCRIPTION changes the delayed time of the apply
> worker # (1 day 1 minute).
Okay, added the comment. In general, TAP tests
fail if we wait for more than 3 minutes. Then,
we should think setting the maximum consumed time
more than 3 minutes is safe. For example, if
(which should not happen usually, but)
we consumed more than 1 minutes between this ALTER SUBSCRIPTION SET
and below check_apply_delay_log() then, the test will fail.

So made the extra time bigger.
> check_apply_delay_log($node_subscriber, "logical replication apply delay",
> "80000000");
> 
> IMO the expectation of 1 day (86460000 ms) wait time might be a better number
> for your "expected" value.
> 
> check_apply_delay_log($node_subscriber, "logical replication apply delay",
> "86400000");
Updated the comment and the function call.

Kindly have a look at the updated patch v17.

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Thu, 19 Jan 2023 07:12:14 +0000

On Wednesday, January 18, 2023 4:06 PM Peter Smith <[email-masked]> wrote:
>  Here are my review comments for the latest patch v16-0001. (excluding the
> test code)
Hi, thank you for your review !

> ======
> 
> General
> 
> 1.
> 
> Since the value of min_apply_delay cannot be < 0,  I was thinking probably it
> should have been declared everywhere in this patch as a
No, we won't be able to adopt this idea.

It seems that we are not able to use uint for catalog type.
So, can't applying it to the pg_subscription.h definitions
and then similarly Int64GetDatum to store catalog variables
and the argument variable of Int64GetDatum.

Plus, there is a possibility that type Interval becomes negative value,
then we are not able to change the int64 variable to get
the return value of interval2ms().

> ======
> 
> Commit message
> 
> 2.
> 
> If the subscription sets min_apply_delay parameter, the logical replication
> worker will delay the transaction commit for min_apply_delay milliseconds.
> 
> ~
> 
> IMO there should be another sentence before this just to say that a new
> parameter is being added:
> 
> e.g.
Added.

> This paragraph seemed confusing. I think it needs to be reworded to change all
> of the "this parameter" references because there are at least 3 different
> parameters mentioned in this paragraph. e.g. maybe just change them to
> explicitly name the parameter you are talking about.
> 
> I also think it needs to mention the ‘min_apply_delay’ subscription parameter
> up-front and then refer to it appropriately.
> 
> The end result might be something like I wrote below (this is just my guess ?
> probably you can word it better).
> 
> SUGGESTION
> For time-delayed logical replication (i.e. when the subscription is created with
> parameter min_apply_delay > 0), the apply worker sends a Standby Status
> Update message to the publisher with a period of wal_receiver_status_interval .
> Make sure to set wal_receiver_status_interval less than the
> wal_sender_timeout on the publisher, otherwise, the walsender will repeatedly
> terminate due to the timeout errors. If wal_receiver_status_interval is set to zero,
> the apply worker doesn't send any feedback messages during the subscriber’s
> min_apply_delay period.
Applied. Also, I added one reference for min_apply_delay parameter
at the end of this description.

> As with the physical replication feature (recovery_min_apply_delay), it can be
> useful to have a time-delayed logical replica.
> 
> IMO not sure that the above sentence is necessary. It seems only to be saying
> that this parameter can be useful. Why do we need to say that?
Removed the sentence.

> "This parameter lets the user to delay" -> "This parameter allows the user to
> delay"
Fixed.

> ~
> 
> 4c.
> "If this value is specified without units" -> "If the value is specified without
> units"
Fixed.
 
Fixed.

> I think the first part can be reworded slightly. See what you think about the
> suggestion below.
> 
> SUGGESTION
> Any delay occurs only on WAL records for transaction begins after all initial
> table synchronization has finished.  The delay is calculated 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.
> 'min_apply_delay' value, in which case no additional wait is necessary. If the
> system clocks...
Addressed.

> A non-zero min_apply_delay parameter is not allowed when streaming in
Applied.

> @@ -404,6 +445,17 @@ parse_subscription_options(ParseState *pstate, List
> *stmt_options,
> + /* Test the combination of streaming mode and min_apply_delay */ if
> + (IsSet(supported_opts, SUBOPT_MIN_APPLY_DELAY) &&
> ereport(ERROR,
> + errcode(ERRCODE_SYNTAX_ERROR), errmsg("%s and %s are mutually
> + exclusive options",
> +    "min_apply_delay > 0", "streaming = parallel")); }
> 
> SUGGESTION (comment)
Fixed.

> ERROR:  min_apply_delay > 0 and streaming = parallel are mutually exclusive
> options
> 
> If you try to ALTER the min_apply_delay when already streaming = parallel you
> can get this error:
Yes. This is because the existing error message styles
in AlterSubscription and parse_subscription_options.

The former uses "mutually exclusive" messages consistently,
while the latter does "cannot enable ..." ones.
> ~
> 
> IMO there is no need to have 3 different error message texts.  I think all these
> cases are explained by just the first text (ERROR:
Then, we followed this kind of formats.

> + if (opts.streaming == LOGICALREP_STREAM_PARALLEL) if
> + ((IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY) &&
Fixed.

> ~
> 
> 9b.
> (see AlterSubscription general review comment #8 above) Here you can use the
> same comment error message that says min_apply_delay > 0 and streaming =
> parallel are mutually exclusive options.
As described above, we followed the current style in the existing functions.

> + if (IsSet(opts.specified_opts, SUBOPT_MIN_APPLY_DELAY)) {
> + /*
> + replaces[Anum_pg_subscription_subminapplydelay - 1] = true; }
> 
> 10a.
> SUGGESTION (comment)
Fixed.

> ~
> 
> 10b.
> (see AlterSubscription general review comment #8 above) Here you can use the
> same comment error message that says min_apply_delay > 0 and streaming =
> parallel are mutually exclusive options.
Same as 9b.

> IMO this passing of 0 is a bit strange because it is currently acting like a dummy
> value since the apply_spooled_messages will never make use of the 'finish_ts'
> anyway (since this call is from a parallel apply worker).
> 
> I think a better way to code this might be to pass the 0 (same as you are doing
> here) but inside the apply_spooled_messages change the code:
> 
> FROM
Adopted.

> @@ -318,6 +318,17 @@ static List *on_commit_wakeup_workers_subids =
> NIL;  bool in_remote_transaction = false;  static XLogRecPtr
> remote_final_lsn = InvalidXLogRecPtr;
> 
> +/*
> + * In order to avoid walsender's timeout during time-delayed
> +replication,
> + * it's necessary to keep sending feedback messages during the delay
> +from the
> + * worker process. Meanwhile, the feature delays the apply before
> +starting the
> + * transaction and thus we don't write WALs for the suspended changes
> +during
> + * the wait. Hence, in the case the worker process sends a feedback
> +message
> + * during the delay, we should not make positions of the flushed and
> +apply LSN
> In order to avoid walsender's timeout during time-delayed replication, it's
> necessary to keep sending feedback messages during the delay from the
> worker process.
> 
> AFTER
> In order to avoid walsender timeout for time-delayed replication the worker
> process keeps sending feedback messages during the delay period.
Fixed.

Fixed.

> "in_delaying_apply" -> "in_delayed_apply” (??)
Changed. The initial intention to append the "in_"
prefix is to make the variable name aligned with
some other variables such as "in_remote_transaction" and
"in_streamed_transaction" that mean the current status
for the transaction. So, until there is a better name proposed,
we can keep it.

> + if (MySubscription->minapplydelay <= 0) return;
> 
Fixed.

> + * phase, once we reach the limit of tablesync workers, it will impose
> + a
> + * delay for each subsequent worker. It means it will take a long time
> + to
> The min_apply_delay parameter is ignored until all tablesync workers have
> reached READY state. This is because if we allowed the delay during the
> catchup phase, then once we reached the limit of tablesync workers it would
> impose a delay for each subsequent worker. That would cause initial table
> synchronization completion to take a long time.
Fixed.

> IMO there should be some small explanatory comment here at the top of the
> while loop.
Added.

> + * STREAM START message does not contain a commit/prepare time (it will
> + be
> + * available when the in-progress transaction finishes). Hence, it's
> + not
> seems misplaced here.  Perhaps this part of the comment is better put into
> the function header where the meaning of 'finish_ts' is explained?
Moved it to the header comment for maybe_delay_apply.

> OTOH if you want to still have the parallel apply comment then maybe reword it
> like this:
> "It is not allowed to combine time-delayed replication with the parallel apply
> feature."
Changed and now I don't mention the parallel apply feature.

> 2 identical conditions (!am_parallel_apply_worker()); In this case, I was
> wondering if it would be better to refactor to use a single condition instead.
I applied #11 comment. Now, the conditions are not identical.

> Maybe change the new param name to “in_delayed_apply”?
Changed.

> + * During the delay of time-delayed replication, do not tell the
> + publisher
> + * stage, since we don't apply the transaction yet. If we do so, it
> + leads
> + * to a wrong assumption of logical replication progress on the
> + publisher
> replication) then do not tell the publisher that the received latest LSN is already
> applied and flushed, otherwise, it leads to the publisher side making a wrong
> assumption of logical replication progress. Instead, we just send a feedback
> message to avoid a publisher timeout during the delay.
Adopted.

> + appendPQExpBuffer(query, " '%s' AS suborigin,\n",
> + LOGICALREP_ORIGIN_ANY); appendPQExpBufferStr(query, " 0 AS
> + subminapplydelay\n"); }
> 
Adopted.

Fixed.

Fixed.

Kindly have a look at the latest v17 patch in [1].

[1] - https://www.postgresql.org/message-id/TYCPR01MB8373F5162C7A0E6224670CF0EDC49%40TYCPR01MB8373.jpnprd01.prod.outlook.com

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Thu, 19 Jan 2023 08:16:11 +0000

On Thursday, January 19, 2023 10:42 AM Peter Smith <[email-masked]> wrote:
> After checking the regression test output I can see the merit of your separate
> error messages like this, even if they are maybe not strictly necessary. So feel
> free to ignore my previous review comment.
Thank you for your notification.

I wrote another reason why we wrote those messages in [1].
So, please have a look at it.

[1] - https://www.postgresql.org/message-id/TYCPR01MB8373447440202B248BB63805EDC49%40TYCPR01MB8373.jpnprd01.prod.outlook.com

Best Regards,
	Takamichi Osumi

---

From: vignesh C <[email-masked]>
Date: Thu, 19 Jan 2023 16:24:50 +0530

> 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

---

From: Amit Kapila <[email-masked]>
Date: Thu, 19 Jan 2023 18:29:05 +0530

> alter subscription sub1 set (min_apply_delay = ' 600 m s h');

I think here we should have specs similar to recovery_min_apply_delay.

> will be applied in the subscription. If you agree this can be 0002

Do we have any similar stats for recovery_min_apply_delay? If not, I
suggest let's postpone this to see if users really need such a
parameter.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Thu, 19 Jan 2023 18:47:22 +0530

> Kindly have a look at the updated patch v17.

Can we try to optimize the test time for this test? On my machine, it
is the second highest time-consuming test in src/test/subscription. It
seems you are waiting twice for apply_delay and both are for streaming
cases by varying the number of changes. I think it should be just once
and that too for the non-streaming case. I think it would be good to
test streaming code path interaction but not sure if it is important
enough to have two test cases for apply_delay.

One minor comment that I observed while going through the patch.
+ /*
+ * 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)

I think it would be good if you can specify the reason for not
allowing this combination in the comments.

-- 
With Regards,
Amit Kapila.

---

From: vignesh C <[email-masked]>
Date: Thu, 19 Jan 2023 18:53:37 +0530

> suggest let's postpone this to see if users really need such a

I did not find any statistics for recovery_min_apply_delay, ok it can
be delayed to a later time.

Regards,
Vignesh

---

From: shveta malik <[email-masked]>
Date: Fri, 20 Jan 2023 09:16:47 +0530

> [1] - https://www.postgresql.org/message-id/TYCPR01MB8373F5162C7A0E6224670CF0EDC49%40TYCPR01MB8373.jpnprd01.prod.outlook.com

1)
Tried different variations of altering 'min_apply_delay'. All passed
except one below:

postgres=# alter subscription mysubnew set (min_apply_delay = '10.9min 1ms');
ALTER SUBSCRIPTION
postgres=# alter subscription mysubnew set (min_apply_delay = '10.9min 2s 1ms');
ALTER SUBSCRIPTION
--very similar to above but fails,
postgres=# alter subscription mysubnew set (min_apply_delay = '10.9s 1ms');
ERROR:  invalid input syntax for type interval: "10.9s 1ms"

2)
 Logging:
2023-01-19 17:33:16.202 IST [404797] DEBUG:  logical replication apply
delay: 19979 ms
2023-01-19 17:33:26.212 IST [404797] DEBUG:  logical replication apply
delay: 9969 ms
2023-01-19 17:34:25.730 IST [404962] DEBUG:  logical replication apply
delay: 179988 ms-->previous wait over, started for next txn
2023-01-19 17:34:35.737 IST [404962] DEBUG:  logical replication apply
delay: 169981 ms
2023-01-19 17:34:45.746 IST [404962] DEBUG:  logical replication apply
delay: 159972 ms

Is there a way to distinguish between these logs? Maybe dumping xids along-with?

thanks
Shveta

---

From: Peter Smith <[email-masked]>
Date: Fri, 20 Jan 2023 17:55:39 +1100

Hi Osumi-san, here are my review comments for the latest patch v17-0001.

======
Commit Message

1.
Prohibit the combination of this feature and parallel streaming mode.

SUGGESTION (using the same wording as in the code comments)
The combination of parallel streaming mode and min_apply_delay is not allowed.

======

2.
+         <para>
+          By default, the subscriber applies changes as soon as possible. This
+          parameter allows the user to delay the application of changes by a
+          specified amount of time. If the value is specified without units, it
+          is taken as milliseconds. The default is zero (no delay).
+         </para>

Looking at this again, it seemed a  bit strange to repeat "specified"
twice in 2 sentences. Maybe change one of them.

I’ve also suggested using the word "interval" because I don’t think
docs yet mentioned anywhere (except in the example) that using
intervals is possible.

SUGGESTION (for the 2nd sentence)
This parameter allows the user to delay the application of changes by
a given time interval.

~~~

3.
+         <para>
+          Any delay occurs only on WAL records for transaction begins after all
+          initial table synchronization has finished. The delay is calculated
+          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 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
+          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>

3a.
Typo "execeeds" (I think Vignesh reported this already)

~

3b.
SUGGESTION (for the 2nd sentence)
BEFORE
The delay is calculated between the WAL timestamp...
AFTER
The delay is calculated as the difference between the WAL timestamp...

~~~

4.
+         <warning>
+           <para>
+            Delaying the replication can mean there is a much longer
time between making
+            a change on the publisher, and that change being
committed on the subscriber.
+            v
+            See <xref linkend="guc-synchronous-commit"/>.
+           </para>
+         </warning>

IMO maybe there is a better way to express the 2nd sentence:

BEFORE
This can have a big impact on synchronous replication.
AFTER
This can impact the performance of synchronous replication.

======

5. parse_subscription_options

@@ -324,6 +328,43 @@ parse_subscription_options(ParseState *pstate,
List *stmt_options,
  opts->specified_opts |= SUBOPT_LSN;
  opts->lsn = lsn;
  }
+ else if (IsSet(supported_opts, SUBOPT_MIN_APPLY_DELAY) &&
+ strcmp(defel->defname, "min_apply_delay") == 0)
+ {
+ char    *val,
+    *tmp;
+ Interval   *interval;
+ int64 ms;

IMO 'delay_ms' (or similar) would be a friendlier variable name than just 'ms'

~~~

6.
@@ -404,6 +445,20 @@ parse_subscription_options(ParseState *pstate,
List *stmt_options,

This could be expressed as a single condition using &&, maybe also
with the brackets eliminated. (Unless you feel the current code is
more readable)

~~~

7.

These nested ifs could instead be a single "if" with && condition.
(Unless you feel the current code is more readable)

======

8. maybe_delay_apply

+ * Hence, it's not appropriate to apply a delay at the time.
+ */
+static void
+maybe_delay_apply(TimestampTz finish_ts)

That last sentence "Hence,... delay at the time" does not sound
correct. Is there a typo or missing words here?

Maybe it meant to say "... at the STREAM START time."?

~~~

9.
+ /* This might change wal_receiver_status_interval */
+ if (ConfigReloadPending)
+ {
+ ConfigReloadPending = false;
+ ProcessConfigFile(PGC_SIGHUP);
+ }

I was unsure why did you make a special mention of
'wal_receiver_status_interval' here. I mean, Aren't there also other
GUCs that might change and affect something here so was there some
special reason only this one was mentioned?

======

10.
+
+# Compare inserted time on the publisher with applied time on the subscriber to
+# confirm the latter is applied after expected time.
+sub check_apply_delay_time

Maybe the comment could also mention that the time is automatically
stored in the table column 'c'.

~~~

11.
+# Confirm the suspended record doesn't get applied expectedly by the ALTER
+# DISABLE command.
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(a) FROM test_tab WHERE a = 0;");
+is($result, qq(0), "check if the delayed transaction doesn't get
applied expectedly");

The use of "doesn't get applied expectedly" (in 2 places here) seemed
strange. Maybe it's better to say like

SUGGESTION
# Confirm disabling the subscription by ALTER DISABLE did not cause
the delayed transaction to be applied.
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(a) FROM test_tab WHERE a = 0;");
is($result, qq(0), "check the delayed transaction was not applied");

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

---

From: Peter Smith <[email-masked]>
Date: Fri, 20 Jan 2023 18:38:08 +1100

On Fri, Jan 20, 2023 at 2:47 PM shveta malik <[email-masked]> wrote:
...
> Is there a way to distinguish between these logs? Maybe dumping xids along-with?

+1

Also, I was thinking of some other logging enhancements

a) the message should say that this is the *remaining* time to left to wait.

b) it might be convenient to know from the log what was the original
min_apply_delay value in the 1st place.

For example, the logs might look something like this:

DEBUG: time-delayed replication for txid 1234, min_apply_delay =
160000 ms. Remaining wait time: 159972 ms
DEBUG: time-delayed replication for txid 1234, min_apply_delay =
160000 ms. Remaining wait time: 142828 ms
DEBUG: time-delayed replication for txid 1234, min_apply_delay =
160000 ms. Remaining wait time: 129994 ms
DEBUG: time-delayed replication for txid 1234, min_apply_delay =
160000 ms. Remaining wait time: 110001 ms
...

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

---

From: shveta malik <[email-masked]>
Date: Fri, 20 Jan 2023 14:23:45 +0530

> 160000 ms. Remaining wait time: 110001 ms

+1
This will also help when min_apply_delay is set to a new value in
between the current wait. Lets say, I started with min_apply_delay=5
min, when the worker was half way through this, I changed
min_apply_delay to 3 min or say 10min, I see the impact of that change
i.e. new wait-time is adjusted, but log becomes confusing. So, please
keep this scenario as well in mind while improving logging.

thanks
Shveta

---

From: shveta malik <[email-masked]>
Date: Fri, 20 Jan 2023 14:43:08 +0530

> keep this scenario as well in mind while improving logging.

when we send-feedback during apply-delay after every
wal_receiver_status_interval , the log comes as:
023-01-19 17:12:56.000 IST [404795] DEBUG:  sending feedback (force 1)
to recv 0/1570840, write 0/1570840, flush 0/1570840

Shall we have some info here to indicate that it is sent while waiting
for apply_delay to distinguish it from other such send-feedback logs?
It will
make apply_delay flow clear in logs.

thanks
Shveta

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Fri, 20 Jan 2023 18:36:29 +0000

> Hi Osumi-san, here are my review comments for the latest patch v17-0001.
Thanks for your review !

> SUGGESTION (using the same wording as in the code comments) The
> combination of parallel streaming mode and min_apply_delay is not allowed.
Okay. Fixed.

> I’ve also suggested using the word "interval" because I don’t think docs yet
> mentioned anywhere (except in the example) that using intervals is possible.
> 
> This parameter allows the user to delay the application of changes by a given
> time interval.
Adopted.

Fixed.

Fixed.

Fixed.

> @@ -324,6 +328,43 @@ parse_subscription_options(ParseState *pstate, List
> *stmt_options,
> + strcmp(defel->defname, "min_apply_delay") == 0) {
> + char    *val,
> +    *tmp;
> + Interval   *interval;
> + int64 ms;
> 
The variable name has been changed which is more clear to the feature.

> ~~~
> 
> 6.
> @@ -404,6 +445,20 @@ parse_subscription_options(ParseState *pstate, List
> *stmt_options,
> + * The combination of parallel streaming mode and min_apply_delay is
> + not
> + * allowed.
> + */
> This could be expressed as a single condition using &&, maybe also with the
> brackets eliminated. (Unless you feel the current code is more readable)
The current style is intentional. We feel the code is more readable.

> (Unless you feel the current code is more readable)
Same as #6.

> That last sentence "Hence,... delay at the time" does not sound correct. Is there
> a typo or missing words here?
> 
Yes. Fixed.

> ~~~
> 
> 9.
> + /* This might change wal_receiver_status_interval */ if
> + (ConfigReloadPending) { ConfigReloadPending = false;
> + ProcessConfigFile(PGC_SIGHUP); }
> 
> 'wal_receiver_status_interval' here. I mean, Aren't there also other GUCs that
> might change and affect something here so was there some special reason only
> this one was mentioned?
This should be similar to the recoveryApplyDelay for physical replication.
It mentions the GUC used in the same function.

> +# Compare inserted time on the publisher with applied time on the
> +subscriber to # confirm the latter is applied after expected time.
> Maybe the comment could also mention that the time is automatically stored in
> the table column 'c'.
Added.

> ~~~
> 
> 11.
> +# Confirm the suspended record doesn't get applied expectedly by the
> +ALTER # DISABLE command.
> + "SELECT count(a) FROM test_tab WHERE a = 0;"); is($result, qq(0),
> +"check if the delayed transaction doesn't get
> applied expectedly");
> 
> The use of "doesn't get applied expectedly" (in 2 places here) seemed strange.
> Maybe it's better to say like
> 
> SUGGESTION
> # Confirm disabling the subscription by ALTER DISABLE did not cause the
> delayed transaction to be applied.
> "SELECT count(a) FROM test_tab WHERE a = 0;"); is($result, qq(0), "check
> the delayed transaction was not applied");
Fixed.

Kindly have a look at the patch v18.

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Fri, 20 Jan 2023 18:41:38 +0000

Hi,

On Friday, January 20, 2023 6:13 PM shveta malik <[email-masked]> wrote:
> > > 160000 ms. Remaining wait time: 110001 ms ...
> > >
> >
> > +1
> 023-01-19 17:12:56.000 IST [404795] DEBUG:  sending feedback (force 1) to
> recv 0/1570840, write 0/1570840, flush 0/1570840
> 
> Shall we have some info here to indicate that it is sent while waiting for
> apply_delay to distinguish it from other such send-feedback logs?
> It will
This additional tip of log information has been added in the latest v18.
Kindly have a look at it in [1].

[1] - https://www.postgresql.org/message-id/TYCPR01MB8373BED9E390C4839AF56685EDC59%40TYCPR01MB8373.jpnprd01.prod.outlook.com

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Fri, 20 Jan 2023 18:46:53 +0000

On Friday, January 20, 2023 5:54 PM shveta malik <[email-masked]> wrote:
> This will also help when min_apply_delay is set to a new value in between the
> current wait. Lets say, I started with min_apply_delay=5 min, when the worker
> was half way through this, I changed min_apply_delay to 3 min or say 10min, I
> see the impact of that change i.e. new wait-time is adjusted, but log becomes
> confusing. So, please keep this scenario as well in mind while improving
> logging.
Yes, now the change of min_apply_delay value can be detected
since I followed the format provided above. So, this scenario is also covered.

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Fri, 20 Jan 2023 18:50:39 +0000

On Friday, January 20, 2023 12:47 PM shveta malik <[email-masked]> wrote:
> 1)
> Tried different variations of altering 'min_apply_delay'. All passed except one
> below:
> 
> postgres=# alter subscription mysubnew set (min_apply_delay = '10.9min
> 1ms'); ALTER SUBSCRIPTION postgres=# alter subscription mysubnew set
> (min_apply_delay = '10.9min 2s 1ms'); ALTER SUBSCRIPTION --very similar to
> above but fails, postgres=# alter subscription mysubnew set
> (min_apply_delay = '10.9s 1ms');
FYI, this was because the interval type couldn't accept this format.
But now we changed the input format from interval to integer alinged
with recovery_min_apply_delay. Thus, we don't face this issue now.

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Fri, 20 Jan 2023 18:57:37 +0000

Hi, 

On Thursday, January 19, 2023 10:17 PM Amit Kapila <[email-masked]> wrote:
> Can we try to optimize the test time for this test? On my machine, it is the
> second highest time-consuming test in src/test/subscription. It seems you are
> waiting twice for apply_delay and both are for streaming cases by varying the
> number of changes. I think it should be just once and that too for the
> non-streaming case. I think it would be good to test streaming code path
> interaction but not sure if it is important enough to have two test cases for
> apply_delay.
The first insert test is for non-streaming case and we need both cases
for coverage. Regarding the time of test, conducted some optimization
such as turning off the initial table sync, shortening the time of wait, and so on.

> I think it would be good if you can specify the reason for not allowing this
> combination in the comments.
Added.

Please have a look at the latest v18 patch in [1].

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Fri, 20 Jan 2023 19:07:30 +0000

Hi,

On Thursday, January 19, 2023 7:55 PM vignesh C <[email-masked]> wrote:
> 1) min_apply_delay was accepting values like '600 m s h', I was not sure if we
> should allow this:
FYI, the input can be accepted by the interval type.
Now we changed the direction of the type from interval to integer
but plus some unit can be added like recovery_min_apply_delay.
Please check.

> 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:
> /*
> 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)) {
> /*
> replaces[Anum_pg_subscription_subminapplydelay - 1] = true; }
We can't. For create subscription, we need to check the patch
from parse_subscription_options, while for alter subscription,
we need to refer the current MySubscription value for those tests
in AlterSubscription.

> +          are not synchronized, this may lead to apply changes earlier
> + than
Fixed.

Kindly have a look at the v18 patch in [1].

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Sun, 22 Jan 2023 12:42:00 +0000

> Kindly have a look at the patch v18.
I've conducted some refactoring for v18.
Now the latest patch should be tidier and
the comments would be clearer and more aligned as a whole.

Attached the updated patch v19.

Best Regards,
	Takamichi Osumi

---

From: Peter Smith <[email-masked]>
Date: Mon, 23 Jan 2023 19:06:41 +1100

Here are my review comments for v19-0001.

======
Commit message

1.
The combination of parallel streaming mode and min_apply_delay is not
allowed. The subscriber in the parallel streaming mode applies each
stream on arrival without the time of commit/prepare. So, the
subscriber needs to depend on the arrival time of the stream in this
case, if we apply the time-delayed feature for such transactions. Then
there is a possibility where some unnecessary delay will be added on
the subscriber by network communication break between nodes or other
heavy work load on the publisher. On the other hand, applying the delay
at the end of transaction with parallel apply also can cause issues of
used resource bloat and locks kept in open for a long time. Thus, those
features can't work together.

~

I think the above is just cut/paste from a code comment within
subscriptioncmds.c. See review comments #5 below -- so if the code is
changed then this commit message should also change to match it.

======

2.
+       <varlistentry>
+        <term><literal>min_apply_delay</literal> (<type>integer</type>)</term>
+        <listitem>
+         <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 interval. If the value is specified without units, it is
+          taken as milliseconds. The default is zero (no delay).
+         </para>

2a.
The pgdocs says this is an integer default to “ms” unit. Also, the
example on this same page shows it is set to '4h'. But I did not see
any mention of what other units are available to the user. Maybe other
time units should be mentioned here, or maybe a link should be given
to the section  “20.1.1. Parameter Names and Values".

~

2b.
Previously the word "interval" was deliberately used because this
parameter had interval support. But maybe now it should be changed so
it is not misleading.

"a given time interval" --> "a given time period" ??

======

3. Forward declare

+static int defGetMinApplyDelay(DefElem *def);

If the new function is implemented as static near the top of this
source file then this forward declare would not even be necessary,
right?

~~~

4. parse_subscription_options

@@ -324,6 +328,12 @@ parse_subscription_options(ParseState *pstate,
List *stmt_options,
  opts->specified_opts |= SUBOPT_LSN;
  opts->lsn = lsn;
  }
+ else if (IsSet(supported_opts, SUBOPT_MIN_APPLY_DELAY) &&
+ strcmp(defel->defname, "min_apply_delay") == 0)
+ {
+ opts->specified_opts |= SUBOPT_MIN_APPLY_DELAY;
+ opts->min_apply_delay = defGetMinApplyDelay(defel);
+ }

Should this code fragment be calling errorConflictingDefElem so it
will report an error if the same min_apply_delay parameter is
redundantly repeated?  (IIUC, this appears to be the code pattern for
other parameters nearby).

~~~

+ /*
+ * The combination of parallel streaming mode and min_apply_delay is not
+ * allowed. The subscriber in the parallel streaming mode applies each
+ * stream on arrival without the time of commit/prepare. So, the
+ * subscriber needs to depend on the arrival time of the stream in this
+ * case, if we apply the time-delayed feature for such transactions. Then
+ * there is a possibility where some unnecessary delay will be added on
+ * the subscriber by network communication break between nodes or other
+ * heavy work load on the publisher. On the other hand, applying the delay
+ * at the end of transaction with parallel apply also can cause issues of
+ * used resource bloat and locks kept in open for a long time. Thus, those
+ * features can't work together.
+ */

IMO some re-wording might be warranted here. I am not sure quite how
to do it. Perhaps like below?

SUGGESTION

Here are some reasons why these features are incompatible:
a. In the parallel streaming mode the subscriber applies each stream
on arrival without knowledge of the commit/prepare 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.
b. If we apply the delay at the end of the transaction of the parallel
apply then that would cause issues related to resource bloat and locks
being held for a long time.

~~~

6. defGetMinApplyDelay

+
+
+/*
+ * Extract the min_apply_delay mode value from a DefElem. This is very similar
+ * to PGC_INT case of parse_and_validate_value(), because min_apply_delay
+ * accepts the same string as recovery_min_apply_delay.
+ */
+int
+defGetMinApplyDelay(DefElem *def)

6a.
"same string" -> "same parameter format" ??

~

6b.
I thought this function should be implemented as static and located at
the top of the subscriptioncmds.c source file.

======

7. maybe_delay_apply

+static void maybe_delay_apply(TransactionId xid, TimestampTz finish_ts);

Is there a reason why this is here? AFAIK the static implementation
precedes any usage so I doubt this forward declaration is required.

~~~

8. send_feedback

@@ -3775,11 +3912,12 @@ send_feedback(XLogRecPtr recvpos, bool force,
bool requestReply)
  pq_sendint64(reply_message, now); /* sendTime */
  pq_sendbyte(reply_message, requestReply); /* replyRequested */

- elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write
%X/%X, flush %X/%X",
+ elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write
%X/%X, flush %X/%X in-delayed: %d",
  force,
  LSN_FORMAT_ARGS(recvpos),
  LSN_FORMAT_ARGS(writepos),
- LSN_FORMAT_ARGS(flushpos));
+ LSN_FORMAT_ARGS(flushpos),
+ in_delayed_apply);

Wondering if it is better to write this as:
"sending feedback (force %d, in_delayed_apply %d) to recv %X/%X, write
%X/%X, flush %X/%X"

======

9. Add new test?

Should there be an additional test to check redundant parameter
setting -- eg. "... WITH (min_apply_delay=123, min_apply_delay=456)"

(this is related to the review comment #4)

~

10. Add new tests?

Should there be other tests just to verify different units (like 'd',
'h', 'min') are working OK?

======

11.
+# Confirm the time-delayed replication has been effective from the server log
+# message where the apply worker emits for applying delay. Moreover, verifies
+# that the current worker's delayed time is sufficiently bigger than the
+# expected value, in order to check any update of the min_apply_delay.
+sub check_apply_delay_log

"the current worker's delayed time..." --> "the current worker's
remaining wait time..." ??

~~~

12.
+ # Get the delay time from the server log
+ my $contents = slurp_file($node_subscriber->logfile, $offset);

"Get the delay time...." --> "Get the remaining wait time..."

~~~

13.
+# Create a subscription that applies the trasaction after 50 milliseconds delay
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr
application_name=$appname' PUBLICATION tap_pub WITH (copy_data = off,
min_apply_delay = '50ms', streaming = 'on')"
+);

13a.
typo: "trasaction"

~

13b
50ms seems an extremely short time – How do you even know if this is
testing anything related to the time delay? You may just be detecting
the normal lag between publisher and subscriber without time delay
having much to do with anything.

~

14.

+# Note that we cannot call check_apply_delay_log() here because there is a
+# possibility that the delay is skipped. The event happens when the WAL
+# replication between publisher and subscriber is delayed due to a mechanical
+# problem. The log output will be checked later - substantial delay-time case.
+
+# Verify that the subscriber lags the publisher by at least 50 milliseconds
+check_apply_delay_time($node_publisher, $node_subscriber, '2', '0.05');

14a.
"The event happens..." ??

Did you mean "This might happen if the WAL..."

~

14b.
The log output will be checked later - substantial delay-time case.

I think that needs re-wording to clarify.
e.g1. you have nothing called a "substantial delay-time" case.
e.g2. the word "later" confused me. Originally, I thought you meant it
is not tested yet but that you will check it "later", but now IIUC you
are just  referring to the "1 day 5 minutes" test that comes below in
this location TAP file (??)

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

---

From: Amit Kapila <[email-masked]>
Date: Mon, 23 Jan 2023 16:14:44 +0530

...
> being held for a long time.

How about something like:
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.

> the top of the subscriptioncmds.c source file.

I agree that this should be a static function but I think its current
location is a better place as other similar function is just above it.

> setting -- eg. "... WITH (min_apply_delay=123, min_apply_delay=456)"

I don't think that will be of much help. We don't seem to have other
tests for subscription parameters.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 23 Jan 2023 17:36:13 +0530

> Attached the updated patch v19.

Few comments:
=============
1.
}
+
+
+/*

Only one empty line is sufficient between different functions.

2.
+ opts->min_apply_delay > 0 && opts->streaming == LOGICALREP_STREAM_PARALLEL)

I think here we should add a comment for the translator as we are
doing in some other nearby cases.

3.
+ /*
+ * The combination of parallel streaming mode and
+ * min_apply_delay is not allowed.
+ */
+ 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))
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot enable %s mode for subscription with %s",
+    "streaming = parallel", "min_apply_delay"));
+

A. When can second condition ((!IsSet(opts.specified_opts,
SUBOPT_MIN_APPLY_DELAY) && sub->minapplydelay > 0)) in above check be
true?
B. In comments, you can say "See parse_subscription_options."

4.
+/*
+ * When min_apply_delay parameter is set on the subscriber, we wait long enough
+ * to make sure a transaction is applied at least that interval behind the
+ * publisher.

Shouldn't this part of the comment needs to be updated after the patch
has stopped using interval?

5. How does this feature interacts with the SKIP feature? Currently,
it doesn't care whether the changes of a particular xact are skipped
or not. I think that might be okay because anyway the purpose of this
feature is to make subscriber lag from publishers. What do you think?
I feel we can add some comments to indicate the same.

-- 
With Regards,
Amit Kapila.

---

From: Peter Smith <[email-masked]>
Date: Tue, 24 Jan 2023 09:15:53 +1100

> bloat and locks being held for a long time.

+1. That's better.

...
> location is a better place as other similar function is just above it.

But, why not do everything, instead of settling on a half-fix?

e.g.
1. Change the new function (defGetMinApplyDelay) to be static as it should be
2. And move defGetMinApplyDelay to the top of the file where IMO it
really belongs
3. And then remove the (now) redundant forward declaration of
defGetMinApplyDelay
4. And also move the existing function (defGetStreamingMode) to the
top of the file so that those similar functions (defGetMinApplyDelay
and defGetStreamingMode) can remain together

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

---

From: "Euler Taveira" <[email-masked]>
Date: Mon, 23 Jan 2023 20:32:08 -0300

> Attached the updated patch v19.
[I haven't been following this thread for a long time...]

Good to know that you keep improving this patch. I have a few suggestions that
were easier to provide a patch on top of your latest patch than to provide an
inline suggestions.

There are a few documentation polishing. Let me comment some of them above.

-       The length of time (ms) to delay the application of changes.
+       Total time spent delaying the application of changes, in milliseconds

I don't remember if I suggested this description for catalog but IMO the
suggestion reads better for me.

-       For time-delayed logical replication (i.e. when the subscription is
-       created with parameter min_apply_delay > 0), the apply worker sends a
-       Standby Status Update message to the publisher with a period of
-       <literal>wal_receiver_status_interval</literal>. Make sure to set
-       <literal>wal_receiver_status_interval</literal> less than the
-       <literal>wal_sender_timeout</literal> on the publisher, otherwise, the
-       walsender will repeatedly terminate due to the timeout errors. If
-       <literal>wal_receiver_status_interval</literal> is set to zero, the apply
-       worker doesn't send any feedback messages during the subscriber's
-       <literal>min_apply_delay</literal> period. See
-       <xref linkend="sql-createsubscription"/> for details.
+       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. If <varname>wal_receiver_status_interval</varname> is set to
+       zero, the apply worker doesn't send any feedback messages during the
+       <literal>min_apply_delay</literal> interval.

I removed the parenthesis explanation about time-delayed logical replication.
If you are reading the documentation and does not know what it means you should
(a) read the logical replication chapter or (b) check the glossary (maybe a new
entry should be added). I also removed the Standby status Update message but it
is a low level detail; let's refer to it as feedback message as the other
sentences do. I changed "literal" to "varname" that's the correct tag for
parameters. I replace "period" with "interval" that was the previous
terminology. IMO we should be uniform, use one or the other.

-   The subscriber replication can be instructed to lag behind the publisher
-   side changes by specifying the <literal>min_apply_delay</literal>
-   subscription parameter. See <xref linkend="sql-createsubscription"/> for
-   details.
+   A logical replication subscription can delay the application of changes by
+   specifying the <literal>min_apply_delay</literal> subscription parameter.
+   See <xref linkend="sql-createsubscription"/> for details.

This feature refers to a specific subscription, hence, "logical replication
subscription" instead of "subscriber replication".

+           if (IsSet(opts->specified_opts, SUBOPT_MIN_APPLY_DELAY))
+               errorConflictingDefElem(defel, pstate);
+

Peter S referred to this missing piece of code too.

-int
+static int
defGetMinApplyDelay(DefElem *def)
{

It seems you forgot static keyword.

-       elog(DEBUG2, "time-delayed replication for txid %u, min_apply_delay = %lld ms, Remaining wait time: %ld ms",
-            xid, (long long) MySubscription->minapplydelay, diffms);
+       elog(DEBUG2, "time-delayed replication for txid %u, min_apply_delay = " INT64_FORMAT " ms, remaining wait time: %ld ms",
+            xid, MySubscription->minapplydelay, diffms);

int64 should use format modifier INT64_FORMAT.

-                     (long) wal_receiver_status_interval * 1000,
+                     wal_receiver_status_interval * 1000L,

Cast is not required. I added a suffix to the constant.

-   elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X in-delayed: %d",
+   elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X, apply delay: %s",
         force,
         LSN_FORMAT_ARGS(recvpos),
         LSN_FORMAT_ARGS(writepos),
         LSN_FORMAT_ARGS(flushpos),
-        in_delayed_apply);
+        in_delayed_apply? "yes" : "no");

It is better to use a string to represent the yes/no option.

-                             gettext_noop("Min apply delay (ms)"));
+                             gettext_noop("Min apply delay"));

I don't know if it was discussed but we don't add units to headers. When I
think about this parameter representation (internal and external), I decided to
use the previous code because it provides a unit for external representation. I
understand that using the same representation as recovery_min_apply_delay is
good but the current code does not handle the external representation
accordingly. (recovery_min_apply_delay uses the GUC machinery to adds the unit
but for min_apply_delay, it doesn't).

# Setup for streaming case
-$node_publisher->append_conf('postgres.conf',
+$node_publisher->append_conf('postgresql.conf',
    'logical_decoding_mode = immediate');
$node_publisher->reload;

Fix configuration file name.

Maybe tests should do a better job. I think check_apply_delay_time is fragile
because it does not guarantee that time is not shifted. Time-delayed
replication is a subscriber feature and to check its correctness it should
check the logs.

# Note that we cannot call check_apply_delay_log() here because there is a
# possibility that the delay is skipped. The event happens when the WAL
# replication between publisher and subscriber is delayed due to a mechanical
# problem. The log output will be checked later - substantial delay-time case.

If you might not use the logs for it, it should adjust the min_apply_delay, no?

It does not exercise the min_apply_delay vs parallel streaming mode.

+                   /*
+                    * The combination of parallel streaming mode and
+                    * min_apply_delay is not allowed.
+                    */
+                   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))
+                           ereport(ERROR,
+                                   errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+                                   errmsg("cannot enable %s mode for subscription with %s",
+                                          "streaming = parallel", "min_apply_delay"));
+

Is this code correct? I also didn't like this message. "cannot enable streaming
= parallel mode for subscription with min_apply_delay" is far from a good error
message. How about refer parallelism to "parallel streaming mode".

--
Euler Taveira
EDB   https://www.enterprisedb.com/
From 5024325284ee3b4a4dc0a6a1cc6457ed5608cb46 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Mon, 23 Jan 2023 15:52:55 -0300
Subject: [PATCH] Euler's review

---
 doc/src/sgml/catalogs.sgml                 |   2 +-
 doc/src/sgml/config.sgml                   |  20 ++-
 doc/src/sgml/logical-replication.sgml      |   7 +-
 doc/src/sgml/ref/create_subscription.sgml  |  13 +-
 src/backend/commands/subscriptioncmds.c    |  13 +-
 src/backend/replication/logical/worker.c   |  40 +++---
 src/bin/psql/describe.c                    |   2 +-
 src/test/regress/expected/subscription.out | 160 ++++++++++-----------
 src/test/subscription/t/032_apply_delay.pl |   8 +-
 9 files changed, 133 insertions(+), 132 deletions(-)

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index bf3c05241c..0bdb683296 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7878,7 +7878,7 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
        <structfield>subminapplydelay</structfield> <type>int8</type>
       </para>
       <para>
-       The length of time (ms) to delay the application of changes.
+       Total time spent delaying the application of changes, in milliseconds
       </para></entry>
      </row>
 
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 39244bf64a..a15723d74f 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4788,17 +4788,15 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
        command line.
       </para>
       <para>
-       For time-delayed logical replication (i.e. when the subscription is
-       created with parameter min_apply_delay > 0), the apply worker sends a
-       Standby Status Update message to the publisher with a period of
-       <literal>wal_receiver_status_interval</literal>. Make sure to set
-       <literal>wal_receiver_status_interval</literal> less than the
-       <literal>wal_sender_timeout</literal> on the publisher, otherwise, the
-       walsender will repeatedly terminate due to the timeout errors. If
-       <literal>wal_receiver_status_interval</literal> is set to zero, the apply
-       worker doesn't send any feedback messages during the subscriber's
-       <literal>min_apply_delay</literal> period. See
-       <xref linkend="sql-createsubscription"/> for details.
+       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. If <varname>wal_receiver_status_interval</varname> is set to
+       zero, the apply worker doesn't send any feedback messages during the
+       <literal>min_apply_delay</literal> interval.
       </para>
       </listitem>
      </varlistentry>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 863af11a47..d8ae93f88d 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -248,10 +248,9 @@
   </para>
 
   <para>
-   The subscriber replication can be instructed to lag behind the publisher
-   side changes by specifying the <literal>min_apply_delay</literal>
-   subscription parameter. See <xref linkend="sql-createsubscription"/> for
-   details.
+   A logical replication subscription can delay the application of changes by
+   specifying the <literal>min_apply_delay</literal> subscription parameter.
+   See <xref linkend="sql-createsubscription"/> for details.
   </para>
 
   <sect2 id="logical-replication-subscription-slot">
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 76ee9c0b3d..97ca9f8d9e 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -378,10 +378,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
          </para>
          <warning>
            <para>
-            Delaying the replication can mean 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"/>.
+            Delaying the replication can mean 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>
         </listitem>
@@ -452,8 +453,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
   </para>
 
   <para>
-   A non-zero <literal>min_apply_delay</literal> parameter is not allowed when streaming
-   in parallel mode.
+   A non-zero <literal>min_apply_delay</literal> parameter is not allowed when
+   streaming in parallel mode.
   </para>
 
   <para>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 11e9e9160a..d5fa7a95a9 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -331,6 +331,9 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		else if (IsSet(supported_opts, SUBOPT_MIN_APPLY_DELAY) &&
 				 strcmp(defel->defname, "min_apply_delay") == 0)
 		{
+			if (IsSet(opts->specified_opts, SUBOPT_MIN_APPLY_DELAY))
+				errorConflictingDefElem(defel, pstate);
+
 			opts->specified_opts |= SUBOPT_MIN_APPLY_DELAY;
 			opts->min_apply_delay = defGetMinApplyDelay(defel);
 		}
@@ -2261,11 +2264,11 @@ defGetStreamingMode(DefElem *def)

 /*
- * Extract the min_apply_delay mode value from a DefElem. This is very similar
- * to PGC_INT case of parse_and_validate_value(), because min_apply_delay
+ * Extract the min_apply_delay value from a DefElem. This is very similar to
+ * parse_and_validate_value() for integer values, because min_apply_delay
  * accepts the same string as recovery_min_apply_delay.
  */
-int
+static int
 defGetMinApplyDelay(DefElem *def)
 {
 	char	   *value;
@@ -2294,8 +2297,8 @@ defGetMinApplyDelay(DefElem *def)
 				 hintmsg ? errhint("%s", _(hintmsg)) : 0));
 
 	/*
-	 * Check lower bound. parse_int() has been already confirmed that result
-	 * is equal to or smaller than INT_MAX.
+	 * Check lower bound. parse_int() has already been confirmed that result
+	 * is less than or equal to INT_MAX.
 	 */
 	if (result < 0)
 		ereport(ERROR,
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index eeac69ea13..00fe29fc20 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -320,11 +320,11 @@ bool		in_remote_transaction = false;
 static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
 
 /*
- * In order to avoid walsender timeout for time-delayed replication the worker
- * process keeps sending feedback messages during the delay period.
- * Meanwhile, the feature delays the apply before starting the
- * transaction and thus we don't write WALs for the suspended changes during
- * the wait. When the worker process sends a feedback message
+ * In order to avoid walsender timeout for time-delayed logical replication the
+ * apply worker keeps sending feedback messages during the delay interval.
+ * 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.
  */
@@ -1090,20 +1090,20 @@ maybe_delay_apply(TransactionId xid, TimestampTz finish_ts)
 		if (diffms <= 0)
 			break;
 
-		elog(DEBUG2, "time-delayed replication for txid %u, min_apply_delay = %lld ms, Remaining wait time: %ld ms",
-			 xid, (long long) MySubscription->minapplydelay, diffms);
+		elog(DEBUG2, "time-delayed replication for txid %u, min_apply_delay = " INT64_FORMAT " ms, remaining wait time: %ld ms",
+			 xid, MySubscription->minapplydelay, diffms);
 
 		/*
 		 * Call send_feedback() to prevent the publisher from exiting by
 		 * timeout during the delay, when wal_receiver_status_interval is
 		 * available.
 		 */
-		if (wal_receiver_status_interval > 0
-			&& diffms > wal_receiver_status_interval * 1000)
+		if (wal_receiver_status_interval > 0 &&
+			diffms > wal_receiver_status_interval * 1000L)
 		{
 			WaitLatch(MyLatch,
 					  WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
-					  (long) wal_receiver_status_interval * 1000,
+					  wal_receiver_status_interval * 1000L,
 					  WAIT_EVENT_RECOVERY_APPLY_DELAY);
 			send_feedback(last_received, true, false, true);
 		}
@@ -2135,8 +2135,8 @@ ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno,
 /*
  * Common spoolfile processing.
  *
- * The commit/prepare time for streaming transaction is required to achieve
- * time-delayed replication.
+ * The commit/prepare time (finish_ts) for streamed transactions is required
+ * for time-delayed logical replication.
  */
 void
 apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
@@ -3869,12 +3869,12 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply, bool in_delayed
 	 * No outstanding transactions to flush, we can report the latest received
 	 * position. This is important for synchronous replication.
 	 *
-	 * If the subscriber side apply is delayed (because of time-delayed
-	 * replication) then do not tell the publisher that the received latest
-	 * LSN is already applied and flushed, otherwise, it leads to the
-	 * publisher side making a wrong assumption of logical replication
-	 * progress. Instead, we just send a feedback message to avoid a publisher
-	 * timeout during the delay.
+	 * If the logical replication subscription is delayed (min_apply_delay
+	 * parameter) 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.
 	 */
 	if (!have_pending_txes && !in_delayed_apply)
 		flushpos = writepos = recvpos;
@@ -3913,12 +3913,12 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply, bool in_delayed
 	pq_sendint64(reply_message, now);	/* sendTime */
 	pq_sendbyte(reply_message, requestReply);	/* replyRequested */
 
-	elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X in-delayed: %d",
+	elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X, apply delay: %s",
 		 force,
 		 LSN_FORMAT_ARGS(recvpos),
 		 LSN_FORMAT_ARGS(writepos),
 		 LSN_FORMAT_ARGS(flushpos),
-		 in_delayed_apply);
+		 in_delayed_apply? "yes" : "no");
 
 	walrcv_send(LogRepWorkerWalRcvConn,
 				reply_message->data, reply_message->len);
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 8a27063bed..81d4607a1c 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6533,7 +6533,7 @@ describeSubscriptions(const char *pattern, bool verbose)
 							  ", suborigin AS \"%s\"\n"
 							  ", subminapplydelay AS \"%s\"\n",
 							  gettext_noop("Origin"),
-							  gettext_noop("Min apply delay (ms)"));
+							  gettext_noop("Min apply delay"));
 
 		appendPQExpBuffer(&buf,
 						  ",  subsynccommit AS \"%s\"\n"
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 1230bcb096..977f73fe9b 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -114,18 +114,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+ regress_testsub4
-                                                                                                     List of subscriptions
-       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
 \dRs+ regress_testsub4
-                                                                                                     List of subscriptions
-       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub3;
@@ -135,10 +135,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
 ERROR:  invalid connection string syntax: missing "=" after "foobar" in connection info string
 
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -155,10 +155,10 @@ ERROR:  unrecognized subscription parameter: "create_slot"
 -- ok
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
 \dRs+
-                                                                                                         List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |           Conninfo           | Skip LSN 
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist2 | 0/12345
+                                                                                                      List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |           Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist2 | 0/12345
 (1 row)
 
 -- ok - with lsn = NONE
@@ -167,10 +167,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
 ERROR:  invalid WAL location (LSN): 0/0
 \dRs+
-                                                                                                         List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |           Conninfo           | Skip LSN 
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist2 | 0/0
+                                                                                                      List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |           Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist2 | 0/0
 (1 row)
 
 BEGIN;
@@ -202,10 +202,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
 ERROR:  invalid value for parameter "synchronous_commit": "foobar"
 HINT:  Available values: local, remote_write, remote_apply, on, off.
 \dRs+
-                                                                                                           List of subscriptions
-        Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |           Conninfo           | Skip LSN 
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    |                    0 | local              | dbname=regress_doesnotexist2 | 0/0
+                                                                                                        List of subscriptions
+        Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |           Conninfo           | Skip LSN 
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    |               0 | local              | dbname=regress_doesnotexist2 | 0/0
 (1 row)
 
 -- rename back to keep the rest simple
@@ -239,19 +239,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | t      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | t      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (binary = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -263,27 +263,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 -- fail - publication already exists
@@ -298,10 +298,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
 ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
 ERROR:  publication "testpub1" is already in subscription "regress_testsub"
 \dRs+
-                                                                                                            List of subscriptions
-      Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                          List of subscriptions
+      Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 -- fail - publication used more than once
@@ -316,10 +316,10 @@ ERROR:  publication "testpub3" is not in subscription "regress_testsub"
 -- ok - delete publications
 ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -355,10 +355,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 --fail - alter of two_phase option not supported.
@@ -367,10 +367,10 @@ ERROR:  unrecognized subscription parameter: "two_phase"
 -- but can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -380,10 +380,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -396,18 +396,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    |                    0 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    |               0 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -425,19 +425,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |                  123 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |             123 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 -- success -- min_apply_delay value with unit is converted into ms and stored as an integer
 ALTER SUBSCRIPTION regress_testsub SET (min_apply_delay = '1 d');
 \dRs+
-                                                                                                    List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay (ms) | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |             86400000 | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Min apply delay | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    |        86400000 | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
 -- fail - alter subscription with streaming = parallel should fail when time-delayed replication is set
diff --git a/src/test/subscription/t/032_apply_delay.pl b/src/test/subscription/t/032_apply_delay.pl
index 37388e474f..1b6bc1ef80 100644
--- a/src/test/subscription/t/032_apply_delay.pl
+++ b/src/test/subscription/t/032_apply_delay.pl
@@ -16,7 +16,7 @@ sub check_apply_delay_log
 {
 	my ($node_subscriber, $offset, $expected) = @_;
 
-	my $log_location = $node_subscriber->wait_for_log(qr/time-delayed replication for txid (\d+), min_apply_delay = (\d+) ms, Remaining wait time: (\d+) ms/, $offset);
+	my $log_location = $node_subscriber->wait_for_log(qr/time-delayed replication for txid (\d+), min_apply_delay = (\d+) ms, remaining wait time: (\d+) ms/, $offset);
 
 	cmp_ok($log_location, '>', $offset,
 		"logfile contains triggered logical replication apply delay"
@@ -25,7 +25,7 @@ sub check_apply_delay_log
 	# Get the delay time from the server log
 	my $contents = slurp_file($node_subscriber->logfile, $offset);
 	$contents =~
-	qr/time-delayed replication for txid (\d+), min_apply_delay = (\d+) ms, Remaining wait time: (\d+) ms/
+	qr/time-delayed replication for txid (\d+), min_apply_delay = (\d+) ms, remaining wait time: (\d+) ms/
 	or die "could not get the apply worker wait time";
 	my $logged_delay = $3;
 
@@ -87,7 +87,7 @@ $node_publisher->safe_psql('postgres',
 
 my $appname = 'tap_sub';
 
-# Create a subscription that applies the trasaction after 50 milliseconds delay
+# Create a subscription that applies the transaction after 50 milliseconds delay
 $node_subscriber->safe_psql('postgres',
 	"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (copy_data = off, min_apply_delay = '50ms', streaming = 'on')"
 );
@@ -114,7 +114,7 @@ is($result, qq(2|1|2), 'check if the new rows were applied to subscriber');
 check_apply_delay_time($node_publisher, $node_subscriber, '2', '0.05');
 
 # Setup for streaming case
-$node_publisher->append_conf('postgres.conf',
+$node_publisher->append_conf('postgresql.conf',
 	'logical_decoding_mode = immediate');
 $node_publisher->reload;
 
-- 
2.30.2

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Tue, 24 Jan 2023 09:45:19 +0900 (JST)

At Mon, 23 Jan 2023 17:36:13 +0530, Amit Kapila <[email-masked]> wrote in 
> doing in some other nearby cases.

IMHO "foo > bar" is not an "option".  I think we say "foo and bar are
mutually exclusive options" but I think don't say "foo = x and bar = y
are.. options".  I wrote a comment as "this should be more like
human-speaking" and Euler seems having the same feeling for another
error message.

Concretely I would spell this as "min_apply_delay cannot be enabled
when parallel streaming mode is enabled" or something. And the
opposite-direction message nearby would be "parallel streaming mode
cannot be enabled when min_apply_delay is enabled."

regards.

---

From: Amit Kapila <[email-masked]>
Date: Tue, 24 Jan 2023 10:48:42 +0530

> were easier to provide a patch on top of your latest patch than to provide an

Euler, thanks for your comments. We have an existing problem related
to shutdown which impacts this patch. The problem is that during
shutdown on the publisher, we wait for all the WAL to be sent and
flushed on the subscriber. Now, if we user has configured a long value
for min_apply_delay on the subscriber then the shutdown won't be
successful. This can happen even today if the subscriber waits for
some lock during the apply. This is not so much a problem with
physical replication because there we have a separate process to first
flush the WAL. This problem has been discussed in a separate thread as
well. See [1]. It is important to reach conclusion even if we just
want to document it. So, your thoughts on that other thread can help
us to make it move forward.

[1] - https://www.postgresql.org/message-id/TYAPR01MB586668E50FC2447AD7F92491F5E89%40TYAPR01MB5866.jpnprd01.prod.outlook.com

-- 
With Regards,
Amit Kapila.

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Tue, 24 Jan 2023 14:03:09 +0000

On Monday, January 23, 2023 7:45 PM Amit Kapila <[email-masked]> wrote:
> > + * allowed. The subscriber in the parallel streaming mode applies
> > + each
> > + * subscriber needs to depend on the arrival time of the stream in
> > + this
> > + * case, if we apply the time-delayed feature for such transactions.
> > + Then
> > + * there is a possibility where some unnecessary delay will be added
> > + on
> > + * the subscriber by network communication break between nodes or
> > + other
> > + * heavy work load on the publisher. On the other hand, applying the
> > + delay
> > + * at the end of transaction with parallel apply also can cause
> > + issues of
> > + * used resource bloat and locks kept in open for a long time. Thus,
> > + those
> allowed.
> 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'
> 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
Thank you for providing a good description ! Adopted.
The latest patch can be seen in [1].

[1] - https://www.postgresql.org/message-id/TYCPR01MB8373DC1881F382B4703F26E0EDC99%40TYCPR01MB8373.jpnprd01.prod.outlook.com

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Tue, 24 Jan 2023 14:22:58 +0000

Thanks for your review !

> allowed. The subscriber in the parallel streaming mode applies each stream on
> arrival without the time of commit/prepare. So, the subscriber needs to depend
> on the arrival time of the stream in this case, if we apply the time-delayed
> feature for such transactions. Then there is a possibility where some
> unnecessary delay will be added on the subscriber by network communication
> break between nodes or other heavy work load on the publisher. On the other
> hand, applying the delay at the end of transaction with parallel apply also can
> cause issues of used resource bloat and locks kept in open for a long time.
> Thus, those features can't work together.
> ~
> 
Now, updated this. Kindly have a look at the latest patch in [1].

> The pgdocs says this is an integer default to “ms” unit. Also, the example on
> this same page shows it is set to '4h'. But I did not see any mention of what
> other units are available to the user. Maybe other time units should be
> mentioned here, or maybe a link should be given to the section  “20.1.1.
> Parameter Names and Values".
Added.

> ~
> 
> 2b.
> Previously the word "interval" was deliberately used because this parameter
> had interval support. But maybe now it should be changed so it is not
> misleading.
> 
Fixed.

> If the new function is implemented as static near the top of this source file then
> this forward declare would not even be necessary, right?
This declaration has been kept as discussed.

> @@ -324,6 +328,12 @@ parse_subscription_options(ParseState *pstate, List
> *stmt_options,
> + opts->specified_opts |= SUBOPT_MIN_APPLY_DELAY; min_apply_delay =
> + opts->defGetMinApplyDelay(defel);
> + }
> 
> Should this code fragment be calling errorConflictingDefElem so it will report
> an error if the same min_apply_delay parameter is redundantly repeated?
> (IIUC, this appears to be the code pattern for other parameters nearby).
Added.

> + * at the end of transaction with parallel apply also can cause issues
> + of
> IMO some re-wording might be warranted here. I am not sure quite how to do it.
> Perhaps like below?
> 
> SUGGESTION
> 
> a. In the parallel streaming mode the subscriber applies each stream on arrival
> without knowledge of the commit/prepare 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'
> b. If we apply the delay at the end of the transaction of the parallel apply then
> that would cause issues related to resource bloat and locks being held for a
> long time.
Now, this has been changed to the one suggested by Amit-san.
Thanks for your help.

> ~~~
> 
> 6. defGetMinApplyDelay
> 
> +
> +
> +/*
> + * Extract the min_apply_delay mode value from a DefElem. This is very
> +similar
> + * to PGC_INT case of parse_and_validate_value(), because
> +min_apply_delay
Fixed.

> ~
> 
> 6b.
> I thought this function should be implemented as static and located at the top
> of the subscriptioncmds.c source file.
Made it static but didn't change the place, as Amit-san mentioned.

> +static void maybe_delay_apply(TransactionId xid, TimestampTz
> +finish_ts);
> 
> Is there a reason why this is here? AFAIK the static implementation precedes
> any usage so I doubt this forward declaration is required.
Removed.

> - elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X,
> flush %X/%X",
> "sending feedback (force %d, in_delayed_apply %d) to recv %X/%X,
> write %X/%X, flush %X/%X"
Adopted and merged with the modification Euler-san provided.

> ~
> 
> 10. Add new tests?
> 
> Should there be other tests just to verify different units (like 'd', 'h', 'min') are
> working OK?
No need. The current subscription.sql does the check
of "invalid value for parameter..." error message, which ensures we call
the defGetMinApplyDelay(). Additionally, we have the test of one unit 'd'
for unit iteration loopin convert_to_base_unit().
So, the current test sets should suffice.

> +# Confirm the time-delayed replication has been effective from the
> +server log # message where the apply worker emits for applying delay.
> +Moreover, verifies # that the current worker's delayed time is
> +sufficiently bigger than the # expected value, in order to check any update of
> the min_apply_delay.
> "the current worker's delayed time..." --> "the current worker's remaining wait
> time..." ??
Fixed.

> ~~~
> 
> 12.
> + # Get the delay time from the server log my $contents =
> + slurp_file($node_subscriber->logfile, $offset);
> 
Fixed.

> ~~~
> 
> 13.
> +# Create a subscription that applies the trasaction after 50
> +milliseconds delay $node_subscriber->safe_psql('postgres',
Fixed.

> ~
> 
> 13b
> 50ms seems an extremely short time – How do you even know if this is testing
> anything related to the time delay? You may just be detecting the normal lag
> between publisher and subscriber without time delay having much to do with
> anything.
The wait time has been updated to 1 second now.
Also, the TAP tests now search for the emitted logs by the apply worker.
The path to emit the log is in the maybe_apply_delay and
it does writes the log only if the "diffms" is bigger than zero,
which invokes the wait. So, this will ensure we use the feature
by this flow.

> ~
> 
> 14.
> 
> +# Note that we cannot call check_apply_delay_log() here because there
> +is a # possibility that the delay is skipped. The event happens when
> +the WAL # replication between publisher and subscriber is delayed due
> +to a mechanical # problem. The log output will be checked later - substantial
> delay-time case.
> +
> +# Verify that the subscriber lags the publisher by at least 50
> +milliseconds check_apply_delay_time($node_publisher, $node_subscriber,
> +'2', '0.05');
> 
> 14a.
This part has been removed.

> e.g2. the word "later" confused me. Originally, I thought you meant it is not
> tested yet but that you will check it "later", but now IIUC you are just  referring
> to the "1 day 5 minutes" test that comes below in this location TAP file (??)
Also, removed.

Best Regards,
	Takamichi Osumi

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Tue, 24 Jan 2023 14:57:22 +0000

> were easier to provide a patch on top of your latest patch than to provide an
Thanks for your review ! We basically adopted your suggestions.

> suggestion reads better for me.
Adopted the above change.

Adopted.

Also, I added the glossary for time-delayed replication (one for
applicable to both physical replication and logical replication).
Plus, I united the term "interval" to period, because it would clarify the type for this feature.
I think this is better.
Adopted.

Added.

Fixed.

Fixed.

Fixed.

Fixed.

Adopted.

Fixed.

> If you might not use the logs for it, it should adjust the min_apply_delay, no?
Yes. Adjusted.

> message. How about refer parallelism to "parallel streaming mode".
Yes. opts is the input for alter command and sub object is the existing definition.
We need to check those combinations like when streaming is set to parallel
and min_apply_delay also gets set, then, min_apply_delay should not be bigger than 0, for example.
Besides, adopted your suggestion to improve the comments.

Attach the patch in [1]. Kindly have a look at it.

Best Regards,
	Takamichi Osumi

---

From: Amit Kapila <[email-masked]>
Date: Mon, 6 Feb 2023 17:26:33 +0530

> It is better to use a string to represent the yes/no option.

I think it is better to be consistent with the existing force
parameter which is also boolean, otherwise, it will look odd.

-- 
With Regards,
Amit Kapila.

---

From: "Takamichi Osumi (Fujitsu)" <[email-masked]>
Date: Mon, 6 Feb 2023 13:21:13 +0000

Hi,

On Monday, February 6, 2023 8:57 PM Amit Kapila <[email-masked]> wrote:
> > -   elog(DEBUG2, "sending feedback (force %d) to recv %X/%X,
> write %X/%X, flush %X/%X in-delayed: %d",
> > + %X/%X, flush %X/%X, apply delay: %s",
> >          force,
> I think it is better to be consistent with the existing force parameter which is
> also boolean, otherwise, it will look odd.
Agreed. The latest patch v29 posted in [1] followed this suggestion.

Kindly have a look at it.

[1] - https://www.postgresql.org/message-id/TYCPR01MB8373A59E7B74AA4F96B62BEAEDDA9%40TYCPR01MB8373.jpnprd01.prod.outlook.com

Best Regards,
	Takamichi Osumi