test2.pl

application/octet-stream

Download

use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
use Time::HiRes;

my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->append_conf(
	'postgresql.conf', q[
shared_buffers = 10GB
max_worker_processes = 32
max_parallel_maintenance_workers = 24
max_parallel_workers = 32
synchronous_commit = on
checkpoint_timeout = 1d
max_wal_size = 24GB
min_wal_size = 15GB
autovacuum = off
max_connections = 20
]);
$node_publisher->start;

$node_publisher->safe_psql(
	'postgres', qq(
	CREATE TABLE tab_conc1(a int);
	CREATE TABLE tp(a int);
	CREATE PUBLICATION regress_pub2 for table tp;
	CREATE PUBLICATION regress_pub1 for table tab_conc1;
    SELECT pg_create_logical_replication_slot('slot1', 'pgoutput');
    --SELECT pg_create_logical_replication_slot('slot1', 'test_decoding');
));

my $psql_timeout_secs = 4 * $PostgreSQL::Test::Utils::timeout_default;

my $background_psql1 = $node_publisher->background_psql(
	'postgres',
	on_error_stop => 0,
	timeout => $psql_timeout_secs);
$background_psql1->set_query_timer_restart();

my $count = 1;
my $count2 = $count + 3;

$background_psql1->query_safe('BEGIN;');
foreach my $i (1 .. $count)
{
    $background_psql1->query_safe(qq[INSERT INTO tab_conc1 VALUES ($i);]);
    if ($i % 2 == 1)
    {
        $node_publisher->safe_psql('postgres', 'ALTER PUBLICATION regress_pub2 DROP TABLE tp;');
    }
    else
    {
        $node_publisher->safe_psql('postgres', 'ALTER PUBLICATION regress_pub2 ADD TABLE tp;');
    }
}
$background_psql1->query_safe('COMMIT;');
$background_psql1->quit;

my $start = Time::HiRes::gettimeofday();
my $result = $node_publisher->safe_psql('postgres', qq(select * from pg_logical_slot_get_binary_changes('slot1', NULL, NULL, 'proto_version', '1', 'publication_names', 'regress_pub1')));
is($result, qq($count2), 'check replicated update on subscriber');
my $end = Time::HiRes::gettimeofday();

printf("time elapsed %.2f\n", $end - $start);
$node_publisher->stop('fast');

done_testing();