From de8a800b48829ef619200a71dba9028be3ea09e9 Mon Sep 17 00:00:00 2001 From: Thomas Munro Date: Wed, 12 Nov 2025 16:49:57 +1300 Subject: [PATCH 2/2] Add tests for read_stream_{pause,resume,yield}(). --- src/test/modules/test_aio/Makefile | 3 +- src/test/modules/test_aio/meson.build | 1 + src/test/modules/test_aio/t/001_aio.pl | 30 +++ src/test/modules/test_aio/test_aio--1.0.sql | 13 ++ src/test/modules/test_aio/test_read_stream.c | 181 +++++++++++++++++++ src/tools/pgindent/typedefs.list | 1 + 6 files changed, 228 insertions(+), 1 deletion(-) create mode 100644 src/test/modules/test_aio/test_read_stream.c diff --git a/src/test/modules/test_aio/Makefile b/src/test/modules/test_aio/Makefile index f53cc64671a..465eb09ee4f 100644 --- a/src/test/modules/test_aio/Makefile +++ b/src/test/modules/test_aio/Makefile @@ -5,7 +5,8 @@ PGFILEDESC = "test_aio - test code for AIO" MODULE_big = test_aio OBJS = \ $(WIN32RES) \ - test_aio.o + test_aio.o \ + test_read_stream.o EXTENSION = test_aio DATA = test_aio--1.0.sql diff --git a/src/test/modules/test_aio/meson.build b/src/test/modules/test_aio/meson.build index 73d2fd68eaa..6e6fcbfdad9 100644 --- a/src/test/modules/test_aio/meson.build +++ b/src/test/modules/test_aio/meson.build @@ -2,6 +2,7 @@ test_aio_sources = files( 'test_aio.c', + 'test_read_stream.c', ) if host_system == 'windows' diff --git a/src/test/modules/test_aio/t/001_aio.pl b/src/test/modules/test_aio/t/001_aio.pl index 3f0453619e8..2a2c6523a6b 100644 --- a/src/test/modules/test_aio/t/001_aio.pl +++ b/src/test/modules/test_aio/t/001_aio.pl @@ -1489,6 +1489,35 @@ SELECT read_rel_block_ll('tbl_cs_fail', 3, nblocks=>1, zero_on_error=>true);), $psql->quit(); } +# Read stream tests +sub test_read_stream +{ + my $io_method = shift; + my $node = shift; + my ($ret, $output); + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + + $psql->query_safe( + qq( +CREATE TEMPORARY TABLE tmp_read_stream(data int not null); +INSERT INTO tmp_read_stream SELECT generate_series(1, 10000); +SELECT test_read_stream_resume('tmp_read_stream', 0); +DROP TABLE tmp_read_stream; +)); + + $psql->query_safe( + qq( +CREATE TEMPORARY TABLE tmp_read_stream(data int not null); +INSERT INTO tmp_read_stream SELECT generate_series(1, 10000); +SELECT test_read_stream_yield('tmp_read_stream', 0); +DROP TABLE tmp_read_stream; +)); + + $psql->quit(); +} + + # Run all tests that are supported for all io_methods sub test_generic @@ -1525,6 +1554,7 @@ CHECKPOINT; test_checksum($io_method, $node); test_ignore_checksum($io_method, $node); test_checksum_createdb($io_method, $node); + test_read_stream($io_method, $node); SKIP: { diff --git a/src/test/modules/test_aio/test_aio--1.0.sql b/src/test/modules/test_aio/test_aio--1.0.sql index e495481c41e..e37810b7273 100644 --- a/src/test/modules/test_aio/test_aio--1.0.sql +++ b/src/test/modules/test_aio/test_aio--1.0.sql @@ -106,3 +106,16 @@ AS 'MODULE_PATHNAME' LANGUAGE C; CREATE FUNCTION inj_io_reopen_detach() RETURNS pg_catalog.void STRICT AS 'MODULE_PATHNAME' LANGUAGE C; + + + +/* + * Read stream related functions + */ +CREATE FUNCTION test_read_stream_resume(rel regclass, blockno int4) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION test_read_stream_yield(rel regclass, blockno int4) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; diff --git a/src/test/modules/test_aio/test_read_stream.c b/src/test/modules/test_aio/test_read_stream.c new file mode 100644 index 00000000000..d1d436a90b7 --- /dev/null +++ b/src/test/modules/test_aio/test_read_stream.c @@ -0,0 +1,181 @@ +/*------------------------------------------------------------------------- + * + * test_read_stream.c + * Helpers to write tests for read_stream.c + * + * Copyright (c) 2020-2025, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/test/modules/test_aio/test_read_stream.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "access/relation.h" +#include "fmgr.h" +#include "storage/bufmgr.h" +#include "storage/read_stream.h" + +typedef struct +{ + BlockNumber blkno; + int count; +} test_read_stream_resume_state; + +static BlockNumber +test_read_stream_resume_cb(ReadStream *stream, + void *callback_private_data, + void *per_buffer_data) +{ + test_read_stream_resume_state *state = callback_private_data; + + /* Periodic end-of-stream. */ + if (++state->count % 3 == 0) + return read_stream_pause(stream); + + return state->blkno; +} + +/* + * Test read_stream_resume(), allowing a stream to end temporarily and then + * continue where it left off. + */ +PG_FUNCTION_INFO_V1(test_read_stream_resume); +Datum +test_read_stream_resume(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + BlockNumber blkno = PG_GETARG_UINT32(1); + Relation rel; + Buffer buf; + ReadStream *stream; + test_read_stream_resume_state state = {.blkno = blkno}; + + rel = relation_open(relid, AccessShareLock); + stream = read_stream_begin_relation(READ_STREAM_DEFAULT, + NULL, + rel, + MAIN_FORKNUM, + test_read_stream_resume_cb, + &state, + 0); + + for (int i = 0; i < 3; ++i) + { + /* Same block twice. */ + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + + /* End-of-stream. */ + buf = read_stream_next_buffer(stream, NULL); + Assert(buf == InvalidBuffer); + buf = read_stream_next_buffer(stream, NULL); + Assert(buf == InvalidBuffer); + + /* Resume. */ + read_stream_resume(stream); + } + + read_stream_end(stream); + relation_close(rel, NoLock); + + PG_RETURN_VOID(); +} + +typedef struct +{ + BlockNumber blkno; + int count; + int yields; + int blocks; +} test_read_stream_yield_state; + +static BlockNumber +test_read_stream_yield_cb(ReadStream *stream, + void *callback_private_data, + void *per_buffer_data) +{ + test_read_stream_yield_state *state = callback_private_data; + + /* Yield every third call. */ + if (++state->count % 3 == 2) + { + state->yields++; + return read_stream_yield(stream); + } + + state->blocks++; + return state->blkno; +} + +/* + * Test read_stream_yield(), allowing control to be yielded temporarily from + * the lookahead loop and returned to the caller of read_stream_next_buffer(). + */ +PG_FUNCTION_INFO_V1(test_read_stream_yield); +Datum +test_read_stream_yield(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + BlockNumber blkno = PG_GETARG_UINT32(1); + Relation rel; + Buffer buf; + ReadStream *stream; + test_read_stream_yield_state state = {.blkno = blkno}; + + rel = relation_open(relid, AccessShareLock); + stream = read_stream_begin_relation(READ_STREAM_DEFAULT, + NULL, + rel, + MAIN_FORKNUM, + test_read_stream_yield_cb, + &state, + 0); + + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + Assert(state.blocks == 1); + Assert(state.yields == 1); + + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + Assert(state.blocks == 3); + Assert(state.yields == 1); + + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + Assert(state.blocks == 3); + Assert(state.yields == 2); + + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + Assert(state.blocks == 5); + Assert(state.yields == 2); + + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + Assert(state.blocks == 5); + Assert(state.yields == 3); + + buf = read_stream_next_buffer(stream, NULL); + Assert(BufferGetBlockNumber(buf) == blkno); + ReleaseBuffer(buf); + Assert(state.blocks == 7); + Assert(state.yields == 3); + + read_stream_end(stream); + relation_close(rel, NoLock); + + PG_RETURN_VOID(); +} diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index cf3f6a7dafd..7396e9ce14b 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -4158,6 +4158,7 @@ td_entry teSection temp_tablespaces_extra test_re_flags +test_read_stream_resume_state test_regex_ctx test_shm_mq_header test_spec -- 2.51.2