test3.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 tconc_1(a int);
	CREATE TABLE tp(a int) PARTITION BY RANGE(a);
	CREATE PUBLICATION regress_pub1 for table tconc_1;
	CREATE PUBLICATION regress_pub2 for table tp;
));

# create tcount no. of partitions
my $tcount = 1000;
foreach my $i (1 .. $tcount)
{	
	my $st = 10 * $i;
	my $ed = 10 * ($i + 1) - 1;
	$node_publisher->safe_psql('postgres', qq(CREATE TABLE tp$i PARTITION OF tp FOR VALUES FROM ($st) TO ($ed)));
}

# create replication slot
$node_publisher->safe_psql('postgres', qq(SELECT pg_create_logical_replication_slot('slot1', 'pgoutput')));

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;

$background_psql1->query_safe('BEGIN;');
$background_psql1->query_safe(qq(INSERT INTO tconc_1 values(1)));
$node_publisher->safe_psql('postgres', 'TRUNCATE TABLE tp;');
$background_psql1->query_safe(qq(INSERT INTO tconc_1 values(2)));
$background_psql1->query_safe('COMMIT;');

$background_psql1->quit;

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

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

done_testing();