0001-Make-ReadStream-struct-non-opaque.v1.patch

From ea4bb194e0dcccac8465b3aa13950f721bde3860 Mon Sep 17 00:00:00 2001
From: Mats Kindahl <mats@timescale.com>
Date: Thu, 29 Aug 2024 10:39:34 +0200
Subject: Make ReadStream struct non-opaque

Move the ReadStream struct and two utility functions from read_stream.c
to read_stream.h to allow extensions to modify the structure if they
need to.
---
 src/backend/storage/aio/read_stream.c |  59 ---------------
 src/include/storage/read_stream.h     | 105 ++++++++++++++++++++++++++
 2 files changed, 105 insertions(+), 59 deletions(-)

diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index a83c18c2a4..bf2a679037 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -97,65 +97,6 @@
 #include "utils/rel.h"
 #include "utils/spccache.h"
 
-typedef struct InProgressIO
-{
-	int16		buffer_index;
-	ReadBuffersOperation op;
-} InProgressIO;
-
-/*
- * State for managing a stream of reads.
- */
-struct ReadStream
-{
-	int16		max_ios;
-	int16		ios_in_progress;
-	int16		queue_size;
-	int16		max_pinned_buffers;
-	int16		pinned_buffers;
-	int16		distance;
-	bool		advice_enabled;
-
-	/*
-	 * Small buffer of block numbers, useful for 'ungetting' to resolve flow
-	 * control problems when I/Os are split.  Also useful for batch-loading
-	 * block numbers in the fast path.
-	 */
-	BlockNumber blocknums[16];
-	int16		blocknums_count;
-	int16		blocknums_next;
-
-	/*
-	 * The callback that will tell us which block numbers to read, and an
-	 * opaque pointer that will be pass to it for its own purposes.
-	 */
-	ReadStreamBlockNumberCB callback;
-	void	   *callback_private_data;
-
-	/* Next expected block, for detecting sequential access. */
-	BlockNumber seq_blocknum;
-
-	/* The read operation we are currently preparing. */
-	BlockNumber pending_read_blocknum;
-	int16		pending_read_nblocks;
-
-	/* Space for buffers and optional per-buffer private data. */
-	size_t		per_buffer_data_size;
-	void	   *per_buffer_data;
-
-	/* Read operations that have been started but not waited for yet. */
-	InProgressIO *ios;
-	int16		oldest_io_index;
-	int16		next_io_index;
-
-	bool		fast_path;
-
-	/* Circular queue of buffers. */
-	int16		oldest_buffer_index;	/* Next pinned buffer to return */
-	int16		next_buffer_index;	/* Index of next buffer to pin */
-	Buffer		buffers[FLEXIBLE_ARRAY_MEMBER];
-};
-
 /*
  * Return a pointer to the per-buffer data by index.
  */
diff --git a/src/include/storage/read_stream.h b/src/include/storage/read_stream.h
index 4e599904f2..006ec3feb1 100644
--- a/src/include/storage/read_stream.h
+++ b/src/include/storage/read_stream.h
@@ -50,6 +50,111 @@ typedef BlockNumber (*ReadStreamBlockNumberCB) (ReadStream *stream,
 												void *callback_private_data,
 												void *per_buffer_data);
 
+/*
+ * State of an in-progress read buffer operation.
+ */
+typedef struct InProgressIO
+{
+	int16		buffer_index;
+	ReadBuffersOperation op;
+} InProgressIO;
+
+/*
+ * State for managing a stream of reads.
+ */
+struct ReadStream
+{
+	int16		max_ios;
+	int16		ios_in_progress;
+	int16		queue_size;
+	int16		max_pinned_buffers;
+	int16		pinned_buffers;
+	int16		distance;
+	bool		advice_enabled;
+
+	/*
+	 * Small buffer of block numbers, useful for 'ungetting' to resolve flow
+	 * control problems when I/Os are split.  Also useful for batch-loading
+	 * block numbers in the fast path.
+	 */
+	BlockNumber blocknums[16];
+	int16		blocknums_count;
+	int16		blocknums_next;
+
+	/*
+	 * The callback that will tell us which block numbers to read, and an
+	 * opaque pointer that will be pass to it for its own purposes.
+	 */
+	ReadStreamBlockNumberCB callback;
+	void	   *callback_private_data;
+
+	/* Next expected block, for detecting sequential access. */
+	BlockNumber seq_blocknum;
+
+	/* The read operation we are currently preparing. */
+	BlockNumber pending_read_blocknum;
+	int16		pending_read_nblocks;
+
+	/* Space for buffers and optional per-buffer private data. */
+	size_t		per_buffer_data_size;
+	void	   *per_buffer_data;
+
+	/* Read operations that have been started but not waited for yet. */
+	InProgressIO *ios;
+	int16		oldest_io_index;
+	int16		next_io_index;
+
+	bool		fast_path;
+
+	/* Circular queue of buffers. */
+	int16		oldest_buffer_index;	/* Next pinned buffer to return */
+	int16		next_buffer_index;	/* Index of next buffer to pin */
+	Buffer		buffers[FLEXIBLE_ARRAY_MEMBER];
+};
+
+/*
+ * Ask the callback which block it would like us to read next, with a small
+ * buffer in front to allow read_stream_unget_block() to work and to allow the
+ * fast path to skip this function and work directly from the array.
+ */
+static inline BlockNumber
+read_stream_get_block(ReadStream *stream, void *per_buffer_data)
+{
+	if (stream->blocknums_next < stream->blocknums_count)
+		return stream->blocknums[stream->blocknums_next++];
+
+	/*
+	 * We only bother to fetch one at a time here (but see the fast path which
+	 * uses more).
+	 */
+	return stream->callback(stream,
+							stream->callback_private_data,
+							per_buffer_data);
+}
+
+/*
+ * In order to deal with short reads in StartReadBuffers(), we sometimes need
+ * to defer handling of a block until later.
+ */
+static inline void
+read_stream_unget_block(ReadStream *stream, BlockNumber blocknum)
+{
+	if (stream->blocknums_next == stream->blocknums_count)
+	{
+		/* Never initialized or entirely consumed.  Re-initialize. */
+		stream->blocknums[0] = blocknum;
+		stream->blocknums_count = 1;
+		stream->blocknums_next = 0;
+	}
+	else
+	{
+		/* Must be the last value return from blocknums array. */
+		Assert(stream->blocknums_next > 0);
+		stream->blocknums_next--;
+		Assert(stream->blocknums[stream->blocknums_next] == blocknum);
+	}
+}
+
 extern ReadStream *read_stream_begin_relation(int flags,
 											  BufferAccessStrategy strategy,
 											  Relation rel,
-- 
2.34.1