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)