Hi,
On 4/27/23 5:37 AM, Amit Kapila wrote:
On Wed, Apr 26, 2023 at 4:41 PM Drouvot, Bertrand
<bertranddrouvot...@gmail.com> wrote:
+When in recovery, the default value of target_lsn is $node->lsn('replay')
+instead. This is needed when the publisher passed to
wait_for_subscription_sync()
+is a standby node.
I think this will be useful whenever wait_for_catchup has been called
for a standby node (where self is a standby node). I have tried even
by commenting wait_for_subscription_sync in the new test then it fails
for $node_standby->wait_for_catchup('tap_sub');. So instead, how about
a comment like: "When in recovery, the default value of target_lsn is
$node->lsn('replay') instead which ensures that the cascaded standby
has caught up to what has been replayed on the standby."?
I did it that way because wait_for_subscription_sync() was the first case I had
to work on but I do agree that your wording better describe the intend of the
new
code.
Changed in V7 attached.
Regards,
--
Bertrand Drouvot
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
From af8ab8460a70fe9b70836be021a1a98b2ba6ddd3 Mon Sep 17 00:00:00 2001
From: Bertrand Drouvot <bertranddrouvot...@gmail.com>
Date: Mon, 24 Apr 2023 05:13:23 +0000
Subject: [PATCH v7] Add subscription to the standby test in
035_standby_logical_decoding.pl
Adding one test, to verify that subscription to the standby is possible.
---
src/test/perl/PostgreSQL/Test/Cluster.pm | 21 ++++-
.../t/035_standby_logical_decoding.pl | 91 ++++++++++++++++++-
2 files changed, 107 insertions(+), 5 deletions(-)
18.9% src/test/perl/PostgreSQL/Test/
81.0% src/test/recovery/t/
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm
b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 6f7f4e5de4..bc9b5dc644 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -2611,8 +2611,14 @@ When doing physical replication, the standby is usually
identified by
passing its PostgreSQL::Test::Cluster instance. When doing logical
replication, standby_name identifies a subscription.
-The default value of target_lsn is $node->lsn('write'), which ensures
-that the standby has caught up to what has been committed on the primary.
+When not in recovery, the default value of target_lsn is $node->lsn('write'),
+which ensures that the standby has caught up to what has been committed on
+the primary.
+
+When in recovery, the default value of target_lsn is $node->lsn('replay')
+instead which ensures that the cascaded standby has caught up to what has been
+replayed on the standby.
+
If you pass an explicit value of target_lsn, it should almost always be
the primary's write LSN; so this parameter is seldom needed except when
querying some intermediate replication node rather than the primary.
@@ -2644,7 +2650,16 @@ sub wait_for_catchup
}
if (!defined($target_lsn))
{
- $target_lsn = $self->lsn('write');
+ my $isrecovery = $self->safe_psql('postgres', "SELECT
pg_is_in_recovery()");
+ chomp($isrecovery);
+ if ($isrecovery eq 't')
+ {
+ $target_lsn = $self->lsn('replay');
+ }
+ else
+ {
+ $target_lsn = $self->lsn('write');
+ }
}
print "Waiting for replication conn "
. $standby_name . "'s "
diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl
b/src/test/recovery/t/035_standby_logical_decoding.pl
index b8f5311fe9..f6d6447412 100644
--- a/src/test/recovery/t/035_standby_logical_decoding.pl
+++ b/src/test/recovery/t/035_standby_logical_decoding.pl
@@ -10,12 +10,17 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
-my ($stdin, $stdout, $stderr, $cascading_stdout, $cascading_stderr, $ret,
$handle, $slot);
+my ($stdin, $stdout, $stderr,
+ $cascading_stdout, $cascading_stderr, $subscriber_stdin,
+ $subscriber_stdout, $subscriber_stderr, $ret,
+ $handle, $slot);
my $node_primary = PostgreSQL::Test::Cluster->new('primary');
my $node_standby = PostgreSQL::Test::Cluster->new('standby');
my $node_cascading_standby =
PostgreSQL::Test::Cluster->new('cascading_standby');
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
my $default_timeout = $PostgreSQL::Test::Utils::timeout_default;
+my $psql_timeout = IPC::Run::timer($default_timeout);
my $res;
# Name for the physical slot on primary
@@ -267,7 +272,8 @@ $node_standby->init_from_backup(
has_streaming => 1,
has_restoring => 1);
$node_standby->append_conf('postgresql.conf',
- qq[primary_slot_name = '$primary_slotname']);
+ qq[primary_slot_name = '$primary_slotname'
+ max_replication_slots = 5]);
$node_standby->start;
$node_primary->wait_for_replay_catchup($node_standby);
$node_standby->safe_psql('testdb', qq[SELECT * FROM
pg_create_physical_replication_slot('$standby_physical_slotname');]);
@@ -285,6 +291,26 @@ $node_cascading_standby->append_conf('postgresql.conf',
$node_cascading_standby->start;
$node_standby->wait_for_replay_catchup($node_cascading_standby, $node_primary);
+#######################
+# Initialize subscriber node
+#######################
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+my %psql_subscriber = (
+ 'subscriber_stdin' => '',
+ 'subscriber_stdout' => '',
+ 'subscriber_stderr' => '');
+$psql_subscriber{run} = IPC::Run::start(
+ [ 'psql', '-XA', '-f', '-', '-d', $node_subscriber->connstr('postgres')
],
+ '<',
+ \$psql_subscriber{subscriber_stdin},
+ '>',
+ \$psql_subscriber{subscriber_stdout},
+ '2>',
+ \$psql_subscriber{subscriber_stderr},
+ $psql_timeout);
+
##################################################
# Test that logical decoding on the standby
# behaves correctly.
@@ -365,6 +391,67 @@ is( $node_primary->psql(
3,
'replaying logical slot from another database fails');
+##################################################
+# Test that we can subscribe on the standby with the publication
+# created on the primary.
+##################################################
+
+# Create a table on the primary
+$node_primary->safe_psql('postgres',
+ "CREATE TABLE tab_rep (a int primary key)");
+
+# Create a table (same structure) on the subscriber node
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rep (a int primary key)");
+
+# Create a publication on the primary
+$node_primary->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub for table tab_rep");
+
+$node_primary->wait_for_replay_catchup($node_standby);
+
+# Subscribe on the standby
+my $standby_connstr = $node_standby->connstr . ' dbname=postgres';
+
+# Not using safe_psql() here as it would wait for activity on the primary
+# and we wouldn't be able to launch pg_log_standby_snapshot() on the primary
+# while waiting.
+# psql_subscriber() allows to not wait synchronously.
+$psql_subscriber{subscriber_stdin} .=
+ qq[CREATE SUBSCRIPTION tap_sub
+ CONNECTION '$standby_connstr'
+ PUBLICATION tap_pub
+ WITH (copy_data = off);];
+$psql_subscriber{subscriber_stdin} .= "\n";
+
+$psql_subscriber{run}->pump_nb();
+
+# Speed up the subscription creation
+$node_primary->safe_psql('postgres', "SELECT pg_log_standby_snapshot()");
+
+# Explicitly shut down psql instance gracefully - to avoid hangs
+# or worse on windows
+$psql_subscriber{subscriber_stdin} .= "\\q\n";
+$psql_subscriber{run}->finish;
+
+$node_subscriber->wait_for_subscription_sync($node_standby, 'tap_sub');
+
+# Insert some rows on the primary
+$node_primary->safe_psql('postgres',
+ qq[INSERT INTO tab_rep select generate_series(1,10);]);
+
+$node_primary->wait_for_replay_catchup($node_standby);
+$node_standby->wait_for_catchup('tap_sub');
+
+# Check that the subscriber can see the rows inserted in the primary
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_rep");
+is($result, qq(10), 'check replicated inserts after subscription on standby');
+
+# We do not need the subscription and the subscriber anymore
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->stop;
+
##################################################
# Recovery conflict: Invalidate conflicting slots, including in-use slots
# Scenario 1: hot_standby_feedback off and vacuum FULL
--
2.34.1
