=pod
+=item $node->wait_for_subscription_sync(publisher, subname, dbname)
+
+Wait for all tables in pg_subscription_rel to complete the initial
+synchronization (i.e to be either in 'syncdone' or 'ready' state).
+
+If the publisher node is given, additionally, check if the subscriber has
+caught up to what has been committed on the primary. This is useful to
+ensure that the initial data synchronization has been completed after
+creating a new subscription.
+
+If there is no active replication connection from this peer, wait until
+poll_query_until timeout.
+
+This is not a test. It die()s on failure.
+
+=cut
+
+sub wait_for_subscription_sync
+{
+ my ($self, $publisher, $subname, $dbname) = @_;
+ my $name = $self->name;
+
+ $dbname = defined($dbname) ? $dbname : 'postgres';
+
+ # Wait for all tables to finish initial sync.
+ print "Waiting for all subscriptions in \"$name\" to synchronize data\n";
+ my $query =
+ qq[SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');];
+ $self->poll_query_until($dbname, $query)
+ or croak "timed out waiting for subscriber to synchronize data";
+
+ # Then, wait for the replication to catchup if required.
+ if (defined($publisher))
+ {
+ croak 'subscription name must be specified' unless defined($subname);
+ $publisher->wait_for_catchup($subname);
+ }
+
+ print "done\n";
+ return;
+}
+
+=pod
+
=item $node->wait_for_log(regexp, offset)
Waits for the contents of the server log file, starting at the given offset, to
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub, tap_pub_ins_only"
);
-$node_publisher->wait_for_catchup('tap_sub');
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
my $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_notrep");
"CREATE SUBSCRIPTION tap_sub_temp1 CONNECTION '$publisher_connstr' PUBLICATION tap_pub_temp1, tap_pub_temp2"
);
-$node_publisher->wait_for_catchup('tap_sub_temp1');
-
-# Also wait for initial table sync to finish
-$synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_temp1');
# Subscriber table will have no rows initially
$result =
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_slot)"
);
-$node_publisher->wait_for_catchup('tap_sub');
-
# Wait for initial sync to finish as well
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
# Insert initial test data
$node_publisher->safe_psql(
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub"
);
-$node_publisher->wait_for_catchup('tap_sub');
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
my $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_rep");
$node_subscriber->safe_psql('postgres', "DELETE FROM tab_rep;");
# wait for sync to finish this time
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
# check that all data is synced
$result =
);
# and wait for data sync to finish again
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
# check that all data is synced
$result =
"ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION");
# wait for sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM tab_rep_next");
"CREATE SUBSCRIPTION mysub CONNECTION '$publisher_connstr' PUBLICATION mypub;"
);
-$node_publisher->wait_for_catchup('mysub');
-
-# Wait for initial sync to finish as well
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'mysub');
$node_publisher->safe_psql('postgres',
q{INSERT INTO test1 VALUES (1, E'Mot\xc3\xb6rhead')}); # hand-rolled UTF-8
"CREATE SUBSCRIPTION mysub CONNECTION '$publisher_connstr' PUBLICATION mypub;"
);
-$node_publisher->wait_for_catchup('mysub');
-
-# Wait for initial sync to finish as well
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'mysub');
$node_publisher->safe_psql('postgres',
q{INSERT INTO test1 (a, b) VALUES (1, 'one'), (2, 'two');});
m/WARNING: publication "non_existent_pub" does not exist in the publisher/,
"Create subscription throws warning for non-existent publication");
-$node_publisher->wait_for_catchup('mysub1');
-
-# Also wait for initial table sync to finish.
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish.
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'mysub1');
# Specifying non-existent publication along with add publication.
($ret, $stdout, $stderr) = $node_subscriber->psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub"
);
-$node_publisher->wait_for_catchup('tap_sub');
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
my $result =
$node_subscriber->safe_psql('postgres',
$node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION");
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
# Add replica identity column. (The serial is not necessary, but it's
# a convenient way to get a default on the new column so that rows
);
# Wait for initial sync of all subscriptions
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
# insert data to truncate
);
# wait for initial data sync
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
# insert data to truncate
);
# Wait for initial sync of all subscriptions
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
my $result = $node_subscriber->safe_psql('postgres', "SELECT a, b FROM tab1");
is( $result, qq(1|22
});
# Wait for initial sync of all subscriptions
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber1->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-$node_subscriber2->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber1->wait_for_subscription_sync;
+$node_subscriber2->wait_for_subscription_sync;
# Tests for replication using leaf partition identity and schema
"ALTER SUBSCRIPTION sub2 SET PUBLICATION pub_lower_level, pub_all");
# Wait for initial sync of all subscriptions
-$node_subscriber1->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-$node_subscriber2->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber1->wait_for_subscription_sync;
+$node_subscriber2->wait_for_subscription_sync;
# check that data is synced correctly
$result = $node_subscriber1->safe_psql('postgres', "SELECT c, a FROM tab2");
# make sure the subscription on the second subscriber is synced, before
# continuing
-$node_subscriber2->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber2->wait_for_subscription_sync;
# Insert a change into the leaf partition, should be replicated through
# the partition root (thanks to the FOR ALL TABLES partition).
$node_subscriber2->safe_psql('postgres',
"ALTER SUBSCRIPTION sub2 REFRESH PUBLICATION");
-$node_subscriber2->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber2->wait_for_subscription_sync;
# Make partition map cache
$node_publisher->safe_psql('postgres', "INSERT INTO tab5 VALUES (1, 1)");
. "PUBLICATION tpub WITH (slot_name = tpub_slot, binary = true)");
# Ensure nodes are in sync with each other
-$node_publisher->wait_for_catchup('tsub');
-$node_subscriber->poll_query_until('postgres',
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');"
-) or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
# Insert some content and make sure it's replicated across
$node_publisher->safe_psql(
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
my $result =
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
my $result =
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
my $result =
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
my $result =
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
my $result =
$node_subscriber->safe_psql('postgres',
PUBLICATION tap_pub
WITH (two_phase = on)");
-# Wait for subscriber to finish initialization
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
# Also wait for two-phase to be enabled
my $twophase_query =
PUBLICATION tap_pub_copy
WITH (two_phase=on, copy_data=false);");
-# Wait for subscriber to finish initialization
-$node_publisher->wait_for_catchup($appname_copy);
-
-# Also wait for initial table sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
# Also wait for two-phase to be enabled
$node_subscriber->poll_query_until('postgres', $twophase_query)
PUBLICATION tap_pub
WITH (streaming = on, two_phase = on)");
-# Wait for subscriber to finish initialization
-$node_publisher->wait_for_catchup($appname);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
# Also wait for two-phase to be enabled
my $twophase_query =
);
# Wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-
-$node_publisher->wait_for_catchup('tap_sub');
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
# Check the initial data of tab_1 is copied to subscriber
my $result = $node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tap_sub DROP PUBLICATION tap_pub_1");
# Wait for initial table sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-
-$node_publisher->wait_for_catchup('tap_sub');
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
# Check the initial data of tab_drop_refresh was copied to subscriber
$result = $node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tap_sub ADD PUBLICATION tap_pub_1");
# Wait for initial table sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-
-$node_publisher->wait_for_catchup('tap_sub');
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
# Check the initial data of tab_1 was copied to subscriber again
$result = $node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub_schema CONNECTION '$publisher_connstr' PUBLICATION tap_pub_schema"
);
-$node_publisher->wait_for_catchup('tap_sub_schema');
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_schema');
# Check the schema table data is synced up
my $result = $node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tap_sub_schema REFRESH PUBLICATION");
# Wait for sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql('postgres', "INSERT INTO sch1.tab3 VALUES(11)");
"ALTER SUBSCRIPTION tap_sub_schema REFRESH PUBLICATION");
# Wait for sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_subscription_rel WHERE srsubid IN (SELECT oid FROM pg_subscription WHERE subname = 'tap_sub_schema')"
"ALTER SUBSCRIPTION tap_sub_schema REFRESH PUBLICATION");
# Wait for sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_subscription_rel WHERE srsubid IN (SELECT oid FROM pg_subscription WHERE subname = 'tap_sub_schema')"
CREATE SUBSCRIPTION admin_sub CONNECTION '$publisher_connstr' PUBLICATION alice;
));
-$node_publisher->wait_for_catchup('admin_sub');
-
-# Wait for initial sync to finish as well
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');";
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'admin_sub');
# Verify that "regress_admin" can replicate into the tables
#
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->start;
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
my $appname = 'tap_sub';
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_x, tap_pub_forall"
);
-$node_publisher->wait_for_catchup($appname);
# wait for initial table synchronization to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
# The subscription of the FOR ALL TABLES publication means there should be no
# filtering on the tablesync COPY, so all expect all 5 will be present.
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_x, tap_pub_allinschema"
);
-$node_publisher->wait_for_catchup($appname);
# wait for initial table synchronization to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
# The subscription of the ALL TABLES IN SCHEMA publication means there should be
# no filtering on the tablesync COPY, so expect all 5 will be present.
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_1, tap_pub_2, tap_pub_3, tap_pub_4a, tap_pub_4b, tap_pub_5a, tap_pub_5b, tap_pub_toast, tap_pub_inherits, tap_pub_viaroot_2, tap_pub_viaroot_1"
);
-$node_publisher->wait_for_catchup($appname);
-
# wait for initial table synchronization to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
# Check expected replicated rows for tab_rowfilter_1
# tap_pub_1 filter is: (a > 1000 AND b <> 'filtered')
"ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION WITH (copy_data = true)");
# wait for table synchronization to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql('postgres',
"INSERT INTO tab_rowfilter_partitioned (a, b) VALUES(4000, 400),(4001, 401),(4002, 402)"
$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE");
# Wait for the data to replicate.
-$node_publisher->wait_for_catchup('sub');
-$node_subscriber->poll_query_until('postgres',
- "SELECT COUNT(1) = 0 FROM pg_subscription_rel sr WHERE sr.srsubstate NOT IN ('s', 'r') AND sr.srrelid = 'tbl'::regclass"
-);
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'sub');
# Confirm that we have finished the table sync.
my $result =
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
my $offset = 0;
-sub wait_for_subscription_sync
-{
- my ($node) = @_;
-
- # Also wait for initial table sync to finish
- my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-
- $node->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-}
-
# setup tables on both nodes
# tab1: simple 1:1 replication
CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
# tab1: only (a,b) is replicated
$result =
# wait for the tablesync to complete, add a bit more data and then check
# the results of the replication
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub2, pub3
));
-wait_for_subscription_sync($node_subscriber);
-
-$node_publisher->wait_for_catchup('sub1');
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'sub1');
# insert data and make sure the columns in column list get fully replicated
$node_publisher->safe_psql(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub4
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub5
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub6
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub7
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub8;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
TRUNCATE test_part_c;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub9
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub_mix_5, pub_mix_6;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub_root_true;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub1, pub2;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub2, pub1;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub3;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION sub1 SET PUBLICATION pub4;
));
-wait_for_subscription_sync($node_subscriber);
+$node_subscriber->wait_for_subscription_sync;
$node_publisher->safe_psql(
'postgres', qq(
# We cannot rely solely on wait_for_catchup() here; it isn't sufficient
# when tablesync workers might still be running. So in addition to that,
# verify that tables are synced.
-# XXX maybe this should be integrated in wait_for_catchup() itself.
-$node_twoways->wait_for_catchup('testsub');
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_twoways->poll_query_until('d2', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+$node_twoways->wait_for_subscription_sync($node_twoways, 'testsub', 'd2');
is($node_twoways->safe_psql('d2', "SELECT count(f) FROM t"),
$rows * 2, "2x$rows rows in t");
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub"
);
-$node_publisher->wait_for_catchup('tap_sub');
-
-# Also wait for initial table sync to finish
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
is( $node_subscriber->safe_psql(
'postgres', "SELECT * FROM tab_replidentity_index"),
projects / postgresql.git / commitdiff