0002-aio-Add-READ_STREAM_RESET_CONTINUE-flag.txt

From 5c3cf2018845093041dcf23bd6e8ae90b5c5906a Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
Date: Mon, 21 Jul 2025 14:38:32 +1200
Subject: [PATCH 2/3] aio: Add READ_STREAM_RESET_CONTINUE flag.

For users that sometimes run out of block numbers temporarily but then
later want to continue, provide a way to reset the stream and continue
without forgetting the internal look-ahead distance.

This is done as a bitmask, just in case we think of more reasons for
fine-grained control of what is reset or not in the future.

XXX Name debatable; is it better to use a 'high level' concept like
'continuing an interrupted stream', or a lower level concept explicitly
referring to what is reset or not?

XXX Where existing users pass 0 or the new flag not thought about very
hard

Discussion: https://postgr.es/m/CA%2BhUKG%2BWWr4-8TYemyU%3DucQsNe6bUBN_Sq3mCnBoBtxaJ9w3ug%40mail.gmail.com
---
 src/backend/access/gist/gistvacuum.c  |  2 +-
 src/backend/access/heap/heapam.c      |  4 ++--
 src/backend/access/nbtree/nbtree.c    |  2 +-
 src/backend/access/spgist/spgvacuum.c |  2 +-
 src/backend/storage/aio/read_stream.c | 13 +++++++++----
 src/include/storage/read_stream.h     | 14 +++++++++++++-
 6 files changed, 27 insertions(+), 10 deletions(-)

diff --git a/src/backend/access/gist/gistvacuum.c b/src/backend/access/gist/gistvacuum.c
index dca236b6e57..45044e3cc84 100644
--- a/src/backend/access/gist/gistvacuum.c
+++ b/src/backend/access/gist/gistvacuum.c
@@ -260,7 +260,7 @@ gistvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 		 * InvalidBuffer, the read stream API won't invoke our callback again
 		 * until the stream has been reset.
 		 */
-		read_stream_reset(stream);
+		read_stream_reset(stream, READ_STREAM_RESET_CONTINUE);
 	}
 
 	read_stream_end(stream);
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 0dcd6ee817e..1ffc4a872db 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -667,7 +667,7 @@ heap_fetch_next_buffer(HeapScanDesc scan, ScanDirection dir)
 	if (unlikely(scan->rs_dir != dir))
 	{
 		scan->rs_prefetch_block = scan->rs_cblock;
-		read_stream_reset(scan->rs_read_stream);
+		read_stream_reset(scan->rs_read_stream, 0);
 	}
 
 	scan->rs_dir = dir;
@@ -1284,7 +1284,7 @@ heap_rescan(TableScanDesc sscan, ScanKey key, bool set_params,
 	 * in initscan().
 	 */
 	if (scan->rs_read_stream)
-		read_stream_reset(scan->rs_read_stream);
+		read_stream_reset(scan->rs_read_stream, 0);
 
 	/*
 	 * reinitialize scan descriptor
diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index fdff960c130..596df227c20 100644
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -1318,7 +1318,7 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 		 * InvalidBuffer, the read stream API won't invoke our callback again
 		 * until the stream has been reset.
 		 */
-		read_stream_reset(stream);
+		read_stream_reset(stream, READ_STREAM_RESET_CONTINUE);
 	}
 
 	read_stream_end(stream);
diff --git a/src/backend/access/spgist/spgvacuum.c b/src/backend/access/spgist/spgvacuum.c
index 2678f7ab782..c4e4a89680e 100644
--- a/src/backend/access/spgist/spgvacuum.c
+++ b/src/backend/access/spgist/spgvacuum.c
@@ -885,7 +885,7 @@ spgvacuumscan(spgBulkDeleteState *bds)
 		 * InvalidBuffer, the read stream API won't invoke our callback again
 		 * until the stream has been reset.
 		 */
-		read_stream_reset(stream);
+		read_stream_reset(stream, READ_STREAM_RESET_CONTINUE);
 	}
 
 	read_stream_end(stream);
diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index 5de6f83c253..f242b373b22 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -1011,9 +1011,13 @@ read_stream_next_block(ReadStream *stream, BufferAccessStrategy *strategy)
  * to be used again for different blocks.  This can be used to clear an
  * end-of-stream condition and start again, or to throw away blocks that were
  * speculatively read and read some different blocks instead.
+ *
+ * READ_STREAM_RESET_CONTINUE can be pass to flags to indicate that a stream
+ * was temporarily interrupted but internal look-ahead distance heuristics
+ * should not be reset, because a similar access pattern is expected.
  */
 void
-read_stream_reset(ReadStream *stream)
+read_stream_reset(ReadStream *stream, int flags)
 {
 	int16		index;
 	Buffer		buffer;
@@ -1050,8 +1054,9 @@ read_stream_reset(ReadStream *stream)
 	Assert(stream->pinned_buffers == 0);
 	Assert(stream->ios_in_progress == 0);
 
-	/* Start off assuming data is cached. */
-	stream->distance = 1;
+	/* Start off like a newly initialized stream, unless asked not to. */
+	if ((flags & READ_STREAM_RESET_CONTINUE) == 0)
+		stream->distance = 1;
 	stream->end_of_stream = false;
 }
 
@@ -1061,6 +1066,6 @@ read_stream_reset(ReadStream *stream)
 void
 read_stream_end(ReadStream *stream)
 {
-	read_stream_reset(stream);
+	read_stream_reset(stream, 0);
 	pfree(stream);
 }
diff --git a/src/include/storage/read_stream.h b/src/include/storage/read_stream.h
index 9b0d65161d0..90ec278117f 100644
--- a/src/include/storage/read_stream.h
+++ b/src/include/storage/read_stream.h
@@ -17,6 +17,8 @@
 #include "storage/bufmgr.h"
 #include "storage/smgr.h"
 
+/* Flags controlling stream initialization. */
+
 /* Default tuning, reasonable for many users. */
 #define READ_STREAM_DEFAULT 0x00
 
@@ -42,6 +44,16 @@
  */
 #define READ_STREAM_FULL 0x04
 
+/* Flags controlling read_stream_reset(). */
+
+/*
+ * If the callback reports end-of-stream or higher levels decide to abandon
+ * blocks that it generated speculatively, the stream can be reset and allowed
+ * to try to fetch blocks again without forgetting internal heuristics by
+ * passing this flag.
+ */
+#define READ_STREAM_RESET_CONTINUE 0x01
+
 /* ---
  * Opt-in to using AIO batchmode.
  *
@@ -99,7 +111,7 @@ extern ReadStream *read_stream_begin_smgr_relation(int flags,
 												   ReadStreamBlockNumberCB callback,
 												   void *callback_private_data,
 												   size_t per_buffer_data_size);
-extern void read_stream_reset(ReadStream *stream);
+extern void read_stream_reset(ReadStream *stream, int flags);
 extern void read_stream_end(ReadStream *stream);
 
 #endif							/* READ_STREAM_H */
-- 
2.39.5 (Apple Git-154)