Add a pg_recvlogical wrapper to PostgresNode

Allows testing of logical decoding using SQL interface and/or pg_recvlogical
Most logical decoding tests are in contrib/test_decoding. This module
is for work that doesn't fit well there, like where server restarts
are required.

Craig Ringer

diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 7e530676b298fc333d2cd0ad487749f6f7603024..5ef007f7d44bf61001221571c2bb93516d1595e9 100644 (file)
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -1505,6 +1505,84 @@ sub slot
 
 =pod
 
+=item $node->pg_recvlogical_upto(self, dbname, slot_name, endpos, timeout_secs, ...)
+
+Invoke pg_recvlogical to read from slot_name on dbname until LSN endpos, which
+corresponds to pg_recvlogical --endpos.  Gives up after timeout (if nonzero).
+
+Disallows pg_recvlogical from internally retrying on error by passing --no-loop.
+
+Plugin options are passed as additional keyword arguments.
+
+If called in scalar context, returns stdout, and die()s on timeout or nonzero return.
+
+If called in array context, returns a tuple of (retval, stdout, stderr, timeout).
+timeout is the IPC::Run::Timeout object whose is_expired method can be tested
+to check for timeout. retval is undef on timeout.
+
+=cut
+
+sub pg_recvlogical_upto
+{
+       my ($self, $dbname, $slot_name, $endpos, $timeout_secs, %plugin_options) = @_;
+       my ($stdout, $stderr);
+
+       my $timeout_exception = 'pg_recvlogical timed out';
+
+       die 'slot name must be specified' unless defined($slot_name);
+       die 'endpos must be specified' unless defined($endpos);
+
+       my @cmd = ('pg_recvlogical', '-S', $slot_name, '--dbname', $self->connstr($dbname));
+       push @cmd, '--endpos', $endpos;
+       push @cmd, '-f', '-', '--no-loop', '--start';
+
+       while (my ($k, $v) = each %plugin_options)
+       {
+               die "= is not permitted to appear in replication option name" if ($k =~ qr/=/);
+               push @cmd, "-o", "$k=$v";
+       }
+
+       my $timeout;
+       $timeout = IPC::Run::timeout($timeout_secs, exception => $timeout_exception ) if $timeout_secs;
+       my $ret = 0;
+
+       do {
+               local $@;
+               eval {
+                       IPC::Run::run(\@cmd, ">", \$stdout, "2>", \$stderr, $timeout);
+                       $ret = $?;
+               };
+               my $exc_save = $@;
+               if ($exc_save)
+               {
+                       # IPC::Run::run threw an exception. re-throw unless it's a
+                       # timeout, which we'll handle by testing is_expired
+                       die $exc_save
+                         if (blessed($exc_save) || $exc_save !~ qr/$timeout_exception/);
+
+                       $ret = undef;
+
+                       die "Got timeout exception '$exc_save' but timer not expired?!"
+                         unless $timeout->is_expired;
+
+                       die "$exc_save waiting for endpos $endpos with stdout '$stdout', stderr '$stderr'"
+                               unless wantarray;
+               }
+       };
+
+       if (wantarray)
+       {
+               return ($ret, $stdout, $stderr, $timeout);
+       }
+       else
+       {
+               die "pg_recvlogical exited with code '$ret', stdout '$stdout' and stderr '$stderr'" if $ret;
+               return $stdout;
+       }
+}
+
+=pod
+
 =back
 
 =cut

diff --git a/src/test/recovery/t/006_logical_decoding.pl b/src/test/recovery/t/006_logical_decoding.pl
index 1716360a170f78970fd36d83272abc1b3e0453f8..3f249cd3439353165bcdbcc59f1b2bf777782f32 100644 (file)
--- a/src/test/recovery/t/006_logical_decoding.pl
+++ b/src/test/recovery/t/006_logical_decoding.pl
@@ -1,9 +1,13 @@
 # Testing of logical decoding using SQL interface and/or pg_recvlogical
+#
+# Most logical decoding tests are in contrib/test_decoding. This module
+# is for work that doesn't fit well there, like where server restarts
+# are required.
 use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 2;
+use Test::More tests => 5;
 
 # Initialize master node
 my $node_master = get_new_node('master');
@@ -35,5 +39,30 @@ $result = $node_master->safe_psql('postgres', qq[SELECT pg_logical_slot_get_chan
 chomp($result);
 is($result, '', 'Decoding after fast restart repeats no rows');
 
+# Insert some rows and verify that we get the same results from pg_recvlogical
+# and the SQL interface.
+$node_master->safe_psql('postgres', qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]);
+
+my $expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT};
+
+my $stdout_sql = $node_master->safe_psql('postgres', qq[SELECT data FROM pg_logical_slot_peek_changes('test_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]);
+is($stdout_sql, $expected, 'got expected output from SQL decoding session');
+
+my $endpos = $node_master->safe_psql('postgres', "SELECT location FROM pg_logical_slot_peek_changes('test_slot', NULL, NULL) ORDER BY location DESC LIMIT 1;");
+diag "waiting to replay $endpos";
+
+my $stdout_recv = $node_master->pg_recvlogical_upto('postgres', 'test_slot', $endpos, 10, 'include-xids' => '0', 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, $expected, 'got same expected output from pg_recvlogical decoding session');
+
+$stdout_recv = $node_master->pg_recvlogical_upto('postgres', 'test_slot', $endpos, 10, 'include-xids' => '0', 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, '', 'pg_recvlogical acknowledged changes, nothing pending on slot');
+
 # done with the node
 $node_master->stop;