1# Testing of logical decoding using SQL interface and/or pg_recvlogical
2#
3# Most logical decoding tests are in contrib/test_decoding. This module
4# is for work that doesn't fit well there, like where server restarts
5# are required.
6use strict;
7use warnings;
8use PostgresNode;
9use TestLib;
10use Test::More tests => 10;
11use Config;
12
13# Initialize master node
14my $node_master = get_new_node('master');
15$node_master->init(allows_streaming => 1);
16$node_master->append_conf(
17	'postgresql.conf', qq(
18wal_level = logical
19));
20$node_master->start;
21my $backup_name = 'master_backup';
22
23$node_master->safe_psql('postgres',
24	qq[CREATE TABLE decoding_test(x integer, y text);]);
25
26$node_master->safe_psql('postgres',
27qq[SELECT pg_create_logical_replication_slot('test_slot', 'test_decoding');]);
28
29$node_master->safe_psql('postgres',
30qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,10) s;]
31);
32
33# Basic decoding works
34my ($result) = $node_master->safe_psql('postgres',
35	qq[SELECT pg_logical_slot_get_changes('test_slot', NULL, NULL);]);
36is(scalar(my @foobar = split /^/m, $result),
37	12, 'Decoding produced 12 rows inc BEGIN/COMMIT');
38
39# If we immediately crash the server we might lose the progress we just made
40# and replay the same changes again. But a clean shutdown should never repeat
41# the same changes when we use the SQL decoding interface.
42$node_master->restart('fast');
43
44# There are no new writes, so the result should be empty.
45$result = $node_master->safe_psql('postgres',
46	qq[SELECT pg_logical_slot_get_changes('test_slot', NULL, NULL);]);
47chomp($result);
48is($result, '', 'Decoding after fast restart repeats no rows');
49
50# Insert some rows and verify that we get the same results from pg_recvlogical
51# and the SQL interface.
52$node_master->safe_psql('postgres',
53qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
54);
55
56my $expected = q{BEGIN
57table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
58table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
59table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
60table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
61COMMIT};
62
63my $stdout_sql = $node_master->safe_psql('postgres',
64qq[SELECT data FROM pg_logical_slot_peek_changes('test_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
65);
66is($stdout_sql, $expected, 'got expected output from SQL decoding session');
67
68my $endpos = $node_master->safe_psql('postgres',
69"SELECT lsn FROM pg_logical_slot_peek_changes('test_slot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
70);
71print "waiting to replay $endpos\n";
72
73# Insert some rows after $endpos, which we won't read.
74$node_master->safe_psql('postgres',
75	qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,50) s;]
76);
77
78my $stdout_recv = $node_master->pg_recvlogical_upto(
79	'postgres', 'test_slot', $endpos, 180,
80	'include-xids'     => '0',
81	'skip-empty-xacts' => '1');
82chomp($stdout_recv);
83is($stdout_recv, $expected,
84	'got same expected output from pg_recvlogical decoding session');
85
86$node_master->poll_query_until('postgres',
87"SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_slot' AND active_pid IS NULL)"
88)
89  or die "slot never became inactive";
90
91$stdout_recv = $node_master->pg_recvlogical_upto(
92	'postgres', 'test_slot', $endpos, 180,
93	'include-xids'     => '0',
94	'skip-empty-xacts' => '1');
95chomp($stdout_recv);
96is($stdout_recv, '',
97	'pg_recvlogical acknowledged changes');
98
99$node_master->safe_psql('postgres', 'CREATE DATABASE otherdb');
100
101is( $node_master->psql(
102		'otherdb',
103"SELECT lsn FROM pg_logical_slot_peek_changes('test_slot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
104	),
105	3,
106	'replaying logical slot from another database fails');
107
108$node_master->safe_psql('otherdb',
109qq[SELECT pg_create_logical_replication_slot('otherdb_slot', 'test_decoding');]
110);
111
112# make sure you can't drop a slot while active
113SKIP:
114{
115
116   # some Windows Perls at least don't like IPC::Run's start/kill_kill regime.
117	skip "Test fails on Windows perl", 2 if $Config{osname} eq 'MSWin32';
118
119	my $pg_recvlogical = IPC::Run::start(
120		[   'pg_recvlogical', '-d', $node_master->connstr('otherdb'),
121			'-S', 'otherdb_slot', '-f', '-', '--start' ]);
122	$node_master->poll_query_until('otherdb',
123"SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'otherdb_slot' AND active_pid IS NOT NULL)"
124	) or die "slot never became active";
125	is($node_master->psql('postgres', 'DROP DATABASE otherdb'),
126		3, 'dropping a DB with active logical slots fails');
127	$pg_recvlogical->kill_kill;
128	is($node_master->slot('otherdb_slot')->{'slot_name'},
129		undef, 'logical slot still exists');
130}
131
132$node_master->poll_query_until('otherdb',
133"SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'otherdb_slot' AND active_pid IS NULL)"
134) or die "slot never became inactive";
135
136is($node_master->psql('postgres', 'DROP DATABASE otherdb'),
137	0, 'dropping a DB with inactive logical slots succeeds');
138is($node_master->slot('otherdb_slot')->{'slot_name'},
139	undef, 'logical slot was actually dropped with DB');
140
141# done with the node
142$node_master->stop;
143