On Mar 15, 2026, Tomas Vondra <[email protected]> wrote:
> The first line "Prefetch" tracks the look-ahead distance, i.e. how many
> blocks ahead the ReadStream is requesting.
> The second line "I/O" is about the I/O requests actually issued - how
> many times we had to wait for the block (when we get to process it),
> average size of a request (in BLCKSZ blocks), and average number of
> in-progress requests.
Our new AI harness for testing found that a rescan can include buffers that
never reach the consumer in the Prefetch average, and prepared the attached
patch. On a clean cluster with io_method=worker, this is a complete
reproducer:
\set ON_ERROR_STOP on
create extension pg_buffercache;
set jit = off;
set max_parallel_workers_per_gather = 0;
set enable_seqscan = off;
set effective_io_concurrency = 16;
create unlogged table reset_t as
select g as id, repeat(md5(g::text), 16) as payload
from generate_series(1, 80000) as g;
vacuum (analyze, freeze) reset_t;
checkpoint;
select pg_buffercache_evict_relation('reset_t');
explain (analyze, buffers, io, timing off, summary off, costs off)
select r.startblock, s.ctid
from (values (0), (300), (600), (900), (1200),
(1500), (1800), (2100), (2400), (2700)) r(startblock)
cross join lateral (
select t.ctid
from reset_t t
where t.ctid >= format('(%s,1)', r.startblock)::tid
and t.ctid < format('(%s,1)', r.startblock + 200)::tid
offset 0 limit 1
) s;
The inner TID Range Scan reports:
Prefetch: avg=1.32 max=2
I/O: count=20 waits=19 size=1.50 in-progress=1.00
Buffers: shared read=30
Each of the ten loops returns one buffer to the consumer at distance one.
A single-loop control reports avg=1.00 max=1.
read_stream_reset() drains unread buffers by calling
read_stream_next_buffer(), which also calls read_stream_count_prefetch().
The nine rescans above add 18 cleanup samples with distance sum 27, so the
reported average is (10 + 27) / (10 + 18) = 1.321428... and max becomes 2.
The attached patch preserves prefetch_count, distance_sum, and distance_max
around that internal drain, while leaving the real I/O statistics
cumulative. It adds a test that checks all three fields with worker and
sync I/O.
On master d39fda1c the test fails without the read_stream.c change and
passes with it; the full test_aio and core regression suites pass. The
same patch applies to REL_19_STABLE b73d13c3, where the reproducer shows
the same avg=1.32 max=2 result and test_aio passes with the fix.
Nik
diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index fddf6ef441d814abf3dbbbebfa344ae9d827d678..852cc958560832270b726c3bf620169fbc13eaa8 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -1439,6 +1439,9 @@ read_stream_reset(ReadStream *stream)
{
int16 index;
Buffer buffer;
+ uint64 prefetch_count = 0;
+ uint64 distance_sum = 0;
+ int16 distance_max = 0;
/* Stop looking ahead. */
stream->readahead_distance = 0;
@@ -1451,10 +1454,31 @@ read_stream_reset(ReadStream *stream)
stream->buffered_blocknum = InvalidBlockNumber;
stream->fast_path = false;
+ /*
+ * Draining the stream below is an implementation detail, not consumption
+ * by the stream's caller. Preserve the prefetch distance statistics so
+ * that the drained buffers are not counted as having been returned to the
+ * caller. I/O statistics still need to reflect any work done while
+ * draining.
+ */
+ if (stream->stats)
+ {
+ prefetch_count = stream->stats->prefetch_count;
+ distance_sum = stream->stats->distance_sum;
+ distance_max = stream->stats->distance_max;
+ }
+
/* Unpin anything that wasn't consumed. */
while ((buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
ReleaseBuffer(buffer);
+ if (stream->stats)
+ {
+ stream->stats->prefetch_count = prefetch_count;
+ stream->stats->distance_sum = distance_sum;
+ stream->stats->distance_max = distance_max;
+ }
+
/* Unpin any unused forwarded buffers. */
index = stream->next_buffer_index;
while (index < stream->initialized_buffers &&
diff --git a/src/test/modules/test_aio/t/004_read_stream.pl b/src/test/modules/test_aio/t/004_read_stream.pl
index 32311c07ac02bb3ebe21be687dc54bcd5ced0be3..758a36aadc89b99355e4eb41d07652183d68ba11 100644
--- a/src/test/modules/test_aio/t/004_read_stream.pl
+++ b/src/test/modules/test_aio/t/004_read_stream.pl
@@ -115,6 +115,22 @@ sub test_repeated_blocks
}
+sub test_reset_stats
+{
+ my $io_method = shift;
+ my $node = shift;
+
+ $node->safe_psql('postgres', qq/SELECT evict_rel('largeish');/);
+ is(
+ $node->safe_psql(
+ 'postgres',
+ qq/SELECT read_stream_reset_stats('largeish', ARRAY[0, 1, 2, 3]);/),
+ 't',
+ "$io_method: resetting a stream preserves consumer prefetch statistics"
+ );
+}
+
+
sub test_inject_foreign
{
my $io_method = shift;
@@ -267,6 +283,7 @@ sub test_io_method
is($node->safe_psql('postgres', 'SHOW io_method'),
$io_method, "$io_method: io_method set correctly");
+ test_reset_stats($io_method, $node);
test_repeated_blocks($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 762ac29512f7fc1dabfdc47dfe6840bf840832f3..b5948eb8c50b0ca616be15dcc6b1dcbde0b35c41 100644
--- a/src/test/modules/test_aio/test_aio--1.0.sql
+++ b/src/test/modules/test_aio/test_aio--1.0.sql
@@ -64,6 +64,10 @@ CREATE FUNCTION read_stream_for_blocks(rel regclass, blocks int4[], OUT blockoff
RETURNS SETOF record STRICT
AS 'MODULE_PATHNAME' LANGUAGE C;
+CREATE FUNCTION read_stream_reset_stats(rel regclass, blocks int4[])
+RETURNS pg_catalog.bool STRICT
+AS 'MODULE_PATHNAME' LANGUAGE C;
+
/*
* Handle related functions
diff --git a/src/test/modules/test_aio/test_aio.c b/src/test/modules/test_aio/test_aio.c
index 39d857557cfaf7cabe6c9c4b8b0090dec5c5bc78..7594d2efcce4021154a8fa8d4542bed9d2028c2b 100644
--- a/src/test/modules/test_aio/test_aio.c
+++ b/src/test/modules/test_aio/test_aio.c
@@ -20,6 +20,7 @@
#include "access/relation.h"
#include "catalog/pg_type.h"
+#include "executor/instrument_node.h"
#include "fmgr.h"
#include "funcapi.h"
#include "storage/aio.h"
@@ -916,6 +917,62 @@ read_stream_for_blocks(PG_FUNCTION_ARGS)
}
+PG_FUNCTION_INFO_V1(read_stream_reset_stats);
+Datum
+read_stream_reset_stats(PG_FUNCTION_ARGS)
+{
+ Oid relid = PG_GETARG_OID(0);
+ ArrayType *blocksarray = PG_GETARG_ARRAYTYPE_P(1);
+ Relation rel;
+ BlocksReadStreamData stream_data;
+ ReadStream *stream;
+ IOStats stats = {0};
+ Buffer buf;
+ uint64 prefetch_count;
+ uint64 distance_sum;
+ int16 distance_max;
+
+ if (ARR_NDIM(blocksarray) != 1 ||
+ ARR_HASNULL(blocksarray) ||
+ ARR_ELEMTYPE(blocksarray) != INT4OID ||
+ ARR_DIMS(blocksarray)[0] == 0)
+ elog(ERROR, "expected non-empty 1 dimensional int4 array");
+
+ stream_data.curblock = 0;
+ stream_data.nblocks = ARR_DIMS(blocksarray)[0];
+ stream_data.blocks = (uint32 *) ARR_DATA_PTR(blocksarray);
+
+ rel = relation_open(relid, AccessShareLock);
+ stream = read_stream_begin_relation(READ_STREAM_FULL,
+ NULL,
+ rel,
+ MAIN_FORKNUM,
+ read_stream_for_blocks_cb,
+ &stream_data,
+ 0);
+ read_stream_enable_stats(stream, &stats);
+
+ buf = read_stream_next_buffer(stream, NULL);
+ if (!BufferIsValid(buf))
+ elog(ERROR, "first read_stream_next_buffer() call is unexpectedly invalid");
+ ReleaseBuffer(buf);
+
+ if (stats.prefetch_count != 1)
+ elog(ERROR, "expected exactly one prefetch sample");
+ prefetch_count = stats.prefetch_count;
+ distance_sum = stats.distance_sum;
+ distance_max = stats.distance_max;
+
+ read_stream_reset(stream);
+ read_stream_end(stream);
+ relation_close(rel, NoLock);
+
+ PG_RETURN_BOOL(stats.prefetch_count == prefetch_count &&
+ stats.distance_sum == distance_sum &&
+ stats.distance_max == distance_max);
+}
+
+
PG_FUNCTION_INFO_V1(handle_get);
Datum
handle_get(PG_FUNCTION_ARGS)
