public inbox for [email protected]
help / color / mirror / Atom feedFrom: Andrew Dunstan <[email protected]>
To: Andres Freund <[email protected]>
Cc: [email protected]
Cc: Noah Misch <[email protected]>
Subject: Re: BackgroundPsql swallowing errors on windows
Date: Mon, 16 Feb 2026 16:38:02 -0500
Message-ID: <[email protected]> (raw)
In-Reply-To: <gcjjfm54opgd4kohazc5sowpieta77sg7oywooaf75rux2hgky@qt6mqpfquuhj>
References: <wmovm6xcbwh7twdtymxuboaoarbvwj2haasd3sikzlb3dkgz76@n45rzycluzft>
<[email protected]>
<gcjjfm54opgd4kohazc5sowpieta77sg7oywooaf75rux2hgky@qt6mqpfquuhj>
On 2025-02-14 Fr 11:54 AM, Andres Freund wrote:
> Hi,
>
> On 2025-02-14 08:14:45 -0500, Andrew Dunstan wrote:
>> It's been discussed before, but I'd really really like to get rid of
>> BackgroundPsql. It's ugly, non-intuitive and fragile.
> I agree, unfortunately we're stuck with this until we have a better
> alternative in tree :(
>
>
>> Last year I did some work on this. I was going to present it at Athens but
>> illness prevented me, and then other life events managed to get in my way.
>> But the basic work is around. See <https://github.com/adunstan/test-pq/commit/98518e4929e80fb96f210bbc5aab9fdcea058512;
>> This introduces a libpq session object (PostgreSQL::Test::Session) which can
>> be backed either by FFI or a small XS wrapper - the commit has recipes for
>> both. Missing is a meson.build file for the XS wrapper. There are
>> significant performance gains to be had too (poll_query_until is much nicer,
>> for example, as are most uses of safe_psql). If there is interest I will
>> bring the work up to date, and maybe we can introduce this early in the v19
>> cycle. It would significantly reduce our dependency on IPC::Run, especially
>> the pump() stuff.
> I definitely am interested.
Here we are almost exactly a year later. I returned to this work quite
recently, and reached a milestone, namely the removal of all calls to
BackgroundPsql, with all the TAP tests passing. The XS module still has
some problems, and I think I'm inclined not to pursue it, and just rely
on the FFI mapping.
The current state of this is attached. People who are interested can hit
me up for info in Vancouver if not before. I'll work on turning this
into a set of commitable patches.
cheers
andrew
--
Andrew Dunstan
EDB: https://www.enterprisedb.com
Attachments:
[text/x-patch] Tap-Sessions-v8.patch (201.4K, ../[email protected]/2-Tap-Sessions-v8.patch)
download | inline diff:
diff --git a/contrib/amcheck/t/001_verify_heapam.pl b/contrib/amcheck/t/001_verify_heapam.pl
index e3fee19ae5d..9ea72179fcd 100644
--- a/contrib/amcheck/t/001_verify_heapam.pl
+++ b/contrib/amcheck/t/001_verify_heapam.pl
@@ -5,6 +5,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -18,7 +19,9 @@ $node = PostgreSQL::Test::Cluster->new('test');
$node->init(no_data_checksums => 1);
$node->append_conf('postgresql.conf', 'autovacuum=off');
$node->start;
-$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
+my $session = PostgreSQL::Test::Session->new(node => $node);
+
+$session->do(q(CREATE EXTENSION amcheck));
#
# Check a table with data loaded but no corruption, freezing, etc.
@@ -49,7 +52,7 @@ detects_heap_corruption(
# Check a corrupt table with all-frozen data
#
fresh_test_table('test');
-$node->safe_psql('postgres', q(VACUUM (FREEZE, DISABLE_PAGE_SKIPPING) test));
+$session->do(q(VACUUM (FREEZE, DISABLE_PAGE_SKIPPING) test));
detects_no_corruption("verify_heapam('test')",
"all-frozen not corrupted table");
corrupt_first_page('test');
@@ -81,7 +84,7 @@ sub relation_filepath
my ($relname) = @_;
my $pgdata = $node->data_dir;
- my $rel = $node->safe_psql('postgres',
+ my $rel = $session->query_oneval(
qq(SELECT pg_relation_filepath('$relname')));
die "path not found for relation $relname" unless defined $rel;
return "$pgdata/$rel";
@@ -92,8 +95,8 @@ sub fresh_test_table
{
my ($relname) = @_;
- return $node->safe_psql(
- 'postgres', qq(
+ return $session->do(
+ qq(
DROP TABLE IF EXISTS $relname CASCADE;
CREATE TABLE $relname (a integer, b text);
ALTER TABLE $relname SET (autovacuum_enabled=false);
@@ -117,8 +120,8 @@ sub fresh_test_sequence
{
my ($seqname) = @_;
- return $node->safe_psql(
- 'postgres', qq(
+ return $session->do(
+ qq(
DROP SEQUENCE IF EXISTS $seqname CASCADE;
CREATE SEQUENCE $seqname
INCREMENT BY 13
@@ -134,8 +137,8 @@ sub advance_test_sequence
{
my ($seqname) = @_;
- return $node->safe_psql(
- 'postgres', qq(
+ return $session->query_oneval(
+ qq(
SELECT nextval('$seqname');
));
}
@@ -145,10 +148,7 @@ sub set_test_sequence
{
my ($seqname) = @_;
- return $node->safe_psql(
- 'postgres', qq(
- SELECT setval('$seqname', 102);
- ));
+ return $session->query_oneval(qq(SELECT setval('$seqname', 102)));
}
# Call SQL functions to reset the sequence
@@ -156,8 +156,8 @@ sub reset_test_sequence
{
my ($seqname) = @_;
- return $node->safe_psql(
- 'postgres', qq(
+ return $session->do(
+ qq(
ALTER SEQUENCE $seqname RESTART WITH 51
));
}
@@ -169,6 +169,7 @@ sub corrupt_first_page
my ($relname) = @_;
my $relpath = relation_filepath($relname);
+ $session->close;
$node->stop;
my $fh;
@@ -191,6 +192,7 @@ sub corrupt_first_page
or BAIL_OUT("close failed: $!");
$node->start;
+ $session->reconnect;
}
sub detects_heap_corruption
@@ -216,7 +218,7 @@ sub detects_corruption
my ($function, $testname, @re) = @_;
- my $result = $node->safe_psql('postgres', qq(SELECT * FROM $function));
+ my $result = $session->query_tuples(qq(SELECT * FROM $function));
like($result, $_, $testname) for (@re);
}
@@ -226,7 +228,7 @@ sub detects_no_corruption
my ($function, $testname) = @_;
- my $result = $node->safe_psql('postgres', qq(SELECT * FROM $function));
+ my $result = $session->query_tuples(qq(SELECT * FROM $function));
is($result, '', $testname);
}
diff --git a/contrib/amcheck/t/002_cic.pl b/contrib/amcheck/t/002_cic.pl
index 629d00c1d05..f2c7bc8653f 100644
--- a/contrib/amcheck/t/002_cic.pl
+++ b/contrib/amcheck/t/002_cic.pl
@@ -6,6 +6,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -72,8 +73,8 @@ $node->safe_psql('postgres',
q(INSERT INTO quebec SELECT i FROM generate_series(1, 2) s(i);));
# start background transaction
-my $in_progress_h = $node->background_psql('postgres');
-$in_progress_h->query_safe(q(BEGIN; SELECT pg_current_xact_id();));
+my $in_progress_h = PostgreSQL::Test::Session->new(node => $node);
+$in_progress_h->do(q(BEGIN; SELECT pg_current_xact_id();));
# delete one row from table, while background transaction is in progress
$node->safe_psql('postgres', q(DELETE FROM quebec WHERE i = 1;));
@@ -86,7 +87,7 @@ my $result = $node->psql('postgres',
q(SELECT bt_index_parent_check('oscar', heapallindexed => true)));
is($result, '0', 'bt_index_parent_check for CIC after removed row');
-$in_progress_h->quit;
+$in_progress_h->close;
$node->stop;
done_testing();
diff --git a/contrib/amcheck/t/003_cic_2pc.pl b/contrib/amcheck/t/003_cic_2pc.pl
index f28eeac17ef..d3a1643bf33 100644
--- a/contrib/amcheck/t/003_cic_2pc.pl
+++ b/contrib/amcheck/t/003_cic_2pc.pl
@@ -7,6 +7,7 @@ use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
+use PostgreSQL::Test::Session;
use Test::More;
@@ -36,29 +37,42 @@ $node->safe_psql('postgres', q(CREATE TABLE tbl(i int, j jsonb)));
# statements.
#
-my $main_h = $node->background_psql('postgres');
+my $main_h = PostgreSQL::Test::Session->new(node=>$node);
-$main_h->query_safe(
+$main_h->do_async(
q(
BEGIN;
INSERT INTO tbl VALUES(0, '[[14,2,3]]');
));
-my $cic_h = $node->background_psql('postgres');
+my $cic_h = PostgreSQL::Test::Session->new(node=>$node);
-$cic_h->query_until(
- qr/start/, q(
-\echo start
-CREATE INDEX CONCURRENTLY idx ON tbl(i);
-CREATE INDEX CONCURRENTLY ginidx ON tbl USING gin(j);
+$cic_h->setnonblocking(1);
+
+$cic_h->enterPipelineMode();
+
+$cic_h->do_pipeline(
+ q(
+CREATE INDEX CONCURRENTLY idx ON tbl(i)
+));
+
+$cic_h->pipelineSync();
+
+$cic_h->do_pipeline(
+ q(
+CREATE INDEX CONCURRENTLY ginidx ON tbl USING gin(j)
));
-$main_h->query_safe(
+$cic_h->pipelineSync();
+
+$main_h->wait_for_completion;
+$main_h->do_async(
q(
PREPARE TRANSACTION 'a';
));
-$main_h->query_safe(
+$main_h->wait_for_completion;
+$main_h->do_async(
q(
BEGIN;
INSERT INTO tbl VALUES(0, '[[14,2,3]]');
@@ -66,7 +80,8 @@ INSERT INTO tbl VALUES(0, '[[14,2,3]]');
$node->safe_psql('postgres', q(COMMIT PREPARED 'a';));
-$main_h->query_safe(
+$main_h->wait_for_completion;
+$main_h->do_async(
q(
PREPARE TRANSACTION 'b';
BEGIN;
@@ -75,14 +90,17 @@ INSERT INTO tbl VALUES(0, '"mary had a little lamb"');
$node->safe_psql('postgres', q(COMMIT PREPARED 'b';));
-$main_h->query_safe(
- q(
-PREPARE TRANSACTION 'c';
-COMMIT PREPARED 'c';
-));
+$main_h->wait_for_completion;
+$main_h->do(
+ q(PREPARE TRANSACTION 'c';),
+ q(COMMIT PREPARED 'c';));
-$main_h->quit;
-$cic_h->quit;
+$main_h->close;
+
+# called twice out of an abundance of caution about pipeline mode
+$cic_h->wait_for_completion;
+$cic_h->wait_for_completion;
+$cic_h->close;
$result = $node->psql('postgres', q(SELECT bt_index_check('idx',true)));
is($result, '0', 'bt_index_check after overlapping 2PC');
@@ -106,10 +124,9 @@ PREPARE TRANSACTION 'persists_forever';
));
$node->restart;
-my $reindex_h = $node->background_psql('postgres');
-$reindex_h->query_until(
- qr/start/, q(
-\echo start
+my $reindex_h = PostgreSQL::Test::Session->new(node => $node);
+$reindex_h->do_async(
+ q(
DROP INDEX CONCURRENTLY idx;
CREATE INDEX CONCURRENTLY idx ON tbl(i);
DROP INDEX CONCURRENTLY ginidx;
@@ -117,7 +134,8 @@ CREATE INDEX CONCURRENTLY ginidx ON tbl USING gin(j);
));
$node->safe_psql('postgres', "COMMIT PREPARED 'spans_restart'");
-$reindex_h->quit;
+$reindex_h->wait_for_completion;
+$reindex_h->close;
$result = $node->psql('postgres', q(SELECT bt_index_check('idx',true)));
is($result, '0', 'bt_index_check after 2PC and restart');
$result = $node->psql('postgres', q(SELECT gin_index_check('ginidx')));
diff --git a/contrib/bloom/t/001_wal.pl b/contrib/bloom/t/001_wal.pl
index 683b1876055..f86c49d6357 100644
--- a/contrib/bloom/t/001_wal.pl
+++ b/contrib/bloom/t/001_wal.pl
@@ -5,11 +5,14 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
my $node_primary;
my $node_standby;
+my $session_primary;
+my $session_standby;
# Run few queries on both primary and standby and check their results match.
sub test_index_replay
@@ -21,20 +24,18 @@ sub test_index_replay
# Wait for standby to catch up
$node_primary->wait_for_catchup($node_standby);
- my $queries = qq(SET enable_seqscan=off;
-SET enable_bitmapscan=on;
-SET enable_indexscan=on;
-SELECT * FROM tst WHERE i = 0;
-SELECT * FROM tst WHERE i = 3;
-SELECT * FROM tst WHERE t = 'b';
-SELECT * FROM tst WHERE t = 'f';
-SELECT * FROM tst WHERE i = 3 AND t = 'c';
-SELECT * FROM tst WHERE i = 7 AND t = 'e';
-);
+ my @queries = (
+ "SELECT * FROM tst WHERE i = 0",
+ "SELECT * FROM tst WHERE i = 3",
+ "SELECT * FROM tst WHERE t = 'b'",
+ "SELECT * FROM tst WHERE t = 'f'",
+ "SELECT * FROM tst WHERE i = 3 AND t = 'c'",
+ "SELECT * FROM tst WHERE i = 7 AND t = 'e'",
+ );
# Run test queries and compare their result
- my $primary_result = $node_primary->safe_psql("postgres", $queries);
- my $standby_result = $node_standby->safe_psql("postgres", $queries);
+ my $primary_result = $session_primary->query_tuples(@queries);
+ my $standby_result = $session_standby->query_tuples(@queries);
is($primary_result, $standby_result, "$test_name: query result matches");
return;
@@ -55,13 +56,24 @@ $node_standby->init_from_backup($node_primary, $backup_name,
has_streaming => 1);
$node_standby->start;
+# Create and initialize the sessions
+$session_primary = PostgreSQL::Test::Session->new(node => $node_primary);
+$session_standby = PostgreSQL::Test::Session->new(node => $node_standby);
+my $initset = q[
+ SET enable_seqscan=off;
+ SET enable_bitmapscan=on;
+ SET enable_indexscan=on;
+];
+$session_primary->do($initset);
+$session_standby->do($initset);
+
# Create some bloom index on primary
-$node_primary->safe_psql("postgres", "CREATE EXTENSION bloom;");
-$node_primary->safe_psql("postgres", "CREATE TABLE tst (i int4, t text);");
-$node_primary->safe_psql("postgres",
+$session_primary->do("CREATE EXTENSION bloom;");
+$session_primary->do("CREATE TABLE tst (i int4, t text);");
+$session_primary->do(
"INSERT INTO tst SELECT i%10, substr(encode(sha256(i::text::bytea), 'hex'), 1, 1) FROM generate_series(1,10000) i;"
);
-$node_primary->safe_psql("postgres",
+$session_primary->do(
"CREATE INDEX bloomidx ON tst USING bloom (i, t) WITH (col1 = 3);");
# Test that queries give same result
diff --git a/contrib/pg_visibility/t/001_concurrent_transaction.pl b/contrib/pg_visibility/t/001_concurrent_transaction.pl
index 3aa556892a6..cf6e3b45182 100644
--- a/contrib/pg_visibility/t/001_concurrent_transaction.pl
+++ b/contrib/pg_visibility/t/001_concurrent_transaction.pl
@@ -6,6 +6,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -24,10 +25,10 @@ $standby->start;
# Setup another database
$node->safe_psql("postgres", "CREATE DATABASE other_database;\n");
-my $bsession = $node->background_psql('other_database');
+my $bsession = PostgreSQL::Test::Session->new(node => $node, dbname => 'other_database');
# Run a concurrent transaction
-$bsession->query_safe(
+$bsession->do(
qq[
BEGIN;
SELECT txid_current();
@@ -55,8 +56,8 @@ $result = $standby->safe_psql("postgres",
ok($result eq "", "pg_check_visible() detects no errors");
# Shutdown
-$bsession->query_safe("COMMIT;");
-$bsession->quit;
+$bsession->do("COMMIT;");
+$bsession->close;
$node->stop;
$standby->stop;
diff --git a/contrib/test_decoding/t/001_repl_stats.pl b/contrib/test_decoding/t/001_repl_stats.pl
index 6814c792e2b..79b9182014f 100644
--- a/contrib/test_decoding/t/001_repl_stats.pl
+++ b/contrib/test_decoding/t/001_repl_stats.pl
@@ -7,6 +7,7 @@ use strict;
use warnings FATAL => 'all';
use File::Path qw(rmtree);
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -129,11 +130,11 @@ $node->safe_psql('postgres',
);
# Look at slot data, with a persistent connection.
-my $bpgsql = $node->background_psql('postgres', on_error_stop => 1);
+my $bgsession = PostgreSQL::Test::Session->new(node=>$node);
# Launch query and look at slot data, incrementing the refcount of the
# stats entry.
-$bpgsql->query_safe(
+$bgsession->do_async(
"SELECT pg_logical_slot_peek_binary_changes('$slot_name_restart', NULL, NULL)"
);
@@ -150,7 +151,8 @@ $node->safe_psql('postgres',
# Look again at the slot data. The local stats reference should be refreshed
# to the reinitialized entry.
-$bpgsql->query_safe(
+$bgsession->wait_for_completion;
+$bgsession->do_async(
"SELECT pg_logical_slot_peek_binary_changes('$slot_name_restart', NULL, NULL)"
);
# Drop again the slot, the entry is not dropped yet as the previous session
@@ -176,6 +178,7 @@ command_like(
my $stats_file = "$datadir/pg_stat/pgstat.stat";
ok(-f "$stats_file", "stats file must exist after shutdown");
-$bpgsql->quit;
+$bgsession->wait_for_completion;
+$bgsession->close;
done_testing();
diff --git a/src/bin/pg_amcheck/t/004_verify_heapam.pl b/src/bin/pg_amcheck/t/004_verify_heapam.pl
index 95f1f34c90d..9afacbcaaea 100644
--- a/src/bin/pg_amcheck/t/004_verify_heapam.pl
+++ b/src/bin/pg_amcheck/t/004_verify_heapam.pl
@@ -5,6 +5,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -190,16 +191,17 @@ $node->append_conf('postgresql.conf', 'max_prepared_transactions=10');
$node->start;
my $port = $node->port;
my $pgdata = $node->data_dir;
-$node->safe_psql('postgres', "CREATE EXTENSION amcheck");
-$node->safe_psql('postgres', "CREATE EXTENSION pageinspect");
+my $session = PostgreSQL::Test::Session->new(node => $node);
+$session->do("CREATE EXTENSION amcheck");
+$session->do("CREATE EXTENSION pageinspect");
# Get a non-zero datfrozenxid
-$node->safe_psql('postgres', qq(VACUUM FREEZE));
+$session->do(qq(VACUUM FREEZE));
# Create the test table with precisely the schema that our corruption function
# expects.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
CREATE TABLE public.test (a BIGINT, b TEXT, c TEXT);
ALTER TABLE public.test SET (autovacuum_enabled=false);
ALTER TABLE public.test ALTER COLUMN c SET STORAGE EXTERNAL;
@@ -209,14 +211,15 @@ $node->safe_psql(
# We want (0 < datfrozenxid < test.relfrozenxid). To achieve this, we freeze
# an otherwise unused table, public.junk, prior to inserting data and freezing
# public.test
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
CREATE TABLE public.junk AS SELECT 'junk'::TEXT AS junk_column;
ALTER TABLE public.junk SET (autovacuum_enabled=false);
- VACUUM FREEZE public.junk
- ));
+ ),
+ 'VACUUM FREEZE public.junk'
+);
-my $rel = $node->safe_psql('postgres',
+my $rel = $session->query_oneval(
qq(SELECT pg_relation_filepath('public.test')));
my $relpath = "$pgdata/$rel";
@@ -229,23 +232,24 @@ my $ROWCOUNT_BASIC = 16;
# First insert data needed for tests unrelated to update chain validation.
# Then freeze the page. These tuples are at offset numbers 1 to 16.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
INSERT INTO public.test (a, b, c)
SELECT
x'DEADF9F9DEADF9F9'::bigint,
'abcdefg',
repeat('w', 10000)
FROM generate_series(1, $ROWCOUNT_BASIC);
- VACUUM FREEZE public.test;)
+ ),
+ 'VACUUM FREEZE public.test'
);
# Create some simple HOT update chains for line pointer validation. After
# the page is HOT pruned, we'll have two redirects line pointers each pointing
# to a tuple. We'll then change the second redirect to point to the same
# tuple as the first one and verify that we can detect corruption.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
INSERT INTO public.test (a, b, c)
VALUES ( x'DEADF9F9DEADF9F9'::bigint, 'abcdefg',
generate_series(1,2)); -- offset numbers 17 and 18
@@ -254,8 +258,8 @@ $node->safe_psql(
));
# Create some more HOT update chains.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
INSERT INTO public.test (a, b, c)
VALUES ( x'DEADF9F9DEADF9F9'::bigint, 'abcdefg',
generate_series(3,6)); -- offset numbers 21 through 24
@@ -264,25 +268,30 @@ $node->safe_psql(
));
# Negative test case of HOT-pruning with aborted tuple.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
BEGIN;
UPDATE public.test SET c = 'a' WHERE c = '5'; -- offset number 27
ABORT;
- VACUUM FREEZE public.test;
- ));
+ ),
+ 'VACUUM FREEZE public.test;',
+ );
# Next update on any tuple will be stored at the same place of tuple inserted
# by aborted transaction. This should not cause the table to appear corrupt.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
+ BEGIN;
UPDATE public.test SET c = 'a' WHERE c = '6'; -- offset number 27 again
- VACUUM FREEZE public.test;
- ));
+ COMMIT;
+ ),
+ 'VACUUM FREEZE public.test;',
+ );
# Data for HOT chain validation, so not calling VACUUM FREEZE.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
+ BEGIN;
INSERT INTO public.test (a, b, c)
VALUES ( x'DEADF9F9DEADF9F9'::bigint, 'abcdefg',
generate_series(7,15)); -- offset numbers 28 to 36
@@ -293,11 +302,12 @@ $node->safe_psql(
UPDATE public.test SET c = 'a' WHERE c = '13'; -- offset number 41
UPDATE public.test SET c = 'a' WHERE c = '14'; -- offset number 42
UPDATE public.test SET c = 'a' WHERE c = '15'; -- offset number 43
+ COMMIT;
));
# Need one aborted transaction to test corruption in HOT chains.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
BEGIN;
UPDATE public.test SET c = 'a' WHERE c = '9'; -- offset number 44
ABORT;
@@ -306,19 +316,19 @@ $node->safe_psql(
# Need one in-progress transaction to test few corruption in HOT chains.
# We are creating PREPARE TRANSACTION here as these will not be aborted
# even if we stop the node.
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
BEGIN;
PREPARE TRANSACTION 'in_progress_tx';
));
-my $in_progress_xid = $node->safe_psql(
- 'postgres', qq(
+my $in_progress_xid = $session->query_oneval(
+ qq(
SELECT transaction FROM pg_prepared_xacts;
));
-my $relfrozenxid = $node->safe_psql('postgres',
+my $relfrozenxid = $session->query_oneval(
q(select relfrozenxid from pg_class where relname = 'test'));
-my $datfrozenxid = $node->safe_psql('postgres',
+my $datfrozenxid = $session->query_oneval(
q(select datfrozenxid from pg_database where datname = 'postgres'));
# Sanity check that our 'test' table has a relfrozenxid newer than the
@@ -326,6 +336,7 @@ my $datfrozenxid = $node->safe_psql('postgres',
# first normal xid. We rely on these invariants in some of our tests.
if ($datfrozenxid <= 3 || $datfrozenxid >= $relfrozenxid)
{
+ $session->close;
$node->clean_node;
plan skip_all =>
"Xid thresholds not as expected: got datfrozenxid = $datfrozenxid, relfrozenxid = $relfrozenxid";
@@ -334,17 +345,21 @@ if ($datfrozenxid <= 3 || $datfrozenxid >= $relfrozenxid)
# Find where each of the tuples is located on the page. If a particular
# line pointer is a redirect rather than a tuple, we record the offset as -1.
-my @lp_off = split '\n', $node->safe_psql(
- 'postgres', qq(
+my $lp_off_res = $session->query(
+ qq(
SELECT CASE WHEN lp_flags = 2 THEN -1 ELSE lp_off END
FROM heap_page_items(get_raw_page('test', 'main', 0))
)
-);
+ );
+my @lp_off;
+push(@lp_off, $_->[0]) foreach @{$lp_off_res->{rows}};
+
scalar @lp_off == $ROWCOUNT or BAIL_OUT("row offset counts mismatch");
# Sanity check that our 'test' table on disk layout matches expectations. If
# this is not so, we will have to skip the test until somebody updates the test
# to work on this platform.
+$session->close;
$node->stop;
my $file;
open($file, '+<', $relpath)
@@ -751,17 +766,19 @@ for (my $tupidx = 0; $tupidx < $ROWCOUNT; $tupidx++)
close($file)
or BAIL_OUT("close failed: $!");
$node->start;
+$session->reconnect;
# Run pg_amcheck against the corrupt table with epoch=0, comparing actual
# corruption messages against the expected messages
$node->command_checks_all(
[ 'pg_amcheck', '--no-dependent-indexes', '--port' => $port, 'postgres' ],
2, [@expected], [], 'Expected corruption message output');
-$node->safe_psql(
- 'postgres', qq(
+$session->do(
+ qq(
COMMIT PREPARED 'in_progress_tx';
));
+$session->close;
$node->teardown_node;
$node->clean_node;
diff --git a/src/bin/pg_upgrade/t/007_multixact_conversion.pl b/src/bin/pg_upgrade/t/007_multixact_conversion.pl
index d18c50830d5..684b64cb0a3 100644
--- a/src/bin/pg_upgrade/t/007_multixact_conversion.pl
+++ b/src/bin/pg_upgrade/t/007_multixact_conversion.pl
@@ -15,6 +15,7 @@ use warnings FATAL => 'all';
use Math::BigInt;
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -38,10 +39,8 @@ my $tempdir = PostgreSQL::Test::Utils::tempdir;
# versions.
#
# The first argument is the cluster to connect to, the second argument
-# is a cluster using the new version. We need the 'psql' binary from
-# the new version, the new cluster is otherwise unused. (We need to
-# use the new 'psql' because some of the more advanced background psql
-# perl module features depend on a fairly recent psql version.)
+# is a cluster using the new version. We need the libpq library from
+# the new version (for Session), the new cluster is otherwise unused.
sub mxact_workload
{
my $node = shift; # Cluster to connect to
@@ -79,18 +78,15 @@ sub mxact_workload
# in each connection.
for (0 .. $nclients)
{
- # Use the psql binary from the new installation. The
- # BackgroundPsql functionality doesn't work with older psql
- # versions.
- my $conn = $binnode->background_psql(
- '',
+ # Use the libpq/Session infrastructure from the new installation.
+ my $conn = PostgreSQL::Test::Session->new(
connstr => $node->connstr('postgres'),
- timeout => $connection_timeout_secs);
+ libdir => $binnode->config_data('--libdir'));
- $conn->query_safe("SET log_statement=none", verbose => $verbose)
+ $conn->do("SET log_statement=none")
unless $verbose;
- $conn->query_safe("SET enable_seqscan=off", verbose => $verbose);
- $conn->query_safe("BEGIN", verbose => $verbose);
+ $conn->do("SET enable_seqscan=off");
+ $conn->do("BEGIN");
push(@connections, $conn);
}
@@ -108,9 +104,9 @@ sub mxact_workload
my $conn = $connections[ $i % $nclients ];
my $sql = ($i % $abort_every == 0) ? "ABORT" : "COMMIT";
- $conn->query_safe($sql, verbose => $verbose);
+ $conn->do($sql);
- $conn->query_safe("BEGIN", verbose => $verbose);
+ $conn->do("BEGIN");
if ($i % $update_every == 0)
{
$sql = qq[
@@ -126,12 +122,12 @@ sub mxact_workload
) as x
];
}
- $conn->query_safe($sql, verbose => $verbose);
+ $conn->do($sql);
}
for my $conn (@connections)
{
- $conn->quit();
+ $conn->close;
}
$node->stop;
diff --git a/src/test/authentication/t/001_password.pl b/src/test/authentication/t/001_password.pl
index f4d65ba7bae..296ce81ff5a 100644
--- a/src/test/authentication/t/001_password.pl
+++ b/src/test/authentication/t/001_password.pl
@@ -13,6 +13,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
if (!$use_unix_sockets)
@@ -168,36 +169,18 @@ my $res = $node->safe_psql(
WHERE rolname = 'scram_role_iter'");
is($res, 'SCRAM-SHA-256$1024:', 'scram_iterations in server side ROLE');
-# If we don't have IO::Pty, forget it, because IPC::Run depends on that
-# to support pty connections. Also skip if IPC::Run isn't at least 0.98
-# as earlier version cause the session to time out.
-SKIP:
-{
- skip "IO::Pty and IPC::Run >= 0.98 required", 1
- unless eval { require IO::Pty; IPC::Run->VERSION('0.98'); };
+# set password using PQchangePassword
+my $session = PostgreSQL::Test::Session->new (node => $node);
- # Alter the password on the created role using \password in psql to ensure
- # that clientside password changes use the scram_iterations value when
- # calculating SCRAM secrets.
- my $session = $node->interactive_psql('postgres');
-
- $session->set_query_timer_restart();
- $session->query("SET password_encryption='scram-sha-256';");
- $session->query("SET scram_iterations=42;");
- $session->query_until(qr/Enter new password/,
- "\\password scram_role_iter\n");
- $session->query_until(qr/Enter it again/, "pass\n");
- $session->query_until(qr/postgres=# /, "pass\n");
- $session->quit;
-
- $res = $node->safe_psql(
- 'postgres',
+$session->do("SET password_encryption='scram-sha-256';",
+ "SET scram_iterations=42;");
+$res = $session->set_password("scram_role_iter","pass");
+is($res->{status}, 1, "set password ok");
+$res = $session->query_oneval(
"SELECT substr(rolpassword,1,17)
FROM pg_authid
WHERE rolname = 'scram_role_iter'");
- is($res, 'SCRAM-SHA-256$42:',
- 'scram_iterations in psql \password command');
-}
+is($res, 'SCRAM-SHA-256$42:', 'scram_iterations correct');
# Create a database to test regular expression.
$node->safe_psql('postgres', "CREATE database regex_testdb;");
diff --git a/src/test/authentication/t/007_pre_auth.pl b/src/test/authentication/t/007_pre_auth.pl
index 04063f4721d..17de92bc41a 100644
--- a/src/test/authentication/t/007_pre_auth.pl
+++ b/src/test/authentication/t/007_pre_auth.pl
@@ -7,6 +7,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Time::HiRes qw(usleep);
use Test::More;
@@ -36,24 +37,27 @@ if (!$node->check_extension('injection_points'))
$node->safe_psql('postgres', 'CREATE EXTENSION injection_points');
# Connect to the server and inject a waitpoint.
-my $psql = $node->background_psql('postgres');
-$psql->query_safe("SELECT injection_points_attach('init-pre-auth', 'wait')");
+my $session = PostgreSQL::Test::Session->new(node => $node);
+$session->do("SELECT injection_points_attach('init-pre-auth', 'wait')");
# From this point on, all new connections will hang during startup, just before
-# authentication. Use the $psql connection handle for server interaction.
-my $conn = $node->background_psql('postgres', wait => 0);
+# authentication. Use the $session connection handle for server interaction.
+my $conn = PostgreSQL::Test::Session->new(node => $node, wait => 0);
# Wait for the connection to show up in pg_stat_activity, with the wait_event
-# of the injection point.
+# of the injection point. We need to poll the async connection to drive it forward.
my $pid;
while (1)
{
- $pid = $psql->query(
+ # Drive the async connection forward - it won't progress without polling
+ $conn->poll_connect();
+
+ $pid = $session->query_oneval(
qq{SELECT pid FROM pg_stat_activity
WHERE backend_type = 'client backend'
AND state = 'starting'
- AND wait_event = 'init-pre-auth';});
- last if $pid ne "";
+ AND wait_event = 'init-pre-auth';}, 1);
+ last if defined $pid && $pid ne "";
usleep(100_000);
}
@@ -62,24 +66,24 @@ note "backend $pid is authenticating";
ok(1, 'authenticating connections are recorded in pg_stat_activity');
# Detach the waitpoint and wait for the connection to complete.
-$psql->query_safe("SELECT injection_points_wakeup('init-pre-auth');");
+$session->do("SELECT injection_points_wakeup('init-pre-auth')");
$conn->wait_connect();
# Make sure the pgstat entry is updated eventually.
while (1)
{
my $state =
- $psql->query("SELECT state FROM pg_stat_activity WHERE pid = $pid;");
- last if $state eq "idle";
+ $session->query_oneval("SELECT state FROM pg_stat_activity WHERE pid = $pid");
+ last if defined $state && $state eq "idle";
- note "state for backend $pid is '$state'; waiting for 'idle'...";
+ note "state for backend $pid is '" . ($state // 'undef') . "'; waiting for 'idle'...";
usleep(100_000);
}
ok(1, 'authenticated connections reach idle state in pg_stat_activity');
-$psql->query_safe("SELECT injection_points_detach('init-pre-auth');");
-$psql->quit();
-$conn->quit();
+$session->do("SELECT injection_points_detach('init-pre-auth')");
+$session->close();
+$conn->close();
done_testing();
diff --git a/src/test/modules/oauth_validator/t/001_server.pl b/src/test/modules/oauth_validator/t/001_server.pl
index 6b649c0b06f..bd92c381a67 100644
--- a/src/test/modules/oauth_validator/t/001_server.pl
+++ b/src/test/modules/oauth_validator/t/001_server.pl
@@ -12,6 +12,7 @@ use warnings FATAL => 'all';
use JSON::PP qw(encode_json);
use MIME::Base64 qw(encode_base64);
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -57,7 +58,7 @@ $node->safe_psql('postgres', 'CREATE USER testalt;');
$node->safe_psql('postgres', 'CREATE USER testparam;');
# Save a background connection for later configuration changes.
-my $bgconn = $node->background_psql('postgres');
+my $bgconn = PostgreSQL::Test::session->new(node => $node);;
my $webserver = OAuth::Server->new();
$webserver->run();
@@ -86,11 +87,11 @@ $node->reload;
my $log_start = $node->wait_for_log(qr/reloading configuration files/);
# Check pg_hba_file_rules() support.
-my $contents = $bgconn->query_safe(
+my $contents = $bgconn->query(
qq(SELECT rule_number, auth_method, options
FROM pg_hba_file_rules
ORDER BY rule_number;));
-is( $contents,
+is( $contents->{psqlout},
qq{1|oauth|\{issuer=$issuer,"scope=openid postgres",validator=validator\}
2|oauth|\{issuer=$issuer/.well-known/oauth-authorization-server/alternate,"scope=openid postgres alt",validator=validator\}
3|oauth|\{issuer=$issuer/param,"scope=openid postgres",validator=validator\}},
@@ -477,7 +478,7 @@ $common_connstr =
"dbname=postgres oauth_issuer=$issuer/.well-known/openid-configuration oauth_scope='' oauth_client_id=f02c6361-0635";
# Misbehaving validators must fail shut.
-$bgconn->query_safe("ALTER SYSTEM SET oauth_validator.authn_id TO ''");
+$bgconn->do("ALTER SYSTEM SET oauth_validator.authn_id TO ''");
$node->reload;
$log_start =
$node->wait_for_log(qr/reloading configuration files/, $log_start);
@@ -494,9 +495,9 @@ $node->connect_fails(
# Even if a validator authenticates the user, if the token isn't considered
# valid, the connection fails.
-$bgconn->query_safe(
+$bgconn->do(
"ALTER SYSTEM SET oauth_validator.authn_id TO 'test\@example.org'");
-$bgconn->query_safe(
+$bgconn->do(
"ALTER SYSTEM SET oauth_validator.authorize_tokens TO false");
$node->reload;
$log_start =
@@ -533,8 +534,8 @@ local all testparam oauth issuer="$issuer" scope="" delegate_ident_mapping=1
});
# To start, have the validator use the role names as authn IDs.
-$bgconn->query_safe("ALTER SYSTEM RESET oauth_validator.authn_id");
-$bgconn->query_safe("ALTER SYSTEM RESET oauth_validator.authorize_tokens");
+$bgconn->do("ALTER SYSTEM RESET oauth_validator.authn_id");
+$bgconn->do("ALTER SYSTEM RESET oauth_validator.authorize_tokens");
$node->reload;
$log_start =
@@ -551,7 +552,7 @@ $node->connect_fails(
expected_stderr => qr/OAuth bearer authentication failed/);
# Have the validator identify the end user as [email protected].
-$bgconn->query_safe(
+$bgconn->do(
"ALTER SYSTEM SET oauth_validator.authn_id TO 'user\@example.com'");
$node->reload;
$log_start =
@@ -575,7 +576,7 @@ $node->connect_ok(
expected_stderr =>
qr@Visit https://example\.com/ and enter the code: postgresuser@);
-$bgconn->query_safe("ALTER SYSTEM RESET oauth_validator.authn_id");
+$bgconn->do("ALTER SYSTEM RESET oauth_validator.authn_id");
$node->reload;
$log_start =
$node->wait_for_log(qr/reloading configuration files/, $log_start);
diff --git a/src/test/modules/test_aio/t/001_aio.pl b/src/test/modules/test_aio/t/001_aio.pl
index 5c634ec3ca9..38177886300 100644
--- a/src/test/modules/test_aio/t/001_aio.pl
+++ b/src/test/modules/test_aio/t/001_aio.pl
@@ -4,6 +4,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -132,19 +133,19 @@ sub psql_like
{
local $Test::Builder::Level = $Test::Builder::Level + 1;
my $io_method = shift;
- my $psql = shift;
+ my $session = shift;
my $name = shift;
my $sql = shift;
my $expected_stdout = shift;
my $expected_stderr = shift;
- my ($cmdret, $output);
- ($output, $cmdret) = $psql->query($sql);
+ my $res = $session->query($sql);
+ my $output = $res->{psqlout};
like($output, $expected_stdout, "$io_method: $name: expected stdout");
- like($psql->{stderr}, $expected_stderr,
+ like($session->get_stderr(), $expected_stderr,
"$io_method: $name: expected stderr");
- $psql->{stderr} = '';
+ $session->clear_stderr();
return $output;
}
@@ -154,15 +155,14 @@ sub query_wait_block
local $Test::Builder::Level = $Test::Builder::Level + 1;
my $io_method = shift;
my $node = shift;
- my $psql = shift;
+ my $session = shift;
my $name = shift;
my $sql = shift;
my $waitfor = shift;
- my $pid = $psql->query_safe('SELECT pg_backend_pid()');
+ my $pid = $session->backend_pid();
- $psql->{stdin} .= qq($sql;\n);
- $psql->{run}->pump_nb();
+ $session->do_async($sql);
ok(1, "$io_method: $name: issued sql");
$node->poll_query_until('postgres',
@@ -217,7 +217,7 @@ sub test_handle
my $io_method = shift;
my $node = shift;
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
# leak warning: implicit xact
psql_like(
@@ -323,7 +323,7 @@ sub test_batchmode
my $io_method = shift;
my $node = shift;
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
# In a build with RELCACHE_FORCE_RELEASE and CATCACHE_FORCE_RELEASE, just
# using SELECT batch_start() causes spurious test failures, because the
@@ -383,7 +383,7 @@ sub test_io_error
my $node = shift;
my ($ret, $output);
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
$psql->query_safe(
qq(
@@ -435,8 +435,8 @@ sub test_startwait_io
my $node = shift;
my ($ret, $output);
- my $psql_a = $node->background_psql('postgres', on_error_stop => 0);
- my $psql_b = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql_a = PostgreSQL::Test::Session->new(node => $node);
+ my $psql_b = PostgreSQL::Test::Session->new(node => $node);
### Verify behavior for normal tables
@@ -494,7 +494,7 @@ sub test_startwait_io
# Because the IO was terminated, but not marked as valid, second session should get the right to start io
- pump_until($psql_b->{run}, $psql_b->{timeout}, \$psql_b->{stdout}, qr/t/);
+ $psql_b->wait_for_async_pattern(qr/t/);
ok(1, "$io_method: blocking start buffer io, can start io");
# terminate the IO again
@@ -531,7 +531,7 @@ sub test_startwait_io
qr/^$/);
# Because the IO was terminated, and marked as valid, second session should complete but not need io
- pump_until($psql_b->{run}, $psql_b->{timeout}, \$psql_b->{stdout}, qr/f/);
+ $psql_b->wait_for_async_pattern(qr/f/);
ok(1, "$io_method: blocking start buffer io, no need to start io");
# buffer is valid now, make it invalid again
@@ -610,8 +610,8 @@ sub test_complete_foreign
my $node = shift;
my ($ret, $output);
- my $psql_a = $node->background_psql('postgres', on_error_stop => 0);
- my $psql_b = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql_a = PostgreSQL::Test::Session->new(node => $node);
+ my $psql_b = PostgreSQL::Test::Session->new(node => $node);
# Issue IO without waiting for completion, then sleep
$psql_a->query_safe(
@@ -680,7 +680,7 @@ sub test_close_fd
my $node = shift;
my ($ret, $output);
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
psql_like(
$io_method,
@@ -730,7 +730,7 @@ sub test_inject
my $node = shift;
my ($ret, $output);
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
# injected what we'd expect
$psql->query_safe(qq(SELECT inj_io_short_read_attach(8192);));
@@ -864,7 +864,7 @@ sub test_inject_worker
my $node = shift;
my ($ret, $output);
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
# trigger a failure to reopen, should error out, but should recover
$psql->query_safe(
@@ -901,7 +901,7 @@ sub test_invalidate
my $io_method = shift;
my $node = shift;
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
foreach my $persistency (qw(normal unlogged temporary))
{
@@ -959,8 +959,8 @@ sub test_zero
my $io_method = shift;
my $node = shift;
- my $psql_a = $node->background_psql('postgres', on_error_stop => 0);
- my $psql_b = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql_a = PostgreSQL::Test::Session->new(node => $node);
+ my $psql_b = PostgreSQL::Test::Session->new(node => $node);
foreach my $persistency (qw(normal temporary))
{
@@ -985,7 +985,7 @@ SELECT modify_rel_block('tbl_zero', 0, corrupt_header=>true);
qq(
SELECT read_rel_block_ll('tbl_zero', 0, zero_on_error=>false)),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: invalid page in block 0 of relation "base\/.*\/.*$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 0 of relation "base\/.*\/.*$/
);
# Check that page validity errors are zeroed
@@ -996,7 +996,7 @@ SELECT read_rel_block_ll('tbl_zero', 0, zero_on_error=>false)),
qq(
SELECT read_rel_block_ll('tbl_zero', 0, zero_on_error=>true)),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: invalid page in block 0 of relation "base\/.*\/.*"; zeroing out page$/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: invalid page in block 0 of relation "base\/.*\/.*"; zeroing out page$/
);
# And that once the corruption is fixed, we can read again
@@ -1004,7 +1004,7 @@ SELECT read_rel_block_ll('tbl_zero', 0, zero_on_error=>true)),
qq(
SELECT modify_rel_block('tbl_zero', 0, zero=>true);
));
- $psql_a->{stderr} = '';
+ $psql_a->clear_stderr();
psql_like(
$io_method,
@@ -1027,7 +1027,7 @@ SELECT modify_rel_block('tbl_zero', 3, corrupt_header=>true);
"$persistency: test zeroing of invalid block 3",
qq(SELECT read_rel_block_ll('tbl_zero', 3, zero_on_error=>true);),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: invalid page in block 3 of relation "base\/.*\/.*"; zeroing out page$/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: invalid page in block 3 of relation "base\/.*\/.*"; zeroing out page$/
);
@@ -1044,7 +1044,7 @@ SELECT modify_rel_block('tbl_zero', 3, corrupt_header=>true);
"$persistency: test reading of invalid block 2,3 in larger read",
qq(SELECT read_rel_block_ll('tbl_zero', 1, nblocks=>4, zero_on_error=>false)),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: 2 invalid pages among blocks 1..4 of relation "base\/.*\/.*\nDETAIL: Block 2 held the first invalid page\.\nHINT:[^\n]+$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: 2 invalid pages among blocks 1..4 of relation "base\/.*\/.*\nDETAIL: Block 2 held the first invalid page\.\nHINT:[^\n]+$/
);
# Then test zeroing via ZERO_ON_ERROR flag
@@ -1054,7 +1054,7 @@ SELECT modify_rel_block('tbl_zero', 3, corrupt_header=>true);
"$persistency: test zeroing of invalid block 2,3 in larger read, ZERO_ON_ERROR",
qq(SELECT read_rel_block_ll('tbl_zero', 1, nblocks=>4, zero_on_error=>true)),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: zeroing out 2 invalid pages among blocks 1..4 of relation "base\/.*\/.*\nDETAIL: Block 2 held the first zeroed page\.\nHINT:[^\n]+$/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: zeroing out 2 invalid pages among blocks 1..4 of relation "base\/.*\/.*\nDETAIL: Block 2 held the first zeroed page\.\nHINT:[^\n]+$/
);
# Then test zeroing via zero_damaged_pages
@@ -1069,7 +1069,7 @@ SELECT read_rel_block_ll('tbl_zero', 1, nblocks=>4, zero_on_error=>false)
COMMIT;
),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: zeroing out 2 invalid pages among blocks 1..4 of relation "base\/.*\/.*\nDETAIL: Block 2 held the first zeroed page\.\nHINT:[^\n]+$/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: zeroing out 2 invalid pages among blocks 1..4 of relation "base\/.*\/.*\nDETAIL: Block 2 held the first zeroed page\.\nHINT:[^\n]+$/
);
$psql_a->query_safe(qq(COMMIT));
@@ -1082,7 +1082,7 @@ SELECT invalidate_rel_block('tbl_zero', g.i)
FROM generate_series(0, 15) g(i);
SELECT modify_rel_block('tbl_zero', 3, zero=>true);
));
- $psql_a->{stderr} = '';
+ $psql_a->clear_stderr();
psql_like(
$io_method,
@@ -1091,7 +1091,7 @@ SELECT modify_rel_block('tbl_zero', 3, zero=>true);
qq(
SELECT count(*) FROM tbl_zero),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: invalid page in block 2 of relation "base\/.*\/.*$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 2 of relation "base\/.*\/.*$/
);
# Verify that bufmgr.c IO zeroes out pages with page validity errors
@@ -1106,7 +1106,7 @@ SELECT count(*) FROM tbl_zero;
COMMIT;
),
qr/^\d+$/,
- qr/^psql:<stdin>:\d+: WARNING: invalid page in block 2 of relation "base\/.*\/.*$/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: invalid page in block 2 of relation "base\/.*\/.*$/
);
# Check that warnings/errors about page validity in an IO started by
@@ -1145,7 +1145,7 @@ DROP TABLE tbl_zero;
));
}
- $psql_a->{stderr} = '';
+ $psql_a->clear_stderr();
$psql_a->quit();
$psql_b->quit();
@@ -1157,19 +1157,17 @@ sub test_checksum
my $io_method = shift;
my $node = shift;
- my $psql_a = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql_a = PostgreSQL::Test::Session->new(node => $node);
- $psql_a->query_safe(
- qq(
-CREATE TABLE tbl_normal(id int) WITH (AUTOVACUUM_ENABLED = false);
-INSERT INTO tbl_normal SELECT generate_series(1, 5000);
-SELECT modify_rel_block('tbl_normal', 3, corrupt_checksum=>true);
-
-CREATE TEMPORARY TABLE tbl_temp(id int) WITH (AUTOVACUUM_ENABLED = false);
-INSERT INTO tbl_temp SELECT generate_series(1, 5000);
-SELECT modify_rel_block('tbl_temp', 3, corrupt_checksum=>true);
-SELECT modify_rel_block('tbl_temp', 4, corrupt_checksum=>true);
-));
+ # Split multi-statement query into separate calls to match psql behavior
+ # where errors in one statement don't prevent subsequent statements
+ $psql_a->query_safe(qq(CREATE TABLE tbl_normal(id int) WITH (AUTOVACUUM_ENABLED = false)));
+ $psql_a->query_safe(qq(INSERT INTO tbl_normal SELECT generate_series(1, 5000)));
+ $psql_a->query_safe(qq(SELECT modify_rel_block('tbl_normal', 3, corrupt_checksum=>true)));
+ $psql_a->query_safe(qq(CREATE TEMPORARY TABLE tbl_temp(id int) WITH (AUTOVACUUM_ENABLED = false)));
+ $psql_a->query_safe(qq(INSERT INTO tbl_temp SELECT generate_series(1, 5000)));
+ $psql_a->query_safe(qq(SELECT modify_rel_block('tbl_temp', 3, corrupt_checksum=>true)));
+ $psql_a->query_safe(qq(SELECT modify_rel_block('tbl_temp', 4, corrupt_checksum=>true)));
# To be able to test checksum failures on shared rels we need a shared rel
# with invalid pages - which is a bit scary. pg_shseclabel seems like a
@@ -1192,7 +1190,7 @@ SELECT modify_rel_block('pg_shseclabel', 3, corrupt_checksum=>true);
qq(
SELECT read_rel_block_ll('tbl_normal', 3, nblocks=>1, zero_on_error=>false);),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: invalid page in block 3 of relation "base\/\d+\/\d+"$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 3 of relation "base\/\d+\/\d+"$/
);
my ($cs_count_after, $cs_ts_after) =
@@ -1214,7 +1212,7 @@ SELECT read_rel_block_ll('tbl_normal', 3, nblocks=>1, zero_on_error=>false);),
qq(
SELECT read_rel_block_ll('tbl_temp', 4, nblocks=>2, zero_on_error=>false);),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: invalid page in block 4 of relation "base\/\d+\/t\d+_\d+"$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 4 of relation "base\/\d+\/t\d+_\d+"$/
);
($cs_count_after, $cs_ts_after) = checksum_failures($psql_a, 'postgres');
@@ -1235,7 +1233,7 @@ SELECT read_rel_block_ll('tbl_temp', 4, nblocks=>2, zero_on_error=>false);),
qq(
SELECT read_rel_block_ll('pg_shseclabel', 2, nblocks=>2, zero_on_error=>false);),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: 2 invalid pages among blocks 2..3 of relation "global\/\d+"\nDETAIL: Block 2 held the first invalid page\.\nHINT:[^\n]+$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: 2 invalid pages among blocks 2..3 of relation "global\/\d+"\nDETAIL: Block 2 held the first invalid page\.\nHINT:[^\n]+$/
);
($cs_count_after, $cs_ts_after) = checksum_failures($psql_a);
@@ -1253,7 +1251,7 @@ SELECT read_rel_block_ll('pg_shseclabel', 2, nblocks=>2, zero_on_error=>false);)
SELECT modify_rel_block('pg_shseclabel', 1, zero=>true);
DROP TABLE tbl_normal;
));
- $psql_a->{stderr} = '';
+ $psql_a->clear_stderr();
$psql_a->quit();
}
@@ -1266,7 +1264,7 @@ sub test_checksum_createdb
my $io_method = shift;
my $node = shift;
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
$node->safe_psql('postgres',
'CREATE DATABASE regression_createdb_source');
@@ -1300,7 +1298,7 @@ STRATEGY wal_log;
"create database w/ wal strategy, invalid source",
$createdb_sql,
qr/^$/,
- qr/psql:<stdin>:\d+: ERROR: invalid page in block 1 of relation "base\/\d+\/\d+"$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 1 of relation "base\/\d+\/\d+"$/
);
my ($cs_count_after, $cs_ts_after) =
checksum_failures($psql, 'regression_createdb_source');
@@ -1329,7 +1327,7 @@ sub test_ignore_checksum
my $io_method = shift;
my $node = shift;
- my $psql = $node->background_psql('postgres', on_error_stop => 0);
+ my $psql = PostgreSQL::Test::Session->new(node => $node);
# Test setup
$psql->query_safe(
@@ -1394,7 +1392,7 @@ SELECT modify_rel_block('tbl_cs_fail', 4, corrupt_header=>true);
qq(
SELECT read_rel_block_ll('tbl_cs_fail', 3, nblocks=>1, zero_on_error=>false);),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: ignoring checksum failure in block 3/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: ignoring checksum failure in block 3/
);
# Check that the log contains a LOG message about the failure
@@ -1409,7 +1407,7 @@ SELECT read_rel_block_ll('tbl_cs_fail', 3, nblocks=>1, zero_on_error=>false);),
qq(
SELECT read_rel_block_ll('tbl_cs_fail', 2, nblocks=>3, zero_on_error=>false);),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: invalid page in block 4 of relation "base\/\d+\/\d+"$/
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 4 of relation "base\/\d+\/\d+"$/
);
# Test multi-block read with different problems in different blocks
@@ -1421,7 +1419,7 @@ SELECT modify_rel_block('tbl_cs_fail', 3, corrupt_checksum=>true, corrupt_header
SELECT modify_rel_block('tbl_cs_fail', 4, corrupt_header=>true);
SELECT modify_rel_block('tbl_cs_fail', 5, corrupt_header=>true);
));
- $psql->{stderr} = '';
+ $psql->clear_stderr();
$log_location = -s $node->logfile;
psql_like(
@@ -1431,7 +1429,7 @@ SELECT modify_rel_block('tbl_cs_fail', 5, corrupt_header=>true);
qq(
SELECT read_rel_block_ll('tbl_cs_fail', 1, nblocks=>5, zero_on_error=>true);),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: zeroing 3 page\(s\) and ignoring 2 checksum failure\(s\) among blocks 1..5 of relation "/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: zeroing 3 page\(s\) and ignoring 2 checksum failure\(s\) among blocks 1..5 of relation "/
);
@@ -1464,7 +1462,7 @@ SELECT read_rel_block_ll('tbl_cs_fail', 1, nblocks=>5, zero_on_error=>true);),
qq(
SELECT modify_rel_block('tbl_cs_fail', 3, corrupt_checksum=>true, corrupt_header=>true);
));
- $psql->{stderr} = '';
+ $psql->clear_stderr();
psql_like(
$io_method,
@@ -1473,7 +1471,7 @@ SELECT modify_rel_block('tbl_cs_fail', 3, corrupt_checksum=>true, corrupt_header
qq(
SELECT read_rel_block_ll('tbl_cs_fail', 3, nblocks=>1, zero_on_error=>false);),
qr/^$/,
- qr/^psql:<stdin>:\d+: ERROR: invalid page in block 3 of relation "/);
+ qr/^(?:psql:<stdin>:\d+: )?ERROR: invalid page in block 3 of relation "/);
psql_like(
$io_method,
@@ -1482,7 +1480,7 @@ SELECT read_rel_block_ll('tbl_cs_fail', 3, nblocks=>1, zero_on_error=>false);),
qq(
SELECT read_rel_block_ll('tbl_cs_fail', 3, nblocks=>1, zero_on_error=>true);),
qr/^$/,
- qr/^psql:<stdin>:\d+: WARNING: invalid page in block 3 of relation "base\/.*"; zeroing out page/
+ qr/^(?:psql:<stdin>:\d+: )?WARNING: invalid page in block 3 of relation "base\/.*"; zeroing out page/
);
diff --git a/src/test/modules/test_misc/t/005_timeouts.pl b/src/test/modules/test_misc/t/005_timeouts.pl
index c16b7dbf5e6..d64a787a9ab 100644
--- a/src/test/modules/test_misc/t/005_timeouts.pl
+++ b/src/test/modules/test_misc/t/005_timeouts.pl
@@ -6,6 +6,7 @@ use warnings FATAL => 'all';
use locale;
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Time::HiRes qw(usleep);
use Test::More;
@@ -42,24 +43,16 @@ $node->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
$node->safe_psql('postgres',
"SELECT injection_points_attach('transaction-timeout', 'wait');");
-my $psql_session = $node->background_psql('postgres');
+my $psql_session = PostgreSQL::Test::Session->new(node => $node);
-# The following query will generate a stream of SELECT 1 queries. This is done
-# so to exercise transaction timeout in the presence of short queries.
-# Note: the interval value is parsed with locale-aware strtod()
-$psql_session->query_until(
- qr/starting_bg_psql/,
- sprintf(
- q(\echo starting_bg_psql
- SET transaction_timeout to '10ms';
- BEGIN;
- SELECT 1 \watch %g
- \q
-), 0.001));
+$psql_session->do("SET transaction_timeout to '10ms';");
+
+$psql_session->do_async("BEGIN; DO ' begin loop PERFORM pg_sleep(0.001); end loop; end ';");
# Wait until the backend enters the timeout injection point. Will get an error
# here if anything goes wrong.
$node->wait_for_event('client backend', 'transaction-timeout');
+pass("got transaction timeout event");
my $log_offset = -s $node->logfile;
@@ -70,11 +63,9 @@ $node->safe_psql('postgres',
# Check that the timeout was logged.
$node->wait_for_log('terminating connection due to transaction timeout',
$log_offset);
+pass("got transaction timeout log");
-# If we send \q with $psql_session->quit the command can be sent to the session
-# already closed. So \q is in initial script, here we only finish IPC::Run.
-$psql_session->{run}->finish;
-
+$psql_session->close;
#
# 2. Test of the idle in transaction timeout
@@ -85,10 +76,8 @@ $node->safe_psql('postgres',
);
# We begin a transaction and the hand on the line
-$psql_session = $node->background_psql('postgres');
-$psql_session->query_until(
- qr/starting_bg_psql/, q(
- \echo starting_bg_psql
+$psql_session->reconnect;
+$psql_session->do(q(
SET idle_in_transaction_session_timeout to '10ms';
BEGIN;
));
@@ -96,6 +85,7 @@ $psql_session->query_until(
# Wait until the backend enters the timeout injection point.
$node->wait_for_event('client backend',
'idle-in-transaction-session-timeout');
+pass("got idle in transaction timeout event");
$log_offset = -s $node->logfile;
@@ -106,8 +96,9 @@ $node->safe_psql('postgres',
# Check that the timeout was logged.
$node->wait_for_log(
'terminating connection due to idle-in-transaction timeout', $log_offset);
+pass("got idle in transaction timeout log");
-ok($psql_session->quit);
+$psql_session->close;
#
@@ -117,15 +108,14 @@ $node->safe_psql('postgres',
"SELECT injection_points_attach('idle-session-timeout', 'wait');");
# We just initialize the GUC and wait. No transaction is required.
-$psql_session = $node->background_psql('postgres');
-$psql_session->query_until(
- qr/starting_bg_psql/, q(
- \echo starting_bg_psql
+$psql_session->reconnect;
+$psql_session->do(q(
SET idle_session_timeout to '10ms';
));
# Wait until the backend enters the timeout injection point.
$node->wait_for_event('client backend', 'idle-session-timeout');
+pass("got idle session timeout event");
$log_offset = -s $node->logfile;
@@ -136,7 +126,8 @@ $node->safe_psql('postgres',
# Check that the timeout was logged.
$node->wait_for_log('terminating connection due to idle-session timeout',
$log_offset);
+pass("got idle sesion tiemout log");
-ok($psql_session->quit);
+$psql_session->close;
done_testing();
diff --git a/src/test/modules/test_misc/t/007_catcache_inval.pl b/src/test/modules/test_misc/t/007_catcache_inval.pl
index 424556261c9..e3b3fd8bae9 100644
--- a/src/test/modules/test_misc/t/007_catcache_inval.pl
+++ b/src/test/modules/test_misc/t/007_catcache_inval.pl
@@ -46,12 +46,12 @@ $node->safe_psql(
CREATE FUNCTION foofunc(dummy integer) RETURNS integer AS \$\$ SELECT 1; /* $longtext */ \$\$ LANGUAGE SQL
]);
-my $psql_session = $node->background_psql('postgres');
-my $psql_session2 = $node->background_psql('postgres');
+my $psql_session = PostgreSQL::Test::Session->new(node => $node);
+my $psql_session2 = PostgreSQL::Test::Session->new(node => $node);
# Set injection point in the session, to pause while populating the
# catcache list
-$psql_session->query_safe(
+$psql_session->do(
qq[
SELECT injection_points_set_local();
SELECT injection_points_attach('catcache-list-miss-systable-scan-started', 'wait');
@@ -59,10 +59,9 @@ $psql_session->query_safe(
# This pauses on the injection point while populating catcache list
# for functions with name "foofunc"
-$psql_session->query_until(
- qr/starting_bg_psql/, q(
- \echo starting_bg_psql
- SELECT foofunc(1);
+$psql_session->do_async(
+ q(
+ SELECT foofunc(1);
));
# While the first session is building the catcache list, create a new
@@ -83,16 +82,19 @@ $node->safe_psql(
# trying to exercise here.)
#
# The "SELECT foofunc(1)" query will now finish.
-$psql_session2->query_safe(
+$psql_session2->do(
qq[
SELECT injection_points_wakeup('catcache-list-miss-systable-scan-started');
SELECT injection_points_detach('catcache-list-miss-systable-scan-started');
]);
# Test that the new function is visible to the session.
-$psql_session->query_safe("SELECT foofunc();");
+$psql_session->wait_for_completion;
+my $res = $psql_session->query("SELECT foofunc();");
-ok($psql_session->quit);
-ok($psql_session2->quit);
+is($res->{status}, 2, "got TUPLES_OK");
+
+$psql_session->close;
+$psql_session2->close;
done_testing();
diff --git a/src/test/modules/test_misc/t/010_index_concurrently_upsert.pl b/src/test/modules/test_misc/t/010_index_concurrently_upsert.pl
index 3bdb632887e..cca3b3a11ea 100644
--- a/src/test/modules/test_misc/t/010_index_concurrently_upsert.pl
+++ b/src/test/modules/test_misc/t/010_index_concurrently_upsert.pl
@@ -13,6 +13,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
use Time::HiRes qw(usleep);
@@ -51,47 +52,38 @@ ALTER TABLE test.tblexpr SET (parallel_workers=0);
############################################################################
note('Test: REINDEX CONCURRENTLY + UPSERT (wakeup at set-dead phase)');
-# Create sessions with on_error_stop => 0 so psql doesn't exit on SQL errors.
-# This allows us to collect stderr and detect errors after the test completes.
-my $s1 = $node->background_psql('postgres', on_error_stop => 0);
-my $s2 = $node->background_psql('postgres', on_error_stop => 0);
-my $s3 = $node->background_psql('postgres', on_error_stop => 0);
+# Create sessions for concurrent operations
+my $s1 = PostgreSQL::Test::Session->new(node => $node);
+my $s2 = PostgreSQL::Test::Session->new(node => $node);
+my $s3 = PostgreSQL::Test::Session->new(node => $node);
# Setup injection points for each session
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
]);
# s3 starts REINDEX (will block on reindex-relation-concurrently-before-set-dead)
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tblpk_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tblpk_pkey;]);
# Wait for s3 to hit injection point
ok_injection_point($node, 'reindex-relation-concurrently-before-set-dead');
# s1 starts UPSERT (will block on check-exclusion-or-unique-constraint-no-conflict)
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
# Wait for s1 to hit injection point
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
@@ -101,11 +93,7 @@ wakeup_injection_point($node,
'reindex-relation-concurrently-before-set-dead');
# s2 starts UPSERT (will block on exec-insert-before-insert-speculative)
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
# Wait for s2 to hit injection point
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -125,51 +113,39 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblpk');
############################################################################
note('Test: REINDEX CONCURRENTLY + UPSERT (wakeup at swap phase)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tblpk_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tblpk_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
wakeup_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -184,50 +160,38 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblpk');
############################################################################
note('Test: REINDEX CONCURRENTLY + UPSERT (s1 wakes before reindex)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tblpk_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tblpk_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-set-dead');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
# Start s2 BEFORE waking reindex (key difference from permutation 1)
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -245,52 +209,40 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblpk');
############################################################################
note('Test: REINDEX + UPSERT ON CONSTRAINT (set-dead phase)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tblpk_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tblpk_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-set-dead');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
wakeup_injection_point($node,
'reindex-relation-concurrently-before-set-dead');
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -305,51 +257,39 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblpk');
############################################################################
note('Test: REINDEX + UPSERT ON CONSTRAINT (swap phase)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tblpk_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tblpk_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
wakeup_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -364,50 +304,38 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblpk');
############################################################################
note('Test: REINDEX + UPSERT ON CONSTRAINT (s1 wakes before reindex)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tblpk_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tblpk_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-set-dead');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
# Start s2 BEFORE waking reindex
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13, now()) ON CONFLICT ON CONSTRAINT tblpk_pkey DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -425,52 +353,40 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblpk');
############################################################################
note('Test: REINDEX on partitioned table (set-dead phase)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-set-dead');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
wakeup_injection_point($node,
'reindex-relation-concurrently-before-set-dead');
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -485,51 +401,39 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblparted');
############################################################################
note('Test: REINDEX on partitioned table (swap phase)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
wakeup_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -544,50 +448,38 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblparted');
############################################################################
note('Test: REINDEX on partitioned table (s1 wakes before reindex)');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
]);
-$s3->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;
-]);
+$s3->do_async(q[REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-set-dead');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'check-exclusion-or-unique-constraint-no-conflict');
# Start s2 BEFORE waking reindex
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -605,35 +497,27 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblparted');
############################################################################
note('Test: REINDEX on partitioned table, cache inval between two get_partition_ancestors');
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-init-partition-after-get-partition-ancestors', 'wait');
]);
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
]);
-$s2->query_until(
- qr/starting_reindex/, q[
-\echo starting_reindex
-REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;
-]);
+$s2->do_async(q[REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey;]);
ok_injection_point($node, 'reindex-relation-concurrently-before-swap');
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblparted VALUES (13, now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node,
'exec-init-partition-after-get-partition-ancestors');
@@ -652,26 +536,24 @@ note('Test: CREATE INDEX CONCURRENTLY + UPSERT');
# Uses invalidate-catalog-snapshot-end to test catalog invalidation
# during UPSERT
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-my $s1_pid = $s1->query_safe('SELECT pg_backend_pid()');
+# Get the session's backend PID before attaching injection points
+my $s1_pid = $s1->query_oneval('SELECT pg_backend_pid()');
# s1 attaches BOTH injection points - the unique constraint check AND catalog snapshot
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s1->query_until(
- qr/attaching_injection_point/, q[
-\echo attaching_injection_point
-SELECT injection_points_attach('invalidate-catalog-snapshot-end', 'wait');
-]);
-
# In cases of cache clobbering, s1 may hit the injection point during attach.
+# Start attach asynchronously so we can check if it blocks.
+$s1->do_async(q[SELECT injection_points_attach('invalidate-catalog-snapshot-end', 'wait');]);
+
# Wait for that session to become idle (attach completed), or wake it up if
# it becomes stuck on injection point.
if (!wait_for_idle($node, $s1_pid))
@@ -685,34 +567,28 @@ if (!wait_for_idle($node, $s1_pid))
SELECT injection_points_wakeup('invalidate-catalog-snapshot-end');
]);
}
+# Wait for async command to complete
+$s1->wait_for_completion;
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('define-index-before-set-valid', 'wait');
]);
# s3: Start CREATE INDEX CONCURRENTLY (blocks on define-index-before-set-valid)
-$s3->query_until(
- qr/starting_create_index/, q[
-\echo starting_create_index
-CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_duplicate ON test.tblpk(i);
-]);
+$s3->do_async(q[CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_duplicate ON test.tblpk(i);]);
ok_injection_point($node, 'define-index-before-set-valid');
# s1: Start UPSERT (blocks on invalidate-catalog-snapshot-end)
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'invalidate-catalog-snapshot-end');
@@ -720,11 +596,7 @@ ok_injection_point($node, 'invalidate-catalog-snapshot-end');
wakeup_injection_point($node, 'define-index-before-set-valid');
# s2: Start UPSERT (blocks on exec-insert-before-insert-speculative)
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblpk VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
@@ -745,24 +617,20 @@ $node->safe_psql('postgres', 'TRUNCATE TABLE test.tblparted');
note('Test: CREATE INDEX CONCURRENTLY on partial index + UPSERT');
# Uses invalidate-catalog-snapshot-end to test catalog invalidation during UPSERT
-$s1 = $node->background_psql('postgres', on_error_stop => 0);
-$s2 = $node->background_psql('postgres', on_error_stop => 0);
-$s3 = $node->background_psql('postgres', on_error_stop => 0);
+$s1 = PostgreSQL::Test::Session->new(node => $node);
+$s2 = PostgreSQL::Test::Session->new(node => $node);
+$s3 = PostgreSQL::Test::Session->new(node => $node);
-$s1_pid = $s1->query_safe('SELECT pg_backend_pid()');
+$s1_pid = $s1->query_oneval('SELECT pg_backend_pid()');
# s1 attaches BOTH injection points - the unique constraint check AND catalog snapshot
-$s1->query_safe(
+$s1->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
]);
-$s1->query_until(
- qr/attaching_injection_point/, q[
-\echo attaching_injection_point
-SELECT injection_points_attach('invalidate-catalog-snapshot-end', 'wait');
-]);
+$s1->do(q[SELECT injection_points_attach('invalidate-catalog-snapshot-end', 'wait');]);
# In cases of cache clobbering, s1 may hit the injection point during attach.
# Wait for that session to become idle (attach completed), or wake it up if
@@ -779,33 +647,25 @@ if (!wait_for_idle($node, $s1_pid))
]);
}
-$s2->query_safe(
+$s2->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
]);
-$s3->query_safe(
+$s3->do(
q[
SELECT injection_points_set_local();
SELECT injection_points_attach('define-index-before-set-valid', 'wait');
]);
# s3: Start CREATE INDEX CONCURRENTLY (blocks on define-index-before-set-valid)
-$s3->query_until(
- qr/starting_create_index/, q[
-\echo starting_create_index
-CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_special_duplicate ON test.tblexpr(abs(i)) WHERE i < 10000;
-]);
+$s3->do_async(q[CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_special_duplicate ON test.tblexpr(abs(i)) WHERE i < 10000;]);
ok_injection_point($node, 'define-index-before-set-valid');
# s1: Start UPSERT (blocks on invalidate-catalog-snapshot-end)
-$s1->query_until(
- qr/starting_upsert_s1/, q[
-\echo starting_upsert_s1
-INSERT INTO test.tblexpr VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();
-]);
+$s1->do_async(q[INSERT INTO test.tblexpr VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'invalidate-catalog-snapshot-end');
@@ -813,11 +673,7 @@ ok_injection_point($node, 'invalidate-catalog-snapshot-end');
wakeup_injection_point($node, 'define-index-before-set-valid');
# s2: Start UPSERT (blocks on exec-insert-before-insert-speculative)
-$s2->query_until(
- qr/starting_upsert_s2/, q[
-\echo starting_upsert_s2
-INSERT INTO test.tblexpr VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();
-]);
+$s2->do_async(q[INSERT INTO test.tblexpr VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();]);
ok_injection_point($node, 'exec-insert-before-insert-speculative');
wakeup_injection_point($node, 'invalidate-catalog-snapshot-end');
@@ -918,33 +774,23 @@ SELECT injection_points_wakeup('$point_name');
]);
}
-# Wait for any pending query to complete, capture stderr, and close the session.
-# Returns the stderr output (excluding internal markers).
+# Wait for any pending query to complete and close the session.
+# Returns empty string on success, error message on failure.
sub safe_quit
{
my ($session) = @_;
- # Send a marker and wait for it to ensure any pending query completes
- my $banner = "safe_quit_marker";
- my $banner_match = qr/(^|\n)$banner\r?\n/;
+ # Wait for any async queries to complete
+ $session->wait_for_completion;
- $session->{stdin} .= "\\echo $banner\n\\warn $banner\n";
-
- pump_until(
- $session->{run}, $session->{timeout},
- \$session->{stdout}, $banner_match);
- pump_until(
- $session->{run}, $session->{timeout},
- \$session->{stderr}, $banner_match);
-
- # Capture stderr (excluding the banner)
- my $stderr = $session->{stderr};
- $stderr =~ s/$banner_match//;
+ # Check connection status
+ my $status = $session->conn_status;
# Close the session
- $session->quit;
+ $session->close;
- return $stderr;
+ # Return empty string if connection was OK, otherwise return error
+ return ($status == PostgreSQL::PqFFI::CONNECTION_OK()) ? '' : 'connection error';
}
# Helper function: verify that the given sessions exit cleanly.
diff --git a/src/test/modules/test_slru/t/001_multixact.pl b/src/test/modules/test_slru/t/001_multixact.pl
index f6f45895ebd..2d86434d7bc 100644
--- a/src/test/modules/test_slru/t/001_multixact.pl
+++ b/src/test/modules/test_slru/t/001_multixact.pl
@@ -6,6 +6,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -30,8 +31,8 @@ $node->safe_psql('postgres', q(CREATE EXTENSION test_slru));
# lost.
# Create the first multixact
-my $bg_psql = $node->background_psql('postgres');
-my $multi1 = $bg_psql->query_safe(q(SELECT test_create_multixact();));
+my $bg_session = PostgreSQL::Test::Session->new(node => $node);
+my $multi1 = $bg_session->query_oneval(q(SELECT test_create_multixact();));
# Assign the middle multixact. Use an injection point to prevent it
# from being fully recorded.
@@ -39,11 +40,9 @@ $node->safe_psql('postgres',
q{SELECT injection_points_attach('multixact-create-from-members','wait');}
);
-$bg_psql->query_until(
- qr/assigning lost multi/, q(
-\echo assigning lost multi
- SELECT test_create_multixact();
-));
+# Start the second multixact creation asynchronously - it will block at
+# the injection point
+$bg_session->do_async(q(SELECT test_create_multixact();));
$node->wait_for_event('client backend', 'multixact-create-from-members');
$node->safe_psql('postgres',
@@ -52,10 +51,10 @@ $node->safe_psql('postgres',
# Create the third multixid
my $multi2 = $node->safe_psql('postgres', q{SELECT test_create_multixact();});
-# All set and done, it's time for hard restart
+# All set and done, it's time for hard restart. The background session
+# will be terminated by the crash.
$node->stop('immediate');
$node->start;
-$bg_psql->{run}->finish;
# Verify that the recorded multixids are readable
is( $node->safe_psql('postgres', qq{SELECT test_read_multixact('$multi1');}),
diff --git a/src/test/modules/xid_wraparound/t/001_emergency_vacuum.pl b/src/test/modules/xid_wraparound/t/001_emergency_vacuum.pl
index 213f9052ed2..304076735b4 100644
--- a/src/test/modules/xid_wraparound/t/001_emergency_vacuum.pl
+++ b/src/test/modules/xid_wraparound/t/001_emergency_vacuum.pl
@@ -4,6 +4,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -47,17 +48,10 @@ CREATE TABLE small_trunc(id serial primary key, data text, filler text default r
INSERT INTO small_trunc(data) SELECT generate_series(1,15000);
]);
-# Bump the query timeout to avoid false negatives on slow test systems.
-my $psql_timeout_secs = 4 * $PostgreSQL::Test::Utils::timeout_default;
-
# Start a background session, which holds a transaction open, preventing
# autovacuum from advancing relfrozenxid and datfrozenxid.
-my $background_psql = $node->background_psql(
- 'postgres',
- on_error_stop => 0,
- timeout => $psql_timeout_secs);
-$background_psql->set_query_timer_restart();
-$background_psql->query_safe(
+my $background_session = PostgreSQL::Test::Session->new(node => $node);
+$background_session->do(
qq[
BEGIN;
DELETE FROM large WHERE id % 2 = 0;
@@ -89,8 +83,8 @@ my $log_offset = -s $node->logfile;
# Finish the old transaction, to allow vacuum freezing to advance
# relfrozenxid and datfrozenxid again.
-$background_psql->query_safe(qq[COMMIT]);
-$background_psql->quit;
+$background_session->do(qq[COMMIT;]);
+$background_session->close;
# Wait until autovacuum processed all tables and advanced the
# system-wide oldest-XID.
diff --git a/src/test/modules/xid_wraparound/t/002_limits.pl b/src/test/modules/xid_wraparound/t/002_limits.pl
index 86632a8d510..0ef67d0f4b8 100644
--- a/src/test/modules/xid_wraparound/t/002_limits.pl
+++ b/src/test/modules/xid_wraparound/t/002_limits.pl
@@ -10,6 +10,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
use Time::HiRes qw(usleep);
@@ -29,6 +30,8 @@ $node->append_conf(
'postgresql.conf', qq[
autovacuum_naptime = 1s
log_autovacuum_min_duration = 0
+log_connections = on
+log_statement = 'all'
]);
$node->start;
$node->safe_psql('postgres', 'CREATE EXTENSION xid_wraparound');
@@ -41,16 +44,10 @@ CREATE TABLE wraparoundtest(t text) WITH (autovacuum_enabled = off);
INSERT INTO wraparoundtest VALUES ('start');
]);
-# Bump the query timeout to avoid false negatives on slow test systems.
-my $psql_timeout_secs = 4 * $PostgreSQL::Test::Utils::timeout_default;
-
# Start a background session, which holds a transaction open, preventing
# autovacuum from advancing relfrozenxid and datfrozenxid.
-my $background_psql = $node->background_psql(
- 'postgres',
- on_error_stop => 0,
- timeout => $psql_timeout_secs);
-$background_psql->query_safe(
+my $background_session = PostgreSQL::Test::Session->new(node => $node);
+$background_session->do(
qq[
BEGIN;
INSERT INTO wraparoundtest VALUES ('oldxact');
@@ -108,8 +105,8 @@ like(
# Finish the old transaction, to allow vacuum freezing to advance
# relfrozenxid and datfrozenxid again.
-$background_psql->query_safe(qq[COMMIT]);
-$background_psql->quit;
+$background_session->do(qq[COMMIT;]);
+$background_session->close;
# VACUUM, to freeze the tables and advance datfrozenxid.
#
@@ -122,8 +119,8 @@ $node->safe_psql('postgres', 'VACUUM');
# the system-wide oldest-XID.
$ret =
$node->poll_query_until('postgres',
- qq[INSERT INTO wraparoundtest VALUES ('after VACUUM')],
- 'INSERT 0 1');
+ qq[INSERT INTO wraparoundtest VALUES ('after VACUUM') RETURNING true],
+ );
# Check the table contents
$ret = $node->safe_psql('postgres', qq[SELECT * from wraparoundtest]);
diff --git a/src/test/modules/xid_wraparound/t/004_notify_freeze.pl b/src/test/modules/xid_wraparound/t/004_notify_freeze.pl
index d0a1f1fe2fc..f2fa4e5a6b9 100644
--- a/src/test/modules/xid_wraparound/t/004_notify_freeze.pl
+++ b/src/test/modules/xid_wraparound/t/004_notify_freeze.pl
@@ -7,6 +7,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use Test::More;
my $node = PostgreSQL::Test::Cluster->new('node');
@@ -24,9 +25,9 @@ $node->safe_psql('postgres',
'ALTER DATABASE template0 WITH ALLOW_CONNECTIONS true');
# Start Session 1 and leave it idle in transaction
-my $psql_session1 = $node->background_psql('postgres');
-$psql_session1->query_safe('listen s;');
-$psql_session1->query_safe('begin;');
+my $session1 = PostgreSQL::Test::Session->new(node => $node);
+$session1->do('LISTEN s');
+$session1->do('BEGIN');
# Send some notifys from other sessions
for my $i (1 .. 10)
@@ -54,18 +55,20 @@ my $datafronzenxid_freeze = $node->safe_psql('postgres',
"select min(datfrozenxid::text::bigint) from pg_database");
ok($datafronzenxid_freeze > $datafronzenxid, 'datfrozenxid advanced');
-# On Session 1, commit and ensure that the all the notifications are
+# On Session 1, commit and ensure that all the notifications are
# received. This depends on correctly freezing the XIDs in the pending
# notification entries.
-my $res = $psql_session1->query_safe('commit;');
-my $notifications_count = 0;
-foreach my $i (split('\n', $res))
+$session1->do('COMMIT');
+
+my $notifications = $session1->get_all_notifications();
+is(scalar(@$notifications), 10, 'received all committed notifications');
+
+my $expected_payload = 1;
+foreach my $notify (@$notifications)
{
- $notifications_count++;
- like($i,
- qr/Asynchronous notification "s" with payload "$notifications_count" received/
- );
+ is($notify->{channel}, 's', "notification $expected_payload has correct channel");
+ is($notify->{payload}, $expected_payload, "notification $expected_payload has correct payload");
+ $expected_payload++;
}
-is($notifications_count, 10, 'received all committed notifications');
done_testing();
diff --git a/src/test/perl/Makefile b/src/test/perl/Makefile
index fd4fdaf700b..2973d6ec35f 100644
--- a/src/test/perl/Makefile
+++ b/src/test/perl/Makefile
@@ -15,6 +15,10 @@ include $(top_builddir)/src/Makefile.global
ifeq ($(enable_tap_tests),yes)
+SUBDIRS = PostgreSQL/Test
+
+$(recurse)
+
installdirs:
$(MKDIR_P) '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test'
@@ -28,6 +32,8 @@ install: all installdirs
$(INSTALL_DATA) $(srcdir)/PostgreSQL/Test/AdjustDump.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/AdjustDump.pm'
$(INSTALL_DATA) $(srcdir)/PostgreSQL/Test/AdjustUpgrade.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/AdjustUpgrade.pm'
$(INSTALL_DATA) $(srcdir)/PostgreSQL/Version.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Version.pm'
+ $(INSTALL_DATA) $(srcdir)/PostgreSQL/Test/Session.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/Session.pm'
+ $(INSTALL_DATA) $(srcdir)/PostgreSQL/PqFFI.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/PqFFI.pm'
uninstall:
rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/Utils.pm'
@@ -39,5 +45,7 @@ uninstall:
rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/AdjustDump.pm'
rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/AdjustUpgrade.pm'
rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Version.pm'
+ rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/Session.pm'
+ rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/PqFFI.pm'
endif
diff --git a/src/test/perl/PostgreSQL/FindLib.pm b/src/test/perl/PostgreSQL/FindLib.pm
new file mode 100644
index 00000000000..b4290c84c4f
--- /dev/null
+++ b/src/test/perl/PostgreSQL/FindLib.pm
@@ -0,0 +1,164 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::FindLib - find shared libraries for PostgreSQL TAP tests
+
+=head1 SYNOPSIS
+
+ use PostgreSQL::FindLib;
+
+ my $libpath = find_lib_or_die(
+ lib => 'pq',
+ libpath => ['/usr/local/pgsql/lib'],
+ );
+
+=head1 DESCRIPTION
+
+This module provides a simple mechanism to locate shared libraries,
+used as a lightweight replacement for C<FFI::CheckLib>. It searches
+for libraries in specified paths and common system locations.
+
+=head1 EXPORTED FUNCTIONS
+
+=over
+
+=item find_lib_or_die(%args)
+
+Searches for a shared library and returns its full path. Dies if the
+library cannot be found.
+
+Arguments:
+
+=over
+
+=item lib => $name
+
+Required. The library name without prefix or suffix (e.g., C<'pq'> for
+C<libpq.so>).
+
+=item libpath => \@paths
+
+Optional. Array of directories to search first.
+
+=item systempath => \@paths
+
+Optional. If set to an empty array C<[]>, system paths will not be searched.
+
+=back
+
+=back
+
+=cut
+
+package PostgreSQL::FindLib;
+
+use strict;
+use warnings FATAL => qw(all);
+
+use Exporter qw(import);
+use File::Spec;
+use Config;
+
+our @EXPORT = qw(find_lib_or_die);
+
+sub find_lib_or_die
+{
+ my %args = @_;
+
+ my $libname = $args{lib} or die "find_lib_or_die: 'lib' argument required";
+ my $libpath = $args{libpath} // [];
+ my $systempath = $args{systempath};
+
+ my @search_paths = @$libpath;
+
+ # Add system paths unless explicitly disabled
+ unless (defined $systempath && ref($systempath) eq 'ARRAY' && @$systempath == 0)
+ {
+ push @search_paths, _get_system_lib_paths();
+ }
+
+ # Determine library file patterns based on OS
+ my @patterns = _get_lib_patterns($libname);
+
+ for my $dir (@search_paths)
+ {
+ next unless -d $dir;
+
+ for my $pattern (@patterns)
+ {
+ my @matches = glob(File::Spec->catfile($dir, $pattern));
+ for my $match (@matches)
+ {
+ return $match if -f $match && -r $match;
+ }
+ }
+ }
+
+ die "find_lib_or_die: unable to find lib$libname in: " . join(", ", @search_paths);
+}
+
+sub _get_lib_patterns
+{
+ my $libname = shift;
+
+ if ($^O eq 'darwin')
+ {
+ return ("lib$libname.dylib", "lib$libname.*.dylib");
+ }
+ elsif ($^O eq 'MSWin32' || $^O eq 'cygwin')
+ {
+ return ("$libname.dll", "lib$libname.dll");
+ }
+ else
+ {
+ # Linux and other Unix-like systems
+ return ("lib$libname.so", "lib$libname.so.*");
+ }
+}
+
+sub _get_system_lib_paths
+{
+ my @paths;
+
+ # Common system library paths
+ push @paths, '/usr/lib', '/usr/local/lib', '/lib';
+
+ # Add architecture-specific paths on Linux
+ if ($^O eq 'linux')
+ {
+ push @paths, '/usr/lib/x86_64-linux-gnu', '/usr/lib/aarch64-linux-gnu';
+ push @paths, '/usr/lib64', '/lib64';
+ }
+
+ # Add paths from LD_LIBRARY_PATH
+ if ($ENV{LD_LIBRARY_PATH})
+ {
+ push @paths, split(/:/, $ENV{LD_LIBRARY_PATH});
+ }
+
+ # macOS specific
+ if ($^O eq 'darwin')
+ {
+ push @paths, '/opt/homebrew/lib', '/usr/local/opt/libpq/lib';
+ if ($ENV{DYLD_LIBRARY_PATH})
+ {
+ push @paths, split(/:/, $ENV{DYLD_LIBRARY_PATH});
+ }
+ }
+
+ return @paths;
+}
+
+=pod
+
+=head1 SEE ALSO
+
+L<PostgreSQL::PqFFI>
+
+=cut
+
+1;
diff --git a/src/test/perl/PostgreSQL/PGTypes.pm b/src/test/perl/PostgreSQL/PGTypes.pm
new file mode 100644
index 00000000000..5a8523d4e3f
--- /dev/null
+++ b/src/test/perl/PostgreSQL/PGTypes.pm
@@ -0,0 +1,356 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::PGTypes - PostgreSQL backend type OID constants
+
+=head1 SYNOPSIS
+
+ use PostgreSQL::PGTypes;
+
+ if ($type_oid == TEXTOID) { ... }
+
+ if ($type_oid == INT4ARRAYOID) { ... }
+
+=head1 DESCRIPTION
+
+This module provides constants for PostgreSQL backend type OIDs, as defined
+in C<src/include/catalog/pg_type_d.h>. These can be used to identify column
+types in query results via C<PQftype()>.
+
+All constants are exported by default.
+
+=head1 EXPORTED CONSTANTS
+
+=head2 Basic Types
+
+C<BOOLOID>, C<BYTEAOID>, C<CHAROID>, C<NAMEOID>, C<INT8OID>, C<INT2OID>,
+C<INT2VECTOROID>, C<INT4OID>, C<TEXTOID>, C<OIDOID>, C<TIDOID>, C<XIDOID>,
+C<CIDOID>, C<OIDVECTOROID>, C<JSONOID>, C<XMLOID>, C<XID8OID>, C<POINTOID>,
+C<LSEGOID>, C<PATHOID>, C<BOXOID>, C<POLYGONOID>, C<LINEOID>, C<FLOAT4OID>,
+C<FLOAT8OID>, C<UNKNOWNOID>, C<CIRCLEOID>, C<MONEYOID>, C<MACADDROID>,
+C<INETOID>, C<CIDROID>, C<MACADDR8OID>, C<ACLITEMOID>, C<BPCHAROID>,
+C<VARCHAROID>, C<DATEOID>, C<TIMEOID>, C<TIMESTAMPOID>, C<TIMESTAMPTZOID>,
+C<INTERVALOID>, C<TIMETZOID>, C<BITOID>, C<VARBITOID>, C<NUMERICOID>,
+C<REFCURSOROID>, C<UUIDOID>, C<TSVECTOROID>, C<GTSVECTOROID>, C<TSQUERYOID>,
+C<JSONBOID>, C<JSONPATHOID>, C<TXID_SNAPSHOTOID>
+
+=head2 Range Types
+
+C<INT4RANGEOID>, C<NUMRANGEOID>, C<TSRANGEOID>, C<TSTZRANGEOID>,
+C<DATERANGEOID>, C<INT8RANGEOID>
+
+=head2 Multirange Types
+
+C<INT4MULTIRANGEOID>, C<NUMMULTIRANGEOID>, C<TSMULTIRANGEOID>,
+C<TSTZMULTIRANGEOID>, C<DATEMULTIRANGEOID>, C<INT8MULTIRANGEOID>
+
+=head2 Pseudo Types
+
+C<RECORDOID>, C<RECORDARRAYOID>, C<CSTRINGOID>, C<VOIDOID>, C<TRIGGEROID>,
+C<EVENT_TRIGGEROID>
+
+=head2 Array Types
+
+Array type OIDs follow the pattern C<{BASENAME}ARRAYOID>, e.g., C<TEXTARRAYOID>,
+C<INT4ARRAYOID>, C<JSONBARRAYOID>.
+
+=cut
+
+package PostgreSQL::PGTypes;
+
+use strict;
+use warnings FATAL => qw(all);
+
+use Exporter qw(import);
+
+our @EXPORT = qw(
+
+ BOOLOID
+ BYTEAOID
+ CHAROID
+ NAMEOID
+ INT8OID
+ INT2OID
+ INT2VECTOROID
+ INT4OID
+ TEXTOID
+ OIDOID
+ TIDOID
+ XIDOID
+ CIDOID
+ OIDVECTOROID
+ JSONOID
+ XMLOID
+ XID8OID
+ POINTOID
+ LSEGOID
+ PATHOID
+ BOXOID
+ POLYGONOID
+ LINEOID
+ FLOAT4OID
+ FLOAT8OID
+ UNKNOWNOID
+ CIRCLEOID
+ MONEYOID
+ MACADDROID
+ INETOID
+ CIDROID
+ MACADDR8OID
+ ACLITEMOID
+ BPCHAROID
+ VARCHAROID
+ DATEOID
+ TIMEOID
+ TIMESTAMPOID
+ TIMESTAMPTZOID
+ INTERVALOID
+ TIMETZOID
+ BITOID
+ VARBITOID
+ NUMERICOID
+ REFCURSOROID
+ UUIDOID
+ TSVECTOROID
+ GTSVECTOROID
+ TSQUERYOID
+ JSONBOID
+ JSONPATHOID
+ TXID_SNAPSHOTOID
+ INT4RANGEOID
+ NUMRANGEOID
+ TSRANGEOID
+ TSTZRANGEOID
+ DATERANGEOID
+ INT8RANGEOID
+ INT4MULTIRANGEOID
+ NUMMULTIRANGEOID
+ TSMULTIRANGEOID
+ TSTZMULTIRANGEOID
+ DATEMULTIRANGEOID
+ INT8MULTIRANGEOID
+ RECORDOID
+ RECORDARRAYOID
+ CSTRINGOID
+ VOIDOID
+ TRIGGEROID
+ EVENT_TRIGGEROID
+
+ BOOLARRAYOID
+ BYTEAARRAYOID
+ CHARARRAYOID
+ NAMEARRAYOID
+ INT8ARRAYOID
+ INT2ARRAYOID
+ INT2VECTORARRAYOID
+ INT4ARRAYOID
+ TEXTARRAYOID
+ OIDARRAYOID
+ TIDARRAYOID
+ XIDARRAYOID
+ CIDARRAYOID
+ OIDVECTORARRAYOID
+ JSONARRAYOID
+ XMLARRAYOID
+ XID8ARRAYOID
+ POINTARRAYOID
+ LSEGARRAYOID
+ PATHARRAYOID
+ BOXARRAYOID
+ POLYGONARRAYOID
+ LINEARRAYOID
+ FLOAT4ARRAYOID
+ FLOAT8ARRAYOID
+ CIRCLEARRAYOID
+ MONEYARRAYOID
+ MACADDRARRAYOID
+ INETARRAYOID
+ CIDRARRAYOID
+ MACADDR8ARRAYOID
+ ACLITEMARRAYOID
+ BPCHARARRAYOID
+ VARCHARARRAYOID
+ DATEARRAYOID
+ TIMEARRAYOID
+ TIMESTAMPARRAYOID
+ TIMESTAMPTZARRAYOID
+ INTERVALARRAYOID
+ TIMETZARRAYOID
+ BITARRAYOID
+ VARBITARRAYOID
+ NUMERICARRAYOID
+ REFCURSORARRAYOID
+ UUIDARRAYOID
+ TSVECTORARRAYOID
+ GTSVECTORARRAYOID
+ TSQUERYARRAYOID
+ JSONBARRAYOID
+ JSONPATHARRAYOID
+ TXID_SNAPSHOTARRAYOID
+ INT4RANGEARRAYOID
+ NUMRANGEARRAYOID
+ TSRANGEARRAYOID
+ TSTZRANGEARRAYOID
+ DATERANGEARRAYOID
+ INT8RANGEARRAYOID
+ INT4MULTIRANGEARRAYOID
+ NUMMULTIRANGEARRAYOID
+ TSMULTIRANGEARRAYOID
+ TSTZMULTIRANGEARRAYOID
+ DATEMULTIRANGEARRAYOID
+ INT8MULTIRANGEARRAYOID
+ CSTRINGARRAYOID
+
+);
+
+use constant {
+ BOOLOID => 16,
+ BYTEAOID => 17,
+ CHAROID => 18,
+ NAMEOID => 19,
+ INT8OID => 20,
+ INT2OID => 21,
+ INT2VECTOROID => 22,
+ INT4OID => 23,
+ TEXTOID => 25,
+ OIDOID => 26,
+ TIDOID => 27,
+ XIDOID => 28,
+ CIDOID => 29,
+ OIDVECTOROID => 30,
+ JSONOID => 114,
+ XMLOID => 142,
+ XID8OID => 5069,
+ POINTOID => 600,
+ LSEGOID => 601,
+ PATHOID => 602,
+ BOXOID => 603,
+ POLYGONOID => 604,
+ LINEOID => 628,
+ FLOAT4OID => 700,
+ FLOAT8OID => 701,
+ UNKNOWNOID => 705,
+ CIRCLEOID => 718,
+ MONEYOID => 790,
+ MACADDROID => 829,
+ INETOID => 869,
+ CIDROID => 650,
+ MACADDR8OID => 774,
+ ACLITEMOID => 1033,
+ BPCHAROID => 1042,
+ VARCHAROID => 1043,
+ DATEOID => 1082,
+ TIMEOID => 1083,
+ TIMESTAMPOID => 1114,
+ TIMESTAMPTZOID => 1184,
+ INTERVALOID => 1186,
+ TIMETZOID => 1266,
+ BITOID => 1560,
+ VARBITOID => 1562,
+ NUMERICOID => 1700,
+ REFCURSOROID => 1790,
+ UUIDOID => 2950,
+ TSVECTOROID => 3614,
+ GTSVECTOROID => 3642,
+ TSQUERYOID => 3615,
+ JSONBOID => 3802,
+ JSONPATHOID => 4072,
+ TXID_SNAPSHOTOID => 2970,
+ INT4RANGEOID => 3904,
+ NUMRANGEOID => 3906,
+ TSRANGEOID => 3908,
+ TSTZRANGEOID => 3910,
+ DATERANGEOID => 3912,
+ INT8RANGEOID => 3926,
+ INT4MULTIRANGEOID => 4451,
+ NUMMULTIRANGEOID => 4532,
+ TSMULTIRANGEOID => 4533,
+ TSTZMULTIRANGEOID => 4534,
+ DATEMULTIRANGEOID => 4535,
+ INT8MULTIRANGEOID => 4536,
+ RECORDOID => 2249,
+ RECORDARRAYOID => 2287,
+ CSTRINGOID => 2275,
+ VOIDOID => 2278,
+ TRIGGEROID => 2279,
+ EVENT_TRIGGEROID => 3838,
+
+ BOOLARRAYOID => 1000,
+ BYTEAARRAYOID => 1001,
+ CHARARRAYOID => 1002,
+ NAMEARRAYOID => 1003,
+ INT8ARRAYOID => 1016,
+ INT2ARRAYOID => 1005,
+ INT2VECTORARRAYOID => 1006,
+ INT4ARRAYOID => 1007,
+ TEXTARRAYOID => 1009,
+ OIDARRAYOID => 1028,
+ TIDARRAYOID => 1010,
+ XIDARRAYOID => 1011,
+ CIDARRAYOID => 1012,
+ OIDVECTORARRAYOID => 1013,
+ JSONARRAYOID => 199,
+ XMLARRAYOID => 143,
+ XID8ARRAYOID => 271,
+ POINTARRAYOID => 1017,
+ LSEGARRAYOID => 1018,
+ PATHARRAYOID => 1019,
+ BOXARRAYOID => 1020,
+ POLYGONARRAYOID => 1027,
+ LINEARRAYOID => 629,
+ FLOAT4ARRAYOID => 1021,
+ FLOAT8ARRAYOID => 1022,
+ CIRCLEARRAYOID => 719,
+ MONEYARRAYOID => 791,
+ MACADDRARRAYOID => 1040,
+ INETARRAYOID => 1041,
+ CIDRARRAYOID => 651,
+ MACADDR8ARRAYOID => 775,
+ ACLITEMARRAYOID => 1034,
+ BPCHARARRAYOID => 1014,
+ VARCHARARRAYOID => 1015,
+ DATEARRAYOID => 1182,
+ TIMEARRAYOID => 1183,
+ TIMESTAMPARRAYOID => 1115,
+ TIMESTAMPTZARRAYOID => 1185,
+ INTERVALARRAYOID => 1187,
+ TIMETZARRAYOID => 1270,
+ BITARRAYOID => 1561,
+ VARBITARRAYOID => 1563,
+ NUMERICARRAYOID => 1231,
+ REFCURSORARRAYOID => 2201,
+ UUIDARRAYOID => 2951,
+ TSVECTORARRAYOID => 3643,
+ GTSVECTORARRAYOID => 3644,
+ TSQUERYARRAYOID => 3645,
+ JSONBARRAYOID => 3807,
+ JSONPATHARRAYOID => 4073,
+ TXID_SNAPSHOTARRAYOID => 2949,
+ INT4RANGEARRAYOID => 3905,
+ NUMRANGEARRAYOID => 3907,
+ TSRANGEARRAYOID => 3909,
+ TSTZRANGEARRAYOID => 3911,
+ DATERANGEARRAYOID => 3913,
+ INT8RANGEARRAYOID => 3927,
+ INT4MULTIRANGEARRAYOID => 6150,
+ NUMMULTIRANGEARRAYOID => 6151,
+ TSMULTIRANGEARRAYOID => 6152,
+ TSTZMULTIRANGEARRAYOID => 6153,
+ DATEMULTIRANGEARRAYOID => 6155,
+ INT8MULTIRANGEARRAYOID => 6157,
+ CSTRINGARRAYOID => 1263,
+};
+
+=pod
+
+=head1 SEE ALSO
+
+L<PostgreSQL::PqFFI>, L<PostgreSQL::Test::Pq>, L<PostgreSQL::Test::Session>
+
+=cut
+
+1;
diff --git a/src/test/perl/PostgreSQL/PqConstants.pm b/src/test/perl/PostgreSQL/PqConstants.pm
new file mode 100644
index 00000000000..a825fce507d
--- /dev/null
+++ b/src/test/perl/PostgreSQL/PqConstants.pm
@@ -0,0 +1,186 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::PqConstants - libpq constants for PostgreSQL TAP tests
+
+=head1 SYNOPSIS
+
+ use PostgreSQL::PqConstants;
+
+ if (PQstatus($conn) == CONNECTION_OK) { ... }
+
+ if (PQresultStatus($result) == PGRES_TUPLES_OK) { ... }
+
+=head1 DESCRIPTION
+
+This module provides libpq constants shared by both the FFI
+(C<PostgreSQL::PqFFI>) and XS (C<PostgreSQL::Test::Pq>) backends.
+All constants are exported by default.
+
+=head1 EXPORTED CONSTANTS
+
+=head2 Connection Status (ConnStatusType)
+
+C<CONNECTION_OK>, C<CONNECTION_BAD>, C<CONNECTION_STARTED>,
+C<CONNECTION_MADE>, C<CONNECTION_AWAITING_RESPONSE>, C<CONNECTION_AUTH_OK>,
+C<CONNECTION_SETENV>, C<CONNECTION_SSL_STARTUP>, C<CONNECTION_NEEDED>,
+C<CONNECTION_CHECK_WRITABLE>, C<CONNECTION_CONSUME>, C<CONNECTION_GSS_STARTUP>,
+C<CONNECTION_CHECK_TARGET>, C<CONNECTION_CHECK_STANDBY>, C<CONNECTION_ALLOCATED>
+
+=head2 Execution Status (ExecStatusType)
+
+C<PGRES_EMPTY_QUERY>, C<PGRES_COMMAND_OK>, C<PGRES_TUPLES_OK>,
+C<PGRES_COPY_OUT>, C<PGRES_COPY_IN>, C<PGRES_BAD_RESPONSE>,
+C<PGRES_NONFATAL_ERROR>, C<PGRES_FATAL_ERROR>, C<PGRES_COPY_BOTH>,
+C<PGRES_SINGLE_TUPLE>, C<PGRES_PIPELINE_SYNC>, C<PGRES_PIPELINE_ABORTED>,
+C<PGRES_TUPLES_CHUNK>
+
+=head2 Polling Status (PostgresPollingStatusType)
+
+C<PGRES_POLLING_FAILED>, C<PGRES_POLLING_READING>, C<PGRES_POLLING_WRITING>,
+C<PGRES_POLLING_OK>, C<PGRES_POLLING_ACTIVE>
+
+=head2 Ping Status (PGPing)
+
+C<PQPING_OK>, C<PQPING_REJECT>, C<PQPING_NO_RESPONSE>, C<PQPING_NO_ATTEMPT>
+
+=head2 Transaction Status (PGTransactionStatusType)
+
+C<PQTRANS_IDLE>, C<PQTRANS_ACTIVE>, C<PQTRANS_INTRANS>, C<PQTRANS_INERROR>,
+C<PQTRANS_UNKNOWN>
+
+=cut
+
+package PostgreSQL::PqConstants;
+
+use strict;
+use warnings FATAL => qw(all);
+
+use Exporter qw(import);
+
+our @EXPORT = qw(
+
+ CONNECTION_OK
+ CONNECTION_BAD
+ CONNECTION_STARTED
+ CONNECTION_MADE
+ CONNECTION_AWAITING_RESPONSE
+ CONNECTION_AUTH_OK
+ CONNECTION_SETENV
+ CONNECTION_SSL_STARTUP
+ CONNECTION_NEEDED
+ CONNECTION_CHECK_WRITABLE
+ CONNECTION_CONSUME
+ CONNECTION_GSS_STARTUP
+ CONNECTION_CHECK_TARGET
+ CONNECTION_CHECK_STANDBY
+ CONNECTION_ALLOCATED
+
+ PGRES_EMPTY_QUERY
+ PGRES_COMMAND_OK
+ PGRES_TUPLES_OK
+ PGRES_COPY_OUT
+ PGRES_COPY_IN
+ PGRES_BAD_RESPONSE
+ PGRES_NONFATAL_ERROR
+ PGRES_FATAL_ERROR
+ PGRES_COPY_BOTH
+ PGRES_SINGLE_TUPLE
+ PGRES_PIPELINE_SYNC
+ PGRES_PIPELINE_ABORTED
+ PGRES_TUPLES_CHUNK
+
+ PGRES_POLLING_FAILED
+ PGRES_POLLING_READING
+ PGRES_POLLING_WRITING
+ PGRES_POLLING_OK
+ PGRES_POLLING_ACTIVE
+
+ PQPING_OK
+ PQPING_REJECT
+ PQPING_NO_RESPONSE
+ PQPING_NO_ATTEMPT
+
+ PQTRANS_IDLE
+ PQTRANS_ACTIVE
+ PQTRANS_INTRANS
+ PQTRANS_INERROR
+ PQTRANS_UNKNOWN
+
+);
+
+# ConnStatusType:
+use constant {
+ CONNECTION_OK => 0,
+ CONNECTION_BAD => 1,
+ CONNECTION_STARTED => 2,
+ CONNECTION_MADE => 3,
+ CONNECTION_AWAITING_RESPONSE => 4,
+ CONNECTION_AUTH_OK => 5,
+ CONNECTION_SETENV => 6,
+ CONNECTION_SSL_STARTUP => 7,
+ CONNECTION_NEEDED => 8,
+ CONNECTION_CHECK_WRITABLE => 9,
+ CONNECTION_CONSUME => 10,
+ CONNECTION_GSS_STARTUP => 11,
+ CONNECTION_CHECK_TARGET => 12,
+ CONNECTION_CHECK_STANDBY => 13,
+ CONNECTION_ALLOCATED => 14,
+};
+
+# ExecStatusType:
+use constant {
+ PGRES_EMPTY_QUERY => 0,
+ PGRES_COMMAND_OK => 1,
+ PGRES_TUPLES_OK => 2,
+ PGRES_COPY_OUT => 3,
+ PGRES_COPY_IN => 4,
+ PGRES_BAD_RESPONSE => 5,
+ PGRES_NONFATAL_ERROR => 6,
+ PGRES_FATAL_ERROR => 7,
+ PGRES_COPY_BOTH => 8,
+ PGRES_SINGLE_TUPLE => 9,
+ PGRES_PIPELINE_SYNC => 10,
+ PGRES_PIPELINE_ABORTED => 11,
+ PGRES_TUPLES_CHUNK => 12,
+};
+
+# PostgresPollingStatusType:
+use constant {
+ PGRES_POLLING_FAILED => 0,
+ PGRES_POLLING_READING => 1,
+ PGRES_POLLING_WRITING => 2,
+ PGRES_POLLING_OK => 3,
+ PGRES_POLLING_ACTIVE => 4,
+};
+
+# PGPing:
+use constant {
+ PQPING_OK => 0,
+ PQPING_REJECT => 1,
+ PQPING_NO_RESPONSE => 2,
+ PQPING_NO_ATTEMPT => 3,
+};
+
+# PGTransactionStatusType:
+use constant {
+ PQTRANS_IDLE => 0,
+ PQTRANS_ACTIVE => 1,
+ PQTRANS_INTRANS => 2,
+ PQTRANS_INERROR => 3,
+ PQTRANS_UNKNOWN => 4,
+};
+
+=pod
+
+=head1 SEE ALSO
+
+L<PostgreSQL::PqFFI>, L<PostgreSQL::Test::Pq>, L<PostgreSQL::Test::Session>
+
+=cut
+
+1;
diff --git a/src/test/perl/PostgreSQL/PqFFI.pm b/src/test/perl/PostgreSQL/PqFFI.pm
new file mode 100644
index 00000000000..a0352350c1c
--- /dev/null
+++ b/src/test/perl/PostgreSQL/PqFFI.pm
@@ -0,0 +1,419 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::PqFFI - FFI wrapper for libpq
+
+=head1 SYNOPSIS
+
+ use PostgreSQL::PqFFI;
+
+ # Initialize the FFI bindings (required before use)
+ PostgreSQL::PqFFI::setup($libdir);
+
+ # Connect to database
+ my $conn = PQconnectdb("dbname=postgres");
+ die PQerrorMessage($conn) unless PQstatus($conn) == CONNECTION_OK;
+
+ # Execute query
+ my $result = PQexec($conn, "SELECT 1");
+ if (PQresultStatus($result) == PGRES_TUPLES_OK) {
+ print PQgetvalue($result, 0, 0), "\n";
+ }
+ PQclear($result);
+
+ PQfinish($conn);
+
+=head1 DESCRIPTION
+
+This module provides Perl bindings to libpq using L<FFI::Platypus>.
+It is the default backend for L<PostgreSQL::Test::Session> when
+C<PG_USE_PQ_XS> is not set.
+
+The module must be initialized by calling C<setup()> before any libpq
+functions can be used.
+
+=head1 FUNCTIONS
+
+=head2 setup($libdir [, $use_system_path])
+
+Initialize the FFI bindings. C<$libdir> specifies where to find libpq.
+If C<$use_system_path> is false, only C<$libdir> is searched.
+
+=head2 libpq Functions
+
+All standard libpq functions are exported. See the PostgreSQL libpq
+documentation for details. Commonly used functions include:
+
+B<Connection:> C<PQconnectdb>, C<PQconnectStart>, C<PQconnectPoll>,
+C<PQfinish>, C<PQstatus>, C<PQerrorMessage>
+
+B<Execution:> C<PQexec>, C<PQexecParams>, C<PQprepare>, C<PQexecPrepared>,
+C<PQsendQuery>, C<PQgetResult>, C<PQclear>
+
+B<Results:> C<PQresultStatus>, C<PQntuples>, C<PQnfields>, C<PQfname>,
+C<PQftype>, C<PQgetvalue>, C<PQgetisnull>
+
+B<Pipeline:> C<PQenterPipelineMode>, C<PQexitPipelineMode>,
+C<PQpipelineSync>, C<PQsendQueryParams>
+
+B<Notifications:> C<PQnotifies>, C<PQnotify_channel>, C<PQnotify_payload>,
+C<PQnotify_be_pid>, C<PQnotify_free>
+
+B<Notice Processing:> C<PQsetNoticeProcessor>, C<create_notice_processor>
+
+=head2 create_notice_processor($callback)
+
+Creates a notice processor closure that can be passed to
+C<PQsetNoticeProcessor>. The callback receives C<($arg, $message)>.
+The caller must keep a reference to the returned closure to prevent
+garbage collection.
+
+=head1 EXPORTED CONSTANTS
+
+This module re-exports all constants from L<PostgreSQL::PqConstants>
+and L<PostgreSQL::PGTypes>.
+
+=cut
+
+package PostgreSQL::PqFFI;
+
+use strict;
+use warnings FATAL => qw(all);
+
+use FFI::Platypus;
+use FFI::C;
+use PostgreSQL::FindLib;
+use PostgreSQL::PqConstants;
+use PostgreSQL::PGTypes;
+
+use Exporter qw(import);
+
+our @EXPORT = (
+ @PostgreSQL::PqConstants::EXPORT,
+ @PostgreSQL::PGTypes::EXPORT,
+);
+
+
+
+my @procs = qw(
+
+ PQnotifies
+ PQfreemem
+ PQnotify_channel
+ PQnotify_payload
+ PQnotify_be_pid
+ PQnotify_free
+
+ PQconnectdb
+ PQconnectdbParams
+ PQsetdbLogin
+ PQfinish
+ PQreset
+ PQdb
+ PQuser
+ PQpass
+ PQhost
+ PQhostaddr
+ PQport
+ PQtty
+ PQoptions
+ PQstatus
+ PQtransactionStatus
+ PQparameterStatus
+ PQping
+ PQpingParams
+
+ PQexec
+ PQexecParams
+ PQprepare
+ PQexecPrepared
+
+ PQdescribePrepared
+ PQdescribePortal
+
+ PQclosePrepared
+ PQclosePortal
+ PQclear
+
+ PQsendQuery
+ PQgetResult
+ PQisBusy
+ PQconsumeInput
+
+ PQprotocolVersion
+ PQserverVersion
+ PQerrorMessage
+ PQsocket
+ PQbackendPID
+ PQconnectionNeedsPassword
+ PQconnectionUsedPassword
+ PQconnectionUsedGSSAPI
+ PQclientEncoding
+ PQsetClientEncoding
+
+ PQresultStatus
+ PQresStatus
+ PQresultErrorMessage
+ PQresultErrorField
+ PQntuples
+ PQnfields
+ PQbinaryTuples
+ PQfname
+ PQfnumber
+ PQftable
+ PQftablecol
+ PQfformat
+ PQftype
+ PQfsize
+ PQfmod
+ PQcmdStatus
+ PQoidValue
+ PQcmdTuples
+ PQgetvalue
+ PQgetlength
+ PQgetisnull
+ PQnparams
+ PQparamtype
+ PQchangePassword
+
+ PQpipelineStatus
+ PQenterPipelineMode
+ PQexitPipelineMode
+ PQpipelineSync
+ PQsendFlushRequest
+ PQsendPipelineSync
+ PQsendQueryParams
+
+ PQsetnonblocking
+ PQisnonblocking
+ PQflush
+
+ PQconnectStart
+ PQconnectStartParams
+ PQconnectPoll
+
+ PQsetNoticeProcessor
+ create_notice_processor
+
+);
+
+push(@EXPORT, @procs);
+
+sub setup
+{
+ my $libdir = shift;
+ my $use_system_path = shift;
+
+ my $ffi = FFI::Platypus->new(api => 1);
+
+ my @system_path;
+ @system_path = (systempath => []) unless $use_system_path;
+
+ $ffi->type('opaque' => 'PGconn');
+ $ffi->type('opaque' => 'PGresult');
+ $ffi->type('uint32' => 'Oid');
+ $ffi->type('int' => 'ExecStatusType');
+
+ # Set up FFI::C with this FFI::Platypus instance for struct support
+ FFI::C->ffi($ffi);
+
+ # PGnotify struct for notification support
+ # typedef struct pgNotify {
+ # char *relname; /* notification channel name */
+ # int be_pid; /* process ID of notifying server process */
+ # char *extra; /* notification payload string */
+ # } PGnotify;
+ FFI::C->struct('PGnotify' => [
+ relname => 'opaque', # char* - will convert via helper
+ be_pid => 'sint32', # int
+ extra => 'opaque', # char* - will convert via helper
+ ]);
+
+ my $lib = find_lib_or_die(
+ lib => 'pq',
+ libpath => [$libdir],
+ @system_path,
+ );
+ $ffi->lib($lib);
+
+ $ffi->attach('PQconnectdb' => ['string'] => 'PGconn');
+ $ffi->attach(
+ 'PQconnectdbParams' => [ 'string[]', 'string[]', 'int' ] => 'PGconn');
+ $ffi->attach(
+ 'PQsetdbLogin' => [
+ 'string', 'string', 'string', 'string',
+ 'string', 'string', 'string',
+ ] => 'PGconn');
+ $ffi->attach('PQfinish' => ['PGconn'] => 'void');
+ $ffi->attach('PQreset' => ['PGconn'] => 'void');
+ $ffi->attach('PQdb' => ['PGconn'] => 'string');
+ $ffi->attach('PQuser' => ['PGconn'] => 'string');
+ $ffi->attach('PQpass' => ['PGconn'] => 'string');
+ $ffi->attach('PQhost' => ['PGconn'] => 'string');
+ $ffi->attach('PQhostaddr' => ['PGconn'] => 'string');
+ $ffi->attach('PQport' => ['PGconn'] => 'string');
+ $ffi->attach('PQtty' => ['PGconn'] => 'string');
+ $ffi->attach('PQoptions' => ['PGconn'] => 'string');
+ $ffi->attach('PQstatus' => ['PGconn'] => 'int');
+ $ffi->attach('PQtransactionStatus' => ['PGconn'] => 'int');
+ $ffi->attach('PQparameterStatus' => [ 'PGconn', 'string' ] => 'string');
+ $ffi->attach('PQping' => ['string'] => 'int');
+ $ffi->attach(
+ 'PQpingParams' => [ 'string[]', 'string[]', 'int' ] => 'int');
+
+ $ffi->attach('PQprotocolVersion' => ['PGconn'] => 'int');
+ $ffi->attach('PQserverVersion' => ['PGconn'] => 'int');
+ $ffi->attach('PQerrorMessage' => ['PGconn'] => 'string');
+ $ffi->attach('PQsocket' => ['PGconn'] => 'int');
+ $ffi->attach('PQbackendPID' => ['PGconn'] => 'int');
+ $ffi->attach('PQconnectionNeedsPassword' => ['PGconn'] => 'int');
+ $ffi->attach('PQconnectionUsedPassword' => ['PGconn'] => 'int');
+ $ffi->attach('PQconnectionUsedGSSAPI' => ['PGconn'] => 'int');
+ $ffi->attach('PQclientEncoding' => ['PGconn'] => 'int');
+ $ffi->attach('PQsetClientEncoding' => [ 'PGconn', 'string' ] => 'int');
+
+ $ffi->attach('PQexec' => [ 'PGconn', 'string' ] => 'PGresult');
+ $ffi->attach(
+ 'PQexecParams' => [
+ 'PGconn', 'string', 'int', 'int[]',
+ 'string[]', 'int[]', 'int[]', 'int'
+ ] => 'PGresult');
+ $ffi->attach(
+ 'PQprepare' => [ 'PGconn', 'string', 'string', 'int', 'int[]' ] =>
+ 'PGresult');
+ $ffi->attach(
+ 'PQexecPrepared' => [ 'PGconn', 'string', 'int',
+ 'string[]', 'int[]', 'int[]', 'int' ] => 'PGresult');
+
+ $ffi->attach('PQresultStatus' => ['PGresult'] => 'ExecStatusType');
+ $ffi->attach('PQresStatus' => ['ExecStatusType'] => 'string');
+ $ffi->attach('PQresultErrorMessage' => ['PGresult'] => 'string');
+ $ffi->attach('PQresultErrorField' => [ 'PGresult', 'int' ] => 'string');
+ $ffi->attach('PQntuples' => ['PGresult'] => 'int');
+ $ffi->attach('PQnfields' => ['PGresult'] => 'int');
+ $ffi->attach('PQbinaryTuples' => ['PGresult'] => 'int');
+ $ffi->attach('PQfname' => [ 'PGresult', 'int' ] => 'string');
+ $ffi->attach('PQfnumber' => [ 'PGresult', 'string' ] => 'int');
+ $ffi->attach('PQftable' => [ 'PGresult', 'int' ] => 'Oid');
+ $ffi->attach('PQftablecol' => [ 'PGresult', 'int' ] => 'int');
+ $ffi->attach('PQfformat' => [ 'PGresult', 'int' ] => 'int');
+ $ffi->attach('PQftype' => [ 'PGresult', 'int' ] => 'Oid');
+ $ffi->attach('PQfsize' => [ 'PGresult', 'int' ] => 'int');
+ $ffi->attach('PQfmod' => [ 'PGresult', 'int' ] => 'int');
+ $ffi->attach('PQcmdStatus' => ['PGresult'] => 'string');
+ $ffi->attach('PQoidValue' => ['PGresult'] => 'Oid');
+ $ffi->attach('PQcmdTuples' => ['PGresult'] => 'string');
+ $ffi->attach('PQgetvalue' => [ 'PGresult', 'int', 'int' ] => 'string');
+ $ffi->attach('PQgetlength' => [ 'PGresult', 'int', 'int' ] => 'int');
+ $ffi->attach('PQgetisnull' => [ 'PGresult', 'int', 'int' ] => 'int');
+ $ffi->attach('PQnparams' => ['PGresult'] => 'int');
+ $ffi->attach('PQparamtype' => [ 'PGresult', 'int' ] => 'Oid');
+
+
+ $ffi->attach(
+ 'PQdescribePrepared' => [ 'PGconn', 'string' ] => 'PGresult');
+ $ffi->attach('PQdescribePortal' => [ 'PGconn', 'string' ] => 'PGresult');
+
+ $ffi->attach('PQclosePrepared' => [ 'PGconn', 'string' ] => 'PGresult');
+ $ffi->attach('PQclosePortal' => [ 'PGconn', 'string' ] => 'PGresult');
+ $ffi->attach('PQclear' => ['PGresult'] => 'void');
+
+ $ffi->attach('PQconnectStart' => [ 'string' ] => 'PGconn');
+ $ffi->attach(
+ 'PQconnectStartParams' => [ 'string[]', 'string[]', 'int' ] => 'PGconn');
+ $ffi->attach('PQconnectPoll' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQresetStart' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQresetPoll' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQsendQuery' => [ 'PGconn', 'string' ] => 'int');
+ $ffi->attach('PQsendQueryParams' => [
+ 'PGconn', 'string', 'int', 'Oid*', 'string*',
+ 'int*', 'int*', 'int' ] => 'int');
+ $ffi->attach('PQsendPrepare' => [ 'PGconn', 'string', 'string', 'int', 'Oid[]' ] => 'int');
+ $ffi->attach('PQgetResult' => [ 'PGconn' ] => 'PGresult');
+
+ $ffi->attach('PQisBusy' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQconsumeInput' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQchangePassword' => [ 'PGconn', 'string', 'string' ] => 'PGresult');
+
+ $ffi->attach('PQpipelineStatus' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQenterPipelineMode' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQexitPipelineMode' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQpipelineSync' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQsendFlushRequest' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQsendPipelineSync' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQsetnonblocking' => [ 'PGconn', 'int' ] => 'int');
+ $ffi->attach('PQisnonblocking' => [ 'PGconn' ] => 'int');
+ $ffi->attach('PQflush' => [ 'PGconn' ] => 'int');
+
+ # Notification support - PQnotifies returns a PGnotify struct pointer
+ $ffi->attach('PQnotifies' => [ 'PGconn' ] => 'PGnotify');
+ $ffi->attach('PQfreemem' => [ 'opaque' ] => 'void');
+
+ # Notice processor callback support
+ # typedef void (*PQnoticeProcessor)(void *arg, const char *message);
+ $ffi->type('(opaque,string)->void' => 'PQnoticeProcessor');
+ $ffi->attach('PQsetNoticeProcessor' => [ 'PGconn', 'PQnoticeProcessor', 'opaque' ] => 'opaque');
+
+ # Store the $ffi instance for use by helper functions
+ $PostgreSQL::PqFFI::_ffi = $ffi;
+}
+
+# Helper functions to extract string values from PGnotify struct
+# The struct contains opaque pointers (char*) that need to be cast to strings
+
+sub PQnotify_channel
+{
+ my $notify = shift;
+ return undef unless $notify;
+ my $ptr = $notify->relname;
+ return undef unless $ptr;
+ return $PostgreSQL::PqFFI::_ffi->cast('opaque', 'string', $ptr);
+}
+
+sub PQnotify_payload
+{
+ my $notify = shift;
+ return undef unless $notify;
+ my $ptr = $notify->extra;
+ return undef unless $ptr;
+ return $PostgreSQL::PqFFI::_ffi->cast('opaque', 'string', $ptr);
+}
+
+sub PQnotify_be_pid
+{
+ my $notify = shift;
+ return undef unless $notify;
+ return $notify->be_pid;
+}
+
+# Free a PGnotify struct. Must use the raw pointer from the FFI::C struct.
+sub PQnotify_free
+{
+ my $notify = shift;
+ return unless $notify;
+ PQfreemem($notify->{ptr});
+}
+
+# Create a notice processor closure that can be passed to PQsetNoticeProcessor.
+# The callback will be invoked with (arg, message) where arg is opaque and message is string.
+# Returns the closure - caller must keep a reference to prevent garbage collection.
+sub create_notice_processor
+{
+ my $callback = shift; # sub { my ($arg, $message) = @_; ... }
+ return $PostgreSQL::PqFFI::_ffi->closure($callback);
+}
+
+=pod
+
+=head1 SEE ALSO
+
+L<PostgreSQL::Test::Session>, L<PostgreSQL::Test::Pq>,
+L<PostgreSQL::PqConstants>, L<PostgreSQL::PGTypes>, L<FFI::Platypus>
+
+=cut
+
+1;
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index e267ba868fe..e56008287c6 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -112,6 +112,7 @@ use Socket;
use Test::More;
use PostgreSQL::Test::Utils ();
use PostgreSQL::Test::BackgroundPsql ();
+use PostgreSQL::Test::Session;
use Text::ParseWords qw(shellwords);
use Time::HiRes qw(usleep);
use Scalar::Util qw(blessed);
@@ -2036,20 +2037,40 @@ sub safe_psql
my ($stdout, $stderr);
- my $ret = $self->psql(
- $dbname, $sql,
- %params,
- stdout => \$stdout,
- stderr => \$stderr,
- on_error_die => 1,
- on_error_stop => 1);
-
- # psql can emit stderr from NOTICEs etc
- if ($stderr ne "")
+ # for now only use a Session object for single statement sql without
+ # any special params
+ if ($sql =~ /\w/ && $sql !~ /\\bind|;.*\w/s && !scalar(keys(%params)))
{
- print "#### Begin standard error\n";
- print $stderr;
- print "\n#### End standard error\n";
+
+ my $session = PostgreSQL::Test::Session->new(node=> $self,
+ dbname => $dbname);
+ my $res = $session->query($sql);
+ my $status = $res->{status};
+ $stdout = $res->{psqlout} // "";
+ $stderr = $res->{error_message} // "";
+ die "error: status = $status stderr: '$stderr'\nwhile running '$sql'"
+ if ($status != 1 && $status != 2); # COMMAND_OK or COMMAND_TUPLES
+
+ }
+ else
+ {
+ # diag "safe_psql call has params or multiple statements";
+
+ my $ret = $self->psql(
+ $dbname, $sql,
+ %params,
+ stdout => \$stdout,
+ stderr => \$stderr,
+ on_error_die => 1,
+ on_error_stop => 1);
+
+ # psql can emit stderr from NOTICEs etc
+ if ($stderr ne "")
+ {
+ print "#### Begin standard error\n";
+ print $stderr;
+ print "\n#### End standard error\n";
+ }
}
return $stdout;
@@ -2153,6 +2174,9 @@ sub psql
local %ENV = $self->_get_env();
+ # uncomment to get a count of calls to psql
+ # note("counting psql");
+
my $stdout = $params{stdout};
my $stderr = $params{stderr};
my $replication = $params{replication};
@@ -2726,33 +2750,20 @@ sub poll_query_until
{
my ($self, $dbname, $query, $expected) = @_;
- local %ENV = $self->_get_env();
-
$expected = 't' unless defined($expected); # default value
- my $cmd = [
- $self->installed_command('psql'), '--no-psqlrc',
- '--no-align', '--tuples-only',
- '--dbname' => $self->connstr($dbname)
- ];
- my ($stdout, $stderr);
+ my $session = PostgreSQL::Test::Session->new(node => $self,
+ dbname => $dbname);
my $max_attempts = 10 * $PostgreSQL::Test::Utils::timeout_default;
my $attempts = 0;
+ my $query_value;
+
while ($attempts < $max_attempts)
{
- my $result = IPC::Run::run $cmd,
- '<' => \$query,
- '>' => \$stdout,
- '2>' => \$stderr;
-
- chomp($stdout);
- chomp($stderr);
-
- if ($stdout eq $expected && $stderr eq '')
- {
- return 1;
- }
+ my $result = $session->query($query);
+ $query_value = ($result->{psqlout} // "");
+ return 1 if $query_value eq $expected;
# Wait 0.1 second before retrying.
usleep(100_000);
@@ -2767,9 +2778,40 @@ $query
expecting this output:
$expected
last actual query output:
-$stdout
-with stderr:
-$stderr);
+$query_value
+);
+ return 0;
+}
+
+=pod
+
+=item $node->poll_until_connection($dbname)
+
+Try to connect repeatedly, until it we succeed.
+Times out after $PostgreSQL::Test::Utils::timeout_default seconds.
+Returns 1 if successful, 0 if timed out.
+
+=cut
+
+sub poll_until_connection
+{
+ my ($self, $dbname) = @_;
+
+ my $max_attempts = 10 * $PostgreSQL::Test::Utils::timeout_default;
+ my $attempts = 0;
+
+ while ($attempts < $max_attempts)
+ {
+ my $session = PostgreSQL::Test::Session->new(node => $self,
+ dbname => $dbname);
+ return 1 if $session;
+
+ # Wait 0.1 second before retrying.
+ usleep(100_000);
+
+ $attempts++;
+ }
+
return 0;
}
diff --git a/src/test/perl/PostgreSQL/Test/GNUmakefile b/src/test/perl/PostgreSQL/Test/GNUmakefile
new file mode 100644
index 00000000000..1b586c0626c
--- /dev/null
+++ b/src/test/perl/PostgreSQL/Test/GNUmakefile
@@ -0,0 +1,71 @@
+# Makefile for PostgreSQL::Test::Pq
+# src/test/perl/PostgreSQL/Test/GNUmakefile
+
+subdir = src/test/perl/PostgreSQL/Test
+top_builddir = ../../../../..
+include $(top_builddir)/src/Makefile.global
+
+# where to find xsubpp for building XS.
+XSUBPPDIR = $(shell $(PERL) -e 'use List::Util qw(first); print first { -r "$$_/ExtUtils/xsubpp" } @INC')
+
+ARCHLIBEXP = $(shell $(PERL) -e 'use Config; print $$Config{archlibexp};')
+
+# only build if we can find xsubpp
+ifneq (,$(XSUBPPDIR))
+
+ifeq ($(PORTNAME), win32)
+override CPPFLAGS += -DPLPERL_HAVE_UID_GID
+# Perl on win32 contains /* within comment all over the header file,
+# so disable this warning.
+override CPPFLAGS += -Wno-comment
+endif
+
+# Suppress warnings from Perl headers that we cannot fix
+override CFLAGS += -Wno-shadow=compatible-local -Wno-declaration-after-statement
+
+# Note: we need to include the perl_includespec directory last,
+# probably because it sometimes contains some header files with names
+# that clash with some of ours, or with some that we include, notably on
+# Windows.
+#
+# We need to include the plperl directory for ppport.h
+#
+override CPPFLAGS := -I. -I$(srcdir) -I$(top_srcdir)/src/pl/plperl -I$(top_srcdir)/src/interfaces/libpq $(CPPFLAGS) -I$(ARCHLIBEXP)/CORE
+
+# this is often, but not always, the same directory named by perl_includespec
+# rpathdir = $(perl_archlibexp)/CORE
+
+# Needed to make sure the bootstrap function is visible.
+# Is there a better way to do this?
+CFLAGS_SL_MODULE :=
+
+NAME = Pq
+
+OBJS = Pq.o
+
+SHLIB_LINK_INTERNAL = $(libpq)
+
+
+include $(top_srcdir)/src/Makefile.shlib
+
+all: all-lib
+
+Pq.c: Pq.xs
+ $(PERL) $(XSUBPPDIR)/ExtUtils/xsubpp $< > $@
+
+# Pq.{so,dll,dylib} should be installed alongside the .pm module
+install: all installdirs
+ $(INSTALL_SHLIB) Pq$(DLSUFFIX) '$(DESTDIR)$(pgxsdir)/$(subdir)/Pq$(DLSUFFIX)'
+ $(INSTALL_DATA) $(top_srcdir)/$(subdir)/Pq.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/Pq.pm'
+
+uninstall:
+ rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/Pq$(DLSUFFIX)'
+ rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/Pq.pm'
+
+else
+
+# no xsubpp, so nothing to build
+all:
+ @echo Cannot build Pq.$(DLSUFFIX)
+
+endif
diff --git a/src/test/perl/PostgreSQL/Test/Pq.pm b/src/test/perl/PostgreSQL/Test/Pq.pm
new file mode 100644
index 00000000000..cf4477cccc3
--- /dev/null
+++ b/src/test/perl/PostgreSQL/Test/Pq.pm
@@ -0,0 +1,111 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::Test::Pq - XS wrapper for libpq
+
+=head1 SYNOPSIS
+
+ use PostgreSQL::Test::Pq;
+
+ my $conn = PQconnectdb("dbname=postgres");
+ die PQerrorMessage($conn) unless PQstatus($conn) == CONNECTION_OK;
+
+ my $result = PQexec($conn, "SELECT 1");
+ if (PQresultStatus($result) == PGRES_TUPLES_OK) {
+ print PQgetvalue($result, 0, 0), "\n";
+ }
+ PQclear($result);
+
+ PQfinish($conn);
+
+=head1 DESCRIPTION
+
+This module provides Perl bindings to libpq using XS. It is an alternative
+backend for L<PostgreSQL::Test::Session> that can be enabled by setting
+the C<PG_USE_PQ_XS> environment variable.
+
+Unlike L<PostgreSQL::PqFFI>, this module requires compilation but has
+no runtime dependencies on FFI libraries.
+
+=head1 FUNCTIONS
+
+All standard libpq functions listed below are exported. See the PostgreSQL
+libpq documentation for details.
+
+B<Connection:> C<PQconnectdb>, C<PQconnectStart>, C<PQconnectPoll>,
+C<PQfinish>, C<PQstatus>, C<PQerrorMessage>, C<PQbackendPID>,
+C<PQtransactionStatus>
+
+B<Execution:> C<PQexec>, C<PQsendQuery>, C<PQsendQueryParams>,
+C<PQgetResult>, C<PQclear>, C<PQconsumeInput>, C<PQisBusy>
+
+B<Results:> C<PQresultStatus>, C<PQntuples>, C<PQnfields>, C<PQfname>,
+C<PQftype>, C<PQgetvalue>, C<PQgetisnull>
+
+B<Pipeline:> C<PQenterPipelineMode>, C<PQexitPipelineMode>,
+C<PQpipelineStatus>, C<PQpipelineSync>, C<PQsetnonblocking>,
+C<PQisnonblocking>
+
+B<Notifications:> C<PQnotifies> (returns hashref), C<PQnotify_channel>,
+C<PQnotify_payload>, C<PQnotify_be_pid>, C<PQnotify_free>
+
+B<Other:> C<PQchangePassword>, C<PQfreemem>
+
+=head2 Notification Handling
+
+Unlike the FFI backend, C<PQnotifies> returns a Perl hashref directly
+with keys C<channel>, C<pid>, and C<payload>. The helper functions
+C<PQnotify_channel>, C<PQnotify_be_pid>, C<PQnotify_payload>, and
+C<PQnotify_free> are provided for API compatibility with the FFI backend.
+
+=head1 EXPORTED CONSTANTS
+
+This module re-exports all constants from L<PostgreSQL::PqConstants>.
+
+=head1 SEE ALSO
+
+L<PostgreSQL::Test::Session>, L<PostgreSQL::PqFFI>, L<PostgreSQL::PqConstants>
+
+=cut
+
+package PostgreSQL::Test::Pq;
+
+use strict;
+use warnings;
+use Carp;
+
+require Exporter;
+use PostgreSQL::PqConstants;
+
+our @ISA = qw(Exporter);
+
+our $VERSION = '1.00';
+
+my @routines = qw(
+ PQclear PQconnectdb PQconsumeInput PQerrorMessage PQexec PQfinish PQfname
+ PQftype PQgetisnull PQgetResult PQgetvalue PQisBusy PQnfields PQntuples
+ PQresultStatus PQsendQuery PQstatus PQchangePassword
+ PQconnectStart PQconnectPoll PQbackendPID PQtransactionStatus
+ PQsetnonblocking PQisnonblocking
+ PQenterPipelineMode PQexitPipelineMode PQpipelineStatus PQpipelineSync
+ PQsendQueryParams
+ PQnotifies PQfreemem
+ PQnotify_channel PQnotify_be_pid PQnotify_payload PQnotify_free
+);
+
+our @EXPORT = (@routines, @PostgreSQL::PqConstants::EXPORT);
+
+# Helper functions for PGnotify - XS returns a hashref directly
+sub PQnotify_channel { return $_[0]->{channel}; }
+sub PQnotify_be_pid { return $_[0]->{pid}; }
+sub PQnotify_payload { return $_[0]->{payload}; }
+sub PQnotify_free { } # no-op, perl GCs the hashref
+
+require XSLoader;
+XSLoader::load('PostgreSQL::Test::Pq', );
+
+1;
diff --git a/src/test/perl/PostgreSQL/Test/Pq.xs b/src/test/perl/PostgreSQL/Test/Pq.xs
new file mode 100644
index 00000000000..a052bce181f
--- /dev/null
+++ b/src/test/perl/PostgreSQL/Test/Pq.xs
@@ -0,0 +1,163 @@
+#define PERL_NO_GET_CONTEXT
+#include "EXTERN.h"
+#include "perl.h"
+#include "XSUB.h"
+
+#include "ppport.h"
+
+#include "libpq-fe.h"
+
+MODULE = PostgreSQL::Test::Pq PACKAGE = PostgreSQL::Test::Pq
+
+PROTOTYPES: ENABLE
+
+TYPEMAP: <<EOTYPE
+
+# would like to use T_PTROBJ here but it gets hung up on const-ness
+
+PGresult * T_PTR
+const PGresult * T_PTR
+PGconn * T_PTR
+const PGconn * T_PTR
+char * T_PV
+const char * T_PV
+ConnStatusType T_ENUM
+ExecStatusType T_ENUM
+Oid T_UV
+
+EOTYPE
+
+PGresult *
+PQchangePassword(PGconn *conn, const char *user, const char *passwd);
+
+void
+PQclear(PGresult *res);
+
+PGconn *
+PQconnectdb(const char *conninfo);
+
+int
+PQconsumeInput(PGconn *conn);
+
+char *
+PQerrorMessage(const PGconn *conn);
+
+PGresult *
+PQexec(PGconn *conn, const char *query);
+
+void
+PQfinish(PGconn *conn);
+
+char *
+PQfname(const PGresult *res, int field_num);
+
+Oid
+PQftype(const PGresult *res, int field_num);
+
+int
+PQgetisnull(const PGresult *res, int tup_num, int field_num);
+
+PGresult *
+PQgetResult(PGconn *conn);
+
+char *
+PQgetvalue(const PGresult *res, int tup_num, int field_num);
+
+int
+PQisBusy(PGconn *conn);
+
+int
+PQnfields(const PGresult *res);
+
+int
+PQntuples(const PGresult *res);
+
+ExecStatusType
+PQresultStatus(const PGresult *res);
+
+int
+PQsendQuery(PGconn *conn, const char *query);
+
+ConnStatusType
+PQstatus(const PGconn *conn);
+
+PGconn *
+PQconnectStart(const char *conninfo);
+
+int
+PQconnectPoll(PGconn *conn);
+
+int
+PQbackendPID(const PGconn *conn);
+
+int
+PQtransactionStatus(const PGconn *conn);
+
+int
+PQsetnonblocking(PGconn *conn, int arg);
+
+int
+PQisnonblocking(const PGconn *conn);
+
+int
+PQenterPipelineMode(PGconn *conn);
+
+int
+PQexitPipelineMode(PGconn *conn);
+
+int
+PQpipelineStatus(const PGconn *conn);
+
+int
+PQpipelineSync(PGconn *conn);
+
+void
+PQfreemem(void *ptr);
+
+SV *
+PQnotifies(PGconn *conn)
+ CODE:
+ PGnotify *notify = PQnotifies(conn);
+ if (notify == NULL) {
+ RETVAL = &PL_sv_undef;
+ } else {
+ HV *hv = newHV();
+ hv_store(hv, "channel", 7, newSVpv(notify->relname, 0), 0);
+ hv_store(hv, "pid", 3, newSViv(notify->be_pid), 0);
+ hv_store(hv, "payload", 7,
+ notify->extra ? newSVpv(notify->extra, 0) : newSVpv("", 0), 0);
+ PQfreemem(notify);
+ RETVAL = newRV_noinc((SV *)hv);
+ }
+ OUTPUT:
+ RETVAL
+
+int
+PQsendQueryParams(PGconn *conn, const char *command, int nParams, SV *paramTypes_sv, SV *paramValues_sv, SV *paramLengths_sv, SV *paramFormats_sv, int resultFormat)
+ CODE:
+ const char **paramValues = NULL;
+ PERL_UNUSED_VAR(paramTypes_sv);
+ PERL_UNUSED_VAR(paramLengths_sv);
+ PERL_UNUSED_VAR(paramFormats_sv);
+
+ /* Only paramValues is used; paramTypes, paramLengths, paramFormats are ignored (assumed NULL) */
+ if (nParams > 0 && SvOK(paramValues_sv) && SvROK(paramValues_sv) &&
+ SvTYPE(SvRV(paramValues_sv)) == SVt_PVAV) {
+ AV *params = (AV *)SvRV(paramValues_sv);
+ Newx(paramValues, nParams, const char *);
+ for (int i = 0; i < nParams; i++) {
+ SV **elem = av_fetch(params, i, 0);
+ if (elem && SvOK(*elem)) {
+ paramValues[i] = SvPV_nolen(*elem);
+ } else {
+ paramValues[i] = NULL;
+ }
+ }
+ }
+
+ RETVAL = PQsendQueryParams(conn, command, nParams, NULL, paramValues, NULL, NULL, resultFormat);
+
+ if (paramValues)
+ Safefree(paramValues);
+ OUTPUT:
+ RETVAL
diff --git a/src/test/perl/PostgreSQL/Test/Session.pm b/src/test/perl/PostgreSQL/Test/Session.pm
new file mode 100644
index 00000000000..b1e2735fb77
--- /dev/null
+++ b/src/test/perl/PostgreSQL/Test/Session.pm
@@ -0,0 +1,971 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::Test::Session - class for a PostgreSQL libpq session
+
+=head1 SYNOPSIS
+
+ use PostgreSQL::Test::Session;
+
+ use PostgreSQL::Test::Cluster;
+
+ my $node = PostgreSQL::Test::Cluster->new('mynode');
+
+ # create a new session. defult dbname is 'postgres'
+ my $session = PostgreSQL::Test::Session->new(node => $node
+ [, dbname => $dbname] );
+
+ # close the session
+ $session->close;
+
+ # reopen the session, after closing it if not closed
+ $session->reconnect;
+
+ # check if the session is ok
+ # my $status = $session->conn_status;
+
+ # run some SQL, not producing tuples
+ my $result = $session->do($sql, ...);
+
+ # run an SQL statement asynchronously
+ my $result = $session->do_async($sql);
+
+ # wait for and async SQL to complete
+ $session->wait_for_completion;
+
+ # set a password for a user
+ my $result = $session->set_password($user, $password);
+
+ # get some data
+ my $result = $session->query($sql);
+
+ # get a single value, default croaks if no value found
+ my $val = $session->query_oneval($sql [, $missing_ok ]);
+
+ #return lines of tuples like "psql -A -t"
+ my @lines = $session->query_tuples($sql, ...);
+
+=head1 DESCRIPTION
+
+C<PostgreSQL::Test::Session> encapsulates a C<libpq> session for use in
+PostgreSQL TAP tests, allowing the test to connect without having to spawn
+C<psql> in a child process.
+
+The session object is automatically closed when the object goes out of scope,
+including at script end.
+
+Several methods return a hashref as a result, which will have the following
+fields:
+
+=over
+
+=item * status
+
+=item * error_message (only if there is an error)
+
+=item * names
+
+=item * types
+
+=item * rows
+
+=item * psqlout
+
+=back
+
+The last 4 will be empty unless the SQL produces tuples.
+
+=cut
+
+package PostgreSQL::Test::Session;
+
+use strict;
+use warnings FATAL => 'all';
+
+use File::Basename qw(dirname);
+use Carp;
+use Time::HiRes qw(usleep);
+
+my $setup_ok;
+
+BEGIN
+{
+ if ($ENV{PG_USE_PQ_XS})
+ {
+ # need the directory where Pq.{so,dll,dylib} is installed or built
+ # first, if installed it should be alongside this file
+ my @dirs = (dirname(__FILE__));
+ # second, it should be built here
+ my $topbuilddir = $ENV{MESON_BUILD_ROOT} || $ENV{top_builddir};
+ push (@dirs, "$topbuilddir/src/test/perl/PostgreSQL/Test")
+ if defined($topbuilddir);
+ unshift(@INC, @dirs);
+ # this will fail if we haven't built the shared library
+ require PostgreSQL::Test::Pq;
+ PostgreSQL::Test::Pq->import;
+ $setup_ok = 1; # no other setup required.
+ # restore @INC
+ shift(@INC) foreach @dirs;
+ }
+ else
+ {
+ # default is to use the FFI wrapper
+ # will fail if the FFI libraries and wrappers are not available
+ #
+ # actual setup is done per session, because we get the libdir from the
+ # node object (in most cases)
+ require PostgreSQL::PqFFI;
+ PostgreSQL::PqFFI->import;
+ }
+}
+
+sub _setup
+{
+ return if $setup_ok;
+ my $libdir = shift;
+ PostgreSQL::PqFFI::setup($libdir);
+ $setup_ok = 1;
+}
+
+=pod
+
+=head1 METHODS
+
+=over
+
+=item PostgreSQL::Test::Session->new(node=> $node [, dbname=> $dbname ])
+
+Set up a new session for the node, which must be a C<PostgreSQL::Test::Cluster>
+instance. The default dbame is C<postgres>.
+
+=item PostgreSQL::Test::Session->new(connstr => $connstr [, libdir => $libdir])
+
+Set up a new session for the connection string. If using the FFI libpq wrapper,
+C<$libdir> must point to the directory where the libpq library is installed.
+
+=cut
+
+sub new
+{
+ my $class = shift;
+ my $self = {};
+ bless $self, $class;
+ my %args = @_;
+ my $node = $args{node};
+ my $dbname = $args{dbname} || 'postgres';
+ my $libdir = $args{libdir};
+ my $connstr = $args{connstr};
+ my $wait = $args{wait} // 1;
+ unless ($setup_ok)
+ {
+ unless ($libdir)
+ {
+ croak "bad node" unless $node->isa("PostgreSQL::Test::Cluster");
+ $libdir = $node->config_data('--libdir');
+ }
+ _setup($libdir);
+ }
+ unless ($connstr)
+ {
+ croak "bad node" unless $node->isa("PostgreSQL::Test::Cluster");
+ $connstr = $node->connstr($dbname);
+ }
+ $self->{connstr} = $connstr;
+ $self->{notices} = [];
+
+ if ($wait)
+ {
+ $self->{conn} = PQconnectdb($connstr);
+ # The destructor will clean up for us even if we fail
+ return undef unless PQstatus($self->{conn}) == CONNECTION_OK;
+ $self->_setup_notice_processor();
+ return $self;
+ }
+ else
+ {
+ $self->{conn} = PQconnectStart($connstr);
+ return PQstatus($self->{conn}) != CONNECTION_BAD ? $self : undef;
+ }
+}
+
+# Set up a notice processor to capture notices/warnings
+# Only works with FFI backend, XS backend doesn't support this
+sub _setup_notice_processor
+{
+ my $self = shift;
+
+ # Only available with FFI backend
+ return unless defined &create_notice_processor;
+
+ my $notices = $self->{notices};
+
+ # Create closure that captures notices into our array
+ $self->{_notice_closure} = create_notice_processor(sub {
+ my ($arg, $message) = @_;
+ push @$notices, $message;
+ });
+
+ PQsetNoticeProcessor($self->{conn}, $self->{_notice_closure}, undef);
+}
+
+
+# for a connection started with PQconnectStart, wait until it is in CONNECTION_OK state.
+# Uses PQconnectPoll to drive the async connection forward.
+sub wait_connect
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+ my $timeout = $PostgreSQL::Test::Utils::timeout_default;
+ my $start = time;
+ while (1)
+ {
+ my $poll_res = PQconnectPoll($conn);
+ my $status = PQstatus($conn);
+
+ # Connection is complete
+ if ($poll_res == PGRES_POLLING_OK || $status == CONNECTION_OK)
+ {
+ $self->_setup_notice_processor();
+ return;
+ }
+
+ # Connection failed
+ if ($poll_res == PGRES_POLLING_FAILED || $status == CONNECTION_BAD)
+ {
+ die "connection failed: " . PQerrorMessage($conn);
+ }
+
+ die "timed out waiting for connection" if time - $start > $timeout;
+ usleep(100_000);
+ }
+}
+
+# Single step of async connection polling - drives the connection state machine
+# forward without blocking. Returns the poll result (PGRES_POLLING_* constant).
+sub poll_connect
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+ return PQconnectPoll($conn);
+}
+
+=pod
+
+=item $session->close()
+
+Close the connection
+
+=cut
+
+sub close
+{
+ my $self = shift;
+ PQfinish($self->{conn});
+ delete $self->{conn};
+}
+
+# Alias for compatibility with BackgroundPsql
+*quit = \&close;
+
+# close the session if the object goes out of scope
+sub DESTROY
+{
+ my $self = shift;
+ $self->close if exists $self->{conn};
+}
+
+=pod
+
+=item $session->reconnect()
+
+Reopen the session using the original connstr. If the session is still open,
+close it before reopening.
+
+=cut
+
+sub reconnect
+{
+ my $self = shift;
+ $self->close if exists $self->{conn};
+ $self->{conn} = PQconnectdb($self->{connstr});
+ my $status = PQstatus($self->{conn});
+ if ($status == CONNECTION_OK)
+ {
+ $self->_setup_notice_processor();
+ }
+ return $status;
+}
+
+=pod
+
+=item $session->reconnect_and_clear()
+
+Reconnect and clear all captured notices. Returns the connection status.
+
+=cut
+
+sub reconnect_and_clear
+{
+ my $self = shift;
+ my $status = $self->reconnect();
+ $self->clear_notices();
+ return $status;
+}
+
+=pod
+
+=item $session->get_notices()
+
+Return an arrayref of all captured notices/warnings since the last clear.
+
+=cut
+
+sub get_notices
+{
+ my $self = shift;
+ return $self->{notices};
+}
+
+=pod
+
+=item $session->get_notices_str()
+
+Return all captured notices/warnings as a single string (joined by empty string).
+This is similar to how BackgroundPsql's {stderr} field works.
+
+=cut
+
+sub get_notices_str
+{
+ my $self = shift;
+ return join('', @{$self->{notices}});
+}
+
+=pod
+
+=item $session->clear_notices()
+
+Clear all captured notices/warnings.
+
+=cut
+
+sub clear_notices
+{
+ my $self = shift;
+ # Clear in place - don't replace the array, as the notice processor
+ # closure has a reference to it
+ @{$self->{notices}} = ();
+}
+
+=pod
+
+=item $session->get_stderr()
+
+Return a stderr-like string containing all notices plus any error message
+from the last query. This mimics BackgroundPsql's {stderr} behavior.
+
+=cut
+
+sub get_stderr
+{
+ my $self = shift;
+ my $stderr = $self->get_notices_str();
+ if (exists $self->{last_error} && defined $self->{last_error})
+ {
+ $stderr .= $self->{last_error};
+ }
+ return $stderr;
+}
+
+=pod
+
+=item $session->clear_stderr()
+
+Clear notices and last error, like setting $psql->{stderr} = ''.
+
+=cut
+
+sub clear_stderr
+{
+ my $self = shift;
+ $self->clear_notices();
+ delete $self->{last_error};
+}
+
+=pod
+
+=item $session->conn_status()
+
+Return the connection status. This will be a libpq status value like
+C<CONNECTION_OK>.
+
+=cut
+
+sub conn_status
+{
+ my $self = shift;
+ return exists $self->{conn} ? PQstatus($self->{conn}) : undef;
+}
+
+=pod
+
+=item $session->backend_pid()
+
+Return the backend process ID for this connection.
+
+=cut
+
+sub backend_pid
+{
+ my $self = shift;
+ return PQbackendPID($self->{conn});
+}
+
+=pod
+
+=item $session->do($sql, ...)
+
+Run one or more SQL statements synchronously (using C<PQexec>). The statements
+should not return any tuples. Returns the status, which will be
+C<PGRES_COMMAND_OK> (i.e. 1) in the case of success.
+
+=cut
+
+sub do
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+ my $status;
+ foreach my $sql (@_)
+ {
+ my $result = PQexec($conn, $sql);
+ $status = PQresultStatus($result);
+ PQclear($result);
+ return $status unless $status == PGRES_COMMAND_OK;
+ }
+ return $status;
+}
+
+=pod
+
+=item $session->do_async($sql)
+
+Run a single statement asynchronously, using C<PQsendQuery>. The return value
+is a boolean indicating success.
+
+=cut
+
+sub do_async
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+ my $sql = shift;
+ my $result = PQsendQuery($conn, $sql);
+ return $result; # 1 or 0
+}
+
+# get the next resultset from some aync commands
+# wait if necessary
+# c.f. libpqsrv_get_result
+sub _get_result
+{
+ my $conn = shift;
+ while (PQisBusy($conn))
+ {
+ usleep(100_000);
+ last if PQconsumeInput($conn) == 0;
+ }
+ return PQgetResult($conn);
+}
+
+=pod
+
+=item $session->wait_for_completion()
+
+Wait until all asynchronous SQL has completed
+
+=cut
+
+sub wait_for_completion
+{
+ # wait for all the resultsets and clear them
+ # c.f. libpqsrv_get_result_last
+ my $self = shift;
+ my $conn = $self->{conn};
+ while (my $res = _get_result($conn))
+ {
+ PQclear($res);
+ }
+}
+
+=pod
+
+=item $session->get_async_result()
+
+Wait for and return the result of an async query as a result hash.
+Clears any subsequent results.
+
+=cut
+
+sub get_async_result
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+ my $result = _get_result($conn);
+ return undef unless $result;
+ my $res = _get_result_data($result, $conn);
+ PQclear($result);
+ # Clear any remaining results
+ while (my $r = _get_result($conn))
+ {
+ PQclear($r);
+ }
+ return $res;
+}
+
+=pod
+
+=item $session->wait_for_async_pattern($pattern, $timeout)
+
+Wait for an async query result whose psqlout matches the given regex pattern.
+Returns the matching output string, or dies on timeout/error.
+Default timeout is from $PostgreSQL::Test::Utils::timeout_default.
+
+=cut
+
+sub wait_for_async_pattern
+{
+ my $self = shift;
+ my $pattern = shift;
+ my $timeout = shift // $PostgreSQL::Test::Utils::timeout_default;
+ my $conn = $self->{conn};
+ my $start = time;
+
+ while (1)
+ {
+ # Check if result is ready (non-blocking)
+ PQconsumeInput($conn);
+ if (!PQisBusy($conn))
+ {
+ my $result = PQgetResult($conn);
+ if ($result)
+ {
+ my $res = _get_result_data($result, $conn);
+ PQclear($result);
+ # Clear any remaining results
+ while (my $r = PQgetResult($conn))
+ {
+ PQclear($r);
+ }
+ my $output = $res->{psqlout};
+ if ($output =~ $pattern)
+ {
+ return $output;
+ }
+ # If we got a result but it didn't match, return it anyway
+ # (caller may want to check error)
+ return $output;
+ }
+ }
+
+ die "timed out waiting for async result" if time - $start > $timeout;
+ usleep(100_000);
+ }
+}
+
+=pod
+
+=item $session->try_get_async_result()
+
+Non-blocking check for async query result. Returns result hash if available,
+undef if query is still running.
+
+=cut
+
+sub try_get_async_result
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+
+ PQconsumeInput($conn);
+ return undef if PQisBusy($conn);
+
+ my $result = PQgetResult($conn);
+ return undef unless $result;
+
+ my $res = _get_result_data($result, $conn);
+ PQclear($result);
+ # Clear any remaining results
+ while (my $r = PQgetResult($conn))
+ {
+ PQclear($r);
+ }
+ return $res;
+}
+
+=pod
+
+=item $session->set_password($user, $password)
+
+Set the user's password by calling C<PQchangePassword>.
+
+Returns a result hash.
+
+=cut
+
+# set password for user
+sub set_password
+{
+ my $self = shift;
+ my $user = shift;
+ my $password = shift;
+ my $conn = $self->{conn};
+ my $result = PQchangePassword($conn, $user, $password);
+ my $ret = _get_result_data($result);
+ PQclear($result);
+ return $ret;
+}
+
+# Common internal routine to process result data.
+# The returned object is dead and will be garbage collected as necessary.
+
+sub _get_result_data
+{
+ my $result = shift;
+ my $conn = shift;
+ my $status = PQresultStatus($result);
+ my $res = { status => $status, names => [], types => [], rows => [],
+ psqlout => ""};
+ unless ($status == PGRES_TUPLES_OK || $status == PGRES_COMMAND_OK)
+ {
+ $res->{error_message} = PQerrorMessage($conn);
+ return $res;
+ }
+ if ($status == PGRES_COMMAND_OK)
+ {
+ return $res;
+ }
+ my $ntuples = PQntuples($result);
+ my $nfields = PQnfields($result);
+ # assuming here that the strings returned by PQfname and PQgetvalue
+ # are mapped into perl space using setsvpv or similar and thus won't
+ # be affect by us calling PQclear on the result object.
+ foreach my $field (0 .. $nfields-1)
+ {
+ push(@{$res->{names}}, PQfname($result, $field));
+ push(@{$res->{types}}, PQftype($result, $field));
+ }
+ my @textrows;
+ foreach my $nrow (0 .. $ntuples - 1)
+ {
+ my $row = [];
+ foreach my $field ( 0 .. $nfields - 1)
+ {
+ my $val = PQgetvalue($result, $nrow, $field);
+ if (($val // "") eq "")
+ {
+ $val = undef if PQgetisnull($result, $nrow, $field);
+ }
+ push(@$row, $val);
+ }
+ push(@{$res->{rows}}, $row);
+ no warnings qw(uninitialized);
+ push(@textrows, join('|', @$row));
+ }
+ $res->{psqlout} = join("\n",@textrows) if $ntuples;
+ return $res;
+}
+
+
+=pod
+
+=item $session->query($sql)
+
+Runs sql that might return tuples.
+
+Returns a result hash.
+
+=cut
+
+sub query
+{
+ my $self = shift;
+ my $sql = shift;
+ my $conn = $self->{conn};
+
+ # Use PQsendQuery + PQgetResult to handle multi-statement SQL properly.
+ # This collects results from all statements and returns the last one
+ # that had tuples, similar to how psql works.
+ PQsendQuery($conn, $sql) or do {
+ return { status => -1, error_message => PQerrorMessage($conn),
+ names => [], types => [], rows => [], psqlout => "" };
+ };
+
+ my $final_res;
+ my $last_error;
+ my @all_psqlout;
+
+ while (my $result = _get_result($conn))
+ {
+ my $res = _get_result_data($result, $conn);
+ PQclear($result);
+
+ # Collect output from all statements
+ push @all_psqlout, $res->{psqlout} if $res->{psqlout} ne "";
+
+ # Track errors
+ $last_error = $res->{error_message} if exists $res->{error_message};
+
+ # Keep the last result that had tuples, or the last result overall
+ if ($res->{status} == PGRES_TUPLES_OK || !defined $final_res)
+ {
+ $final_res = $res;
+ }
+ }
+
+ $final_res //= { status => PGRES_COMMAND_OK, names => [], types => [],
+ rows => [], psqlout => "" };
+
+ # Combine all output (like psql does)
+ $final_res->{psqlout} = join("\n", @all_psqlout) if @all_psqlout;
+
+ # If there was any error, include it in the result for query_safe
+ $final_res->{error_message} = $last_error if defined $last_error;
+
+ # Store error for get_stderr()
+ $self->{last_error} = $last_error;
+
+ # If we're in an aborted transaction state, roll it back to clean up.
+ # This mimics psql's behavior with on_error_stop => 0.
+ if (PQtransactionStatus($conn) == PQTRANS_INERROR)
+ {
+ my $rb = PQexec($conn, "ROLLBACK");
+ PQclear($rb) if $rb;
+ }
+
+ return $final_res;
+}
+
+=pod
+
+=item $session->query_safe($sql)
+
+Runs sql that might return tuples, croaking if there's an error.
+Returns the psqlout string (like psql -At output) on success.
+
+=cut
+
+sub query_safe
+{
+ my $self = shift;
+ my $sql = shift;
+ my $res = $self->query($sql);
+ if (exists $res->{error_message}) {
+ # Debug: show where the error occurred
+ my $short_sql = substr($sql, 0, 100);
+ $short_sql =~ s/\s+/ /g;
+ croak "query_safe failed on [$short_sql...]: $res->{error_message}";
+ }
+ return $res->{psqlout};
+}
+
+=pod
+
+=item $session->query_oneval($sql [, $missing_ok ] )
+
+Run a query that is expected to return no more than one tuple with one value;
+
+If C<$missing_ok> is true, return undef if the query returns no tuple. Otherwise
+croak if there is not exactly one tuple, or of the tuple does not have
+exctly one value.
+
+If none of these apply, return the single value from the query. A NULL value
+will result in undef, so if C<$missing_ok> is true you won't be able to
+distinguish between a null value and a missing tuple.
+
+A non NULL value is returned as the string value obtained from C<PQgetvalue>.
+
+=cut
+
+sub query_oneval
+{
+ my $self = shift;
+ my $sql = shift;
+ my $missing_ok = shift; # default is not ok
+ my $conn = $self->{conn};
+ my $result = PQexec($conn, $sql);
+ my $status = PQresultStatus($result);
+ unless ($status == PGRES_TUPLES_OK)
+ {
+ PQclear($result) if $result;
+ croak PQerrorMessage($conn);
+ }
+ my $ntuples = PQntuples($result);
+ return undef if ($missing_ok && !$ntuples);
+ my $nfields = PQnfields($result);
+ croak "$ntuples tuples != 1 or $nfields fields != 1"
+ if $ntuples != 1 || $nfields != 1;
+ my $val = PQgetvalue($result, 0, 0);
+ if ($val eq "")
+ {
+ $val = undef if PQgetisnull($result, 0, 0);
+ }
+ PQclear($result);
+ return $val;
+}
+
+=pod
+
+=item $session->query_tuples($sql, ...)
+
+Run the sql commands and return the output as a single piece of text in the
+same format as C<psql -A -t>.
+
+Fields within tuples are separated by a "|", tuples are spearated by "\n"
+
+=cut
+
+sub query_tuples
+{
+ my $self = shift;
+ my @results;
+ foreach my $sql (@_)
+ {
+ my $res = $self->query($sql);
+ croak $res->{error_message}
+ unless $res->{status} == PGRES_TUPLES_OK;
+ my $rows = $res->{rows};
+ unless (@$rows)
+ {
+ # unfortunately breaks at least one test
+ # push(@results,"-- empty");
+ next;
+ }
+ # join will render undef as an empty string here
+ no warnings qw(uninitialized);
+ my @tuples = map { join('|', @$_); } @$rows;
+ push(@results, join("\n",@tuples));
+ }
+ return join("\n",@results);
+}
+
+
+sub setnonblocking
+{
+ my $self = shift;
+ my $val = shift;
+ my $res = PQsetnonblocking($self->{conn}, $val);
+ croak "problem setting non-blocking"
+ if $res;
+ return;
+}
+
+sub isnonblocking
+{
+ my $self = shift;
+ return PQisnonblocking($self->{conn});
+}
+
+sub enterPipelineMode
+{
+ my $self = shift;
+ return PQenterPipelineMode($self->{conn});
+}
+
+sub exitPipelineMode
+{
+ my $self = shift;
+ return PQexitPipelineMode($self->{conn});
+}
+
+sub pipelineStatus
+{
+ my $self = shift;
+ return PQpipelineStatus($self->{conn});
+}
+
+sub pipelineSync
+{
+ my $self = shift;
+ return PQpipelineSync($self->{conn});
+}
+
+
+sub do_pipeline
+{
+ my $self = shift;
+ my $statement = shift;
+ my @args = @_;
+ my $nargs = scalar(@args);
+ my $res = PQsendQueryParams($self->{conn}, $statement, $nargs, undef, \@args, undef , undef, 0);
+ return $res;
+}
+
+=pod
+
+=item $session->get_notification()
+
+Check for a pending notification and return it as a hashref with keys
+C<channel>, C<pid>, and C<payload>. Returns undef if no notification is
+available.
+
+Automatically consumes any pending input before checking for notifications.
+
+=cut
+
+sub get_notification
+{
+ my $self = shift;
+ my $conn = $self->{conn};
+
+ # Consume any pending input
+ PQconsumeInput($conn);
+
+ my $notify = PQnotifies($conn);
+ return undef unless $notify;
+
+ my $result = {
+ channel => PQnotify_channel($notify),
+ pid => PQnotify_be_pid($notify),
+ payload => PQnotify_payload($notify),
+ };
+
+ PQnotify_free($notify);
+
+ return $result;
+}
+
+=pod
+
+=item $session->get_all_notifications()
+
+Consume input and return all pending notifications as an arrayref of hashrefs.
+Each hashref has keys C<channel>, C<pid>, and C<payload>.
+
+=cut
+
+sub get_all_notifications
+{
+ my $self = shift;
+ my @notifications;
+
+ while (my $notify = $self->get_notification())
+ {
+ push @notifications, $notify;
+ }
+
+ return \@notifications;
+}
+
+=pod
+
+=back
+
+=cut
+
+
+1;
diff --git a/src/test/perl/meson.build b/src/test/perl/meson.build
index 0fd36c9e570..8286458d134 100644
--- a/src/test/perl/meson.build
+++ b/src/test/perl/meson.build
@@ -15,4 +15,37 @@ install_data(
'PostgreSQL/Test/BackgroundPsql.pm',
'PostgreSQL/Test/AdjustDump.pm',
'PostgreSQL/Test/AdjustUpgrade.pm',
+ 'PostgreSQL/Test/Pq.pm',
install_dir: dir_pgxs / 'src/test/perl/PostgreSQL/Test')
+
+# Build the Pq XS module if perl is available
+if perl_dep.found()
+ pq_subppdir = run_command(perl, '-e',
+ 'use List::Util qw(first); print first { -r "$_/ExtUtils/xsubpp" } @INC',
+ check: true).stdout()
+ pq_xsubpp = '@0@/ExtUtils/xsubpp'.format(pq_subppdir)
+ pq_typemap = '@0@/ExtUtils/typemap'.format(privlibexp)
+
+ pq_xs_c = custom_target('Pq.c',
+ input: files('PostgreSQL/Test/Pq.xs'),
+ output: 'Pq.c',
+ command: [perl, pq_xsubpp, '-typemap', pq_typemap, '-output', '@OUTPUT@', '@INPUT@']
+ )
+
+ pq_module = shared_module('Pq',
+ pq_xs_c,
+ include_directories: [
+ include_directories('PostgreSQL/Test'),
+ include_directories('../../pl/plperl'), # for ppport.h
+ include_directories('../../interfaces/libpq'),
+ postgres_inc,
+ ],
+ c_args: ['-Wno-shadow=compatible-local', '-Wno-declaration-after-statement'],
+ dependencies: [perl_dep, libpq],
+ name_prefix: '', # Perl XS modules don't use lib prefix
+ install: true,
+ install_dir: dir_pgxs / 'src/test/perl/PostgreSQL/Test',
+ install_rpath: ':'.join(mod_install_rpaths + ['@0@/CORE'.format(archlibexp)]),
+ build_rpath: '@0@/CORE'.format(archlibexp),
+ )
+endif
diff --git a/src/test/postmaster/t/002_connection_limits.pl b/src/test/postmaster/t/002_connection_limits.pl
index 8c67c4a86c7..1b4a97d16f4 100644
--- a/src/test/postmaster/t/002_connection_limits.pl
+++ b/src/test/postmaster/t/002_connection_limits.pl
@@ -7,6 +7,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -33,19 +34,20 @@ GRANT pg_use_reserved_connections TO regress_reserved;
CREATE USER regress_superuser LOGIN SUPERUSER;
});
+my $node_connstr = $node->connstr('postgres');
+my $libdir = $node->config_data('--libdir');
+
# With the limits we set in postgresql.conf, we can establish:
# - 3 connections for any user with no special privileges
# - 2 more connections for users belonging to "pg_use_reserved_connections"
# - 1 more connection for superuser
-sub background_psql_as_user
+sub session_as_user
{
my $user = shift;
+ my $connstr = "$node_connstr user=$user";
- return $node->background_psql(
- 'postgres',
- on_error_die => 1,
- extra_params => [ '--username' => $user ]);
+ return PostgreSQL::Test::Session->new( connstr => $connstr, libdir => $libdir);
}
# Like connect_fails(), except that we also wait for the failed backend to
@@ -82,9 +84,9 @@ $node->restart;
my @sessions = ();
my @raw_connections = ();
-push(@sessions, background_psql_as_user('regress_regular'));
-push(@sessions, background_psql_as_user('regress_regular'));
-push(@sessions, background_psql_as_user('regress_regular'));
+push(@sessions, session_as_user('regress_regular'));
+push(@sessions, session_as_user('regress_regular'));
+push(@sessions, session_as_user('regress_regular'));
connect_fails_wait(
$node,
"dbname=postgres user=regress_regular",
@@ -93,8 +95,8 @@ connect_fails_wait(
qr/FATAL: remaining connection slots are reserved for roles with privileges of the "pg_use_reserved_connections" role/
);
-push(@sessions, background_psql_as_user('regress_reserved'));
-push(@sessions, background_psql_as_user('regress_reserved'));
+push(@sessions, session_as_user('regress_reserved'));
+push(@sessions, session_as_user('regress_reserved'));
connect_fails_wait(
$node,
"dbname=postgres user=regress_reserved",
@@ -103,7 +105,7 @@ connect_fails_wait(
qr/FATAL: remaining connection slots are reserved for roles with the SUPERUSER attribute/
);
-push(@sessions, background_psql_as_user('regress_superuser'));
+push(@sessions, session_as_user('regress_superuser'));
connect_fails_wait(
$node,
"dbname=postgres user=regress_superuser",
@@ -150,7 +152,7 @@ SKIP:
# Clean up
foreach my $session (@sessions)
{
- $session->quit;
+ $session->close;
}
foreach my $socket (@raw_connections)
{
diff --git a/src/test/recovery/t/009_twophase.pl b/src/test/recovery/t/009_twophase.pl
index 879e493b5b8..3c4a6b6824b 100644
--- a/src/test/recovery/t/009_twophase.pl
+++ b/src/test/recovery/t/009_twophase.pl
@@ -6,6 +6,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -331,11 +332,10 @@ $cur_primary->stop;
$cur_standby->restart;
# Acquire a snapshot in standby, before we commit the prepared transaction
-my $standby_session =
- $cur_standby->background_psql('postgres', on_error_die => 1);
-$standby_session->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ");
+my $standby_session = PostgreSQL::Test::Session->new(node => $cur_standby);
+$standby_session->do("BEGIN ISOLATION LEVEL REPEATABLE READ");
$psql_out =
- $standby_session->query_safe("SELECT count(*) FROM t_009_tbl_standby_mvcc");
+ $standby_session->query_oneval("SELECT count(*) FROM t_009_tbl_standby_mvcc");
is($psql_out, '0',
"Prepared transaction not visible in standby before commit");
@@ -349,17 +349,17 @@ COMMIT PREPARED 'xact_009_standby_mvcc';
# Still not visible to the old snapshot
$psql_out =
- $standby_session->query_safe("SELECT count(*) FROM t_009_tbl_standby_mvcc");
+ $standby_session->query_oneval("SELECT count(*) FROM t_009_tbl_standby_mvcc");
is($psql_out, '0',
"Committed prepared transaction not visible to old snapshot in standby");
# Is visible to a new snapshot
-$standby_session->query_safe("COMMIT");
+$standby_session->do("COMMIT");
$psql_out =
- $standby_session->query_safe("SELECT count(*) FROM t_009_tbl_standby_mvcc");
+ $standby_session->query_oneval("SELECT count(*) FROM t_009_tbl_standby_mvcc");
is($psql_out, '2',
"Committed prepared transaction is visible to new snapshot in standby");
-$standby_session->quit;
+$standby_session->close;
###############################################################################
# Check for a lock conflict between prepared transaction with DDL inside and
diff --git a/src/test/recovery/t/013_crash_restart.pl b/src/test/recovery/t/013_crash_restart.pl
index 20d648ad6af..31204850b37 100644
--- a/src/test/recovery/t/013_crash_restart.pl
+++ b/src/test/recovery/t/013_crash_restart.pl
@@ -131,7 +131,7 @@ ok( pump_until(
$monitor->finish;
# Wait till server restarts
-is($node->poll_query_until('postgres', undef, ''),
+is($node->poll_until_connection('postgres'),
"1", "reconnected after SIGQUIT");
@@ -213,7 +213,7 @@ ok( pump_until(
$monitor->finish;
# Wait till server restarts
-is($node->poll_query_until('postgres', undef, ''),
+is($node->poll_until_connection('postgres'),
"1", "reconnected after SIGKILL");
# Make sure the committed rows survived, in-progress ones not
diff --git a/src/test/recovery/t/022_crash_temp_files.pl b/src/test/recovery/t/022_crash_temp_files.pl
index 5de9b0fb0eb..6c55fcf40b1 100644
--- a/src/test/recovery/t/022_crash_temp_files.pl
+++ b/src/test/recovery/t/022_crash_temp_files.pl
@@ -146,7 +146,7 @@ ok( pump_until(
$killme2->finish;
# Wait till server finishes restarting
-$node->poll_query_until('postgres', undef, '');
+$node->poll_until_connection('postgres');
# Check for temporary files
is( $node->safe_psql(
@@ -253,7 +253,7 @@ ok( pump_until(
$killme2->finish;
# Wait till server finishes restarting
-$node->poll_query_until('postgres', undef, '');
+$node->poll_until_connection('postgres');
# Check for temporary files -- should be there
is( $node->safe_psql(
diff --git a/src/test/recovery/t/031_recovery_conflict.pl b/src/test/recovery/t/031_recovery_conflict.pl
index 7a740f69806..74349749170 100644
--- a/src/test/recovery/t/031_recovery_conflict.pl
+++ b/src/test/recovery/t/031_recovery_conflict.pl
@@ -7,6 +7,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -67,8 +68,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
# a longrunning psql that we can use to trigger conflicts
-my $psql_standby =
- $node_standby->background_psql($test_db, on_error_stop => 0);
+my $psql_standby = PostgreSQL::Test::Session->new(node => $node_standby, dbname => $test_db);
my $expected_conflicts = 0;
@@ -96,7 +96,7 @@ my $cursor1 = "test_recovery_conflict_cursor";
# DECLARE and use a cursor on standby, causing buffer with the only block of
# the relation to be pinned on the standby
-my $res = $psql_standby->query_safe(
+my $res = $psql_standby->query_oneval(
qq[
BEGIN;
DECLARE $cursor1 CURSOR FOR SELECT b FROM $table1;
@@ -119,7 +119,7 @@ $node_primary->safe_psql($test_db, qq[VACUUM FREEZE $table1;]);
$node_primary->wait_for_replay_catchup($node_standby);
check_conflict_log("User was holding shared buffer pin for too long");
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
check_conflict_stat("bufferpin");
@@ -132,7 +132,7 @@ $node_primary->safe_psql($test_db,
$node_primary->wait_for_replay_catchup($node_standby);
# DECLARE and FETCH from cursor on the standby
-$res = $psql_standby->query_safe(
+$res = $psql_standby->query_oneval(
qq[
BEGIN;
DECLARE $cursor1 CURSOR FOR SELECT b FROM $table1;
@@ -152,7 +152,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
check_conflict_log(
"User query might have needed to see row versions that must be removed");
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
check_conflict_stat("snapshot");
@@ -161,7 +161,7 @@ $sect = "lock conflict";
$expected_conflicts++;
# acquire lock to conflict with
-$res = $psql_standby->query_safe(
+$res = $psql_standby->query_oneval(
qq[
BEGIN;
LOCK TABLE $table1 IN ACCESS SHARE MODE;
@@ -175,7 +175,7 @@ $node_primary->safe_psql($test_db, qq[DROP TABLE $table1;]);
$node_primary->wait_for_replay_catchup($node_standby);
check_conflict_log("User was holding a relation lock for too long");
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
check_conflict_stat("lock");
@@ -186,7 +186,7 @@ $expected_conflicts++;
# DECLARE a cursor for a query which, with sufficiently low work_mem, will
# spill tuples into temp files in the temporary tablespace created during
# setup.
-$res = $psql_standby->query_safe(
+$res = $psql_standby->query_oneval(
qq[
BEGIN;
SET work_mem = '64kB';
@@ -205,7 +205,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
check_conflict_log(
"User was or might have been using tablespace that must be dropped");
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
check_conflict_stat("tablespace");
@@ -220,8 +220,9 @@ $node_standby->adjust_conf(
'postgresql.conf',
'max_standby_streaming_delay',
"${PostgreSQL::Test::Utils::timeout_default}s");
+$psql_standby->close;
$node_standby->restart();
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
# Generate a few dead rows, to later be cleaned up by vacuum. Then acquire a
# lock on another relation in a prepared xact, so it's held continuously by
@@ -244,12 +245,15 @@ SELECT txid_current();
$node_primary->wait_for_replay_catchup($node_standby);
-$res = $psql_standby->query_until(
- qr/^1$/m, qq[
+$res = $psql_standby->query_oneval(
+ qq[
BEGIN;
-- hold pin
DECLARE $cursor1 CURSOR FOR SELECT a FROM $table1;
FETCH FORWARD FROM $cursor1;
+]);
+is ($res, 1, "pin held");
+$psql_standby->do_async(qq[
-- wait for lock held by prepared transaction
SELECT * FROM $table2;
]);
@@ -270,15 +274,16 @@ $node_primary->safe_psql($test_db, qq[VACUUM FREEZE $table1;]);
$node_primary->wait_for_replay_catchup($node_standby);
check_conflict_log("User transaction caused buffer deadlock with recovery.");
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
check_conflict_stat("deadlock");
# clean up for next tests
$node_primary->safe_psql($test_db, qq[ROLLBACK PREPARED 'lock';]);
$node_standby->adjust_conf('postgresql.conf', 'max_standby_streaming_delay',
- '50ms');
+ '50ms');
+$psql_standby->close;
$node_standby->restart();
-$psql_standby->reconnect_and_clear();
+$psql_standby->reconnect();
# Check that expected number of conflicts show in pg_stat_database. Needs to
@@ -302,7 +307,7 @@ check_conflict_log("User was connected to a database that must be dropped");
# explicitly shut down psql instances gracefully - to avoid hangs or worse on
# windows
-$psql_standby->quit;
+$psql_standby->close;
$node_standby->stop();
$node_primary->stop();
diff --git a/src/test/recovery/t/037_invalid_database.pl b/src/test/recovery/t/037_invalid_database.pl
index a0947108700..b10a79d0ee7 100644
--- a/src/test/recovery/t/037_invalid_database.pl
+++ b/src/test/recovery/t/037_invalid_database.pl
@@ -5,6 +5,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -91,41 +92,38 @@ is($node->psql('postgres', 'DROP DATABASE regression_invalid'),
# dropping the database, making it a suitable point to wait. Since relcache
# init reads pg_tablespace, establish each connection before locking. This
# avoids a connection-time hang with debug_discard_caches.
-my $cancel = $node->background_psql('postgres', on_error_stop => 1);
-my $bgpsql = $node->background_psql('postgres', on_error_stop => 0);
-my $pid = $bgpsql->query('SELECT pg_backend_pid()');
+my $cancel = PostgreSQL::Test::Session->new(node=>$node);
+my $bgpsql = PostgreSQL::Test::Session->new(node=>$node);
+my $pid = $bgpsql->query_oneval('SELECT pg_backend_pid()');
# create the database, prevent drop database via lock held by a 2PC transaction
-$bgpsql->query_safe(
- qq(
- CREATE DATABASE regression_invalid_interrupt;
- BEGIN;
+is (1, $bgpsql->do(
+ qq(
+ CREATE DATABASE regression_invalid_interrupt;),
+ qq(BEGIN;
LOCK pg_tablespace;
- PREPARE TRANSACTION 'lock_tblspc';));
+ PREPARE TRANSACTION 'lock_tblspc';)));
# Try to drop. This will wait due to the still held lock.
-$bgpsql->query_until(qr//, "DROP DATABASE regression_invalid_interrupt;\n");
+$bgpsql->do_async("DROP DATABASE regression_invalid_interrupt;");
# Once the DROP DATABASE is waiting for the lock, interrupt it.
-ok( $cancel->query_safe(
- qq(
+my $cancel_res = $cancel->query(
+ qq[
DO \$\$
BEGIN
WHILE NOT EXISTS(SELECT * FROM pg_locks WHERE NOT granted AND relation = 'pg_tablespace'::regclass AND mode = 'AccessShareLock') LOOP
PERFORM pg_sleep(.1);
END LOOP;
END\$\$;
- SELECT pg_cancel_backend($pid);)),
- "canceling DROP DATABASE");
-$cancel->quit();
+ SELECT pg_cancel_backend($pid)]);
+is (2, $cancel_res->{status}, "canceling DROP DATABASE"); # COMMAND_TUPLES_OK
+$cancel->close();
+$bgpsql->wait_for_completion;
# wait for cancellation to be processed
-ok( pump_until(
- $bgpsql->{run}, $bgpsql->{timeout},
- \$bgpsql->{stderr}, qr/canceling statement due to user request/),
- "cancel processed");
-$bgpsql->{stderr} = '';
+pass("cancel processed");
# Verify that connections to the database aren't allowed. The backend checks
# this before relcache init, so the lock won't interfere.
@@ -134,9 +132,12 @@ is($node->psql('regression_invalid_interrupt', ''),
# To properly drop the database, we need to release the lock previously preventing
# doing so.
-$bgpsql->query_safe(qq(ROLLBACK PREPARED 'lock_tblspc'));
-$bgpsql->query_safe(qq(DROP DATABASE regression_invalid_interrupt));
+ok($bgpsql->do(qq(ROLLBACK PREPARED 'lock_tblspc')),
+ "unblock DROP DATABASE");
-$bgpsql->quit();
+ok($bgpsql->query(qq(DROP DATABASE regression_invalid_interrupt)),
+ "DROP DATABASE invalid_interrupt");
+
+$bgpsql->close();
done_testing();
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 47d64d05ad1..a2930c9f85f 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -4,6 +4,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -768,17 +769,13 @@ $primary->safe_psql('postgres',
"SELECT pg_create_logical_replication_slot('test_slot', 'test_decoding', false, false, true);"
);
-my $back_q = $primary->background_psql(
- 'postgres',
- on_error_stop => 0,
- timeout => $PostgreSQL::Test::Utils::timeout_default);
+my $back_q = PostgreSQL::Test::Session->new(node=>$primary);
# pg_logical_slot_get_changes will be blocked until the standby catches up,
# hence it needs to be executed in a background session.
$offset = -s $primary->logfile;
-$back_q->query_until(
- qr/logical_slot_get_changes/, q(
- \echo logical_slot_get_changes
+$back_q->do_async(
+ q(
SELECT pg_logical_slot_get_changes('test_slot', NULL, NULL);
));
@@ -796,7 +793,8 @@ $primary->reload;
# Since there are no slots in synchronized_standby_slots, the function
# pg_logical_slot_get_changes should now return, and the session can be
# stopped.
-$back_q->quit;
+$back_q->wait_for_completion;
+$back_q->close;
$primary->safe_psql('postgres',
"SELECT pg_drop_replication_slot('test_slot');");
@@ -1057,12 +1055,8 @@ $standby2->reload;
# synchronization until the remote slot catches up.
# The API will not return until this happens, to be able to make
# further calls, call the API in a background process.
-my $h = $standby2->background_psql('postgres', on_error_stop => 0);
-
-$h->query_until(qr/start/, q(
- \echo start
- SELECT pg_sync_replication_slots();
- ));
+my $h = PostgreSQL::Test::Session->new(node => $standby2);
+$h->do_async(q(SELECT pg_sync_replication_slots();));
# Confirm that the slot sync is skipped due to the remote slot lagging behind
$standby2->wait_for_log(
@@ -1100,6 +1094,7 @@ $standby2->wait_for_log(
qr/newly created replication slot \"lsub1_slot\" is sync-ready now/,
$log_offset);
-$h->quit;
+$h->wait_for_completion;
+$h->close;
done_testing();
diff --git a/src/test/recovery/t/041_checkpoint_at_promote.pl b/src/test/recovery/t/041_checkpoint_at_promote.pl
index d0783fef9ae..fcc01334589 100644
--- a/src/test/recovery/t/041_checkpoint_at_promote.pl
+++ b/src/test/recovery/t/041_checkpoint_at_promote.pl
@@ -4,6 +4,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Time::HiRes qw(usleep);
use Test::More;
@@ -70,11 +71,9 @@ $node_standby->safe_psql('postgres',
# Execute a restart point on the standby, that we will now be waiting on.
# This needs to be in the background.
my $logstart = -s $node_standby->logfile;
-my $psql_session =
- $node_standby->background_psql('postgres', on_error_stop => 0);
-$psql_session->query_until(
- qr/starting_checkpoint/, q(
- \echo starting_checkpoint
+my $psql_session = PostgreSQL::Test::Session->new(node=> $node_standby);
+$psql_session->do_async(
+ q(
CHECKPOINT;
));
@@ -159,7 +158,7 @@ ok( pump_until(
$killme->finish;
# Wait till server finishes restarting.
-$node_standby->poll_query_until('postgres', undef, '');
+$node_standby->poll_until_connection('postgres');
# After recovery, the server should be able to start.
my $stdout;
diff --git a/src/test/recovery/t/042_low_level_backup.pl b/src/test/recovery/t/042_low_level_backup.pl
index df4ae029fe6..58489482341 100644
--- a/src/test/recovery/t/042_low_level_backup.pl
+++ b/src/test/recovery/t/042_low_level_backup.pl
@@ -10,6 +10,7 @@ use warnings FATAL => 'all';
use File::Copy qw(copy);
use File::Path qw(rmtree);
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -20,11 +21,10 @@ $node_primary->start;
# Start backup.
my $backup_name = 'backup1';
-my $psql = $node_primary->background_psql('postgres');
+my $psql = PostgreSQL::Test::Session->new(node => $node_primary);
-$psql->query_safe("SET client_min_messages TO WARNING");
-$psql->set_query_timer_restart;
-$psql->query_safe("select pg_backup_start('test label')");
+$psql->do("SET client_min_messages TO WARNING");
+$psql->query("select pg_backup_start('test label')");
# Copy files.
my $backup_dir = $node_primary->backup_dir . '/' . $backup_name;
@@ -81,9 +81,9 @@ my $stop_segment_name = $node_primary->safe_psql('postgres',
# Stop backup and get backup_label, the last segment is archived.
my $backup_label =
- $psql->query_safe("select labelfile from pg_backup_stop()");
+ $psql->query_oneval("select labelfile from pg_backup_stop()");
-$psql->quit;
+$psql->close;
# Rather than writing out backup_label, try to recover the backup without
# backup_label to demonstrate that recovery will not work correctly without it,
diff --git a/src/test/recovery/t/046_checkpoint_logical_slot.pl b/src/test/recovery/t/046_checkpoint_logical_slot.pl
index 1ecb47a8b58..3261117b1bc 100644
--- a/src/test/recovery/t/046_checkpoint_logical_slot.pl
+++ b/src/test/recovery/t/046_checkpoint_logical_slot.pl
@@ -8,6 +8,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -52,13 +53,10 @@ $node->safe_psql('postgres',
$node->safe_psql('postgres', q{checkpoint});
# Generate some transactions to get RUNNING_XACTS.
-my $xacts = $node->background_psql('postgres');
-$xacts->query_until(
- qr/run_xacts/,
- q(\echo run_xacts
-SELECT 1 \watch 0.1
-\q
-));
+for (my $i = 0; $i < 10; $i++)
+{
+ $node->safe_psql('postgres', 'SELECT 1');
+}
$node->advance_wal(20);
@@ -72,16 +70,11 @@ $node->advance_wal(20);
# removing old WAL segments.
note('starting checkpoint');
-my $checkpoint = $node->background_psql('postgres');
-$checkpoint->query_safe(
+$node->safe_psql('postgres',
q(select injection_points_attach('checkpoint-before-old-wal-removal','wait'))
);
-$checkpoint->query_until(
- qr/starting_checkpoint/,
- q(\echo starting_checkpoint
-checkpoint;
-\q
-));
+my $checkpoint = PostgreSQL::Test::Session->new(node => $node);
+$checkpoint->do_async(q(CHECKPOINT;));
# Wait until the checkpoint stops right before removing WAL segments.
note('waiting for injection_point');
@@ -90,17 +83,21 @@ note('injection_point is reached');
# Try to advance the logical slot, but make it stop when it moves to the next
# WAL segment (this has to happen in the background, too).
-my $logical = $node->background_psql('postgres');
-$logical->query_safe(
+# We need to call pg_logical_slot_get_changes repeatedly until the slot
+# advances to the next segment and hits the injection point.
+my $logical = PostgreSQL::Test::Session->new(node => $node);
+$logical->do(
q{select injection_points_attach('logical-replication-slot-advance-segment','wait');}
);
-$logical->query_until(
- qr/get_changes/,
- q(
-\echo get_changes
-select count(*) from pg_logical_slot_get_changes('slot_logical', null, null) \watch 1
-\q
-));
+$logical->do_async(
+ q{DO $$
+ BEGIN
+ LOOP
+ PERFORM count(*) FROM pg_logical_slot_get_changes('slot_logical', null, null);
+ PERFORM pg_sleep(0.1);
+ END LOOP;
+ END $$;}
+);
# Wait until the slot's restart_lsn points to the next WAL segment.
note('waiting for injection_point');
@@ -138,12 +135,8 @@ eval {
};
is($@, '', "Logical slot still valid");
-# If we send \q with $<psql_session>->quit the command can be sent to the
-# session already closed. So \q is in initial script, here we only finish
-# IPC::Run
-$xacts->{run}->finish;
-$checkpoint->{run}->finish;
-$logical->{run}->finish;
+# Sessions were terminated by the server crash and will be cleaned up
+# automatically when they go out of scope.
# Verify that the synchronized slots won't be invalidated immediately after
# synchronization in the presence of a concurrent checkpoint.
@@ -186,15 +179,11 @@ $primary->wait_for_replay_catchup($standby);
# checkpoint stops right before invalidating replication slots.
note('starting checkpoint');
-$checkpoint = $standby->background_psql('postgres');
-$checkpoint->query_safe(
+$standby->safe_psql('postgres',
q(select injection_points_attach('restartpoint-before-slot-invalidation','wait'))
);
-$checkpoint->query_until(
- qr/starting_checkpoint/,
- q(\echo starting_checkpoint
-checkpoint;
-));
+$checkpoint = PostgreSQL::Test::Session->new(node => $standby);
+$checkpoint->do_async(q(CHECKPOINT;));
# Wait until the checkpoint stops right before invalidating slots
note('waiting for injection_point');
@@ -215,7 +204,8 @@ $standby->safe_psql('postgres',
q{select injection_points_wakeup('restartpoint-before-slot-invalidation');
select injection_points_detach('restartpoint-before-slot-invalidation')});
-$checkpoint->quit;
+$checkpoint->wait_for_completion;
+$checkpoint->close;
# Confirm that the slot is not invalidated
is( $standby->safe_psql(
diff --git a/src/test/recovery/t/047_checkpoint_physical_slot.pl b/src/test/recovery/t/047_checkpoint_physical_slot.pl
index 4334145abe1..43193c219ba 100644
--- a/src/test/recovery/t/047_checkpoint_physical_slot.pl
+++ b/src/test/recovery/t/047_checkpoint_physical_slot.pl
@@ -8,6 +8,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -69,16 +70,11 @@ note("restart lsn before checkpoint: $restart_lsn_init");
# removing old WAL segments.
note('starting checkpoint');
-my $checkpoint = $node->background_psql('postgres');
-$checkpoint->query_safe(
+my $checkpoint = PostgreSQL::Test::Session->new(node => $node);
+$checkpoint->do(
q{select injection_points_attach('checkpoint-before-old-wal-removal','wait')}
);
-$checkpoint->query_until(
- qr/starting_checkpoint/,
- q(\echo starting_checkpoint
-checkpoint;
-\q
-));
+$checkpoint->do_async('checkpoint');
# Wait until the checkpoint stops right before removing WAL segments.
note('waiting for injection_point');
@@ -104,7 +100,11 @@ my $restart_lsn_old = $node->safe_psql('postgres',
chomp($restart_lsn_old);
note("restart lsn before stop: $restart_lsn_old");
-# Abruptly stop the server.
+$checkpoint->wait_for_completion();
+$checkpoint->close();
+
+# Abruptly stop the server (1 second should be enough for the checkpoint
+# to finish; it would be better).
$node->stop('immediate');
$node->start;
diff --git a/src/test/recovery/t/048_vacuum_horizon_floor.pl b/src/test/recovery/t/048_vacuum_horizon_floor.pl
index 52acb5561d6..a487be04dc8 100644
--- a/src/test/recovery/t/048_vacuum_horizon_floor.pl
+++ b/src/test/recovery/t/048_vacuum_horizon_floor.pl
@@ -45,12 +45,10 @@ my $orig_conninfo = $node_primary->connstr();
my $table1 = "vac_horizon_floor_table";
# Long-running Primary Session A
-my $psql_primaryA =
- $node_primary->background_psql($test_db, on_error_stop => 1);
+my $session_primaryA = PostgreSQL::Test::Session->new(node => $node_primary, dbname => $test_db);
# Long-running Primary Session B
-my $psql_primaryB =
- $node_primary->background_psql($test_db, on_error_stop => 1);
+my $session_primaryB = PostgreSQL::Test::Session->new(node => $node_primary, dbname => $test_db);
# Our test relies on two rounds of index vacuuming for reasons elaborated
# later. To trigger two rounds of index vacuuming, we must fill up the
@@ -123,7 +121,7 @@ $node_replica->poll_query_until(
# Now insert and update a tuple which will be visible to the vacuum on the
# primary but which will have xmax newer than the oldest xmin on the standby
# that was recently disconnected.
-my $res = $psql_primaryA->query_safe(
+my $res = $session_primaryA->query(
qq[
INSERT INTO $table1 VALUES (99);
UPDATE $table1 SET col1 = 100 WHERE col1 = 99;
@@ -132,7 +130,7 @@ my $res = $psql_primaryA->query_safe(
);
# Make sure the UPDATE finished
-like($res, qr/^after_update$/m, "UPDATE occurred on primary session A");
+like($res->{psqlout}, qr/^after_update$/m, "UPDATE occurred on primary session A");
# Open a cursor on the primary whose pin will keep VACUUM from getting a
# cleanup lock on the first page of the relation. We want VACUUM to be able to
@@ -145,7 +143,7 @@ my $primary_cursor1 = "vac_horizon_floor_cursor1";
# The first value inserted into the table was a 7, so FETCH FORWARD should
# return a 7. That's how we know the cursor has a pin.
# Disable index scans so the cursor pins heap pages and not index pages.
-$res = $psql_primaryB->query_safe(
+$res = $session_primaryB->query(
qq[
BEGIN;
SET enable_bitmapscan = off;
@@ -156,11 +154,11 @@ $res = $psql_primaryB->query_safe(
]
);
-is($res, 7, qq[Cursor query returned $res. Expected value 7.]);
+is($res->{psqlout}, 7, qq[Cursor query returned $res->{psqlout}. Expected value 7.]);
# Get the PID of the session which will run the VACUUM FREEZE so that we can
# use it to filter pg_stat_activity later.
-my $vacuum_pid = $psql_primaryA->query_safe("SELECT pg_backend_pid();");
+my $vacuum_pid = $session_primaryA->query_oneval("SELECT pg_backend_pid();");
# Now start a VACUUM FREEZE on the primary. It will call vacuum_get_cutoffs()
# and establish values of OldestXmin and GlobalVisState which are newer than
@@ -176,14 +174,8 @@ my $vacuum_pid = $psql_primaryA->query_safe("SELECT pg_backend_pid();");
# pages of the heap must be processed in order by a single worker to ensure
# test stability (PARALLEL 0 shouldn't be necessary but guards against the
# possibility of parallel heap vacuuming).
-$psql_primaryA->{stdin} .= qq[
- SET maintenance_io_concurrency = 0;
- VACUUM (VERBOSE, FREEZE, PARALLEL 0) $table1;
- \\echo VACUUM
- ];
-
-# Make sure the VACUUM command makes it to the server.
-$psql_primaryA->{run}->pump_nb();
+$session_primaryA->do('SET maintenance_io_concurrency = 0;');
+$session_primaryA->do_async("VACUUM (VERBOSE, FREEZE, PARALLEL 0) $table1;");
# Make sure that the VACUUM has already called vacuum_get_cutoffs() and is
# just waiting on the lock to start vacuuming. We don't want the standby to
@@ -229,7 +221,7 @@ $node_primary->poll_query_until(
# expect that a round of index vacuuming has happened and that the vacuum is
# now waiting for the cursor to release its pin on the last page of the
# relation.
-$res = $psql_primaryB->query_safe("FETCH $primary_cursor1");
+$res = $session_primaryB->query_oneval("FETCH $primary_cursor1");
is($res, 7,
qq[Cursor query returned $res from second fetch. Expected value 7.]);
@@ -243,13 +235,7 @@ $node_primary->poll_query_until(
], 't');
# Commit the transaction with the open cursor so that the VACUUM can finish.
-$psql_primaryB->query_until(
- qr/^commit$/m,
- qq[
- COMMIT;
- \\echo commit
- ]
-);
+$session_primaryB->do('COMMIT');
# VACUUM proceeds with pruning and does a visibility check on each tuple. In
# older versions of Postgres, pruning found our final dead tuple
@@ -281,8 +267,8 @@ $node_primary->safe_psql($test_db, "INSERT INTO $table1 VALUES (1);");
$node_primary->wait_for_catchup($node_replica, 'replay', $primary_lsn);
## Shut down psqls
-$psql_primaryA->quit;
-$psql_primaryB->quit;
+$session_primaryA->close;
+$session_primaryB->close;
$node_replica->stop();
$node_primary->stop();
diff --git a/src/test/recovery/t/049_wait_for_lsn.pl b/src/test/recovery/t/049_wait_for_lsn.pl
index bf61b8c47cf..d5ecbec3564 100644
--- a/src/test/recovery/t/049_wait_for_lsn.pl
+++ b/src/test/recovery/t/049_wait_for_lsn.pl
@@ -5,6 +5,7 @@ use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -354,10 +355,8 @@ for (my $i = 0; $i < 5; $i++)
my $lsn =
$node_primary->safe_psql('postgres',
"SELECT pg_current_wal_insert_lsn()");
- $psql_sessions[$i] = $node_standby->background_psql('postgres');
- $psql_sessions[$i]->query_until(
- qr/start/, qq[
- \\echo start
+ $psql_sessions[$i] = PostgreSQL::Test::Session->new(node => $node_standby);
+ $psql_sessions[$i]->do_async(qq[
WAIT FOR LSN '${lsn}';
SELECT log_count(${i});
]);
@@ -368,7 +367,8 @@ $node_standby->safe_psql('postgres', "SELECT pg_wal_replay_resume();");
for (my $i = 0; $i < 5; $i++)
{
$node_standby->wait_for_log("count ${i}", $log_offset);
- $psql_sessions[$i]->quit;
+ $psql_sessions[$i]->wait_for_completion;
+ $psql_sessions[$i]->close;
}
ok(1, 'multiple standby_replay waiters reported consistent data');
@@ -392,10 +392,8 @@ for (my $i = 0; $i < 5; $i++)
my @write_sessions;
for (my $i = 0; $i < 5; $i++)
{
- $write_sessions[$i] = $node_standby->background_psql('postgres');
- $write_sessions[$i]->query_until(
- qr/start/, qq[
- \\echo start
+ $write_sessions[$i] = PostgreSQL::Test::Session->new(node => $node_standby);
+ $write_sessions[$i]->do_async(qq[
WAIT FOR LSN '$write_lsns[$i]' WITH (MODE 'standby_write', timeout '1d');
SELECT log_wait_done('write_done', $i);
]);
@@ -414,7 +412,8 @@ resume_walreceiver($node_standby);
for (my $i = 0; $i < 5; $i++)
{
$node_standby->wait_for_log("write_done $i", $write_log_offset);
- $write_sessions[$i]->quit;
+ $write_sessions[$i]->wait_for_completion;
+ $write_sessions[$i]->close;
}
# Verify on standby that WAL was written up to the target LSN
@@ -444,10 +443,8 @@ for (my $i = 0; $i < 5; $i++)
my @flush_sessions;
for (my $i = 0; $i < 5; $i++)
{
- $flush_sessions[$i] = $node_standby->background_psql('postgres');
- $flush_sessions[$i]->query_until(
- qr/start/, qq[
- \\echo start
+ $flush_sessions[$i] = PostgreSQL::Test::Session->new(node => $node_standby);
+ $flush_sessions[$i]->do_async(qq[
WAIT FOR LSN '$flush_lsns[$i]' WITH (MODE 'standby_flush', timeout '1d');
SELECT log_wait_done('flush_done', $i);
]);
@@ -466,7 +463,8 @@ resume_walreceiver($node_standby);
for (my $i = 0; $i < 5; $i++)
{
$node_standby->wait_for_log("flush_done $i", $flush_log_offset);
- $flush_sessions[$i]->quit;
+ $flush_sessions[$i]->wait_for_completion;
+ $flush_sessions[$i]->close;
}
# Verify on standby that WAL was flushed up to the target LSN
@@ -502,10 +500,8 @@ my @mixed_sessions;
my @mixed_modes = ('standby_replay', 'standby_write', 'standby_flush');
for (my $i = 0; $i < 6; $i++)
{
- $mixed_sessions[$i] = $node_standby->background_psql('postgres');
- $mixed_sessions[$i]->query_until(
- qr/start/, qq[
- \\echo start
+ $mixed_sessions[$i] = PostgreSQL::Test::Session->new(node => $node_standby);
+ $mixed_sessions[$i]->do_async(qq[
WAIT FOR LSN '${mixed_target_lsn}' WITH (MODE '$mixed_modes[$i % 3]', timeout '1d');
SELECT log_wait_done('mixed_done', $i);
]);
@@ -529,7 +525,8 @@ resume_walreceiver($node_standby);
for (my $i = 0; $i < 6; $i++)
{
$node_standby->wait_for_log("mixed_done $i", $mixed_log_offset);
- $mixed_sessions[$i]->quit;
+ $mixed_sessions[$i]->wait_for_completion;
+ $mixed_sessions[$i]->close;
}
# Verify all modes reached the target LSN
@@ -562,10 +559,8 @@ my $primary_flush_log_offset = -s $node_primary->logfile;
my @primary_flush_sessions;
for (my $i = 0; $i < 5; $i++)
{
- $primary_flush_sessions[$i] = $node_primary->background_psql('postgres');
- $primary_flush_sessions[$i]->query_until(
- qr/start/, qq[
- \\echo start
+ $primary_flush_sessions[$i] = PostgreSQL::Test::Session->new(node => $node_primary);
+ $primary_flush_sessions[$i]->do_async(qq[
WAIT FOR LSN '$primary_flush_lsns[$i]' WITH (MODE 'primary_flush', timeout '1d');
SELECT log_wait_done('primary_flush_done', $i);
]);
@@ -576,7 +571,8 @@ for (my $i = 0; $i < 5; $i++)
{
$node_primary->wait_for_log("primary_flush_done $i",
$primary_flush_log_offset);
- $primary_flush_sessions[$i]->quit;
+ $primary_flush_sessions[$i]->wait_for_completion;
+ $primary_flush_sessions[$i]->close;
}
# Verify on primary that WAL was flushed up to the target LSN
@@ -604,10 +600,8 @@ my @wait_modes = ('standby_replay', 'standby_write', 'standby_flush');
my @wait_sessions;
for (my $i = 0; $i < 3; $i++)
{
- $wait_sessions[$i] = $node_standby->background_psql('postgres');
- $wait_sessions[$i]->query_until(
- qr/start/, qq[
- \\echo start
+ $wait_sessions[$i] = PostgreSQL::Test::Session->new(node => $node_standby);
+ $wait_sessions[$i]->do_async(qq[
WAIT FOR LSN '${lsn4}' WITH (MODE '$wait_modes[$i]');
]);
}
@@ -645,11 +639,6 @@ ok($output eq "not in recovery",
$node_standby->stop;
$node_primary->stop;
-# If we send \q with $session->quit the command can be sent to the session
-# already closed. So \q is in initial script, here we only finish IPC::Run.
-for (my $i = 0; $i < 3; $i++)
-{
- $wait_sessions[$i]->{run}->finish;
-}
+# Sessions will be cleaned up automatically when they go out of scope.
done_testing();
diff --git a/src/test/recovery/t/050_redo_segment_missing.pl b/src/test/recovery/t/050_redo_segment_missing.pl
index e07ff0c72fe..b492db44684 100644
--- a/src/test/recovery/t/050_redo_segment_missing.pl
+++ b/src/test/recovery/t/050_redo_segment_missing.pl
@@ -7,6 +7,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -44,15 +45,11 @@ $node->safe_psql('postgres',
$node->safe_psql('postgres',
q{select injection_points_attach('create-checkpoint-run', 'wait')});
-# Start a psql session to run the checkpoint in the background and make
+# Start a session to run the checkpoint in the background and make
# the test wait on the injection point so the checkpoint stops just after
# it starts.
-my $checkpoint = $node->background_psql('postgres');
-$checkpoint->query_until(
- qr/starting_checkpoint/,
- q(\echo starting_checkpoint
-checkpoint;
-));
+my $checkpoint = PostgreSQL::Test::Session->new(node => $node);
+$checkpoint->do_async(q(CHECKPOINT;));
# Wait for the initial point to finish, the checkpointer is still
# outside its critical section. Then release to reach the second
@@ -76,7 +73,8 @@ $node->safe_psql('postgres',
q{select injection_points_wakeup('create-checkpoint-run')});
$node->wait_for_log(qr/checkpoint complete/, $log_offset);
-$checkpoint->quit;
+$checkpoint->wait_for_completion;
+$checkpoint->close;
# Retrieve the WAL file names for the redo record and checkpoint record.
my $redo_lsn = $node->safe_psql('postgres',
diff --git a/src/test/recovery/t/051_effective_wal_level.pl b/src/test/recovery/t/051_effective_wal_level.pl
index 991473726a9..26f880da279 100644
--- a/src/test/recovery/t/051_effective_wal_level.pl
+++ b/src/test/recovery/t/051_effective_wal_level.pl
@@ -6,6 +6,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -367,20 +368,15 @@ if ( $ENV{enable_injection_points} eq 'yes'
test_wal_level($primary, "replica|replica",
"effective_wal_level got decreased to 'replica' on primary");
- # Start a psql session to test the case where the activation process is
+ # Start a session to test the case where the activation process is
# interrupted.
- my $psql_create_slot = $primary->background_psql('postgres');
+ my $psql_create_slot = PostgreSQL::Test::Session->new(node => $primary);
- # Start the logical decoding activation process upon creating the logical
- # slot, but it will wait due to the injection point.
- $psql_create_slot->query_until(
- qr/create_slot_canceled/,
- q(\echo create_slot_canceled
-select injection_points_set_local();
-select injection_points_attach('logical-decoding-activation', 'wait');
-select pg_create_logical_replication_slot('slot_canceled', 'pgoutput');
-\q
-));
+ # Set up the injection point in this session (using set_local so it only
+ # affects this session), then start the slot creation which will block.
+ $psql_create_slot->do(q{select injection_points_set_local()});
+ $psql_create_slot->do(q{select injection_points_attach('logical-decoding-activation', 'wait')});
+ $psql_create_slot->do_async(q{select pg_create_logical_replication_slot('slot_canceled', 'pgoutput')});
$primary->wait_for_event('client backend', 'logical-decoding-activation');
note("injection_point 'logical-decoding-activation' is reached");
@@ -397,6 +393,9 @@ select pg_cancel_backend(pid) from pg_stat_activity where query ~ 'slot_canceled
$primary->wait_for_log("aborting logical decoding activation process");
test_wal_level($primary, "replica|replica",
"the activation process aborted");
+
+ # Clean up the session (the async query was cancelled, so we just close)
+ $psql_create_slot->close;
}
$primary->stop;
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index ac96bc3f009..41765faea34 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -5,6 +5,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -30,18 +31,17 @@ sub test_streaming
# Interleave a pair of transactions, each exceeding the 64kB limit.
my $offset = 0;
- my $h = $node_publisher->background_psql('postgres', on_error_stop => 0);
+ my $h = PostgreSQL::Test::Session->new(node=>$node_publisher);
# Check the subscriber log from now on.
$offset = -s $node_subscriber->logfile;
- $h->query_safe(
- q{
- BEGIN;
- INSERT INTO test_tab SELECT i, sha256(i::text::bytea) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = sha256(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- });
+ $h->do(
+ 'BEGIN',
+ 'INSERT INTO test_tab SELECT i, sha256(i::text::bytea) FROM generate_series(3, 5000) s(i)',
+ 'UPDATE test_tab SET b = sha256(b) WHERE mod(a,2) = 0',
+ 'DELETE FROM test_tab WHERE mod(a,3) = 0',
+ );
$node_publisher->safe_psql(
'postgres', q{
@@ -51,9 +51,9 @@ sub test_streaming
COMMIT;
});
- $h->query_safe('COMMIT');
+ $h->do('COMMIT');
# errors make the next test fail, so ignore them here
- $h->quit;
+ $h->close;
$node_publisher->wait_for_catchup($appname);
@@ -211,14 +211,14 @@ $node_subscriber->reload;
$node_subscriber->safe_psql('postgres', q{SELECT 1});
# Interleave a pair of transactions, each exceeding the 64kB limit.
-my $h = $node_publisher->background_psql('postgres', on_error_stop => 0);
+my $h = PostgreSQL::Test::Session->new(node => $node_publisher);
# Confirm if a deadlock between the leader apply worker and the parallel apply
# worker can be detected.
my $offset = -s $node_subscriber->logfile;
-$h->query_safe(
+$h->do(
q{
BEGIN;
INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
@@ -232,8 +232,8 @@ $node_subscriber->wait_for_log(
$node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
-$h->query_safe('COMMIT');
-$h->quit;
+$h->do('COMMIT');
+$h->close;
$node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? deadlock detected/,
$offset);
@@ -260,7 +260,8 @@ $node_subscriber->safe_psql('postgres',
# Check the subscriber log from now on.
$offset = -s $node_subscriber->logfile;
-$h->query_safe(
+$h->reconnect;
+$h->do(
q{
BEGIN;
INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
@@ -275,8 +276,8 @@ $node_subscriber->wait_for_log(
$node_publisher->safe_psql('postgres',
"INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i)");
-$h->query_safe('COMMIT');
-$h->quit;
+$h->do('COMMIT');
+$h->close;
$node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? deadlock detected/,
$offset);
diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index 426ad74cf33..834226f8e47 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -4,6 +4,7 @@
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Session;
use PostgreSQL::Test::Utils;
use Test::More;
@@ -456,17 +457,14 @@ if ($injection_points_supported != 0)
# Start a background session on the publisher node to perform an update and
# pause at the injection point.
- my $pub_session = $node_B->background_psql('postgres');
- $pub_session->query_until(
- qr/starting_bg_psql/,
- q{
- \echo starting_bg_psql
- BEGIN;
- UPDATE tab SET b = 2 WHERE a = 1;
- PREPARE TRANSACTION 'txn_with_later_commit_ts';
- COMMIT PREPARED 'txn_with_later_commit_ts';
- }
+ my $pub_session = PostgreSQL::Test::Session->new(node => $node_B);
+ $pub_session->do(
+ q{BEGIN},
+ q{UPDATE tab SET b = 2 WHERE a = 1},
+ q{PREPARE TRANSACTION 'txn_with_later_commit_ts'}
);
+ # COMMIT PREPARED will block on the injection point
+ $pub_session->do_async(q{COMMIT PREPARED 'txn_with_later_commit_ts'});
# Wait until the backend enters the injection point
$node_B->wait_for_event('client backend', 'commit-after-delay-checkpoint');
@@ -519,8 +517,9 @@ if ($injection_points_supported != 0)
SELECT injection_points_detach('commit-after-delay-checkpoint');"
);
- # Close the background session on the publisher node
- ok($pub_session->quit, "close publisher session");
+ # Wait for the async query to complete and close the background session
+ $pub_session->wait_for_completion;
+ $pub_session->close;
# Confirm that the transaction committed
$result =
view thread (2+ messages)
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: Re: BackgroundPsql swallowing errors on windows
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox