Using read stream in autoprewarm

From 2905f74049f77ab9c4f406dfe28b29ef4eafb7b4 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 5 Nov 2024 13:00:11 +0300
Subject: [PATCH v2 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 src/include/storage/buf_internals.h   |  1 +
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 contrib/pg_prewarm/autoprewarm.c      |  4 ++--
 3 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index eda6c699212..d95050adb23 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -432,6 +432,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index dffdd57e9b5..6cad8540bd9 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuosly changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 8291bf3c427..16174bf3d0f 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -518,13 +518,13 @@ autoprewarm_database_main(Datum main_arg)
 	pos = apw_state->prewarm_start_idx;
 
 	p.block_info = block_info;
-	p.max_pos = apw_state->prewarm_stop_idx;
+	p.max_pos = Min(apw_state->prewarm_stop_idx, pos + get_number_of_free_buffers());
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	for (; pos < apw_state->prewarm_stop_idx; pos++)
+	for (; pos < p.max_pos; pos++)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
-- 
2.45.2

From bcdb55f237945072cb5740392515af701cfdc1d2 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 5 Nov 2024 11:40:14 +0300
Subject: [PATCH v2 1/2] Use read stream in autoprewarm

Instead of reading blocks with ReadBufferExtended(), create read stream
object for each possible case and use it.

This change provides about 10% performance improvement.
---
 contrib/pg_prewarm/autoprewarm.c | 112 +++++++++++++++++++++++++++++--
 1 file changed, 107 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index fac4051e1aa..8291bf3c427 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -422,6 +423,68 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+struct apw_read_stream_private
+{
+	bool		first_block;
+	int			max_pos;
+	int			pos;
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks_in_fork;
+
+};
+
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	bool	   *rs_have_free_buffer = per_buffer_data;
+	BlockInfoRecord *old_blk;
+	BlockInfoRecord *cur_blk;
+
+	*rs_have_free_buffer = true;
+
+	/*
+	 * There may still be queued blocks in the stream even when no free
+	 * buffers are available in the buffer pool. This can lead to unnecessary
+	 * I/O operations and buffer evictions. One possible solution is to
+	 * compare the number of free buffers in the buffer pool with the number
+	 * of queued blocks in the stream. However, this approach is considered a
+	 * workaround and would add complexity with minimal benefit, as only a few
+	 * unnecessary I/O operations and buffer evictions are expected.
+	 * Therefore, this solution has not been implemented.
+	 */
+	if (!have_free_buffer())
+	{
+		*rs_have_free_buffer = false;
+		return InvalidBlockNumber;
+	}
+
+	if (p->pos == p->max_pos)
+		return InvalidBlockNumber;
+
+	if (p->first_block)
+	{
+		p->first_block = false;
+		return p->block_info[p->pos++].blocknum;
+	}
+
+	old_blk = &(p->block_info[p->pos - 1]);
+	cur_blk = &(p->block_info[p->pos]);
+
+	if (old_blk->database == cur_blk->database &&
+		old_blk->forknum == cur_blk->forknum &&
+		old_blk->filenumber == cur_blk->filenumber &&
+		cur_blk->blocknum < p->nblocks_in_fork)
+	{
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -435,6 +498,9 @@ autoprewarm_database_main(Datum main_arg)
 	BlockNumber nblocks = 0;
 	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	ReadStream *stream = NULL;
+	struct apw_read_stream_private p;
+	bool	   *rs_have_free_buffer;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,13 +517,16 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
+	p.block_info = block_info;
+	p.max_pos = apw_state->prewarm_stop_idx;
+
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	for (; pos < apw_state->prewarm_stop_idx; pos++)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
+		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
 
 		CHECK_FOR_INTERRUPTS();
@@ -470,6 +539,18 @@ autoprewarm_database_main(Datum main_arg)
 			old_blk->database != 0)
 			break;
 
+		/*
+		 * If stream needs to be created again, end it before closing the old
+		 * relation.
+		 */
+		if (stream && (old_blk == NULL ||
+					   old_blk->filenumber != blk->filenumber ||
+					   old_blk->forknum != blk->forknum))
+		{
+			Assert(read_stream_next_buffer(stream, (void **) &rs_have_free_buffer) == InvalidBuffer);
+			read_stream_end(stream);
+		}
+
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
@@ -506,7 +587,10 @@ autoprewarm_database_main(Datum main_arg)
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
+		/*
+		 * Once per fork, check for fork existence and size. Then create read
+		 * stream if it is suitable.
+		 */
 		if (old_blk == NULL ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
@@ -518,7 +602,21 @@ autoprewarm_database_main(Datum main_arg)
 			if (blk->forknum > InvalidForkNumber &&
 				blk->forknum <= MAX_FORKNUM &&
 				smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
 				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+
+				/* Create read stream. */
+				p.nblocks_in_fork = nblocks;
+				p.pos = pos;
+				p.first_block = true;
+				stream = read_stream_begin_relation(READ_STREAM_FULL,
+													NULL,
+													rel,
+													blk->forknum,
+													apw_read_stream_next_block,
+													&p,
+													sizeof(bool));
+			}
 			else
 				nblocks = 0;
 		}
@@ -532,16 +630,20 @@ autoprewarm_database_main(Datum main_arg)
 		}
 
 		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
+		buf = read_stream_next_buffer(stream, (void **) &rs_have_free_buffer);
 		if (BufferIsValid(buf))
 		{
 			apw_state->prewarmed_blocks++;
 			ReleaseBuffer(buf);
 		}
+		/* There are no free buffers left in shared buffers, break the loop. */
+		else if (!(*rs_have_free_buffer))
+			break;
 
 		old_blk = blk;
 	}
+	Assert(read_stream_next_buffer(stream, (void **) &rs_have_free_buffer) == InvalidBuffer);
+	read_stream_end(stream);
 
 	dsm_detach(seg);
 
-- 
2.45.2

From f5a8faa6ac3f836784c15b776633e7729d95378d Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 5 Nov 2024 11:40:14 +0300
Subject: [PATCH v3 1/2] Use read stream in autoprewarm

Instead of reading blocks with ReadBufferExtended(), create read stream
object for each possible case and use it.

This change provides about 10% performance improvement.
---
 contrib/pg_prewarm/autoprewarm.c | 112 +++++++++++++++++++++++++++++--
 1 file changed, 107 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index fac4051e1aa..8291bf3c427 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -422,6 +423,68 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+struct apw_read_stream_private
+{
+	bool		first_block;
+	int			max_pos;
+	int			pos;
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks_in_fork;
+
+};
+
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	bool	   *rs_have_free_buffer = per_buffer_data;
+	BlockInfoRecord *old_blk;
+	BlockInfoRecord *cur_blk;
+
+	*rs_have_free_buffer = true;
+
+	/*
+	 * There may still be queued blocks in the stream even when no free
+	 * buffers are available in the buffer pool. This can lead to unnecessary
+	 * I/O operations and buffer evictions. One possible solution is to
+	 * compare the number of free buffers in the buffer pool with the number
+	 * of queued blocks in the stream. However, this approach is considered a
+	 * workaround and would add complexity with minimal benefit, as only a few
+	 * unnecessary I/O operations and buffer evictions are expected.
+	 * Therefore, this solution has not been implemented.
+	 */
+	if (!have_free_buffer())
+	{
+		*rs_have_free_buffer = false;
+		return InvalidBlockNumber;
+	}
+
+	if (p->pos == p->max_pos)
+		return InvalidBlockNumber;
+
+	if (p->first_block)
+	{
+		p->first_block = false;
+		return p->block_info[p->pos++].blocknum;
+	}
+
+	old_blk = &(p->block_info[p->pos - 1]);
+	cur_blk = &(p->block_info[p->pos]);
+
+	if (old_blk->database == cur_blk->database &&
+		old_blk->forknum == cur_blk->forknum &&
+		old_blk->filenumber == cur_blk->filenumber &&
+		cur_blk->blocknum < p->nblocks_in_fork)
+	{
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -435,6 +498,9 @@ autoprewarm_database_main(Datum main_arg)
 	BlockNumber nblocks = 0;
 	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	ReadStream *stream = NULL;
+	struct apw_read_stream_private p;
+	bool	   *rs_have_free_buffer;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,13 +517,16 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
+	p.block_info = block_info;
+	p.max_pos = apw_state->prewarm_stop_idx;
+
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	for (; pos < apw_state->prewarm_stop_idx; pos++)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
+		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
 
 		CHECK_FOR_INTERRUPTS();
@@ -470,6 +539,18 @@ autoprewarm_database_main(Datum main_arg)
 			old_blk->database != 0)
 			break;
 
+		/*
+		 * If stream needs to be created again, end it before closing the old
+		 * relation.
+		 */
+		if (stream && (old_blk == NULL ||
+					   old_blk->filenumber != blk->filenumber ||
+					   old_blk->forknum != blk->forknum))
+		{
+			Assert(read_stream_next_buffer(stream, (void **) &rs_have_free_buffer) == InvalidBuffer);
+			read_stream_end(stream);
+		}
+
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
@@ -506,7 +587,10 @@ autoprewarm_database_main(Datum main_arg)
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
+		/*
+		 * Once per fork, check for fork existence and size. Then create read
+		 * stream if it is suitable.
+		 */
 		if (old_blk == NULL ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
@@ -518,7 +602,21 @@ autoprewarm_database_main(Datum main_arg)
 			if (blk->forknum > InvalidForkNumber &&
 				blk->forknum <= MAX_FORKNUM &&
 				smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
 				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+
+				/* Create read stream. */
+				p.nblocks_in_fork = nblocks;
+				p.pos = pos;
+				p.first_block = true;
+				stream = read_stream_begin_relation(READ_STREAM_FULL,
+													NULL,
+													rel,
+													blk->forknum,
+													apw_read_stream_next_block,
+													&p,
+													sizeof(bool));
+			}
 			else
 				nblocks = 0;
 		}
@@ -532,16 +630,20 @@ autoprewarm_database_main(Datum main_arg)
 		}
 
 		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
+		buf = read_stream_next_buffer(stream, (void **) &rs_have_free_buffer);
 		if (BufferIsValid(buf))
 		{
 			apw_state->prewarmed_blocks++;
 			ReleaseBuffer(buf);
 		}
+		/* There are no free buffers left in shared buffers, break the loop. */
+		else if (!(*rs_have_free_buffer))
+			break;
 
 		old_blk = blk;
 	}
+	Assert(read_stream_next_buffer(stream, (void **) &rs_have_free_buffer) == InvalidBuffer);
+	read_stream_end(stream);
 
 	dsm_detach(seg);
 
-- 
2.45.2

From 15345de76b1006cb421d84abe3b03fbe60c754e4 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Wed, 27 Nov 2024 16:53:58 +0300
Subject: [PATCH v3 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 src/include/storage/buf_internals.h   |  1 +
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 contrib/pg_prewarm/autoprewarm.c      |  4 ++--
 3 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index eda6c699212..d95050adb23 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -432,6 +432,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index dffdd57e9b5..3259e6280cf 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 8291bf3c427..16174bf3d0f 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -518,13 +518,13 @@ autoprewarm_database_main(Datum main_arg)
 	pos = apw_state->prewarm_start_idx;
 
 	p.block_info = block_info;
-	p.max_pos = apw_state->prewarm_stop_idx;
+	p.max_pos = Min(apw_state->prewarm_stop_idx, pos + get_number_of_free_buffers());
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	for (; pos < apw_state->prewarm_stop_idx; pos++)
+	for (; pos < p.max_pos; pos++)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
-- 
2.45.2

From ffd37cc5010d2253b3f843e891df0e75ee8d27a6 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 5 Nov 2024 11:40:14 +0300
Subject: [PATCH v4 1/2] Optimize autoprewarm with read streams

We've measured 10% performance improvement, and this arranges to benefit
automatically from future optimizations to the read_stream subsystem.
---
 contrib/pg_prewarm/autoprewarm.c | 114 +++++++++++++++++++++++++++++--
 1 file changed, 109 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index fac4051e1aa..e7a8213209a 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -422,6 +423,70 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+struct apw_read_stream_private
+{
+	bool		first_block;
+	int			max_pos;
+	int			pos;
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks_in_fork;
+
+};
+
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	bool	   *rs_have_free_buffer = per_buffer_data;
+	BlockInfoRecord *old_blk;
+	BlockInfoRecord *cur_blk;
+
+	*rs_have_free_buffer = true;
+
+	/*
+	 * There may still be queued blocks in the stream even when no free
+	 * buffers are available in the buffer pool. This can lead to unnecessary
+	 * I/O operations and buffer evictions. One possible solution is to
+	 * compare the number of free buffers in the buffer pool with the number
+	 * of queued blocks in the stream. However, this approach is considered a
+	 * workaround and would add complexity with minimal benefit, as only a few
+	 * unnecessary I/O operations and buffer evictions are expected.
+	 * Therefore, this solution has not been implemented.
+	 */
+	if (!have_free_buffer())
+	{
+		*rs_have_free_buffer = false;
+		return InvalidBlockNumber;
+	}
+
+	if (p->pos == p->max_pos)
+		return InvalidBlockNumber;
+
+	if (p->first_block)
+	{
+		p->first_block = false;
+		return p->block_info[p->pos++].blocknum;
+	}
+
+	Assert(p->pos > 0 && p->pos < p->max_pos);
+
+	old_blk = &(p->block_info[p->pos - 1]);
+	cur_blk = &(p->block_info[p->pos]);
+
+	if (old_blk->database == cur_blk->database &&
+		old_blk->forknum == cur_blk->forknum &&
+		old_blk->filenumber == cur_blk->filenumber &&
+		cur_blk->blocknum < p->nblocks_in_fork)
+	{
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -435,6 +500,9 @@ autoprewarm_database_main(Datum main_arg)
 	BlockNumber nblocks = 0;
 	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	ReadStream *stream = NULL;
+	struct apw_read_stream_private p;
+	bool	   *rs_have_free_buffer;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,13 +519,16 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
+	p.block_info = block_info;
+	p.max_pos = apw_state->prewarm_stop_idx;
+
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	for (; pos < apw_state->prewarm_stop_idx; pos++)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
+		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
 
 		CHECK_FOR_INTERRUPTS();
@@ -470,6 +541,18 @@ autoprewarm_database_main(Datum main_arg)
 			old_blk->database != 0)
 			break;
 
+		/*
+		 * If stream needs to be created again, end it before closing the old
+		 * relation.
+		 */
+		if (stream && (old_blk == NULL ||
+					   old_blk->filenumber != blk->filenumber ||
+					   old_blk->forknum != blk->forknum))
+		{
+			Assert(read_stream_next_buffer(stream, (void **) &rs_have_free_buffer) == InvalidBuffer);
+			read_stream_end(stream);
+		}
+
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
@@ -506,7 +589,10 @@ autoprewarm_database_main(Datum main_arg)
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
+		/*
+		 * Once per fork, check for fork existence and size. Then create read
+		 * stream if it is suitable.
+		 */
 		if (old_blk == NULL ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
@@ -518,7 +604,21 @@ autoprewarm_database_main(Datum main_arg)
 			if (blk->forknum > InvalidForkNumber &&
 				blk->forknum <= MAX_FORKNUM &&
 				smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
 				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+
+				/* Create read stream. */
+				p.nblocks_in_fork = nblocks;
+				p.pos = pos;
+				p.first_block = true;
+				stream = read_stream_begin_relation(READ_STREAM_FULL,
+													NULL,
+													rel,
+													blk->forknum,
+													apw_read_stream_next_block,
+													&p,
+													sizeof(bool));
+			}
 			else
 				nblocks = 0;
 		}
@@ -532,16 +632,20 @@ autoprewarm_database_main(Datum main_arg)
 		}
 
 		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
+		buf = read_stream_next_buffer(stream, (void **) &rs_have_free_buffer);
 		if (BufferIsValid(buf))
 		{
 			apw_state->prewarmed_blocks++;
 			ReleaseBuffer(buf);
 		}
+		/* There are no free buffers left in shared buffers, break the loop. */
+		else if (!(*rs_have_free_buffer))
+			break;
 
 		old_blk = blk;
 	}
+	Assert(read_stream_next_buffer(stream, (void **) &rs_have_free_buffer) == InvalidBuffer);
+	read_stream_end(stream);
 
 	dsm_detach(seg);
 
-- 
2.45.2

From d35f774b03c8436c0e108b1b445ef61c00eb7180 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Wed, 27 Nov 2024 16:53:58 +0300
Subject: [PATCH v4 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 src/include/storage/buf_internals.h   |  1 +
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 contrib/pg_prewarm/autoprewarm.c      |  4 ++--
 3 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index eda6c699212..d95050adb23 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -432,6 +432,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index dffdd57e9b5..3259e6280cf 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index e7a8213209a..1b5a53e63bf 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -520,13 +520,13 @@ autoprewarm_database_main(Datum main_arg)
 	pos = apw_state->prewarm_start_idx;
 
 	p.block_info = block_info;
-	p.max_pos = apw_state->prewarm_stop_idx;
+	p.max_pos = Min(apw_state->prewarm_stop_idx, pos + get_number_of_free_buffers());
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	for (; pos < apw_state->prewarm_stop_idx; pos++)
+	for (; pos < p.max_pos; pos++)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
-- 
2.45.2

From 8b41e2bb5311072bc3b1d9239049930884e48b87 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 5 Nov 2024 11:40:14 +0300
Subject: [PATCH v5 1/2] Optimize autoprewarm with read streams

We've measured 10% performance improvement, and this arranges to benefit
automatically from future optimizations to the read_stream subsystem.
---
 contrib/pg_prewarm/autoprewarm.c | 120 +++++++++++++++++++++++++++++--
 1 file changed, 115 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..71c4f8a5920 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -422,6 +423,67 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+struct apw_read_stream_private
+{
+	bool		first_block;
+	int			max_pos;
+	int			pos;
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks_in_fork;
+
+};
+
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	BlockInfoRecord *old_blk;
+	BlockInfoRecord *cur_blk;
+
+	/*
+	 * There may still be queued blocks in the stream even when no free
+	 * buffers are available in the buffer pool. This can lead to unnecessary
+	 * I/O operations and buffer evictions. One possible solution is to
+	 * compare the number of free buffers in the buffer pool with the number
+	 * of queued blocks in the stream. However, this approach is considered a
+	 * workaround and would add complexity with minimal benefit, as only a few
+	 * unnecessary I/O operations and buffer evictions are expected.
+	 * Therefore, this solution has not been implemented.
+	 */
+	if (!have_free_buffer())
+		return InvalidBlockNumber;
+
+	if (p->pos == p->max_pos)
+		return InvalidBlockNumber;
+
+	cur_blk = &(p->block_info[p->pos]);
+
+	if (cur_blk->blocknum >= p->nblocks_in_fork)
+		return InvalidBlockNumber;
+
+	if (p->first_block)
+	{
+		p->first_block = false;
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	Assert(p->pos > 0 && p->pos < p->max_pos);
+	old_blk = &(p->block_info[p->pos - 1]);
+
+	if (old_blk->database == cur_blk->database &&
+		old_blk->forknum == cur_blk->forknum &&
+		old_blk->filenumber == cur_blk->filenumber)
+	{
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -435,6 +497,8 @@ autoprewarm_database_main(Datum main_arg)
 	BlockNumber nblocks = 0;
 	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	ReadStream *stream = NULL;
+	struct apw_read_stream_private p;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,13 +515,16 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
+	p.block_info = block_info;
+	p.max_pos = apw_state->prewarm_stop_idx;
+
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	for (; pos < apw_state->prewarm_stop_idx; pos++)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
+		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
 
 		CHECK_FOR_INTERRUPTS();
@@ -470,6 +537,18 @@ autoprewarm_database_main(Datum main_arg)
 			old_blk->database != 0)
 			break;
 
+		/*
+		 * If stream needs to be created again, end it before closing the old
+		 * relation.
+		 */
+		if (stream && (old_blk == NULL ||
+					   old_blk->filenumber != blk->filenumber ||
+					   old_blk->forknum != blk->forknum))
+		{
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+		}
+
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
@@ -506,7 +585,10 @@ autoprewarm_database_main(Datum main_arg)
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
+		/*
+		 * Once per fork, check for fork existence and size. Then create read
+		 * stream if it is suitable.
+		 */
 		if (old_blk == NULL ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
@@ -518,7 +600,27 @@ autoprewarm_database_main(Datum main_arg)
 			if (blk->forknum > InvalidForkNumber &&
 				blk->forknum <= MAX_FORKNUM &&
 				smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
 				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+
+				/* Create read stream. */
+				p.nblocks_in_fork = nblocks;
+				p.pos = pos;
+
+				/*
+				 * There is a special case for the first block in the
+				 * relation. We can't compare it with the previous block as
+				 * there is no previous block yet.
+				 */
+				p.first_block = true;
+				stream = read_stream_begin_relation(READ_STREAM_FULL,
+													NULL,
+													rel,
+													blk->forknum,
+													apw_read_stream_next_block,
+													&p,
+													0);
+			}
 			else
 				nblocks = 0;
 		}
@@ -532,17 +634,25 @@ autoprewarm_database_main(Datum main_arg)
 		}
 
 		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
+		buf = read_stream_next_buffer(stream, NULL);
 		if (BufferIsValid(buf))
 		{
 			apw_state->prewarmed_blocks++;
 			ReleaseBuffer(buf);
 		}
+		/* There are no free buffers left in shared buffers, break the loop. */
+		else if (!have_free_buffer())
+			break;
 
 		old_blk = blk;
 	}
 
+	if (stream)
+	{
+		Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+		read_stream_end(stream);
+	}
+
 	dsm_detach(seg);
 
 	/* Release lock on previous relation. */
-- 
2.43.0

From ffe876140536c89e2d80df1b9711e0970472b3d8 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Wed, 27 Nov 2024 16:53:58 +0300
Subject: [PATCH v5 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 contrib/pg_prewarm/autoprewarm.c      |  4 ++--
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 src/include/storage/buf_internals.h   |  1 +
 3 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 71c4f8a5920..42f2bc88fa9 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -516,13 +516,13 @@ autoprewarm_database_main(Datum main_arg)
 	pos = apw_state->prewarm_start_idx;
 
 	p.block_info = block_info;
-	p.max_pos = apw_state->prewarm_stop_idx;
+	p.max_pos = Min(apw_state->prewarm_stop_idx, pos + get_number_of_free_buffers());
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	for (; pos < apw_state->prewarm_stop_idx; pos++)
+	for (; pos < p.max_pos; pos++)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 336715b6c63..9f26e940426 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 9327f60c44c..197d2db49f5 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -445,6 +445,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
-- 
2.43.0

From 811caa0d543383cc5ec5a4e2d938787cb060cf42 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Mon, 31 Mar 2025 12:21:27 +0300
Subject: [PATCH 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 contrib/pg_prewarm/autoprewarm.c      |  7 +++++--
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 src/include/storage/buf_internals.h   |  1 +
 3 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 142e9d5359b..e96f67adf7c 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -518,6 +518,7 @@ void
 autoprewarm_database_main(Datum main_arg)
 {
 	int			pos;
+	int			stop_idx;
 	BlockInfoRecord *block_info;
 	Relation	rel = NULL;
 	dsm_segment *seg;
@@ -539,11 +540,13 @@ autoprewarm_database_main(Datum main_arg)
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
+	stop_idx = Min(apw_state->prewarm_stop_idx,
+				   pos + get_number_of_free_buffers());
 
 	cur_database = block_info[pos].database;
 
 	/* Loop until we run out of blocks to prewarm. */
-	while (pos < apw_state->prewarm_stop_idx)
+	while (pos < stop_idx)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 
@@ -616,7 +619,7 @@ autoprewarm_database_main(Datum main_arg)
 			nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 			nblocks_processed = autoprewarm_prewarm_relation(rel,
 															 pos,
-															 apw_state->prewarm_stop_idx,
+															 stop_idx,
 															 nblocks_in_fork,
 															 block_info);
 
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 336715b6c63..9f26e940426 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 9327f60c44c..197d2db49f5 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -445,6 +445,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
-- 
2.43.0

From 7e2f610e53cfe097dbc041e09915b8fc5da98c4e Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 5 Nov 2024 11:40:14 +0300
Subject: [PATCH v6 1/2] Optimize autoprewarm with read streams

We've measured 10% performance improvement, and this arranges to benefit
automatically from future optimizations to the read_stream subsystem.
---
 contrib/pg_prewarm/autoprewarm.c | 117 +++++++++++++++++++++++++++++--
 1 file changed, 112 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..a21d9571dff 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -422,6 +423,67 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+struct apw_read_stream_private
+{
+	bool		first_block;
+	int			max_pos;
+	int			pos;
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks_in_fork;
+
+};
+
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	BlockInfoRecord *old_blk;
+	BlockInfoRecord *cur_blk;
+
+	/*
+	 * There may still be queued blocks in the stream even when no free
+	 * buffers are available in the buffer pool. This can lead to unnecessary
+	 * I/O operations and buffer evictions. One possible solution is to
+	 * compare the number of free buffers in the buffer pool with the number
+	 * of queued blocks in the stream. However, this approach is considered a
+	 * workaround and would add complexity with minimal benefit, as only a few
+	 * unnecessary I/O operations and buffer evictions are expected.
+	 * Therefore, this solution has not been implemented.
+	 */
+	if (!have_free_buffer())
+		return InvalidBlockNumber;
+
+	if (p->pos == p->max_pos)
+		return InvalidBlockNumber;
+
+	cur_blk = &(p->block_info[p->pos]);
+
+	if (cur_blk->blocknum >= p->nblocks_in_fork)
+		return InvalidBlockNumber;
+
+	if (p->first_block)
+	{
+		p->first_block = false;
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	Assert(p->pos > 0 && p->pos < p->max_pos);
+	old_blk = &(p->block_info[p->pos - 1]);
+
+	if (old_blk->database == cur_blk->database &&
+		old_blk->forknum == cur_blk->forknum &&
+		old_blk->filenumber == cur_blk->filenumber)
+	{
+		p->pos++;
+		return cur_blk->blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -435,6 +497,8 @@ autoprewarm_database_main(Datum main_arg)
 	BlockNumber nblocks = 0;
 	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	ReadStream *stream = NULL;
+	struct apw_read_stream_private p;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,13 +515,16 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
+	p.block_info = block_info;
+	p.max_pos = apw_state->prewarm_stop_idx;
+
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	for (; pos < apw_state->prewarm_stop_idx && have_free_buffer(); pos++)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
+		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
 
 		CHECK_FOR_INTERRUPTS();
@@ -470,6 +537,18 @@ autoprewarm_database_main(Datum main_arg)
 			old_blk->database != 0)
 			break;
 
+		/*
+		 * If stream needs to be created again, end it before closing the old
+		 * relation.
+		 */
+		if (stream && (old_blk == NULL ||
+					   old_blk->filenumber != blk->filenumber ||
+					   old_blk->forknum != blk->forknum))
+		{
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+		}
+
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
@@ -506,7 +585,10 @@ autoprewarm_database_main(Datum main_arg)
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
+		/*
+		 * Once per fork, check for fork existence and size. Then create read
+		 * stream if it is suitable.
+		 */
 		if (old_blk == NULL ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
@@ -518,7 +600,27 @@ autoprewarm_database_main(Datum main_arg)
 			if (blk->forknum > InvalidForkNumber &&
 				blk->forknum <= MAX_FORKNUM &&
 				smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
 				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+
+				/* Create read stream. */
+				p.nblocks_in_fork = nblocks;
+				p.pos = pos;
+
+				/*
+				 * There is a special case for the first block in the
+				 * relation. We can't compare it with the previous block as
+				 * there is no previous block yet.
+				 */
+				p.first_block = true;
+				stream = read_stream_begin_relation(READ_STREAM_FULL,
+													NULL,
+													rel,
+													blk->forknum,
+													apw_read_stream_next_block,
+													&p,
+													0);
+			}
 			else
 				nblocks = 0;
 		}
@@ -532,8 +634,7 @@ autoprewarm_database_main(Datum main_arg)
 		}
 
 		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
+		buf = read_stream_next_buffer(stream, NULL);
 		if (BufferIsValid(buf))
 		{
 			apw_state->prewarmed_blocks++;
@@ -543,6 +644,12 @@ autoprewarm_database_main(Datum main_arg)
 		old_blk = blk;
 	}
 
+	if (stream)
+	{
+		Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+		read_stream_end(stream);
+	}
+
 	dsm_detach(seg);
 
 	/* Release lock on previous relation. */
-- 
2.43.0

From ac727712a7905ee00a8bc98a31939a7de7bf8493 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Wed, 27 Nov 2024 16:53:58 +0300
Subject: [PATCH v6 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 contrib/pg_prewarm/autoprewarm.c      |  4 ++--
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 src/include/storage/buf_internals.h   |  1 +
 3 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index a21d9571dff..cc0d45fae2b 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -516,13 +516,13 @@ autoprewarm_database_main(Datum main_arg)
 	pos = apw_state->prewarm_start_idx;
 
 	p.block_info = block_info;
-	p.max_pos = apw_state->prewarm_stop_idx;
+	p.max_pos = Min(apw_state->prewarm_stop_idx, pos + get_number_of_free_buffers());
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	for (; pos < apw_state->prewarm_stop_idx && have_free_buffer(); pos++)
+	for (; pos < p.max_pos && have_free_buffer(); pos++)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 		Buffer		buf;
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 336715b6c63..9f26e940426 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 9327f60c44c..197d2db49f5 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -445,6 +445,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
-- 
2.43.0

From 4a1d93b379966e939c1932a5fdd486d89bae4c7a Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Sat, 29 Mar 2025 20:17:42 +0300
Subject: [PATCH 1/2] Optimize autoprewarm with read streams

We've measured 10% performance improvement, and this arranges to benefit
automatically from future optimizations to the read_stream subsystem.
---
 contrib/pg_prewarm/autoprewarm.c | 197 ++++++++++++++++++++++---------
 1 file changed, 141 insertions(+), 56 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..142e9d5359b 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -421,6 +422,93 @@ apw_load_buffers(void)
 				(errmsg("autoprewarm successfully prewarmed %d of %d previously-loaded blocks",
 						apw_state->prewarmed_blocks, num_elements)));
 }
+struct apw_read_stream_private
+{
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks;
+	BlockNumber max_pos;
+	BlockNumber pos;
+	Oid			cur_database;
+	ForkNumber	cur_forknum;
+	RelFileNumber cur_filenumber;
+};
+
+static BlockNumber
+awp_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	BlockInfoRecord cur_blk = p->block_info[p->pos];
+	BlockNumber blocknum = InvalidBlockNumber;
+
+	if (have_free_buffer() &&
+		p->pos < p->max_pos &&
+		cur_blk.blocknum < p->nblocks &&
+		cur_blk.database == p->cur_database &&
+		cur_blk.forknum == p->cur_forknum &&
+		cur_blk.filenumber == p->cur_filenumber)
+	{
+		blocknum = cur_blk.blocknum;
+	}
+
+	(p->pos)++;
+	return blocknum;
+}
+
+/*
+ * Helper function to prewarm buffers in a relation by using read streams.
+ */
+static unsigned int
+autoprewarm_prewarm_relation(Relation rel,
+							 BlockNumber pos,
+							 BlockNumber max_pos,
+							 BlockNumber nblocks_in_fork,
+							 BlockInfoRecord *block_info)
+{
+	struct apw_read_stream_private p;
+	ReadStream *stream;
+	unsigned int blocks_done = 0;
+	BlockInfoRecord first_block = block_info[pos];
+
+	p.pos = pos;
+	p.max_pos = max_pos;
+	p.block_info = block_info;
+	p.nblocks = nblocks_in_fork;
+	p.cur_database = first_block.database;
+	p.cur_forknum = first_block.forknum;
+	p.cur_filenumber = first_block.filenumber;
+
+	stream = read_stream_begin_relation(READ_STREAM_FULL,
+										NULL,
+										rel,
+										first_block.forknum,
+										awp_read_stream_next_block,
+										&p,
+										0);
+
+	while (true)
+	{
+		Buffer		buf;
+
+		CHECK_FOR_INTERRUPTS();
+
+		buf = read_stream_next_buffer(stream, NULL);
+		if (BufferIsValid(buf))
+		{
+			ReleaseBuffer(buf);
+			++blocks_done;
+		}
+		else
+		{
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+			break;
+		}
+	}
+
+	return blocks_done;
+}
 
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
@@ -432,9 +520,10 @@ autoprewarm_database_main(Datum main_arg)
 	int			pos;
 	BlockInfoRecord *block_info;
 	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	Oid			cur_database;
+	Oid			cur_filenumber = InvalidOid;
+	BlockNumber nblocks_in_fork = InvalidBlockNumber;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,32 +540,44 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
-	/*
-	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
-	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	cur_database = block_info[pos].database;
+
+	/* Loop until we run out of blocks to prewarm. */
+	while (pos < apw_state->prewarm_stop_idx)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		BlockInfoRecord *blk = &block_info[pos];
 
-		CHECK_FOR_INTERRUPTS();
+		/* Loop until we run  we run out of free buffers. */
+		if (!have_free_buffer())
+			break;
 
 		/*
 		 * Quit if we've reached records for another database. If previous
 		 * blocks are of some global objects, then continue pre-warming.
 		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		if (cur_database != blk->database)
+		{
+			if (cur_database == 0)
+				cur_database = blk->database;
+			else
+				break;
+		}
+
+		/* Check whether blocknum is valid and within fork file size. */
+		while (cur_filenumber == blk->filenumber &&
+			   blk->blocknum >= nblocks_in_fork)
+		{
+			/* Move to next forknum. */
+			pos++;
+			continue;
+		}
 
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
 		 * previously; in that case, there is nothing to close.
 		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		if (rel && cur_filenumber != blk->filenumber)
 		{
 			relation_close(rel, AccessShareLock);
 			rel = NULL;
@@ -487,70 +588,54 @@ autoprewarm_database_main(Datum main_arg)
 		 * Try to open each new relation, but only once, when we first
 		 * encounter it. If it's been dropped, skip the associated blocks.
 		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
+		if (!rel && cur_filenumber != blk->filenumber)
 		{
 			Oid			reloid;
 
-			Assert(rel == NULL);
 			StartTransactionCommand();
 			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
 			if (OidIsValid(reloid))
 				rel = try_relation_open(reloid, AccessShareLock);
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
-			continue;
-		}
-
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
-		{
 			/*
-			 * smgrexists is not safe for illegal forknum, hence check whether
-			 * the passed forknum is valid before using it in smgrexists.
+			 * Update cur_filenumber although relation may not be opened. If
+			 * not updated and if we can't open the relation when the file
+			 * number is changed; we will end up unnecessarily trying to open
+			 * relation for all the blocks that have the same file number.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			cur_filenumber = blk->filenumber;
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
+			if (!rel)
+				CommitTransactionCommand();
 		}
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
+		if (rel && smgrexists(RelationGetSmgr(rel), blk->forknum))
 		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
+			unsigned int nblocks_processed;
+
+			nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+			nblocks_processed = autoprewarm_prewarm_relation(rel,
+															 pos,
+															 apw_state->prewarm_stop_idx,
+															 nblocks_in_fork,
+															 block_info);
+
+			apw_state->prewarmed_blocks += nblocks_processed;
+			/* Move pos forward by at least one */
+			pos += Max(nblocks_processed, 1);
+			continue;
 		}
 
-		old_blk = blk;
+		pos++;
 	}
 
-	dsm_detach(seg);
-
-	/* Release lock on previous relation. */
 	if (rel)
 	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.43.0

From 19abe210241cf48142c2e1b2e3ae5a7ba34921e4 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Sat, 29 Mar 2025 20:17:42 +0300
Subject: [PATCH v7 1/2] Optimize autoprewarm with read streams

We've measured 10% performance improvement, and this arranges to benefit
automatically from future optimizations to the read_stream subsystem.
---
 contrib/pg_prewarm/autoprewarm.c | 192 ++++++++++++++++++++++---------
 1 file changed, 139 insertions(+), 53 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..68c99664de5 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -421,6 +422,93 @@ apw_load_buffers(void)
 				(errmsg("autoprewarm successfully prewarmed %d of %d previously-loaded blocks",
 						apw_state->prewarmed_blocks, num_elements)));
 }
+struct apw_read_stream_private
+{
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks;
+	BlockNumber max_pos;
+	BlockNumber pos;
+	Oid			cur_database;
+	ForkNumber	cur_forknum;
+	RelFileNumber cur_filenumber;
+};
+
+static BlockNumber
+awp_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	BlockInfoRecord cur_blk = p->block_info[p->pos];
+	BlockNumber blocknum = InvalidBlockNumber;
+
+	if (have_free_buffer() &&
+		p->pos < p->max_pos &&
+		cur_blk.blocknum < p->nblocks &&
+		cur_blk.database == p->cur_database &&
+		cur_blk.forknum == p->cur_forknum &&
+		cur_blk.filenumber == p->cur_filenumber)
+	{
+		blocknum = cur_blk.blocknum;
+	}
+
+	(p->pos)++;
+	return blocknum;
+}
+
+/*
+ * Helper function to prewarm buffers in a relation by using read streams.
+ */
+static unsigned int
+autoprewarm_prewarm_relation(Relation rel,
+							 BlockNumber pos,
+							 BlockNumber max_pos,
+							 BlockNumber nblocks_in_fork,
+							 BlockInfoRecord *block_info)
+{
+	struct apw_read_stream_private p;
+	ReadStream *stream;
+	unsigned int blocks_done = 0;
+	BlockInfoRecord first_block = block_info[pos];
+
+	p.pos = pos;
+	p.max_pos = max_pos;
+	p.block_info = block_info;
+	p.nblocks = nblocks_in_fork;
+	p.cur_database = first_block.database;
+	p.cur_forknum = first_block.forknum;
+	p.cur_filenumber = first_block.filenumber;
+
+	stream = read_stream_begin_relation(READ_STREAM_FULL,
+										NULL,
+										rel,
+										first_block.forknum,
+										awp_read_stream_next_block,
+										&p,
+										0);
+
+	while (true)
+	{
+		Buffer		buf;
+
+		CHECK_FOR_INTERRUPTS();
+
+		buf = read_stream_next_buffer(stream, NULL);
+		if (BufferIsValid(buf))
+		{
+			ReleaseBuffer(buf);
+			++blocks_done;
+		}
+		else
+		{
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+			break;
+		}
+	}
+
+	return blocks_done;
+}
 
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
@@ -432,9 +520,10 @@ autoprewarm_database_main(Datum main_arg)
 	int			pos;
 	BlockInfoRecord *block_info;
 	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	Oid			cur_database;
+	Oid			cur_filenumber = InvalidOid;
+	BlockNumber nblocks_in_fork = InvalidBlockNumber;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,32 +540,44 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
-	/*
-	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
-	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	cur_database = block_info[pos].database;
+
+	/* Loop until we run out of blocks to prewarm. */
+	while (pos < apw_state->prewarm_stop_idx)
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		BlockInfoRecord *blk = &block_info[pos];
 
-		CHECK_FOR_INTERRUPTS();
+		/* Loop until we run  we run out of free buffers. */
+		if (!have_free_buffer())
+			break;
 
 		/*
 		 * Quit if we've reached records for another database. If previous
 		 * blocks are of some global objects, then continue pre-warming.
 		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		if (cur_database != blk->database)
+		{
+			if (cur_database == 0)
+				cur_database = blk->database;
+			else
+				break;
+		}
+
+		/* Check whether blocknum is valid and within fork file size. */
+		if (cur_filenumber == blk->filenumber &&
+			blk->blocknum >= nblocks_in_fork)
+		{
+			/* Move to next forknum. */
+			pos++;
+			continue;
+		}
 
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
 		 * relation. Note that rel will be NULL if try_relation_open failed
 		 * previously; in that case, there is nothing to close.
 		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		if (rel && cur_filenumber != blk->filenumber)
 		{
 			relation_close(rel, AccessShareLock);
 			rel = NULL;
@@ -487,60 +588,45 @@ autoprewarm_database_main(Datum main_arg)
 		 * Try to open each new relation, but only once, when we first
 		 * encounter it. If it's been dropped, skip the associated blocks.
 		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
+		if (!rel && cur_filenumber != blk->filenumber)
 		{
 			Oid			reloid;
 
-			Assert(rel == NULL);
 			StartTransactionCommand();
 			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
 			if (OidIsValid(reloid))
 				rel = try_relation_open(reloid, AccessShareLock);
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
-			continue;
-		}
-
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
-		{
 			/*
-			 * smgrexists is not safe for illegal forknum, hence check whether
-			 * the passed forknum is valid before using it in smgrexists.
+			 * Update cur_filenumber although relation may not be opened. If
+			 * not updated and if we can't open the relation when the file
+			 * number is changed; we will end up unnecessarily trying to open
+			 * relation for all the blocks that have the same file number.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			cur_filenumber = blk->filenumber;
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
+			if (!rel)
+				CommitTransactionCommand();
 		}
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
+		if (rel && smgrexists(RelationGetSmgr(rel), blk->forknum))
 		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
+			unsigned int nblocks_processed;
+
+			nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+			nblocks_processed = autoprewarm_prewarm_relation(rel,
+															 pos,
+															 apw_state->prewarm_stop_idx,
+															 nblocks_in_fork,
+															 block_info);
+
+			apw_state->prewarmed_blocks += nblocks_processed;
+			/* Move pos forward by at least one */
+			pos += Max(nblocks_processed, 1);
+			continue;
 		}
 
-		old_blk = blk;
+		pos++;
 	}
 
 	dsm_detach(seg);
-- 
2.43.0

From c50d0f17e223e9e85421bc2bc897291383cba74d Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Mon, 31 Mar 2025 12:21:27 +0300
Subject: [PATCH v7 2/2] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 contrib/pg_prewarm/autoprewarm.c      |  7 +++++--
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 src/include/storage/buf_internals.h   |  1 +
 3 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 68c99664de5..df10e02358d 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -518,6 +518,7 @@ void
 autoprewarm_database_main(Datum main_arg)
 {
 	int			pos;
+	int			stop_idx;
 	BlockInfoRecord *block_info;
 	Relation	rel = NULL;
 	dsm_segment *seg;
@@ -539,11 +540,13 @@ autoprewarm_database_main(Datum main_arg)
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
+	stop_idx = Min(apw_state->prewarm_stop_idx,
+				   pos + get_number_of_free_buffers());
 
 	cur_database = block_info[pos].database;
 
 	/* Loop until we run out of blocks to prewarm. */
-	while (pos < apw_state->prewarm_stop_idx)
+	while (pos < stop_idx)
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 
@@ -616,7 +619,7 @@ autoprewarm_database_main(Datum main_arg)
 			nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 			nblocks_processed = autoprewarm_prewarm_relation(rel,
 															 pos,
-															 apw_state->prewarm_stop_idx,
+															 stop_idx,
 															 nblocks_in_fork,
 															 block_info);
 
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 336715b6c63..9f26e940426 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 9327f60c44c..197d2db49f5 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -445,6 +445,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
-- 
2.43.0

From 438f13072af060b30485b0dd871c1b26ee503513 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH] pgsr autoprewarm

---
 contrib/pg_prewarm/autoprewarm.c | 218 +++++++++++++++++++------------
 1 file changed, 131 insertions(+), 87 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..459d36a75a9 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -421,6 +422,63 @@ apw_load_buffers(void)
 				(errmsg("autoprewarm successfully prewarmed %d of %d previously-loaded blocks",
 						apw_state->prewarmed_blocks, num_elements)));
 }
+struct apw_read_stream_private
+{
+	BlockInfoRecord *block_info;
+	int pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+};
+
+static BlockNumber
+awp_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+
+	for (int i; (i = p->pos++) < apw_state->prewarm_stop_idx;)
+	{
+		BlockInfoRecord cur_blk = p->block_info[i];
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!have_free_buffer())
+		{
+			p->pos = apw_state->prewarm_stop_idx;
+			return InvalidBlockNumber;
+		}
+
+		if (cur_blk.database != p->database)
+			return InvalidBlockNumber;
+
+		if (cur_blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (cur_blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		if (cur_blk.blocknum >= p->nblocks)
+			continue;
+
+		return cur_blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
+static inline BlockInfoRecord *next_record(BlockInfoRecord *block_info, int *pos)
+{
+	int oldpos;
+	if (*pos >= apw_state->prewarm_stop_idx)
+		return NULL;
+
+	oldpos = *pos;
+	(*pos)++;
+	return &block_info[oldpos];
+}
 
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
@@ -429,12 +487,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int i;
+	BlockInfoRecord *blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -449,108 +506,95 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
 
-	/*
-	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
-	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
-	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+	i = apw_state->prewarm_start_idx;
+	blk = &block_info[i];
+	database = blk->database;
 
-		CHECK_FOR_INTERRUPTS();
-
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+	do
+	{
+		RelFileNumber filenumber = blk->filenumber;
+		Oid reloid;
+		Relation rel;
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
-		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
-			CommitTransactionCommand();
-		}
+		StartTransactionCommand();
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
+		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
+		if (!OidIsValid(reloid))
+			goto skip_relation;
 
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+		if ((rel = try_relation_open(reloid, AccessShareLock)) == NULL)
+			goto skip_relation;
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
+		do
 		{
-			old_blk = blk;
-			continue;
-		}
+			ForkNumber forknum = blk->forknum;
+			BlockNumber nblocks;
+			struct apw_read_stream_private p;
+			ReadStream *stream;
+			Buffer buf;
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
-		{
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk->forknum <= InvalidForkNumber)
+				goto skip_forknumber;
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			if (blk->forknum > MAX_FORKNUM)
+				goto skip_forknumber;
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			if (!smgrexists(RelationGetSmgr(rel), blk->forknum))
+				goto skip_forknumber;
 
-		old_blk = blk;
-	}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-	dsm_detach(seg);
+			p = (struct apw_read_stream_private)
+			{
+				.block_info = block_info,
+				.pos = i,
+				.database = database,
+				.filenumber = filenumber,
+				.forknum = forknum,
+				.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												awp_read_stream_next_block,
+												&p,
+												0);
+
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
+			{
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
+			i = p.pos;
+
+			read_stream_end(stream);
+
+			continue;
+
+		skip_forknumber:;
+			while ((blk = next_record(block_info, &i)) != NULL && blk->database == database && blk->filenumber == filenumber && blk->forknum == forknum);
+		} while ((blk = next_record(block_info, &i)) && blk->database == database && blk->filenumber == filenumber);
+
+		/* Release lock on previous relation. */
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
-	}
+
+		continue;
+
+	skip_relation:
+		CommitTransactionCommand();
+		while ((blk = next_record(block_info, &i)) != NULL && blk->database == database && blk->filenumber == filenumber);
+	} while ((blk = next_record(block_info, &i)) && blk->database == database);
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From 7db3df60698da30bfec4b8cca76342dc6e779f81 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Sat, 29 Mar 2025 20:17:42 +0300
Subject: [PATCH v8 1/3] Optimize autoprewarm with read streams

We've measured 10% performance improvement, and this arranges to benefit
automatically from future optimizations to the read_stream subsystem.
---
 contrib/pg_prewarm/autoprewarm.c | 201 +++++++++++++++++++++++--------
 1 file changed, 148 insertions(+), 53 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..7154303f0ba 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -421,6 +422,93 @@ apw_load_buffers(void)
 				(errmsg("autoprewarm successfully prewarmed %d of %d previously-loaded blocks",
 						apw_state->prewarmed_blocks, num_elements)));
 }
+struct apw_read_stream_private
+{
+	BlockInfoRecord *block_info;
+	BlockNumber nblocks;
+	BlockNumber max_pos;
+	BlockNumber pos;
+	Oid			cur_database;
+	ForkNumber	cur_forknum;
+	RelFileNumber cur_filenumber;
+};
+
+static BlockNumber
+awp_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+	BlockInfoRecord cur_blk = p->block_info[p->pos];
+	BlockNumber blocknum = InvalidBlockNumber;
+
+	if (have_free_buffer() &&
+		p->pos < p->max_pos &&
+		cur_blk.blocknum < p->nblocks &&
+		cur_blk.database == p->cur_database &&
+		cur_blk.filenumber == p->cur_filenumber &&
+		cur_blk.forknum == p->cur_forknum)
+	{
+		blocknum = cur_blk.blocknum;
+	}
+
+	(p->pos)++;
+	return blocknum;
+}
+
+/*
+ * Helper function to prewarm buffers in a relation by using read streams.
+ */
+static unsigned int
+autoprewarm_prewarm_relation(Relation rel,
+							 BlockNumber pos,
+							 BlockNumber max_pos,
+							 BlockNumber nblocks_in_fork,
+							 BlockInfoRecord *block_info)
+{
+	struct apw_read_stream_private p;
+	ReadStream *stream;
+	unsigned int blocks_done = 0;
+	BlockInfoRecord first_block = block_info[pos];
+
+	p.pos = pos;
+	p.max_pos = max_pos;
+	p.block_info = block_info;
+	p.nblocks = nblocks_in_fork;
+	p.cur_database = first_block.database;
+	p.cur_forknum = first_block.forknum;
+	p.cur_filenumber = first_block.filenumber;
+
+	stream = read_stream_begin_relation(READ_STREAM_FULL,
+										NULL,
+										rel,
+										first_block.forknum,
+										awp_read_stream_next_block,
+										&p,
+										0);
+
+	while (true)
+	{
+		Buffer		buf;
+
+		CHECK_FOR_INTERRUPTS();
+
+		buf = read_stream_next_buffer(stream, NULL);
+		if (BufferIsValid(buf))
+		{
+			ReleaseBuffer(buf);
+			++blocks_done;
+		}
+		else
+		{
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+			break;
+		}
+	}
+
+	return blocks_done;
+}
 
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
@@ -432,9 +520,9 @@ autoprewarm_database_main(Datum main_arg)
 	int			pos;
 	BlockInfoRecord *block_info;
 	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
 	dsm_segment *seg;
+	Oid			cur_database;
+	BlockNumber nblocks_in_fork;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -451,32 +539,34 @@ autoprewarm_database_main(Datum main_arg)
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
 
+	cur_database = block_info[pos].database;
+
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
 	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
-
-		CHECK_FOR_INTERRUPTS();
+		BlockInfoRecord *blk = &block_info[pos];
 
 		/*
 		 * Quit if we've reached records for another database. If previous
 		 * blocks are of some global objects, then continue pre-warming.
 		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		if (cur_database != blk->database)
+		{
+			if (cur_database == 0)
+				cur_database = blk->database;
+			else
+				break;
+		}
 
 		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
+		 * Close the old relation. Note that rel will be NULL if
+		 * try_relation_open failed previously; in that case, there is nothing
+		 * to close.
 		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		if (rel)
 		{
 			relation_close(rel, AccessShareLock);
 			rel = NULL;
@@ -487,60 +577,65 @@ autoprewarm_database_main(Datum main_arg)
 		 * Try to open each new relation, but only once, when we first
 		 * encounter it. If it's been dropped, skip the associated blocks.
 		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
+		if (!rel)
 		{
 			Oid			reloid;
 
-			Assert(rel == NULL);
 			StartTransactionCommand();
 			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
 			if (OidIsValid(reloid))
 				rel = try_relation_open(reloid, AccessShareLock);
 
 			if (!rel)
+			{
 				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
-			continue;
-		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
-		{
-			/*
-			 * smgrexists is not safe for illegal forknum, hence check whether
-			 * the passed forknum is valid before using it in smgrexists.
-			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+				/* Move to next filenumber. */
+				while (true)
+				{
+					BlockInfoRecord cur_blk = block_info[pos++];
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+					if (cur_blk.database != blk->database ||
+						cur_blk.filenumber != blk->filenumber)
+						break;
+				}
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
+				continue;
+			}
+			else if (smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
+				unsigned int nblocks_processed;
+
+				nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+				nblocks_processed = autoprewarm_prewarm_relation(rel,
+																 pos,
+																 stop_idx,
+																 nblocks_in_fork,
+																 block_info);
+
+				apw_state->prewarmed_blocks += nblocks_processed;
+
+				/* Move pos forward by at least one. */
+				pos += Max(nblocks_processed, 1);
+
+				/* Move to next forknum. */
+				while (true)
+				{
+					BlockInfoRecord cur_blk = block_info[pos];
+
+					if (cur_blk.database == blk->database &&
+						cur_blk.filenumber == blk->filenumber &&
+						cur_blk.forknum == blk->forknum)
+						pos++;
+					else
+						break;
+				}
+
+				continue;
+			}
 		}
 
-		old_blk = blk;
+		pos++;
 	}
 
 	dsm_detach(seg);
-- 
2.43.0

From 0f640c042458b98c6b7fbd9360561c3623b8d9c7 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Mon, 31 Mar 2025 12:21:27 +0300
Subject: [PATCH v8 2/3] Count free buffers at the start of the autoprewarm

Streamified version of the autoprewarm code may do unnecessary I/O and
buffer evicting. To prevent it at least a little bit, count the number
of free buffers in the buffer pool and queue buffers up to that number
in the callback function of the autoprewarm.
---
 contrib/pg_prewarm/autoprewarm.c      |  5 ++++-
 src/backend/storage/buffer/freelist.c | 17 +++++++++++++++++
 src/include/storage/buf_internals.h   |  1 +
 3 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 7154303f0ba..4f635eaf31c 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -518,6 +518,7 @@ void
 autoprewarm_database_main(Datum main_arg)
 {
 	int			pos;
+	int			stop_idx;
 	BlockInfoRecord *block_info;
 	Relation	rel = NULL;
 	dsm_segment *seg;
@@ -538,6 +539,8 @@ autoprewarm_database_main(Datum main_arg)
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
 	pos = apw_state->prewarm_start_idx;
+	stop_idx = Min(apw_state->prewarm_stop_idx,
+				   pos + get_number_of_free_buffers());
 
 	cur_database = block_info[pos].database;
 
@@ -545,7 +548,7 @@ autoprewarm_database_main(Datum main_arg)
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (pos < stop_idx && have_free_buffer())
 	{
 		BlockInfoRecord *blk = &block_info[pos];
 
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 336715b6c63..9f26e940426 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -180,6 +180,23 @@ have_free_buffer(void)
 		return false;
 }
 
+/*
+ * get_number_of_free_buffers -- a lockless way to get the number of free
+ *								 buffers in buffer pool.
+ *
+ * Note that result continuously changes as free buffers are moved out by other
+ * operations.
+ */
+int
+get_number_of_free_buffers(void)
+{
+	/* All the buffers are free. */
+	if (StrategyControl->firstFreeBuffer < 0)
+		return NBuffers;
+	else
+		return StrategyControl->lastFreeBuffer - StrategyControl->firstFreeBuffer;
+}
+
 /*
  * StrategyGetBuffer
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 72b36a4af26..cdc9936fccd 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -448,6 +448,7 @@ extern void StrategyNotifyBgWriter(int bgwprocno);
 extern Size StrategyShmemSize(void);
 extern void StrategyInitialize(bool init);
 extern bool have_free_buffer(void);
+extern int	get_number_of_free_buffers(void);
 
 /* buf_table.c */
 extern Size BufTableShmemSize(int size);
-- 
2.43.0

From 8136a56cb63ff846b30998d88f211ffff68e0e5b Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavuz81@gmail.com>
Date: Tue, 1 Apr 2025 15:38:51 +0300
Subject: [PATCH v8 3/3] Refactor code more

---
 contrib/pg_prewarm/autoprewarm.c | 116 ++++++++++++++-----------------
 1 file changed, 51 insertions(+), 65 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 4f635eaf31c..1f31d6eaa0e 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -550,6 +550,7 @@ autoprewarm_database_main(Datum main_arg)
 	 */
 	while (pos < stop_idx && have_free_buffer())
 	{
+		Oid			reloid;
 		BlockInfoRecord *blk = &block_info[pos];
 
 		/*
@@ -564,91 +565,76 @@ autoprewarm_database_main(Datum main_arg)
 				break;
 		}
 
-		/*
-		 * Close the old relation. Note that rel will be NULL if
-		 * try_relation_open failed previously; in that case, there is nothing
-		 * to close.
-		 */
-		if (rel)
-		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
-			CommitTransactionCommand();
-		}
+		/* Try to open new relation. */
+		StartTransactionCommand();
+		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
+		if (OidIsValid(reloid))
+			rel = try_relation_open(reloid, AccessShareLock);
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
 		if (!rel)
 		{
-			Oid			reloid;
-
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			CommitTransactionCommand();
 
-			if (!rel)
+			/* Move to next filenumber. */
+			while (pos < stop_idx && have_free_buffer())
 			{
-				CommitTransactionCommand();
-
-				/* Move to next filenumber. */
-				while (true)
-				{
-					BlockInfoRecord cur_blk = block_info[pos++];
+				BlockInfoRecord cur_blk = block_info[pos++];
 
-					if (cur_blk.database != blk->database ||
-						cur_blk.filenumber != blk->filenumber)
-						break;
-				}
-
-				continue;
+				if (cur_blk.database != blk->database ||
+					cur_blk.filenumber != blk->filenumber)
+					break;
 			}
-			else if (smgrexists(RelationGetSmgr(rel), blk->forknum))
-			{
-				unsigned int nblocks_processed;
 
-				nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-				nblocks_processed = autoprewarm_prewarm_relation(rel,
-																 pos,
-																 stop_idx,
-																 nblocks_in_fork,
-																 block_info);
+			continue;
+		}
+		else if (smgrexists(RelationGetSmgr(rel), blk->forknum))
+		{
+			unsigned int nblocks_processed;
 
-				apw_state->prewarmed_blocks += nblocks_processed;
+			nblocks_in_fork = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
+			nblocks_processed = autoprewarm_prewarm_relation(rel,
+															 pos,
+															 stop_idx,
+															 nblocks_in_fork,
+															 block_info);
 
-				/* Move pos forward by at least one. */
-				pos += Max(nblocks_processed, 1);
+			apw_state->prewarmed_blocks += nblocks_processed;
 
-				/* Move to next forknum. */
-				while (true)
-				{
-					BlockInfoRecord cur_blk = block_info[pos];
+			/* Move pos forward by at least one. */
+			pos += Max(nblocks_processed, 1);
 
-					if (cur_blk.database == blk->database &&
-						cur_blk.filenumber == blk->filenumber &&
-						cur_blk.forknum == blk->forknum)
-						pos++;
-					else
-						break;
-				}
+			/* Move to next forknum. */
+			while (pos < stop_idx && have_free_buffer())
+			{
+				BlockInfoRecord cur_blk = block_info[pos];
 
-				continue;
+				if (cur_blk.database == blk->database &&
+					cur_blk.filenumber == blk->filenumber &&
+					cur_blk.forknum == blk->forknum)
+					pos++;
+				else
+					break;
 			}
+
+			/* Close the relation. */
+			relation_close(rel, AccessShareLock);
+			rel = NULL;
+			CommitTransactionCommand();
+
+			continue;
+		}
+		else
+		{
+			/* Close the relation. */
+			relation_close(rel, AccessShareLock);
+			rel = NULL;
+			CommitTransactionCommand();
 		}
 
 		pos++;
 	}
 
 	dsm_detach(seg);
-
-	/* Release lock on previous relation. */
-	if (rel)
-	{
-		relation_close(rel, AccessShareLock);
-		CommitTransactionCommand();
-	}
 }
 
 /*
-- 
2.43.0

From f1c98580825acc63538372e9f67f6ed2bf68b540 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH] pgsr autoprewarm

---
 contrib/pg_prewarm/autoprewarm.c | 244 ++++++++++++++++++++-----------
 1 file changed, 160 insertions(+), 84 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..fd3fb228715 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -421,6 +422,96 @@ apw_load_buffers(void)
 				(errmsg("autoprewarm successfully prewarmed %d of %d previously-loaded blocks",
 						apw_state->prewarmed_blocks, num_elements)));
 }
+struct apw_read_stream_private
+{
+	BlockInfoRecord *block_info;
+	int			pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+};
+
+static BlockNumber
+awp_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	struct apw_read_stream_private *p = callback_private_data;
+
+	for (int i; (i = p->pos++) < apw_state->prewarm_stop_idx;)
+	{
+		BlockInfoRecord cur_blk = p->block_info[i];
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!have_free_buffer())
+		{
+			p->pos = apw_state->prewarm_stop_idx;
+			return InvalidBlockNumber;
+		}
+
+		if (cur_blk.database != p->database)
+			return InvalidBlockNumber;
+
+		if (cur_blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (cur_blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		if (cur_blk.blocknum >= p->nblocks)
+			continue;
+
+		return cur_blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
+static int
+next_fork_idx(BlockInfoRecord *block_info, int start, int stop,
+			  Oid database, RelFileNumber filenumber, ForkNumber forknum)
+{
+	Assert(block_info[start].forknum == forknum);
+
+	for (int i = start; i < stop; i++)
+	{
+		BlockInfoRecord *blk = &block_info[i];
+
+		if (blk->database != database)
+			return i;
+
+		if (blk->filenumber != filenumber)
+			return i;
+
+		if (blk->forknum != forknum)
+			return i;
+	}
+
+	return stop;
+}
+
+static int
+next_rel_idx(BlockInfoRecord *block_info, int start, int stop,
+			 Oid database, RelFileNumber filenumber)
+{
+	Assert(block_info[start].filenumber == filenumber);
+
+	for (int i = start; i < stop; i++)
+	{
+		BlockInfoRecord *blk = &block_info[i];
+
+		if (blk->database != database)
+			return i;
+
+		if (blk->filenumber != filenumber)
+			return i;
+	}
+
+	return stop;
+}
+
 
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
@@ -429,12 +520,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord *blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -449,108 +539,94 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
 
-	/*
-	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
-	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
-	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+	i = apw_state->prewarm_start_idx;
+	blk = &block_info[i];
+	database = blk->database;
 
-		CHECK_FOR_INTERRUPTS();
+	while (i < apw_state->prewarm_stop_idx &&
+		   blk->database == database)
+	{
+		RelFileNumber filenumber = blk->filenumber;
+		Oid			reloid;
+		Relation	rel;
 
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		StartTransactionCommand();
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
+		if (!OidIsValid(reloid) ||
+			(OidIsValid(reloid) &&
+			 (rel = try_relation_open(reloid, AccessShareLock)) == NULL))
 		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
 			CommitTransactionCommand();
-		}
-
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
-
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
-
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			i = next_rel_idx(block_info, i, apw_state->prewarm_stop_idx,
+							 blk->database, blk->filenumber);
+			blk = &block_info[i];
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk->database == database &&
+			   blk->filenumber == filenumber)
 		{
+			ForkNumber	forknum = blk->forknum;
+			BlockNumber nblocks;
+			struct apw_read_stream_private p;
+			ReadStream *stream;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk->forknum <= InvalidForkNumber ||
+				blk->forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
+				i = next_fork_idx(block_info, i, apw_state->prewarm_stop_idx,
+								  blk->database, blk->filenumber, blk->forknum);
+				blk = &block_info[i];
+				continue;
+			}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			p = (struct apw_read_stream_private)
+			{
+				.block_info = block_info,
+					.pos = i,
+					.database = database,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												awp_read_stream_next_block,
+												&p,
+												0);
+
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
+			{
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
 
-		old_blk = blk;
-	}
+			read_stream_end(stream);
 
-	dsm_detach(seg);
+			i = p.pos;
+			blk = &block_info[i];
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
+		/* Release lock on previous relation. */
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From 2d14f37ea824a6ce3605f53b0a9c5622f08bf6e7 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH 1/2] Refactor autoprewarm_database_main() in preparation for
 read stream

TODO: write a commit message

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 166 +++++++++++++++++--------------
 1 file changed, 89 insertions(+), 77 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..84072587ea0 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -429,12 +429,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord *blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -449,108 +448,121 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	database = apw_state->database;
+
+	blk = &block_info[i];
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
+	 * buffers. We'll quit if we've reached records for another database,
+	 * however we do want to prewarm any global objects.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx &&
+		   (blk->database == database || blk->database == 0) &&
+		   have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		RelFileNumber filenumber = blk->filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		StartTransactionCommand();
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
+			/* We failed to open the relation, so there is nothing to close. */
 			CommitTransactionCommand();
-		}
-
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
 
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = &block_info[i];
+				if ((blk->database != database && blk->database != 0) ||
+					blk->filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our new found relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   (blk->database == database || blk->database == 0) &&
+			   blk->filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk->forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk->forknum <= InvalidForkNumber ||
+				blk->forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = &block_info[i];
+					if ((blk->database != database && blk->database != 0) ||
+						blk->filenumber != filenumber ||
+						blk->forknum != forknum)
+						break;
+				}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+				continue;
+			}
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-		old_blk = blk;
-	}
+			/* Check whether blocknum is valid and within fork file size. */
+			if (blk->blocknum >= nblocks)
+			{
+				blk = &block_info[++i];
+				continue;
+			}
 
-	dsm_detach(seg);
+			/* Prewarm buffer. */
+			buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
+									NULL);
+
+			if (BufferIsValid(buf))
+			{
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+
+			blk = &block_info[++i];
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From ff77f60d670f1c25699e969a48f7bbd5a577a405 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH 2/2] streaming read autoprewarm

TODO: write a commit message

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 95 +++++++++++++++++++++++++++-----
 src/tools/pgindent/typedefs.list |  1 +
 2 files changed, 82 insertions(+), 14 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 84072587ea0..667aa895b18 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,21 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callaback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	BlockInfoRecord *block_info;
+	int			pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -422,6 +438,45 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+	BlockInfoRecord blk;
+
+	CHECK_FOR_INTERRUPTS();
+
+	if (!have_free_buffer())
+		p->pos = apw_state->prewarm_stop_idx;
+
+	if (p->pos >= apw_state->prewarm_stop_idx)
+		return InvalidBlockNumber;
+
+	blk = p->block_info[p->pos];
+
+	if (blk.database != p->database && blk.database != 0)
+		return InvalidBlockNumber;
+
+	if (blk.filenumber != p->filenumber)
+		return InvalidBlockNumber;
+
+	if (blk.forknum != p->forknum)
+		return InvalidBlockNumber;
+
+	/* Check whether blocknum is valid and within fork file size. */
+	if (blk.blocknum >= p->nblocks)
+		return InvalidBlockNumber;
+
+	p->pos++;
+	return blk.blocknum;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -467,8 +522,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		StartTransactionCommand();
 
 		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
@@ -509,6 +562,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk->forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -538,24 +593,36 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-			/* Check whether blocknum is valid and within fork file size. */
-			if (blk->blocknum >= nblocks)
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				blk = &block_info[++i];
-				continue;
-			}
-
-			/* Prewarm buffer. */
-			buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-									NULL);
-
-			if (BufferIsValid(buf))
+				.block_info = block_info,
+					.pos = i,
+					.database = database,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
+
+			/* Receive our prewarmed buffers */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
 			{
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
 
-			blk = &block_info[++i];
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+
+			i = p.pos;
+			blk = &block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 449bafc123c..a36e372526c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From b6e8454d632e22ed39db0b343e7be4e9b91900ea Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Wed, 2 Apr 2025 09:52:54 -0400
Subject: [PATCH v9 1/3] Autoprewarm global objects separately

Autoprewarm previously prewarmed global objects while prewarming blocks
from objects from the first valid database it encountered. This was
because you can't read in buffers without being connected to a database.

Prewarming global objects and objects from a single database in the
autoprewarm_database_main() function required a special case. Once we
convert autoprewarm to use a read stream, this special case will have to
be duplicated in multiple places.

Instead, modify apw_load_buffers() to prewarm the shared objects in one
invocation of autoprewarm_database_main() while connected to the first
valid database.

It is a bit fiddly but seems better than the alternative.
---
 contrib/pg_prewarm/autoprewarm.c | 51 ++++++++++++++------------------
 1 file changed, 22 insertions(+), 29 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..172654fee25 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -347,44 +347,41 @@ apw_load_buffers(void)
 	apw_state->prewarm_start_idx = apw_state->prewarm_stop_idx = 0;
 	apw_state->prewarmed_blocks = 0;
 
-	/* Get the info position of the first block of the next database. */
+	/*
+	 * Loop through the records and launch a database worker to process
+	 * objects in each database. We'll stop at the boundary of each new
+	 * database and prewarm those blocks before moving to the next.
+	 */
 	while (apw_state->prewarm_start_idx < num_elements)
 	{
 		int			j = apw_state->prewarm_start_idx;
 		Oid			current_db = blkinfo[j].database;
 
 		/*
-		 * Advance the prewarm_stop_idx to the first BlockInfoRecord that does
-		 * not belong to this database.
+		 * Advance the position to the first BlockInfoRecord that does not
+		 * belong to the current database.
 		 */
-		j++;
-		while (j < num_elements)
+		for (; j < num_elements; j++)
 		{
-			if (current_db != blkinfo[j].database)
-			{
-				/*
-				 * Combine BlockInfoRecords for global objects with those of
-				 * the database.
-				 */
-				if (current_db != InvalidOid)
-					break;
-				current_db = blkinfo[j].database;
-			}
-
-			j++;
+			if (blkinfo[j].database != current_db)
+				break;
 		}
 
 		/*
-		 * If we reach this point with current_db == InvalidOid, then only
-		 * BlockInfoRecords belonging to global objects exist.  We can't
-		 * prewarm without a database connection, so just bail out.
+		 * We can't prewarm without a database connection, so if all of the
+		 * records belong to global objects, we have to bail out.
 		 */
-		if (current_db == InvalidOid)
+		if (current_db == InvalidOid && blkinfo[j].database == InvalidOid)
 			break;
 
+		/* Connect to the first valid db to prewarm global objects. */
+		if (current_db == InvalidOid)
+			current_db = blkinfo[j].database;
+
 		/* Configure stop point and database for next per-database worker. */
 		apw_state->prewarm_stop_idx = j;
 		apw_state->database = current_db;
+
 		Assert(apw_state->prewarm_start_idx < apw_state->prewarm_stop_idx);
 
 		/* If we've run out of free buffers, don't launch another worker. */
@@ -423,8 +420,8 @@ apw_load_buffers(void)
 }
 
 /*
- * Prewarm all blocks for one database (and possibly also global objects, if
- * those got grouped with this database).
+ * Prewarm all blocks for one database or global objects (while connected to a
+ * valid database).
  */
 void
 autoprewarm_database_main(Datum main_arg)
@@ -462,12 +459,8 @@ autoprewarm_database_main(Datum main_arg)
 
 		CHECK_FOR_INTERRUPTS();
 
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
+		/* Quit if we've reached records for another database. */
+		if (old_blk != NULL && old_blk->database != blk->database)
 			break;
 
 		/*
-- 
2.34.1

From e21c0bf3f16dfdd531d22e7da7167ae9a864c6cd Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH v9 3/3] Use streaming read I/O in autoprewarm

Make a read stream for each valid fork of each valid relation
represented in the autoprewarm dump file and prewarm those blocks
through the read stream API instead of by directly invoking
ReadBuffer().

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 105 +++++++++++++++++++++++++++----
 src/tools/pgindent/typedefs.list |   1 +
 2 files changed, 93 insertions(+), 13 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index fb132fbb533..e85931e9c1a 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,28 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	/* The array of records containing the blocks we should prewarm. */
+	BlockInfoRecord *block_info;
+
+	/*
+	 * `pos` is the read stream callback's index into block_info. Because the
+	 * read stream may read ahead, pos is likely to be ahead of the index in
+	 * the main loop in autoprewarm_database_main().
+	 */
+	int			pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -419,6 +442,45 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+	BlockInfoRecord blk;
+
+	CHECK_FOR_INTERRUPTS();
+
+	if (!have_free_buffer())
+		p->pos = apw_state->prewarm_stop_idx;
+
+	if (p->pos >= apw_state->prewarm_stop_idx)
+		return InvalidBlockNumber;
+
+	blk = p->block_info[p->pos];
+
+	if (blk.database != p->database)
+		return InvalidBlockNumber;
+
+	if (blk.filenumber != p->filenumber)
+		return InvalidBlockNumber;
+
+	if (blk.forknum != p->forknum)
+		return InvalidBlockNumber;
+
+	/* Check whether blocknum is valid and within fork file size. */
+	if (blk.blocknum >= p->nblocks)
+		return InvalidBlockNumber;
+
+	p->pos++;
+	return blk.blocknum;
+}
+
 /*
  * Prewarm all blocks for one database or global objects (while connected to a
  * valid database).
@@ -467,8 +529,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		StartTransactionCommand();
 
 		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
@@ -509,6 +569,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk->forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -538,24 +600,41 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-			/* Check whether blocknum is valid and within fork file size. */
-			if (blk->blocknum >= nblocks)
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				blk = &block_info[++i];
-				continue;
-			}
+				.block_info = block_info,
+					.pos = i,
+					.database = database,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
 
-			/* Prewarm buffer. */
-			buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-									 NULL);
-
-			if (BufferIsValid(buf))
+			/*
+			 * Loop until we've prewarmed all the blocks from this fork. The
+			 * read stream callback will check that we still have free buffers
+			 * before requesting each block from the read stream API.
+			 */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
 			{
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
 
-			blk = &block_info[++i];
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+
+			/* Advance i past all the blocks just prewarmed. */
+			i = p.pos;
+			blk = &block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 449bafc123c..a36e372526c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From 8c539abb2a3d9d23fb97e28bd722b85453cf2897 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH v9 2/3] Refactor autoprewarm_database_main() in preparation
 for read stream

The read stream API requires the user to provide a callback that can be
invoked multiple times returning the next block to read.

Autoprewarm uses a sorted array of BlockInfORecords representing each block we
want to try and prewarm from all databases and tables.

autoprewarm_database_main() prewarms all the blocks from a single
database. It is optimized to ensure we don't try to open the same
relation or fork over and over again if it has been dropped or is
invalid. The main loop handles this by carefully setting various local
variables to sentinel values when a run of blocks should be skipped,
allowing it to still only advance one spot in the array on each
iteration.

This method no longer works when using the read stream API. The callback
must be able to advance the position in the array multiple times. Even
duplicating the entirety of the logic in autoprewarm_database_main() in
the read stream callback was not enough to make this continue to work.

Instead, change autoprewarm_database_main() to explicitly fast-forward
in the array past the blocks belonging to an invalid relation or
fork. This makes the structure of autoprewarm_database_main()

loop through each valid relation
   loop through each valid fork
      loop through each valid buffer

This commit only implements the new control flow -- it does not use the
read stream API.

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 165 +++++++++++++++++--------------
 1 file changed, 92 insertions(+), 73 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 172654fee25..fb132fbb533 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -426,12 +426,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord *blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -446,104 +445,124 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	blk = &block_info[i];
+
+	/*
+	 * apw_state->database may differ from blk->database if we are prewarming
+	 * blocks from global objects.
+	 */
+	database = blk->database;
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
+	 * buffers. We'll quit if we've reached records for another database,
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx &&
+		   blk->database == database &&
+		   have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		RelFileNumber filenumber = blk->filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
-		/* Quit if we've reached records for another database. */
-		if (old_blk != NULL && old_blk->database != blk->database)
-			break;
+		StartTransactionCommand();
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
+			/* We failed to open the relation, so there is nothing to close. */
 			CommitTransactionCommand();
-		}
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
-
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = &block_info[i];
+				if (blk->database != database ||
+					blk->filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our new found relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk->database == database &&
+			   blk->filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk->forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk->forknum <= InvalidForkNumber ||
+				blk->forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = &block_info[i];
+					if (blk->database != database ||
+						blk->filenumber != filenumber ||
+						blk->forknum != forknum)
+						break;
+				}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+				continue;
+			}
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-		old_blk = blk;
-	}
+			/* Check whether blocknum is valid and within fork file size. */
+			if (blk->blocknum >= nblocks)
+			{
+				blk = &block_info[++i];
+				continue;
+			}
 
-	dsm_detach(seg);
+			/* Prewarm buffer. */
+			buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
+									 NULL);
+
+			if (BufferIsValid(buf))
+			{
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+
+			blk = &block_info[++i];
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From 96f754d1c65d95cefaf3568d1f1ea0f8f0a0b69c Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH v10 2/3] Refactor autoprewarm_database_main() in preparation
 for read stream

Autoprewarm prewarms blocks from a dump file representing the contents
of shared buffers last time it was dumped. It uses a sorted array of
BlockInfoRecords, each representing a block from one of the cluster's
databases and tables.

autoprewarm_database_main() prewarms all the blocks from a single
database. It is optimized to ensure we don't try to open the same
relation or fork over and over again if it has been dropped or is
invalid. The main loop handled this by carefully setting various local
variables to sentinel values when a run of blocks should be skipped.

This method won't work with the read stream API. A read stream can only
be created for a single relation and fork combination. The callback has
to be able to advance the position in the array to allow for reading
ahead additional blocks, however the callback cannot try to open another
relation or close the current relation. So, the main loop in
autoprewarm_database_main() must also advance the position in the array
of BlockInfoRecords.

To make it compatible with the read stream API, change
autoprewarm_database_main() to explicitly fast-forward in the array past
the blocks belonging to an invalid relation or fork.

This commit only implements the new control flow -- it does not use the
read stream API.

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 174 ++++++++++++++++++-------------
 1 file changed, 101 insertions(+), 73 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 172654fee25..0722d7bf457 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -426,12 +426,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord *blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -446,104 +445,133 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	blk = &block_info[i];
+
+	/*
+	 * apw_state->database may differ from blk->database if we are prewarming
+	 * blocks from global objects.
+	 */
+	database = blk->database;
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
+	 * buffers. We'll quit if we've reached records for another database,
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx &&
+		   blk->database == database &&
+		   have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		RelFileNumber filenumber = blk->filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
-		/* Quit if we've reached records for another database. */
-		if (old_blk != NULL && old_blk->database != blk->database)
-			break;
+		StartTransactionCommand();
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
+			/* We failed to open the relation, so there is nothing to close. */
 			CommitTransactionCommand();
-		}
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
-
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = &block_info[i];
+				if (blk->database != database ||
+					blk->filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our new found relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk->database == database &&
+			   blk->filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk->forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk->forknum <= InvalidForkNumber ||
+				blk->forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk->forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = &block_info[i];
+					if (blk->database != database ||
+						blk->filenumber != filenumber ||
+						blk->forknum != forknum)
+						break;
+				}
+
+				/* Time to check if this newfound fork is valid */
+				continue;
+			}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			/* Check whether blocknum is valid and within fork file size. */
+			if (blk->blocknum >= nblocks)
+			{
+				blk = &block_info[++i];
+				continue;
+			}
 
-		old_blk = blk;
-	}
+			/* Prewarm buffers. */
+			while (i < apw_state->prewarm_stop_idx &&
+				   blk->database == database &&
+				   blk->filenumber == filenumber &&
+				   blk->forknum == forknum &&
+				   have_free_buffer())
+			{
+				CHECK_FOR_INTERRUPTS();
 
-	dsm_detach(seg);
+				buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
+										 NULL);
+
+				blk = &block_info[++i];
+				if (!BufferIsValid(buf))
+					break;
+
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From a945fd408a6c3e073394df5332790a6a0fcf5efe Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Wed, 2 Apr 2025 09:52:54 -0400
Subject: [PATCH v10 1/3] Autoprewarm global objects separately

Autoprewarm previously prewarmed global objects while prewarming blocks
from objects from the first valid database it encountered. This was
because you can't read in buffers without being connected to a database.

Prewarming global objects and objects from a single database in the
autoprewarm_database_main() function required a special case. Once we
convert autoprewarm to use a read stream, this special case will have to
be duplicated in multiple places.

Instead, modify apw_load_buffers() to prewarm the shared objects in one
invocation of autoprewarm_database_main() while connected to the first
valid database.

It is a bit fiddly but seems better than the alternative.

Reviewed-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Discussion: https://postgr.es/m/CAN55FZ0TBBmrJ2vtMQ9rEk-NTL2BWQzavVp%3DiRLOUskm%2BzvNNw%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 51 ++++++++++++++------------------
 1 file changed, 22 insertions(+), 29 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..172654fee25 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -347,44 +347,41 @@ apw_load_buffers(void)
 	apw_state->prewarm_start_idx = apw_state->prewarm_stop_idx = 0;
 	apw_state->prewarmed_blocks = 0;
 
-	/* Get the info position of the first block of the next database. */
+	/*
+	 * Loop through the records and launch a database worker to process
+	 * objects in each database. We'll stop at the boundary of each new
+	 * database and prewarm those blocks before moving to the next.
+	 */
 	while (apw_state->prewarm_start_idx < num_elements)
 	{
 		int			j = apw_state->prewarm_start_idx;
 		Oid			current_db = blkinfo[j].database;
 
 		/*
-		 * Advance the prewarm_stop_idx to the first BlockInfoRecord that does
-		 * not belong to this database.
+		 * Advance the position to the first BlockInfoRecord that does not
+		 * belong to the current database.
 		 */
-		j++;
-		while (j < num_elements)
+		for (; j < num_elements; j++)
 		{
-			if (current_db != blkinfo[j].database)
-			{
-				/*
-				 * Combine BlockInfoRecords for global objects with those of
-				 * the database.
-				 */
-				if (current_db != InvalidOid)
-					break;
-				current_db = blkinfo[j].database;
-			}
-
-			j++;
+			if (blkinfo[j].database != current_db)
+				break;
 		}
 
 		/*
-		 * If we reach this point with current_db == InvalidOid, then only
-		 * BlockInfoRecords belonging to global objects exist.  We can't
-		 * prewarm without a database connection, so just bail out.
+		 * We can't prewarm without a database connection, so if all of the
+		 * records belong to global objects, we have to bail out.
 		 */
-		if (current_db == InvalidOid)
+		if (current_db == InvalidOid && blkinfo[j].database == InvalidOid)
 			break;
 
+		/* Connect to the first valid db to prewarm global objects. */
+		if (current_db == InvalidOid)
+			current_db = blkinfo[j].database;
+
 		/* Configure stop point and database for next per-database worker. */
 		apw_state->prewarm_stop_idx = j;
 		apw_state->database = current_db;
+
 		Assert(apw_state->prewarm_start_idx < apw_state->prewarm_stop_idx);
 
 		/* If we've run out of free buffers, don't launch another worker. */
@@ -423,8 +420,8 @@ apw_load_buffers(void)
 }
 
 /*
- * Prewarm all blocks for one database (and possibly also global objects, if
- * those got grouped with this database).
+ * Prewarm all blocks for one database or global objects (while connected to a
+ * valid database).
  */
 void
 autoprewarm_database_main(Datum main_arg)
@@ -462,12 +459,8 @@ autoprewarm_database_main(Datum main_arg)
 
 		CHECK_FOR_INTERRUPTS();
 
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
+		/* Quit if we've reached records for another database. */
+		if (old_blk != NULL && old_blk->database != blk->database)
 			break;
 
 		/*
-- 
2.34.1

From 3221de7b11f0b1df281b5c9076a639f0731d83bf Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH v10 3/3] Use streaming read I/O in autoprewarm

Make a read stream for each valid fork of each valid relation
represented in the autoprewarm dump file and prewarm those blocks
through the read stream API instead of by directly invoking
ReadBuffer().

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 125 +++++++++++++++++++++++++------
 src/tools/pgindent/typedefs.list |   1 +
 2 files changed, 104 insertions(+), 22 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 0722d7bf457..5d21314120f 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,28 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	/* The array of records containing the blocks we should prewarm. */
+	BlockInfoRecord *block_info;
+
+	/*
+	 * `pos` is the read stream callback's index into block_info. Because the
+	 * read stream may read ahead, pos is likely to be ahead of the index in
+	 * the main loop in autoprewarm_database_main().
+	 */
+	int			pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -419,6 +442,55 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+
+	while (p->pos < apw_state->prewarm_stop_idx)
+	{
+		BlockInfoRecord blk = p->block_info[p->pos];
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!have_free_buffer())
+			p->pos = apw_state->prewarm_stop_idx;
+
+		if (p->pos >= apw_state->prewarm_stop_idx)
+			return InvalidBlockNumber;
+
+		if (blk.database != p->database)
+			return InvalidBlockNumber;
+
+		if (blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		p->pos++;
+
+		/*
+		 * Check whether blocknum is valid and within fork file size.
+		 * Fast-forward through any invalid blocks. We want `p->pos` to
+		 * reflect the location of the next relation or fork before ending the
+		 * stream.
+		 */
+		if (blk.blocknum >= p->nblocks)
+			continue;
+
+		return blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database or global objects (while connected to a
  * valid database).
@@ -467,8 +539,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		StartTransactionCommand();
 
 		reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
@@ -509,6 +579,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk->forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -539,32 +611,41 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
 
-			/* Check whether blocknum is valid and within fork file size. */
-			if (blk->blocknum >= nblocks)
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				blk = &block_info[++i];
-				continue;
-			}
+				.block_info = block_info,
+					.pos = i,
+					.database = database,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
 
-			/* Prewarm buffers. */
-			while (i < apw_state->prewarm_stop_idx &&
-				   blk->database == database &&
-				   blk->filenumber == filenumber &&
-				   blk->forknum == forknum &&
-				   have_free_buffer())
+			/*
+			 * Loop until we've prewarmed all the blocks from this fork. The
+			 * read stream callback will check that we still have free buffers
+			 * before requesting each block from the read stream API.
+			 */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
 			{
-				CHECK_FOR_INTERRUPTS();
-
-				buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-										 NULL);
-
-				blk = &block_info[++i];
-				if (!BufferIsValid(buf))
-					break;
-
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
+
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+
+			/* Advance i past all the blocks just prewarmed. */
+			i = p.pos;
+			blk = &block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8f28d8ff28e..5ac290fae78 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From 3b1657a0ba51299fa0384a0032759f3661a2baca Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH v11 2/3] Refactor autoprewarm_database_main() in preparation
 for read stream

Autoprewarm prewarms blocks from a dump file representing the contents
of shared buffers last time it was dumped. It uses a sorted array of
BlockInfoRecords, each representing a block from one of the cluster's
databases and tables.

autoprewarm_database_main() prewarms all the blocks from a single
database. It is optimized to ensure we don't try to open the same
relation or fork over and over again if it has been dropped or is
invalid. The main loop handled this by carefully setting various local
variables to sentinel values when a run of blocks should be skipped.

This method won't work with the read stream API. A read stream can only
be created for a single relation and fork combination. The callback has
to be able to advance the position in the array to allow for reading
ahead additional blocks, however the callback cannot try to open another
relation or close the current relation. So, the main loop in
autoprewarm_database_main() must also advance the position in the array
of BlockInfoRecords.

To make it compatible with the read stream API, change
autoprewarm_database_main() to explicitly fast-forward in the array past
the blocks belonging to an invalid relation or fork.

This commit only implements the new control flow -- it does not use the
read stream API.

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 174 ++++++++++++++++++-------------
 1 file changed, 101 insertions(+), 73 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 172654fee25..b02859d8ec3 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -426,12 +426,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -446,104 +445,133 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	blk = block_info[i];
+
+	/*
+	 * apw_state->database may differ from blk.database if we are prewarming
+	 * blocks from global objects.
+	 */
+	database = blk.database;
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
+	 * buffers. We'll quit if we've reached records for another database,
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx &&
+		   blk.database == database &&
+		   have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		RelFileNumber filenumber = blk.filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
-		/* Quit if we've reached records for another database. */
-		if (old_blk != NULL && old_blk->database != blk->database)
-			break;
+		StartTransactionCommand();
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
+			/* We failed to open the relation, so there is nothing to close. */
 			CommitTransactionCommand();
-		}
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
-
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = block_info[i];
+				if (blk.database != database ||
+					blk.filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our new found relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk.database == database &&
+			   blk.filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk.forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk.forknum <= InvalidForkNumber ||
+				blk.forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk.forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = block_info[i];
+					if (blk.database != database ||
+						blk.filenumber != filenumber ||
+						blk.forknum != forknum)
+						break;
+				}
+
+				/* Time to check if this newfound fork is valid */
+				continue;
+			}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			/* Check whether blocknum is valid and within fork file size. */
+			if (blk.blocknum >= nblocks)
+			{
+				blk = block_info[++i];
+				continue;
+			}
 
-		old_blk = blk;
-	}
+			/* Prewarm buffers. */
+			while (i < apw_state->prewarm_stop_idx &&
+				   blk.database == database &&
+				   blk.filenumber == filenumber &&
+				   blk.forknum == forknum &&
+				   have_free_buffer())
+			{
+				CHECK_FOR_INTERRUPTS();
 
-	dsm_detach(seg);
+				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
+										 NULL);
+
+				blk = block_info[++i];
+				if (!BufferIsValid(buf))
+					break;
+
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From dbfb2fe6b9f319efd951da93ea9c1f350bac8815 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Wed, 2 Apr 2025 09:52:54 -0400
Subject: [PATCH v11 1/3] Autoprewarm global objects separately

Autoprewarm previously prewarmed global objects while prewarming blocks
from objects from the first valid database it encountered. This was
because you can't read in buffers without being connected to a database.

Prewarming global objects and objects from a single database in the
autoprewarm_database_main() function required a special case. Once we
convert autoprewarm to use a read stream, this special case will have to
be duplicated in multiple places.

Instead, modify apw_load_buffers() to prewarm the shared objects in one
invocation of autoprewarm_database_main() while connected to the first
valid database.

It is a bit fiddly but seems better than the alternative.

Reviewed-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Discussion: https://postgr.es/m/CAN55FZ0TBBmrJ2vtMQ9rEk-NTL2BWQzavVp%3DiRLOUskm%2BzvNNw%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 51 ++++++++++++++------------------
 1 file changed, 22 insertions(+), 29 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..172654fee25 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -347,44 +347,41 @@ apw_load_buffers(void)
 	apw_state->prewarm_start_idx = apw_state->prewarm_stop_idx = 0;
 	apw_state->prewarmed_blocks = 0;
 
-	/* Get the info position of the first block of the next database. */
+	/*
+	 * Loop through the records and launch a database worker to process
+	 * objects in each database. We'll stop at the boundary of each new
+	 * database and prewarm those blocks before moving to the next.
+	 */
 	while (apw_state->prewarm_start_idx < num_elements)
 	{
 		int			j = apw_state->prewarm_start_idx;
 		Oid			current_db = blkinfo[j].database;
 
 		/*
-		 * Advance the prewarm_stop_idx to the first BlockInfoRecord that does
-		 * not belong to this database.
+		 * Advance the position to the first BlockInfoRecord that does not
+		 * belong to the current database.
 		 */
-		j++;
-		while (j < num_elements)
+		for (; j < num_elements; j++)
 		{
-			if (current_db != blkinfo[j].database)
-			{
-				/*
-				 * Combine BlockInfoRecords for global objects with those of
-				 * the database.
-				 */
-				if (current_db != InvalidOid)
-					break;
-				current_db = blkinfo[j].database;
-			}
-
-			j++;
+			if (blkinfo[j].database != current_db)
+				break;
 		}
 
 		/*
-		 * If we reach this point with current_db == InvalidOid, then only
-		 * BlockInfoRecords belonging to global objects exist.  We can't
-		 * prewarm without a database connection, so just bail out.
+		 * We can't prewarm without a database connection, so if all of the
+		 * records belong to global objects, we have to bail out.
 		 */
-		if (current_db == InvalidOid)
+		if (current_db == InvalidOid && blkinfo[j].database == InvalidOid)
 			break;
 
+		/* Connect to the first valid db to prewarm global objects. */
+		if (current_db == InvalidOid)
+			current_db = blkinfo[j].database;
+
 		/* Configure stop point and database for next per-database worker. */
 		apw_state->prewarm_stop_idx = j;
 		apw_state->database = current_db;
+
 		Assert(apw_state->prewarm_start_idx < apw_state->prewarm_stop_idx);
 
 		/* If we've run out of free buffers, don't launch another worker. */
@@ -423,8 +420,8 @@ apw_load_buffers(void)
 }
 
 /*
- * Prewarm all blocks for one database (and possibly also global objects, if
- * those got grouped with this database).
+ * Prewarm all blocks for one database or global objects (while connected to a
+ * valid database).
  */
 void
 autoprewarm_database_main(Datum main_arg)
@@ -462,12 +459,8 @@ autoprewarm_database_main(Datum main_arg)
 
 		CHECK_FOR_INTERRUPTS();
 
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
+		/* Quit if we've reached records for another database. */
+		if (old_blk != NULL && old_blk->database != blk->database)
 			break;
 
 		/*
-- 
2.34.1

From bb812c2d92fe120171d20a284ce21546c5836f29 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH v11 3/3] Use streaming read I/O in autoprewarm

Make a read stream for each valid fork of each valid relation
represented in the autoprewarm dump file and prewarm those blocks
through the read stream API instead of by directly invoking
ReadBuffer().

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 125 +++++++++++++++++++++++++------
 src/tools/pgindent/typedefs.list |   1 +
 2 files changed, 104 insertions(+), 22 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index b02859d8ec3..4b35a19eec1 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,28 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	/* The array of records containing the blocks we should prewarm. */
+	BlockInfoRecord *block_info;
+
+	/*
+	 * `pos` is the read stream callback's index into block_info. Because the
+	 * read stream may read ahead, pos is likely to be ahead of the index in
+	 * the main loop in autoprewarm_database_main().
+	 */
+	int			pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -419,6 +442,55 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+
+	while (p->pos < apw_state->prewarm_stop_idx)
+	{
+		BlockInfoRecord blk = p->block_info[p->pos];
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!have_free_buffer())
+			p->pos = apw_state->prewarm_stop_idx;
+
+		if (p->pos >= apw_state->prewarm_stop_idx)
+			return InvalidBlockNumber;
+
+		if (blk.database != p->database)
+			return InvalidBlockNumber;
+
+		if (blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		p->pos++;
+
+		/*
+		 * Check whether blocknum is valid and within fork file size.
+		 * Fast-forward through any invalid blocks. We want `p->pos` to
+		 * reflect the location of the next relation or fork before ending the
+		 * stream.
+		 */
+		if (blk.blocknum >= p->nblocks)
+			continue;
+
+		return blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database or global objects (while connected to a
  * valid database).
@@ -467,8 +539,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		StartTransactionCommand();
 
 		reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
@@ -509,6 +579,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk.forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -539,32 +611,41 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-			/* Check whether blocknum is valid and within fork file size. */
-			if (blk.blocknum >= nblocks)
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				blk = block_info[++i];
-				continue;
-			}
+				.block_info = block_info,
+					.pos = i,
+					.database = database,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
 
-			/* Prewarm buffers. */
-			while (i < apw_state->prewarm_stop_idx &&
-				   blk.database == database &&
-				   blk.filenumber == filenumber &&
-				   blk.forknum == forknum &&
-				   have_free_buffer())
+			/*
+			 * Loop until we've prewarmed all the blocks from this fork. The
+			 * read stream callback will check that we still have free buffers
+			 * before requesting each block from the read stream API.
+			 */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
 			{
-				CHECK_FOR_INTERRUPTS();
-
-				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
-										 NULL);
-
-				blk = block_info[++i];
-				if (!BufferIsValid(buf))
-					break;
-
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
+
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+
+			/* Advance i past all the blocks just prewarmed. */
+			i = p.pos;
+			blk = block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8f28d8ff28e..5ac290fae78 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From 7fde68340b0a0b80e4e6d2de5c1771a4cd5c7bfe Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH v12 3/3] Use streaming read I/O in autoprewarm

Make a read stream for each valid fork of each valid relation
represented in the autoprewarm dump file and prewarm those blocks
through the read stream API instead of by directly invoking
ReadBuffer().

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 127 +++++++++++++++++++++++++------
 src/tools/pgindent/typedefs.list |   1 +
 2 files changed, 105 insertions(+), 23 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index fe79c8c71bf..4b35a19eec1 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,28 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	/* The array of records containing the blocks we should prewarm. */
+	BlockInfoRecord *block_info;
+
+	/*
+	 * `pos` is the read stream callback's index into block_info. Because the
+	 * read stream may read ahead, pos is likely to be ahead of the index in
+	 * the main loop in autoprewarm_database_main().
+	 */
+	int			pos;
+	Oid			database;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -419,6 +442,55 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+
+	while (p->pos < apw_state->prewarm_stop_idx)
+	{
+		BlockInfoRecord blk = p->block_info[p->pos];
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!have_free_buffer())
+			p->pos = apw_state->prewarm_stop_idx;
+
+		if (p->pos >= apw_state->prewarm_stop_idx)
+			return InvalidBlockNumber;
+
+		if (blk.database != p->database)
+			return InvalidBlockNumber;
+
+		if (blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		p->pos++;
+
+		/*
+		 * Check whether blocknum is valid and within fork file size.
+		 * Fast-forward through any invalid blocks. We want `p->pos` to
+		 * reflect the location of the next relation or fork before ending the
+		 * stream.
+		 */
+		if (blk.blocknum >= p->nblocks)
+			continue;
+
+		return blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database or global objects (while connected to a
  * valid database).
@@ -467,8 +539,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		StartTransactionCommand();
 
 		reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
@@ -509,6 +579,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk.forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -539,32 +611,41 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-			/* Prewarm buffers. */
-			while (i < apw_state->prewarm_stop_idx &&
-				   blk.database == database &&
-				   blk.filenumber == filenumber &&
-				   blk.forknum == forknum &&
-				   have_free_buffer())
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				CHECK_FOR_INTERRUPTS();
-
-				/* Check whether blocknum is valid and within fork file size. */
-				if (blk.blocknum >= nblocks)
-				{
-					blk = block_info[++i];
-					continue;
-				}
-
-				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
-										 NULL);
-
-				blk = block_info[++i];
-				if (!BufferIsValid(buf))
-					break;
+				.block_info = block_info,
+					.pos = i,
+					.database = database,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
 
+			/*
+			 * Loop until we've prewarmed all the blocks from this fork. The
+			 * read stream callback will check that we still have free buffers
+			 * before requesting each block from the read stream API.
+			 */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
+			{
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
+
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+
+			/* Advance i past all the blocks just prewarmed. */
+			i = p.pos;
+			blk = block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8f28d8ff28e..5ac290fae78 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From dbfb2fe6b9f319efd951da93ea9c1f350bac8815 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Wed, 2 Apr 2025 09:52:54 -0400
Subject: [PATCH v12 1/3] Autoprewarm global objects separately

Autoprewarm previously prewarmed global objects while prewarming blocks
from objects from the first valid database it encountered. This was
because you can't read in buffers without being connected to a database.

Prewarming global objects and objects from a single database in the
autoprewarm_database_main() function required a special case. Once we
convert autoprewarm to use a read stream, this special case will have to
be duplicated in multiple places.

Instead, modify apw_load_buffers() to prewarm the shared objects in one
invocation of autoprewarm_database_main() while connected to the first
valid database.

It is a bit fiddly but seems better than the alternative.

Reviewed-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Discussion: https://postgr.es/m/CAN55FZ0TBBmrJ2vtMQ9rEk-NTL2BWQzavVp%3DiRLOUskm%2BzvNNw%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 51 ++++++++++++++------------------
 1 file changed, 22 insertions(+), 29 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..172654fee25 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -347,44 +347,41 @@ apw_load_buffers(void)
 	apw_state->prewarm_start_idx = apw_state->prewarm_stop_idx = 0;
 	apw_state->prewarmed_blocks = 0;
 
-	/* Get the info position of the first block of the next database. */
+	/*
+	 * Loop through the records and launch a database worker to process
+	 * objects in each database. We'll stop at the boundary of each new
+	 * database and prewarm those blocks before moving to the next.
+	 */
 	while (apw_state->prewarm_start_idx < num_elements)
 	{
 		int			j = apw_state->prewarm_start_idx;
 		Oid			current_db = blkinfo[j].database;
 
 		/*
-		 * Advance the prewarm_stop_idx to the first BlockInfoRecord that does
-		 * not belong to this database.
+		 * Advance the position to the first BlockInfoRecord that does not
+		 * belong to the current database.
 		 */
-		j++;
-		while (j < num_elements)
+		for (; j < num_elements; j++)
 		{
-			if (current_db != blkinfo[j].database)
-			{
-				/*
-				 * Combine BlockInfoRecords for global objects with those of
-				 * the database.
-				 */
-				if (current_db != InvalidOid)
-					break;
-				current_db = blkinfo[j].database;
-			}
-
-			j++;
+			if (blkinfo[j].database != current_db)
+				break;
 		}
 
 		/*
-		 * If we reach this point with current_db == InvalidOid, then only
-		 * BlockInfoRecords belonging to global objects exist.  We can't
-		 * prewarm without a database connection, so just bail out.
+		 * We can't prewarm without a database connection, so if all of the
+		 * records belong to global objects, we have to bail out.
 		 */
-		if (current_db == InvalidOid)
+		if (current_db == InvalidOid && blkinfo[j].database == InvalidOid)
 			break;
 
+		/* Connect to the first valid db to prewarm global objects. */
+		if (current_db == InvalidOid)
+			current_db = blkinfo[j].database;
+
 		/* Configure stop point and database for next per-database worker. */
 		apw_state->prewarm_stop_idx = j;
 		apw_state->database = current_db;
+
 		Assert(apw_state->prewarm_start_idx < apw_state->prewarm_stop_idx);
 
 		/* If we've run out of free buffers, don't launch another worker. */
@@ -423,8 +420,8 @@ apw_load_buffers(void)
 }
 
 /*
- * Prewarm all blocks for one database (and possibly also global objects, if
- * those got grouped with this database).
+ * Prewarm all blocks for one database or global objects (while connected to a
+ * valid database).
  */
 void
 autoprewarm_database_main(Datum main_arg)
@@ -462,12 +459,8 @@ autoprewarm_database_main(Datum main_arg)
 
 		CHECK_FOR_INTERRUPTS();
 
-		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
-		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
+		/* Quit if we've reached records for another database. */
+		if (old_blk != NULL && old_blk->database != blk->database)
 			break;
 
 		/*
-- 
2.34.1

From 322e4eb8f54d3f62bdac9a5ca5a6473aa4a53531 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH v12 2/3] Refactor autoprewarm_database_main() in preparation
 for read stream

Autoprewarm prewarms blocks from a dump file representing the contents
of shared buffers last time it was dumped. It uses a sorted array of
BlockInfoRecords, each representing a block from one of the cluster's
databases and tables.

autoprewarm_database_main() prewarms all the blocks from a single
database. It is optimized to ensure we don't try to open the same
relation or fork over and over again if it has been dropped or is
invalid. The main loop handled this by carefully setting various local
variables to sentinel values when a run of blocks should be skipped.

This method won't work with the read stream API. A read stream can only
be created for a single relation and fork combination. The callback has
to be able to advance the position in the array to allow for reading
ahead additional blocks, however the callback cannot try to open another
relation or close the current relation. So, the main loop in
autoprewarm_database_main() must also advance the position in the array
of BlockInfoRecords.

To make it compatible with the read stream API, change
autoprewarm_database_main() to explicitly fast-forward in the array past
the blocks belonging to an invalid relation or fork.

This commit only implements the new control flow -- it does not use the
read stream API.

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 174 ++++++++++++++++++-------------
 1 file changed, 101 insertions(+), 73 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 172654fee25..fe79c8c71bf 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -426,12 +426,11 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord blk;
 	dsm_segment *seg;
+	Oid			database;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
 	pqsignal(SIGTERM, die);
@@ -446,104 +445,133 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	blk = block_info[i];
+
+	/*
+	 * apw_state->database may differ from blk.database if we are prewarming
+	 * blocks from global objects.
+	 */
+	database = blk.database;
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
-	 * buffers.
+	 * buffers. We'll quit if we've reached records for another database,
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx &&
+		   blk.database == database &&
+		   have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		RelFileNumber filenumber = blk.filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
-		/* Quit if we've reached records for another database. */
-		if (old_blk != NULL && old_blk->database != blk->database)
-			break;
+		StartTransactionCommand();
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
-			rel != NULL)
+		reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
+			/* We failed to open the relation, so there is nothing to close. */
 			CommitTransactionCommand();
-		}
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
-		{
-			Oid			reloid;
-
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = block_info[i];
+				if (blk.database != database ||
+					blk.filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our new found relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk.database == database &&
+			   blk.filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk.forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk.forknum <= InvalidForkNumber ||
+				blk.forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk.forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = block_info[i];
+					if (blk.database != database ||
+						blk.filenumber != filenumber ||
+						blk.forknum != forknum)
+						break;
+				}
+
+				/* Time to check if this newfound fork is valid */
+				continue;
+			}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			/* Prewarm buffers. */
+			while (i < apw_state->prewarm_stop_idx &&
+				   blk.database == database &&
+				   blk.filenumber == filenumber &&
+				   blk.forknum == forknum &&
+				   have_free_buffer())
+			{
+				CHECK_FOR_INTERRUPTS();
 
-		old_blk = blk;
-	}
+				/* Check whether blocknum is valid and within fork file size. */
+				if (blk.blocknum >= nblocks)
+				{
+					blk = block_info[++i];
+					continue;
+				}
 
-	dsm_detach(seg);
+				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
+										 NULL);
+
+				blk = block_info[++i];
+				if (!BufferIsValid(buf))
+					break;
+
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From 9d5e4bf6a05d0a3f2838dd9efa8715b05be77423 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Thu, 3 Apr 2025 12:47:19 -0400
Subject: [PATCH v13 1/4] Fix autoprewarm neglect of tablespaces

While prewarming blocks from a dump file, autoprewarm_database_main()
mistakenly ignored tablespace when detecting the beginning of the next
relation to prewarm. Because RelFileNumbers are only unqiue within a
tablespace, autoprewarm could miss prewarming blocks from a
relation with the same RelFileNumber in a different tablespace.

Though this situation is likely rare in practice, it's best to make the
code correct. Do so by explicitly checking for the RelFileNumber when
detecting a new relation.

Reported-by: Heikki Linnakangas <hlinnaka@iki.fi>
Discussion: https://postgr.es/m/97c36982-603b-494a-95f4-aaf2a12ac27e%40iki.fi
---
 contrib/pg_prewarm/autoprewarm.c | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..760b1548eff 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -472,10 +472,15 @@ autoprewarm_database_main(Datum main_arg)
 
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
+		 * relation. RelFileNumbers are only guaranteed to be unique within a
+		 * tablespace, so check that too.
+		 *
+		 * Note that rel will be NULL if try_relation_open failed previously;
+		 * in that case, there is nothing to close.
 		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
+		if (old_blk != NULL &&
+			(old_blk->tablespace != blk->tablespace ||
+			 old_blk->filenumber != blk->filenumber) &&
 			rel != NULL)
 		{
 			relation_close(rel, AccessShareLock);
@@ -487,7 +492,9 @@ autoprewarm_database_main(Datum main_arg)
 		 * Try to open each new relation, but only once, when we first
 		 * encounter it. If it's been dropped, skip the associated blocks.
 		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
+		if (old_blk == NULL ||
+			old_blk->tablespace != blk->tablespace ||
+			old_blk->filenumber != blk->filenumber)
 		{
 			Oid			reloid;
 
@@ -508,6 +515,7 @@ autoprewarm_database_main(Datum main_arg)
 
 		/* Once per fork, check for fork existence and size. */
 		if (old_blk == NULL ||
+			old_blk->tablespace != blk->tablespace ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
 		{
-- 
2.34.1

From 8879473d355dba5c9397c398240db2e1efb81a4c Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Thu, 3 Apr 2025 14:54:09 -0400
Subject: [PATCH v13 2/4] Remove superfluous autoprewarm check

autoprewarm_database_main() prewarms blocks from the same database. It
is passed an array of sorted BlockInfoRecords and a start and stop index
into the array. The range represented should include only blocks
belonging to global objects or blocks from a single database. Remove an
unnecessary check that the current block is from the same database and
add an assert to ensure this invariant remains.
---
 contrib/pg_prewarm/autoprewarm.c | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 760b1548eff..5f6dca57cdd 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -463,12 +463,10 @@ autoprewarm_database_main(Datum main_arg)
 		CHECK_FOR_INTERRUPTS();
 
 		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
+		 * All blocks between prewarm_start_idx and prewarm_stop_idx should
+		 * belong either to global objects or the same database.
 		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		Assert(blk->database == apw_state->database || blk->database == 0);
 
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
-- 
2.34.1

From 54c9d4db9e3c3bddcb02cb515a1dd79ba4a83cf2 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH v13 4/4] Use streaming read I/O in autoprewarm

Make a read stream for each valid fork of each valid relation
represented in the autoprewarm dump file and prewarm those blocks
through the read stream API instead of by directly invoking
ReadBuffer().

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 127 +++++++++++++++++++++++++------
 src/tools/pgindent/typedefs.list |   1 +
 2 files changed, 105 insertions(+), 23 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 0d59fd62e93..37014e648f4 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,28 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	/* The array of records containing the blocks we should prewarm. */
+	BlockInfoRecord *block_info;
+
+	/*
+	 * `pos` is the read stream callback's index into block_info. Because the
+	 * read stream may read ahead, pos is likely to be ahead of the index in
+	 * the main loop in autoprewarm_database_main().
+	 */
+	int			pos;
+	Oid			tablespace;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -422,6 +445,55 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+
+	while (p->pos < apw_state->prewarm_stop_idx)
+	{
+		BlockInfoRecord blk = p->block_info[p->pos];
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!have_free_buffer())
+		{
+			p->pos = apw_state->prewarm_stop_idx;
+			return InvalidBlockNumber;
+		}
+
+		if (blk.tablespace != p->tablespace)
+			return InvalidBlockNumber;
+
+		if (blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		p->pos++;
+
+		/*
+		 * Check whether blocknum is valid and within fork file size.
+		 * Fast-forward through any invalid blocks. We want `p->pos` to
+		 * reflect the location of the next relation or fork before ending the
+		 * stream.
+		 */
+		if (blk.blocknum >= p->nblocks)
+			continue;
+
+		return blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -462,8 +534,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		/*
 		 * All blocks between prewarm_start_idx and prewarm_stop_idx should
 		 * belong either to global objects or the same database.
@@ -510,6 +580,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk.forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -540,32 +612,41 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-			/* Prewarm buffers. */
-			while (i < apw_state->prewarm_stop_idx &&
-				   blk.tablespace == tablespace &&
-				   blk.filenumber == filenumber &&
-				   blk.forknum == forknum &&
-				   have_free_buffer())
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				CHECK_FOR_INTERRUPTS();
-
-				/* Check whether blocknum is valid and within fork file size. */
-				if (blk.blocknum >= nblocks)
-				{
-					blk = block_info[++i];
-					continue;
-				}
-
-				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
-										 NULL);
-
-				blk = block_info[++i];
-				if (!BufferIsValid(buf))
-					break;
+				.block_info = block_info,
+					.pos = i,
+					.tablespace = tablespace,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
 
+			/*
+			 * Loop until we've prewarmed all the blocks from this fork. The
+			 * read stream callback will check that we still have free buffers
+			 * before requesting each block from the read stream API.
+			 */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
+			{
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
+
+			Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+			read_stream_end(stream);
+
+			/* Advance i past all the blocks just prewarmed. */
+			i = p.pos;
+			blk = block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8f28d8ff28e..5ac290fae78 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From a86d64437239b9453143c85a895c7f19b4ab3cb1 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH v13 3/4] Refactor autoprewarm_database_main() in preparation
 for read stream

Autoprewarm prewarms blocks from a dump file representing the contents
of shared buffers last time it was dumped. It uses a sorted array of
BlockInfoRecords, each representing a block from one of the cluster's
databases and tables.

autoprewarm_database_main() prewarms all the blocks from a single
database. It is optimized to ensure we don't try to open the same
relation or fork over and over again if it has been dropped or is
invalid. The main loop handled this by carefully setting various local
variables to sentinel values when a run of blocks should be skipped.

This method won't work with the read stream API. A read stream can only
be created for a single relation and fork combination. The callback has
to be able to advance the position in the array to allow for reading
ahead additional blocks, however the callback cannot try to open another
relation or close the current relation. So, the main loop in
autoprewarm_database_main() must also advance the position in the array
of BlockInfoRecords.

To make it compatible with the read stream API, change
autoprewarm_database_main() to explicitly fast-forward in the array past
the blocks belonging to an invalid relation or fork.

This commit only implements the new control flow -- it does not use the
read stream API.

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 172 +++++++++++++++++--------------
 1 file changed, 94 insertions(+), 78 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 5f6dca57cdd..0d59fd62e93 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -429,11 +429,9 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord blk;
 	dsm_segment *seg;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
@@ -449,16 +447,20 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	blk = block_info[i];
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx && have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		Oid			tablespace = blk.tablespace;
+		RelFileNumber filenumber = blk.filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -466,97 +468,111 @@ autoprewarm_database_main(Datum main_arg)
 		 * All blocks between prewarm_start_idx and prewarm_stop_idx should
 		 * belong either to global objects or the same database.
 		 */
-		Assert(blk->database == apw_state->database || blk->database == 0);
+		Assert(blk.database == apw_state->database || blk.database == 0);
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. RelFileNumbers are only guaranteed to be unique within a
-		 * tablespace, so check that too.
-		 *
-		 * Note that rel will be NULL if try_relation_open failed previously;
-		 * in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL &&
-			(old_blk->tablespace != blk->tablespace ||
-			 old_blk->filenumber != blk->filenumber) &&
-			rel != NULL)
-		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
-			CommitTransactionCommand();
-		}
+		StartTransactionCommand();
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL ||
-			old_blk->tablespace != blk->tablespace ||
-			old_blk->filenumber != blk->filenumber)
+		reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			Oid			reloid;
+			/* We failed to open the relation, so there is nothing to close. */
+			CommitTransactionCommand();
 
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = block_info[i];
+				if ( blk.tablespace != tablespace ||
+					blk.filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our new found relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->tablespace != blk->tablespace ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk.tablespace == tablespace &&
+			   blk.filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk.forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk.forknum <= InvalidForkNumber ||
+				blk.forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk.forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = block_info[i];
+					if (blk.tablespace != tablespace ||
+						blk.filenumber != filenumber ||
+						blk.forknum != forknum)
+						break;
+				}
+
+				/* Time to check if this newfound fork is valid */
+				continue;
+			}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			/* Prewarm buffers. */
+			while (i < apw_state->prewarm_stop_idx &&
+				   blk.tablespace == tablespace &&
+				   blk.filenumber == filenumber &&
+				   blk.forknum == forknum &&
+				   have_free_buffer())
+			{
+				CHECK_FOR_INTERRUPTS();
 
-		old_blk = blk;
-	}
+				/* Check whether blocknum is valid and within fork file size. */
+				if (blk.blocknum >= nblocks)
+				{
+					blk = block_info[++i];
+					continue;
+				}
 
-	dsm_detach(seg);
+				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
+										 NULL);
+
+				blk = block_info[++i];
+				if (!BufferIsValid(buf))
+					break;
+
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From 11f706609998aebbeac06e282a6e1d25d3f38fdd Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Thu, 3 Apr 2025 12:47:19 -0400
Subject: [PATCH v14 1/4] Fix autoprewarm neglect of tablespaces

While prewarming blocks from a dump file, autoprewarm_database_main()
mistakenly ignored tablespace when detecting the beginning of the next
relation to prewarm. Because RelFileNumbers are only unique within a
tablespace, autoprewarm could miss prewarming blocks from a
relation with the same RelFileNumber in a different tablespace.

Though this situation is likely rare in practice, it's best to make the
code correct. Do so by explicitly checking for the RelFileNumber when
detecting a new relation.

Reported-by: Heikki Linnakangas <hlinnaka@iki.fi>
Discussion: https://postgr.es/m/97c36982-603b-494a-95f4-aaf2a12ac27e%40iki.fi
---
 contrib/pg_prewarm/autoprewarm.c | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 73485a2323c..760b1548eff 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -472,10 +472,15 @@ autoprewarm_database_main(Datum main_arg)
 
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. Note that rel will be NULL if try_relation_open failed
-		 * previously; in that case, there is nothing to close.
+		 * relation. RelFileNumbers are only guaranteed to be unique within a
+		 * tablespace, so check that too.
+		 *
+		 * Note that rel will be NULL if try_relation_open failed previously;
+		 * in that case, there is nothing to close.
 		 */
-		if (old_blk != NULL && old_blk->filenumber != blk->filenumber &&
+		if (old_blk != NULL &&
+			(old_blk->tablespace != blk->tablespace ||
+			 old_blk->filenumber != blk->filenumber) &&
 			rel != NULL)
 		{
 			relation_close(rel, AccessShareLock);
@@ -487,7 +492,9 @@ autoprewarm_database_main(Datum main_arg)
 		 * Try to open each new relation, but only once, when we first
 		 * encounter it. If it's been dropped, skip the associated blocks.
 		 */
-		if (old_blk == NULL || old_blk->filenumber != blk->filenumber)
+		if (old_blk == NULL ||
+			old_blk->tablespace != blk->tablespace ||
+			old_blk->filenumber != blk->filenumber)
 		{
 			Oid			reloid;
 
@@ -508,6 +515,7 @@ autoprewarm_database_main(Datum main_arg)
 
 		/* Once per fork, check for fork existence and size. */
 		if (old_blk == NULL ||
+			old_blk->tablespace != blk->tablespace ||
 			old_blk->filenumber != blk->filenumber ||
 			old_blk->forknum != blk->forknum)
 		{
-- 
2.34.1

From 915fe5b0db6ea8599bc21f95980a352ae404463d Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Mon, 31 Mar 2025 22:02:25 -0400
Subject: [PATCH v14 3/4] Refactor autoprewarm_database_main() in preparation
 for read stream

Autoprewarm prewarms blocks from a dump file representing the contents
of shared buffers at the time it was dumped. It uses a sorted array of
BlockInfoRecords, each representing a block from one of the cluster's
databases and tables.

autoprewarm_database_main() prewarms all the blocks from a single
database. It is optimized to ensure we don't try to open the same
relation or fork over and over again if it has been dropped or is
invalid. The main loop handled this by carefully setting various local
variables to sentinel values when a run of blocks should be skipped.

This method won't work with the read stream API. A read stream can only
be created for a single relation and fork combination. The callback has
to be able to advance the position in the array to allow for reading
ahead additional blocks, however the callback cannot try to open another
relation or close the current relation. So, the main loop in
autoprewarm_database_main() must also advance the position in the array
of BlockInfoRecords.

To make it compatible with the read stream API, change
autoprewarm_database_main() to explicitly fast-forward in the array past
the blocks belonging to an invalid relation or fork.

This commit only implements the new control flow -- it does not use the
read stream API.

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Reviewed-by: Heikki Linnakangas <hlinnaka@iki.fi>
Reviewed-by: Daniel Gustafsson <daniel@yesql.se>
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 172 +++++++++++++++++--------------
 1 file changed, 94 insertions(+), 78 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 5f6dca57cdd..761f6a77926 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -429,11 +429,9 @@ apw_load_buffers(void)
 void
 autoprewarm_database_main(Datum main_arg)
 {
-	int			pos;
 	BlockInfoRecord *block_info;
-	Relation	rel = NULL;
-	BlockNumber nblocks = 0;
-	BlockInfoRecord *old_blk = NULL;
+	int			i;
+	BlockInfoRecord blk;
 	dsm_segment *seg;
 
 	/* Establish signal handlers; once that's done, unblock signals. */
@@ -449,16 +447,20 @@ autoprewarm_database_main(Datum main_arg)
 				 errmsg("could not map dynamic shared memory segment")));
 	BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
 	block_info = (BlockInfoRecord *) dsm_segment_address(seg);
-	pos = apw_state->prewarm_start_idx;
+
+	i = apw_state->prewarm_start_idx;
+	blk = block_info[i];
 
 	/*
 	 * Loop until we run out of blocks to prewarm or until we run out of free
 	 * buffers.
 	 */
-	while (pos < apw_state->prewarm_stop_idx && have_free_buffer())
+	while (i < apw_state->prewarm_stop_idx && have_free_buffer())
 	{
-		BlockInfoRecord *blk = &block_info[pos++];
-		Buffer		buf;
+		Oid			tablespace = blk.tablespace;
+		RelFileNumber filenumber = blk.filenumber;
+		Oid			reloid;
+		Relation	rel;
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -466,97 +468,111 @@ autoprewarm_database_main(Datum main_arg)
 		 * All blocks between prewarm_start_idx and prewarm_stop_idx should
 		 * belong either to global objects or the same database.
 		 */
-		Assert(blk->database == apw_state->database || blk->database == 0);
+		Assert(blk.database == apw_state->database || blk.database == 0);
 
-		/*
-		 * As soon as we encounter a block of a new relation, close the old
-		 * relation. RelFileNumbers are only guaranteed to be unique within a
-		 * tablespace, so check that too.
-		 *
-		 * Note that rel will be NULL if try_relation_open failed previously;
-		 * in that case, there is nothing to close.
-		 */
-		if (old_blk != NULL &&
-			(old_blk->tablespace != blk->tablespace ||
-			 old_blk->filenumber != blk->filenumber) &&
-			rel != NULL)
-		{
-			relation_close(rel, AccessShareLock);
-			rel = NULL;
-			CommitTransactionCommand();
-		}
+		StartTransactionCommand();
 
-		/*
-		 * Try to open each new relation, but only once, when we first
-		 * encounter it. If it's been dropped, skip the associated blocks.
-		 */
-		if (old_blk == NULL ||
-			old_blk->tablespace != blk->tablespace ||
-			old_blk->filenumber != blk->filenumber)
+		reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
+		if (!OidIsValid(reloid) ||
+			(rel = try_relation_open(reloid, AccessShareLock)) == NULL)
 		{
-			Oid			reloid;
+			/* We failed to open the relation, so there is nothing to close. */
+			CommitTransactionCommand();
 
-			Assert(rel == NULL);
-			StartTransactionCommand();
-			reloid = RelidByRelfilenumber(blk->tablespace, blk->filenumber);
-			if (OidIsValid(reloid))
-				rel = try_relation_open(reloid, AccessShareLock);
+			/*
+			 * Fast-forward to the next relation. We want to skip all of the
+			 * other records referencing this relation since we know we can't
+			 * open it. That way, we avoid repeatedly trying and failing to
+			 * open the same relation.
+			 */
+			for (; i < apw_state->prewarm_stop_idx; i++)
+			{
+				blk = block_info[i];
+				if (blk.tablespace != tablespace ||
+					blk.filenumber != filenumber)
+					break;
+			}
 
-			if (!rel)
-				CommitTransactionCommand();
-		}
-		if (!rel)
-		{
-			old_blk = blk;
+			/* Time to try and open our newfound relation */
 			continue;
 		}
 
-		/* Once per fork, check for fork existence and size. */
-		if (old_blk == NULL ||
-			old_blk->tablespace != blk->tablespace ||
-			old_blk->filenumber != blk->filenumber ||
-			old_blk->forknum != blk->forknum)
+		/*
+		 * We have a relation; now let's loop until we find a valid fork of
+		 * the relation or we run out of free buffers. Once we've read from
+		 * all valid forks or run out of options, we'll close the relation and
+		 * move on.
+		 */
+		while (i < apw_state->prewarm_stop_idx &&
+			   blk.tablespace == tablespace &&
+			   blk.filenumber == filenumber &&
+			   have_free_buffer())
 		{
+			ForkNumber	forknum = blk.forknum;
+			BlockNumber nblocks;
+			Buffer		buf;
+
 			/*
 			 * smgrexists is not safe for illegal forknum, hence check whether
 			 * the passed forknum is valid before using it in smgrexists.
 			 */
-			if (blk->forknum > InvalidForkNumber &&
-				blk->forknum <= MAX_FORKNUM &&
-				smgrexists(RelationGetSmgr(rel), blk->forknum))
-				nblocks = RelationGetNumberOfBlocksInFork(rel, blk->forknum);
-			else
-				nblocks = 0;
-		}
+			if (blk.forknum <= InvalidForkNumber ||
+				blk.forknum > MAX_FORKNUM ||
+				!smgrexists(RelationGetSmgr(rel), blk.forknum))
+			{
+				/*
+				 * Fast-forward to the next fork. We want to skip all of the
+				 * other records referencing this fork since we already know
+				 * it's not valid.
+				 */
+				for (; i < apw_state->prewarm_stop_idx; i++)
+				{
+					blk = block_info[i];
+					if (blk.tablespace != tablespace ||
+						blk.filenumber != filenumber ||
+						blk.forknum != forknum)
+						break;
+				}
+
+				/* Time to check if this newfound fork is valid */
+				continue;
+			}
 
-		/* Check whether blocknum is valid and within fork file size. */
-		if (blk->blocknum >= nblocks)
-		{
-			/* Move to next forknum. */
-			old_blk = blk;
-			continue;
-		}
+			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-		/* Prewarm buffer. */
-		buf = ReadBufferExtended(rel, blk->forknum, blk->blocknum, RBM_NORMAL,
-								 NULL);
-		if (BufferIsValid(buf))
-		{
-			apw_state->prewarmed_blocks++;
-			ReleaseBuffer(buf);
-		}
+			/* Prewarm buffers. */
+			while (i < apw_state->prewarm_stop_idx &&
+				   blk.tablespace == tablespace &&
+				   blk.filenumber == filenumber &&
+				   blk.forknum == forknum &&
+				   have_free_buffer())
+			{
+				CHECK_FOR_INTERRUPTS();
 
-		old_blk = blk;
-	}
+				/* Check whether blocknum is valid and within fork file size. */
+				if (blk.blocknum >= nblocks)
+				{
+					blk = block_info[++i];
+					continue;
+				}
 
-	dsm_detach(seg);
+				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
+										 NULL);
+
+				blk = block_info[++i];
+				if (!BufferIsValid(buf))
+					break;
+
+				apw_state->prewarmed_blocks++;
+				ReleaseBuffer(buf);
+			}
+		}
 
-	/* Release lock on previous relation. */
-	if (rel)
-	{
 		relation_close(rel, AccessShareLock);
 		CommitTransactionCommand();
 	}
+
+	dsm_detach(seg);
 }
 
 /*
-- 
2.34.1

From b259afd02235ac7189e64ec2d54d59c851eab921 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Tue, 1 Apr 2025 18:07:38 -0400
Subject: [PATCH v14 4/4] Use streaming read I/O in autoprewarm

Make a read stream for each valid fork of each valid relation
represented in the autoprewarm dump file and prewarm those blocks
through the read stream API instead of by directly invoking
ReadBuffer().

Co-authored-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Reviewed-by: Heikki Linnakangas <hlinnaka@iki.fi>
Reviewed-by: Daniel Gustafsson <daniel@yesql.se>
Reviewed-by: Andrey M. Borodin <x4mmm@yandex-team.ru> (earlier versions)
Reviewed-by: Kirill Reshke <reshkekirill@gmail.com>  (earlier versions)
Reviewed-by: Matheus Alcantara <mths.dev@pm.me> (earlier versions)
Discussion: https://postgr.es/m/flat/CAN55FZ3n8Gd%2BhajbL%3D5UkGzu_aHGRqnn%2BxktXq2fuds%3D1AOR6Q%40mail.gmail.com
---
 contrib/pg_prewarm/autoprewarm.c | 125 +++++++++++++++++++++++++------
 src/tools/pgindent/typedefs.list |   1 +
 2 files changed, 103 insertions(+), 23 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 761f6a77926..195dc4b773b 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -41,6 +41,7 @@
 #include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "storage/procsignal.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc.h"
@@ -75,6 +76,28 @@ typedef struct AutoPrewarmSharedState
 	int			prewarmed_blocks;
 } AutoPrewarmSharedState;
 
+/*
+ * Private data passed through the read stream API for our use in the
+ * callback.
+ */
+typedef struct AutoPrewarmReadStreamData
+{
+	/* The array of records containing the blocks we should prewarm. */
+	BlockInfoRecord *block_info;
+
+	/*
+	 * pos is the read stream callback's index into block_info. Because the
+	 * read stream may read ahead, pos is likely to be ahead of the index in
+	 * the main loop in autoprewarm_database_main().
+	 */
+	int			pos;
+	Oid			tablespace;
+	RelFileNumber filenumber;
+	ForkNumber	forknum;
+	BlockNumber nblocks;
+} AutoPrewarmReadStreamData;
+
+
 PGDLLEXPORT void autoprewarm_main(Datum main_arg);
 PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
 
@@ -422,6 +445,54 @@ apw_load_buffers(void)
 						apw_state->prewarmed_blocks, num_elements)));
 }
 
+/*
+ * Return the next block number of a specific relation and fork to read
+ * according to the array of BlockInfoRecord.
+ */
+static BlockNumber
+apw_read_stream_next_block(ReadStream *stream,
+						   void *callback_private_data,
+						   void *per_buffer_data)
+{
+	AutoPrewarmReadStreamData *p = callback_private_data;
+
+	CHECK_FOR_INTERRUPTS();
+
+	while (p->pos < apw_state->prewarm_stop_idx)
+	{
+		BlockInfoRecord blk = p->block_info[p->pos];
+
+		if (!have_free_buffer())
+		{
+			p->pos = apw_state->prewarm_stop_idx;
+			return InvalidBlockNumber;
+		}
+
+		if (blk.tablespace != p->tablespace)
+			return InvalidBlockNumber;
+
+		if (blk.filenumber != p->filenumber)
+			return InvalidBlockNumber;
+
+		if (blk.forknum != p->forknum)
+			return InvalidBlockNumber;
+
+		p->pos++;
+
+		/*
+		 * Check whether blocknum is valid and within fork file size.
+		 * Fast-forward through any invalid blocks. We want p->pos to reflect
+		 * the location of the next relation or fork before ending the stream.
+		 */
+		if (blk.blocknum >= p->nblocks)
+			continue;
+
+		return blk.blocknum;
+	}
+
+	return InvalidBlockNumber;
+}
+
 /*
  * Prewarm all blocks for one database (and possibly also global objects, if
  * those got grouped with this database).
@@ -462,8 +533,6 @@ autoprewarm_database_main(Datum main_arg)
 		Oid			reloid;
 		Relation	rel;
 
-		CHECK_FOR_INTERRUPTS();
-
 		/*
 		 * All blocks between prewarm_start_idx and prewarm_stop_idx should
 		 * belong either to global objects or the same database.
@@ -510,6 +579,8 @@ autoprewarm_database_main(Datum main_arg)
 		{
 			ForkNumber	forknum = blk.forknum;
 			BlockNumber nblocks;
+			struct AutoPrewarmReadStreamData p;
+			ReadStream *stream;
 			Buffer		buf;
 
 			/*
@@ -540,32 +611,40 @@ autoprewarm_database_main(Datum main_arg)
 
 			nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
 
-			/* Prewarm buffers. */
-			while (i < apw_state->prewarm_stop_idx &&
-				   blk.tablespace == tablespace &&
-				   blk.filenumber == filenumber &&
-				   blk.forknum == forknum &&
-				   have_free_buffer())
+			p = (struct AutoPrewarmReadStreamData)
 			{
-				CHECK_FOR_INTERRUPTS();
-
-				/* Check whether blocknum is valid and within fork file size. */
-				if (blk.blocknum >= nblocks)
-				{
-					blk = block_info[++i];
-					continue;
-				}
-
-				buf = ReadBufferExtended(rel, blk.forknum, blk.blocknum, RBM_NORMAL,
-										 NULL);
-
-				blk = block_info[++i];
-				if (!BufferIsValid(buf))
-					break;
+				.block_info = block_info,
+					.pos = i,
+					.tablespace = tablespace,
+					.filenumber = filenumber,
+					.forknum = forknum,
+					.nblocks = nblocks,
+			};
+
+			stream = read_stream_begin_relation(READ_STREAM_FULL,
+												NULL,
+												rel,
+												p.forknum,
+												apw_read_stream_next_block,
+												&p,
+												0);
 
+			/*
+			 * Loop until we've prewarmed all the blocks from this fork. The
+			 * read stream callback will check that we still have free buffers
+			 * before requesting each block from the read stream API.
+			 */
+			while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
+			{
 				apw_state->prewarmed_blocks++;
 				ReleaseBuffer(buf);
 			}
+
+			read_stream_end(stream);
+
+			/* Advance i past all the blocks just prewarmed. */
+			i = p.pos;
+			blk = block_info[i];
 		}
 
 		relation_close(rel, AccessShareLock);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8f28d8ff28e..5ac290fae78 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -175,6 +175,7 @@ AttributeOpts
 AuthRequest
 AuthToken
 AutoPrewarmSharedState
+AutoPrewarmReadStreamData
 AutoVacOpts
 AutoVacuumShmemStruct
 AutoVacuumWorkItem
-- 
2.34.1

From 5a72bd6d00f2068e8f03b956dcafcf2fe3be8987 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplageman@gmail.com>
Date: Thu, 3 Apr 2025 14:54:09 -0400
Subject: [PATCH v14 2/4] Remove superfluous autoprewarm check

autoprewarm_database_main() prewarms blocks from the same database. It
is passed an array of sorted BlockInfoRecords and a start and stop index
into the array. The range represented should include only blocks
belonging to global objects or blocks from a single database. Remove an
unnecessary check that the current block is from the same database and
add an assert to ensure this invariant remains.
---
 contrib/pg_prewarm/autoprewarm.c | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index 760b1548eff..5f6dca57cdd 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -463,12 +463,10 @@ autoprewarm_database_main(Datum main_arg)
 		CHECK_FOR_INTERRUPTS();
 
 		/*
-		 * Quit if we've reached records for another database. If previous
-		 * blocks are of some global objects, then continue pre-warming.
+		 * All blocks between prewarm_start_idx and prewarm_stop_idx should
+		 * belong either to global objects or the same database.
 		 */
-		if (old_blk != NULL && old_blk->database != blk->database &&
-			old_blk->database != 0)
-			break;
+		Assert(blk->database == apw_state->database || blk->database == 0);
 
 		/*
 		 * As soon as we encounter a block of a new relation, close the old
-- 
2.34.1