diff --git a/contrib/pg_buffercache/pg_buffercache_pages.c b/contrib/pg_buffercache/pg_buffercache_pages.c
index 905f14e0a04..1b0004ab42c 100644
--- a/contrib/pg_buffercache/pg_buffercache_pages.c
+++ b/contrib/pg_buffercache/pg_buffercache_pages.c
@@ -60,6 +60,8 @@ typedef struct
 	BufferCacheOsPagesRec *record;
 } BufferCacheOsPagesContext;
 
+static TupleDesc build_buffercache_pages_tupledesc(int natts);
+
 
 /*
  * Function returning data from the shared buffer cache - buffer number,
@@ -87,86 +89,20 @@ pg_buffercache_pages(PG_FUNCTION_ARGS)
 {
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	expected_tupledesc;
+	TupleDesc	actual_tupledesc;
+	MemoryContext oldcontext;
 	int			i;
 
-	if (SRF_IS_FIRSTCALL())
-	{
-		int			i;
-		BEGIN_NBUFFERS_ACCESS(localNBuffers);
-
-		funcctx = SRF_FIRSTCALL_INIT();
-
-		/* Switch context when allocating stuff to be used in later calls */
-		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
-
-		/* Create a user function context for cross-call persistence */
-		fctx = (BufferCachePagesContext *) palloc(sizeof(BufferCachePagesContext));
-
-		/*
-		 * To smoothly support upgrades from version 1.0 of this extension
-		 * transparently handle the (non-)existence of the pinning_backends
-		 * column. We unfortunately have to get the result type for that... -
-		 * we can't use the result type determined by the function definition
-		 * without potentially crashing when somebody uses the old (or even
-		 * wrong) function definition though.
-		 */
-		if (get_call_result_type(fcinfo, NULL, &expected_tupledesc) != TYPEFUNC_COMPOSITE)
-			elog(ERROR, "return type must be a row type");
-
-		if (expected_tupledesc->natts < NUM_BUFFERCACHE_PAGES_MIN_ELEM ||
-			expected_tupledesc->natts > NUM_BUFFERCACHE_PAGES_ELEM)
-			elog(ERROR, "incorrect number of output arguments");
-
-		/* Construct a tuple descriptor for the result rows. */
-		tupledesc = CreateTemplateTupleDesc(expected_tupledesc->natts);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 1, "bufferid",
-						   INT4OID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 2, "relfilenode",
-						   OIDOID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 3, "reltablespace",
-						   OIDOID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 4, "reldatabase",
-						   OIDOID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 5, "relforknumber",
-						   INT2OID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 6, "relblocknumber",
-						   INT8OID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 7, "isdirty",
-						   BOOLOID, -1, 0);
-		TupleDescInitEntry(tupledesc, (AttrNumber) 8, "usage_count",
-						   INT2OID, -1, 0);
-
-		if (expected_tupledesc->natts == NUM_BUFFERCACHE_PAGES_ELEM)
-			TupleDescInitEntry(tupledesc, (AttrNumber) 9, "pinning_backends",
-							   INT4OID, -1, 0);
-
-		fctx->tupdesc = BlessTupleDesc(tupledesc);
-
-
-		/* Allocate NBuffers worth of BufferCachePagesRec records. */
-		fctx->record = (BufferCachePagesRec *)
-			MemoryContextAllocHuge(CurrentMemoryContext,
-								   sizeof(BufferCachePagesRec) * localNBuffers);
-
-		/* Set max calls and remember the user function context. */
-		funcctx->max_calls = localNBuffers;
-		funcctx->user_fctx = fctx;
-
-		/* Return to original context when allocating transient memory */
-		MemoryContextSwitchTo(oldcontext);
-
-		/*
-		 * Scan through all the buffers, saving the relevant fields in the
-		 * fctx->record structure.
-		 *
-		 * We don't hold the partition locks, so we don't get a consistent
-		 * snapshot across all buffers, but we do grab the buffer header
-		 * locks, so the information of each buffer is self-consistent.
-		 */
-		for (i = 0; i < localNBuffers; i++)
-		{
-			BufferDesc *bufHdr;
-			uint32		buf_state;
+	/*
+	 * To smoothly support upgrades from version 1.0 of this extension
+	 * transparently handle the (non-)existence of the pinning_backends
+	 * column. We unfortunately have to get the result type for that... - we
+	 * can't use the result type determined by the function definition without
+	 * potentially crashing when somebody uses the old (or even wrong)
+	 * function definition though.
+	 */
+	if (get_call_result_type(fcinfo, NULL, &expected_tupledesc) != TYPEFUNC_COMPOSITE)
+		elog(ERROR, "return type must be a row type");
 
 	if (expected_tupledesc->natts < NUM_BUFFERCACHE_PAGES_MIN_ELEM ||
 		expected_tupledesc->natts > NUM_BUFFERCACHE_PAGES_ELEM)
@@ -174,6 +110,21 @@ pg_buffercache_pages(PG_FUNCTION_ARGS)
 
 	InitMaterializedSRF(fcinfo, 0);
 
+	oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
+	actual_tupledesc = build_buffercache_pages_tupledesc(expected_tupledesc->natts);
+	MemoryContextSwitchTo(oldcontext);
+
+	/*
+	 * Override the caller-supplied descriptor with the tuple descriptor that
+	 * matches the values we actually return, so executor-side
+	 * tupledesc_match() can verify the caller's row definition.
+	 *
+	 * Do not free the previous rsinfo->setDesc here: for RECORD results it
+	 * can alias rsinfo->expectedDesc, which the executor still needs to
+	 * reference.
+	 */
+	rsinfo->setDesc = actual_tupledesc;
+
 	/*
 	 * Scan through all the buffers, adding one row for each of the buffers to
 	 * the tuplestore.
@@ -182,7 +133,9 @@ pg_buffercache_pages(PG_FUNCTION_ARGS)
 	 * snapshot across all buffers, but we do grab the buffer header locks, so
 	 * the information of each buffer is self-consistent.
 	 */
-	for (i = 0; i < NBuffers; i++)
+	BEGIN_NBUFFERS_ACCESS(localNBuffers);
+
+	for (i = 0; i < localNBuffers; i++)
 	{
 		BufferDesc *bufHdr;
 		uint64		buf_state;
@@ -201,10 +154,18 @@ pg_buffercache_pages(PG_FUNCTION_ARGS)
 
 		CHECK_FOR_INTERRUPTS();
 
-			UnlockBufHdr(bufHdr, buf_state);
-		}
-		END_NBUFFERS_ACCESS(localNBuffers);
-	}
+		bufHdr = GetBufferDescriptor(i);
+		/* Lock each buffer header before inspecting. */
+		buf_state = LockBufHdr(bufHdr);
+
+		bufferid = BufferDescriptorGetBuffer(bufHdr);
+		relfilenumber = BufTagGetRelNumber(&bufHdr->tag);
+		reltablespace = bufHdr->tag.spcOid;
+		reldatabase = bufHdr->tag.dbOid;
+		forknum = BufTagGetForkNum(&bufHdr->tag);
+		blocknum = bufHdr->tag.blockNum;
+		usagecount = BUF_STATE_GET_USAGECOUNT(buf_state);
+		pinning_backends = BUF_STATE_GET_REFCOUNT(buf_state);
 
 		if (buf_state & BM_DIRTY)
 			isdirty = true;
@@ -262,10 +223,43 @@ pg_buffercache_pages(PG_FUNCTION_ARGS)
 
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
 	}
+	END_NBUFFERS_ACCESS(localNBuffers);
 
 	return (Datum) 0;
 }
 
+static TupleDesc
+build_buffercache_pages_tupledesc(int natts)
+{
+	TupleDesc	tupledesc;
+
+	tupledesc = CreateTemplateTupleDesc(natts);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 1, "bufferid",
+					   INT4OID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 2, "relfilenode",
+					   OIDOID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 3, "reltablespace",
+					   OIDOID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 4, "reldatabase",
+					   OIDOID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 5, "relforknumber",
+					   INT2OID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 6, "relblocknumber",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 7, "isdirty",
+					   BOOLOID, -1, 0);
+	TupleDescInitEntry(tupledesc, (AttrNumber) 8, "usagecount",
+					   INT2OID, -1, 0);
+
+	if (natts == NUM_BUFFERCACHE_PAGES_ELEM)
+		TupleDescInitEntry(tupledesc, (AttrNumber) 9, "pinning_backends",
+						   INT4OID, -1, 0);
+
+	TupleDescFinalize(tupledesc);
+
+	return BlessTupleDesc(tupledesc);
+}
+
 /*
  * Inquire about OS pages mappings for shared buffers, with NUMA information,
  * optionally.
@@ -340,24 +334,7 @@ pg_buffercache_os_pages_internal(FunctionCallInfo fcinfo, bool include_numa)
 		 */
 		Assert((os_page_size % BLCKSZ == 0) || (BLCKSZ % os_page_size == 0));
 
-		/*
-		 * How many addresses we are going to query? Simply get the page for
-		 * the first buffer, and first page after the last buffer, and count
-		 * the pages from that.
-		 */
-		startptr = (char *) TYPEALIGN_DOWN(os_page_size,
-										   BufferGetBlock(1));
-		endptr = (char *) TYPEALIGN(os_page_size,
-									(char *) BufferGetBlock(localNBuffers) + BLCKSZ);
-		os_page_count = (endptr - startptr) / os_page_size;
-
-		/* Used to determine the NUMA node for all OS pages at once */
-		os_page_ptrs = palloc0(sizeof(void *) * os_page_count);
-		os_page_status = palloc(sizeof(int) * os_page_count);
-
-		/* Fill pointers for all the memory pages. */
-		idx = 0;
-		for (char *ptr = startptr; ptr < endptr; ptr += os_page_size)
+		if (include_numa)
 		{
 			void	  **os_page_ptrs = NULL;
 
@@ -369,7 +346,7 @@ pg_buffercache_os_pages_internal(FunctionCallInfo fcinfo, bool include_numa)
 			startptr = (char *) TYPEALIGN_DOWN(os_page_size,
 											   BufferGetBlock(1));
 			endptr = (char *) TYPEALIGN(os_page_size,
-										(char *) BufferGetBlock(NBuffers) + BLCKSZ);
+										(char *) BufferGetBlock(localNBuffers) + BLCKSZ);
 			os_page_count = (endptr - startptr) / os_page_size;
 
 			/* Used to determine the NUMA node for all OS pages at once */
@@ -394,8 +371,8 @@ pg_buffercache_os_pages_internal(FunctionCallInfo fcinfo, bool include_numa)
 
 			Assert(idx == os_page_count);
 
-		elog(DEBUG1, "NUMA: NBuffers=%d os_page_count=" UINT64_FORMAT " "
-			 "os_page_size=%zu", localNBuffers, os_page_count, os_page_size);
+			elog(DEBUG1, "NUMA: NBuffers=%d os_page_count=" UINT64_FORMAT " "
+				 "os_page_size=%zu", localNBuffers, os_page_count, os_page_size);
 
 			/*
 			 * If we ever get 0xff back from kernel inquiry, then we probably
@@ -465,10 +442,6 @@ pg_buffercache_os_pages_internal(FunctionCallInfo fcinfo, bool include_numa)
 		 * We don't hold the partition locks, so we don't get a consistent
 		 * snapshot across all buffers, but we do grab the buffer header
 		 * locks, so the information of each buffer is self-consistent.
-		 *
-		 * This loop touches and stores addresses into os_page_ptrs[] as input
-		 * to one big move_pages(2) inquiry system call. Basically we ask for
-		 * all memory pages for localNBuffers.
 		 */
 		startptr = (char *) TYPEALIGN_DOWN(os_page_size, (char *) BufferGetBlock(1));
 		idx = 0;
@@ -863,13 +836,15 @@ pg_buffercache_mark_dirty(PG_FUNCTION_ARGS)
 
 	Buffer		buf = PG_GETARG_INT32(0);
 	bool		buffer_already_dirty;
+	BEGIN_NBUFFERS_ACCESS(localNBuffers);
+	(void) localNBuffers;
 
 	if (get_call_result_type(fcinfo, NULL, &tupledesc) != TYPEFUNC_COMPOSITE)
 		elog(ERROR, "return type must be a row type");
 
 	pg_buffercache_superuser_check("pg_buffercache_mark_dirty");
 
-	if (buf < 1 || buf > NBuffers)
+	if (buf < 1 || buf > GetLowNBuffers())
 		elog(ERROR, "bad buffer ID: %d", buf);
 
 	values[0] = BoolGetDatum(MarkDirtyUnpinnedBuffer(buf, &buffer_already_dirty));
@@ -878,6 +853,8 @@ pg_buffercache_mark_dirty(PG_FUNCTION_ARGS)
 	tuple = heap_form_tuple(tupledesc, values, nulls);
 	result = HeapTupleGetDatum(tuple);
 
+	END_NBUFFERS_ACCESS(localNBuffers);
+
 	PG_RETURN_DATUM(result);
 }
 
diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index b620c053dc6..dd9bd592053 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -372,12 +372,12 @@ apw_load_buffers(void)
 	apw_state->prewarmed_blocks = 0;
 
 	/* Don't prewarm more than we can fit. */
-	if (num_elements > NBuffers)
+	if (num_elements > GetLowNBuffers())
 	{
-		num_elements = NBuffers;
+		num_elements = GetLowNBuffers();
 		ereport(LOG,
 				(errmsg("autoprewarm capping prewarmed blocks to %d (shared_buffers size)",
-						NBuffers)));
+						GetLowNBuffers())));
 	}
 
 	/* Get the info position of the first block of the next database. */
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index bbbc6c2f95d..3f3a6dec51f 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -998,7 +998,10 @@ CheckpointerShmemRequest(void *arg)
 										   MAX_CHECKPOINT_REQUESTS),
 									   sizeof(CheckpointerRequest)));
 
-	return size;
+	ShmemRequestStruct(.name = "Checkpointer Data",
+					   .size = size,
+					   .ptr = (void **) &CheckpointerShmem,
+		);
 }
 
 /*
@@ -1009,7 +1012,18 @@ static void
 CheckpointerShmemInit(void *arg)
 {
 	SpinLockInit(&CheckpointerShmem->ckpt_lck);
-	CheckpointerShmem->max_requests = Min(NBuffers, MAX_CHECKPOINT_REQUESTS);
+
+	/*
+	 * Keep max_requests consistent with the queue size computed in
+	 * CheckpointerShmemRequest: a small fixed cap under
+	 * dynamic_shared_buffers, otherwise the initial buffer pool size capped
+	 * at MAX_CHECKPOINT_REQUESTS.
+	 */
+	if (enable_dynamic_shared_buffers)
+		CheckpointerShmem->max_requests = DSB_CHECKPOINT_REQUESTS;
+	else
+		CheckpointerShmem->max_requests = Min(NBuffersGUC,
+											  MAX_CHECKPOINT_REQUESTS);
 	CheckpointerShmem->head = CheckpointerShmem->tail = 0;
 	ConditionVariableInit(&CheckpointerShmem->start_cv);
 	ConditionVariableInit(&CheckpointerShmem->done_cv);
@@ -1029,21 +1043,27 @@ ExecCheckpoint(ParseState *pstate, CheckPointStmt *stmt)
 
 	foreach_ptr(DefElem, opt, stmt->options)
 	{
-		/*
-		 * First time through, so initialize.  Note that we zero the whole
-		 * requests array; this is so that CompactCheckpointerRequestQueue can
-		 * assume that any pad bytes in the request structs are zeroes.
-		 */
-		MemSet(CheckpointerShmem, 0, size);
-		SpinLockInit(&CheckpointerShmem->ckpt_lck);
-
-		if (enable_dynamic_shared_buffers)
-			CheckpointerShmem->max_requests = DSB_CHECKPOINT_REQUESTS;
+		if (strcmp(opt->defname, "mode") == 0)
+		{
+			char	   *mode = defGetString(opt);
+
+			if (strcmp(mode, "spread") == 0)
+				fast = false;
+			else if (strcmp(mode, "fast") != 0)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("unrecognized value for %s option \"%s\": \"%s\"",
+								"CHECKPOINT", "mode", mode),
+						 parser_errposition(pstate, opt->location)));
+		}
+		else if (strcmp(opt->defname, "flush_unlogged") == 0)
+			unlogged = defGetBoolean(opt);
 		else
-			CheckpointerShmem->max_requests = Min(NBuffersGUC,
-												  MAX_CHECKPOINT_REQUESTS);
-		ConditionVariableInit(&CheckpointerShmem->start_cv);
-		ConditionVariableInit(&CheckpointerShmem->done_cv);
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("unrecognized %s option \"%s\"",
+							"CHECKPOINT", opt->defname),
+					 parser_errposition(pstate, opt->location)));
 	}
 
 	if (!has_privs_of_role(GetUserId(), ROLE_PG_CHECKPOINT))
diff --git a/src/backend/storage/buffer/buf_init.c b/src/backend/storage/buffer/buf_init.c
index 757c00d03d6..93b569cf5e4 100644
--- a/src/backend/storage/buffer/buf_init.c
+++ b/src/backend/storage/buffer/buf_init.c
@@ -24,7 +24,9 @@
 #include "storage/buf_internals.h"
 #include "storage/bufmgr.h"
 #include "storage/pg_shmem.h"
+#include "storage/proclist.h"
 #include "storage/shmem.h"
+#include "storage/subsystems.h"
 #include "utils/memdebug.h"
 
 BufferDescPadded *BufferDescriptors;
@@ -89,22 +91,23 @@ InitializeBuffer(int buf_id)
 	BufferDesc *buf = GetBufferDescriptor(buf_id);
 
 	ClearBufferTag(&buf->tag);
-	pg_atomic_init_u32(&buf->state, 0);
+	pg_atomic_init_u64(&buf->state, 0);
 	buf->wait_backend_pgprocno = INVALID_PROC_NUMBER;
 	buf->buf_id = buf_id;
 	pgaio_wref_clear(&buf->io_wref);
 
-	LWLockInitialize(BufferDescriptorGetContentLock(buf),
-					 LWTRANCHE_BUFFER_CONTENT);
-
+	proclist_init(&buf->lock_waiters);
 	ConditionVariableInit(BufferDescriptorGetIOCV(buf));
 }
 
 /*
  * Page size used both to lay out the buffer-pool arrays in shared memory and
  * to align the per-slice madvise() ranges issued during expand/shrink.
+ *
+ * Exported (not static) so the shared buffer mapping table in buf_table.c can
+ * lay out and resize its own bucket/entry arrays with the same alignment.
  */
-static Size
+Size
 buffer_pool_madvise_alignment(void)
 {
 #ifdef __linux__
@@ -131,7 +134,7 @@ buffer_pool_madvise_alignment(void)
  * so callers should stop expanding rather than continue and risk a SIGBUS on
  * first touch.
  */
-static Size
+Size
 BufferPoolArrayPhysicalExpand(void *baseptr, Size elem_size,
 							  int lowNBuffers, int highNBuffers,
 							  bool *success)
@@ -211,7 +214,7 @@ BufferPoolArrayPhysicalExpand(void *baseptr, Size elem_size,
  * shrinks don't strand page-aligned spans above highNBuffers (see comment
  * inside).
  */
-static Size
+Size
 BufferPoolArrayPhysicalShrink(void *baseptr, Size elem_size,
 							  int lowNBuffers, int highNBuffers,
 							  bool *success)
@@ -284,12 +287,9 @@ BufferPoolArrayPhysicalShrink(void *baseptr, Size elem_size,
 static void
 BufferManagerShmemRequest(void *arg)
 {
-	bool		foundBufs,
-				foundDescs,
-				foundIOCV,
-				foundBufCkpt;
 	int			max_nbuffers;
 	Size		os_page_size = buffer_pool_madvise_alignment();
+
 	Assert(os_page_size != 0);
 
 	if (enable_dynamic_shared_buffers)
@@ -306,30 +306,40 @@ BufferManagerShmemRequest(void *arg)
 							MaxNBuffers, NBuffersGUC)));
 	}
 
+	/*
+	 * The buffer-pool arrays are sized to their upper bound (GetMaxNBuffers())
+	 * so the pool can grow up to max_shared_buffers without reallocating.
+	 * Under dynamic_shared_buffers only the initial [0, NBuffersGUC) slice is
+	 * populated with physical memory at startup (see BufferManagerShmemInit);
+	 * the rest is faulted in on expand.
+	 *
+	 * Each array is aligned to the OS/huge page boundary and given two extra
+	 * pages of slack so that the page-aligned madvise() ranges issued by
+	 * BufferManagerShmemExpand()/Shrink() always stay within the array's own
+	 * reservation and never spill into a neighbouring one.
+	 */
 	max_nbuffers = GetMaxNBuffers();
 
-	/* Align descriptors for madvise (same granularity as buffer blocks). */
-	BufferDescriptors = (BufferDescPadded *)
-		TYPEALIGN(os_page_size,
-				  ShmemInitStruct("Buffer Descriptors",
-								  max_nbuffers * sizeof(BufferDescPadded) + 2 * os_page_size,
-								  &foundDescs));
+	ShmemRequestStruct(.name = "Buffer Descriptors",
+					   .size = add_size(mul_size(max_nbuffers, sizeof(BufferDescPadded)),
+										mul_size(2, os_page_size)),
+					   .alignment = os_page_size,
+					   .ptr = (void **) &BufferDescriptors,
+		);
 
 	ShmemRequestStruct(.name = "Buffer Blocks",
-					   .size = NBuffers * (Size) BLCKSZ,
-	/* Align buffer pool on IO page size boundary. */
-	BufferBlocks = (char *)
-		TYPEALIGN(os_page_size,
-				ShmemInitStruct("Buffer Blocks",
-					max_nbuffers * (Size) BLCKSZ + 2 * os_page_size,
-								&foundBufs));
-
-	/* Align I/O condition variables for madvise. */
-	BufferIOCVArray = (ConditionVariableMinimallyPadded *)
-		TYPEALIGN(os_page_size,
-				  ShmemInitStruct("Buffer IO Condition Variables",
-								  max_nbuffers * sizeof(ConditionVariableMinimallyPadded) + 2 * os_page_size,
-								  &foundIOCV));
+					   .size = add_size(mul_size(max_nbuffers, (Size) BLCKSZ),
+										mul_size(2, os_page_size)),
+					   .alignment = os_page_size,
+					   .ptr = (void **) &BufferBlocks,
+		);
+
+	ShmemRequestStruct(.name = "Buffer IO Condition Variables",
+					   .size = add_size(mul_size(max_nbuffers, sizeof(ConditionVariableMinimallyPadded)),
+										mul_size(2, os_page_size)),
+					   .alignment = os_page_size,
+					   .ptr = (void **) &BufferIOCVArray,
+		);
 
 	/*
 	 * The array used to sort to-be-checkpointed buffer ids is located in
@@ -338,11 +348,13 @@ BufferManagerShmemRequest(void *arg)
 	 * the checkpointer is restarted, memory allocation failures would be
 	 * painful.
 	 */
-	CkptBufferIds = (CkptSortItem *)
-		TYPEALIGN(os_page_size,
-				  ShmemInitStruct("Checkpoint BufferIds",
-								  max_nbuffers * sizeof(CkptSortItem) + 2 * os_page_size,
-								  &foundBufCkpt));
+	ShmemRequestStruct(.name = "Checkpoint BufferIds",
+					   .size = add_size(mul_size(max_nbuffers, sizeof(CkptSortItem)),
+										mul_size(2, os_page_size)),
+					   .alignment = os_page_size,
+					   .ptr = (void **) &CkptBufferIds,
+		);
+}
 
 /*
  * Initialize shared buffer pool
@@ -353,38 +365,32 @@ BufferManagerShmemRequest(void *arg)
 static void
 BufferManagerShmemInit(void *arg)
 {
-	/*
-	 * Initialize all the buffer headers.
-	 */
-	for (int i = 0; i < NBuffers; i++)
+	int			i;
+
+	if (enable_dynamic_shared_buffers)
 	{
-		int			i;
-
-		if (enable_dynamic_shared_buffers)
-		{
-			bool		success = true;
-
-			/*
-			 * Request physical memory for NBuffersGUC. A madvise failure
-			 * here means we cannot eagerly populate the initial buffer
-			 * pool; rather than start with possibly-unbacked memory and
-			 * SIGBUS on first access, we PANIC so the postmaster fails
-			 * to start cleanly.
-			 */
-			BufferManagerShmemExpand(0, NBuffersGUC, &success);
-			if (!success)
-				elog(PANIC, "could not populate initial shared buffer pool: madvise(MADV_POPULATE_WRITE) failed");
-		}
+		bool		success = true;
 
 		/*
-		 * Initialize all the buffer headers for the active pool size.
-		 * The clock sweep is the sole replacement mechanism, so there is
-		 * no freelist to link them into.
+		 * Request physical memory for the initial [0, NBuffersGUC) slice.  A
+		 * madvise failure here means we cannot eagerly populate the initial
+		 * buffer pool; rather than start with possibly-unbacked memory and
+		 * SIGBUS on first access, we PANIC so the postmaster fails to start
+		 * cleanly.
 		 */
-		for (i = 0; i < NBuffersGUC; i++)
-			InitializeBuffer(i);
+		BufferManagerShmemExpand(0, NBuffersGUC, &success);
+		if (!success)
+			elog(PANIC, "could not populate initial shared buffer pool: madvise(MADV_POPULATE_WRITE) failed");
 	}
 
+	/*
+	 * Initialize all the buffer headers for the active pool size.  The clock
+	 * sweep is the sole replacement mechanism, so there is no freelist to link
+	 * them into.
+	 */
+	for (i = 0; i < NBuffersGUC; i++)
+		InitializeBuffer(i);
+
 	/* Initialize per-backend file flush context */
 	WritebackContextInit(&BackendWritebackContext,
 						 &backend_flush_after);
@@ -393,48 +399,19 @@ BufferManagerShmemInit(void *arg)
 	 * Initialize the DSB water marks. DSBCtrl is NULL in special contexts
 	 * such as the WAL redo process, where DSB is not used.
 	 */
-	if (DSBCtrl != NULL && !foundDescs)
+	if (DSBCtrl != NULL)
 	{
 		pg_atomic_write_u32(&DSBCtrl->lowNBuffers, NBuffersGUC);
 		pg_atomic_write_u32(&DSBCtrl->highNBuffers, NBuffersGUC);
 	}
 }
 
-/*
- * BufferManagerShmemSize
- *
- * All buffer arrays are allocated in the single shared-memory heap.  We size
- * them to GetMaxNBuffers() so the pool fits its upper bound.
- */
-Size
-BufferManagerShmemSize(void)
+static void
+BufferManagerShmemAttach(void *arg)
 {
-	Size		size = 0;
-	Size		os_page_size = buffer_pool_madvise_alignment();
-	int			max_nbuffers = GetMaxNBuffers();
-	Assert(os_page_size != 0);
-
-	/* size of buffer descriptors, plus alignment padding for madvise */
-	size = add_size(size, mul_size(max_nbuffers, sizeof(BufferDescPadded)));
-	size = add_size(size, mul_size(2, os_page_size));
-
-	/* size of data pages, plus alignment padding */
-	size = add_size(size, mul_size(2, os_page_size));
-	size = add_size(size, mul_size(max_nbuffers, (Size) BLCKSZ));
-
-	/* size of stuff controlled by freelist.c */
-	size = add_size(size, StrategyShmemSize());
-
-	/* size of I/O condition variables, plus alignment padding for madvise */
-	size = add_size(size, mul_size(max_nbuffers,
-								   sizeof(ConditionVariableMinimallyPadded)));
-	size = add_size(size, mul_size(2, os_page_size));
-
-	/* size of checkpoint sort array in bufmgr.c, plus alignment padding */
-	size = add_size(size, mul_size(max_nbuffers, sizeof(CkptSortItem)));
-	size = add_size(size, mul_size(2, os_page_size));
-
-	return size;
+	/* Initialize per-backend file flush context */
+	WritebackContextInit(&BackendWritebackContext,
+						 &backend_flush_after);
 }
 
 /*
diff --git a/src/backend/storage/buffer/buf_resize.c b/src/backend/storage/buffer/buf_resize.c
index afb5268f636..b7395c8c0f4 100644
--- a/src/backend/storage/buffer/buf_resize.c
+++ b/src/backend/storage/buffer/buf_resize.c
@@ -37,6 +37,8 @@
 #include "utils/builtins.h"
 #include "utils/guc.h"
 #include "utils/injection_point.h"
+#include "utils/tuplestore.h"
+#include "utils/wait_event.h"
 
 PG_FUNCTION_INFO_V1(pg_resize_shared_buffers);
 
@@ -196,6 +198,15 @@ DoShrink(ReturnSetInfo *rsinfo, int old_size, int new_size)
 	 */
 	StrategyReset(old_size, new_size);
 
+	/*
+	 * Lower the low water mark so backends stop allocating buffers in the
+	 * [new_size, old_size) range being reclaimed.  highNBuffers stays at
+	 * old_size for now, so backends may still hold and re-pin buffers in that
+	 * range until the barrier below confirms every backend has observed the
+	 * new lowNBuffers.
+	 */
+	pg_atomic_write_u32(&DSBCtrl->lowNBuffers, new_size);
+
 	elog(LOG, "[Shrink Barrier]: restricting allocations to %d buffers", new_size);
 	INSTR_TIME_SET_CURRENT(phase_start);
 	/*
@@ -243,6 +254,16 @@ DoShrink(ReturnSetInfo *rsinfo, int old_size, int new_size)
 	EmitResizeTimeRow(rsinfo, "Buffer relocation", INSTR_TIME_GET_DOUBLE(phase_end));
 	elog(LOG, "[Shrink]: evicted %d buffers in %f seconds", old_size - new_size, INSTR_TIME_GET_DOUBLE(phase_end));
 
+	/*
+	 * Eviction must have emptied [new_size, old_size) and unlinked it from all
+	 * buffer-table chains; verify that before BufTableEntriesShrink frees the
+	 * range, so a broken eviction protocol trips an assertion instead of
+	 * leaving a dangling chain link into freed memory.
+	 */
+#ifdef USE_ASSERT_CHECKING
+	BufTableAssertShrinkInvariants(old_size, new_size);
+#endif
+
 	CHECK_FOR_INTERRUPTS();
 	/*
 	 * All the victim buffers are now empty and won't be allocated by backends.
@@ -274,6 +295,37 @@ DoShrink(ReturnSetInfo *rsinfo, int old_size, int new_size)
 						old_size, new_size),
 				 errdetail("madvise(MADV_REMOVE) failed while releasing buffer-pool memory; the failure is not recoverable."),
 				 errhint("Check the server log for the underlying madvise() error.")));
+
+	/*
+	 * Shrink the buffer mapping table to match.  highNBuffers has been lowered
+	 * to new_size, so entries[new_size, old_size) is invisible to backends and
+	 * its chain links were already cleared; the bucket array follows the pool
+	 * down (floored at NUM_BUFFER_PARTITIONS -- see BufTableBucketsShrink).
+	 */
+	{
+		Size		bt_bytes;
+		bool		bt_success;
+
+		bt_bytes = BufTableEntriesShrink(old_size, new_size, &bt_success);
+		if (!bt_success)
+			ereport(ERROR,
+					(errcode(ERRCODE_INTERNAL_ERROR),
+					 errmsg("buffer mapping table entries shrink from %d to %d failed",
+							old_size, new_size),
+					 errdetail("madvise(MADV_REMOVE) failed while releasing buffer-table entry memory; the failure is not recoverable."),
+					 errhint("Check the server log for the underlying madvise() error.")));
+		EmitResizeBytesRow(rsinfo, "Shrink buftable entries", (double) bt_bytes);
+
+		bt_bytes = BufTableBucketsShrink(new_size, &bt_success);
+		if (!bt_success)
+			ereport(ERROR,
+					(errcode(ERRCODE_INTERNAL_ERROR),
+					 errmsg("buffer mapping table buckets shrink from %d to %d failed",
+							old_size, new_size),
+					 errdetail("madvise(MADV_REMOVE) failed while releasing buffer-table bucket memory; the failure is not recoverable."),
+					 errhint("Check the server log for the underlying madvise() error.")));
+		EmitResizeBytesRow(rsinfo, "Shrink buftable buckets", (double) bt_bytes);
+	}
 }
 
 /*
@@ -331,6 +383,39 @@ DoExpand(ReturnSetInfo *rsinfo, int old_size, int new_size)
 
 	BufferManagerShmemInitBuffers(old_size, new_size);
 
+	/*
+	 * Grow the buffer mapping table alongside the pool, before publishing the
+	 * new water marks.  Entries for [old_size, new_size) are populated and
+	 * initialized empty; the bucket array grows (dynahash-style splitting,
+	 * rehashing affected chains under their partition locks) to keep chains
+	 * short at the larger pool size.  The new entries are invisible to backends
+	 * until highNBuffers advances below.
+	 */
+	{
+		Size		bt_bytes;
+		bool		bt_success;
+
+		bt_bytes = BufTableEntriesExpand(old_size, new_size, &bt_success);
+		if (!bt_success)
+			ereport(ERROR,
+					(errcode(ERRCODE_INTERNAL_ERROR),
+					 errmsg("buffer mapping table entries expand from %d to %d failed",
+							old_size, new_size),
+					 errdetail("madvise(MADV_POPULATE_WRITE) failed while populating buffer-table entry memory; the new range was not made visible to backends."),
+					 errhint("Check the server log for the underlying madvise() error and retry.")));
+		EmitResizeBytesRow(rsinfo, "Expand buftable entries", (double) bt_bytes);
+
+		bt_bytes = BufTableBucketsExpand(new_size, &bt_success);
+		if (!bt_success)
+			ereport(ERROR,
+					(errcode(ERRCODE_INTERNAL_ERROR),
+					 errmsg("buffer mapping table buckets expand from %d to %d failed",
+							old_size, new_size),
+					 errdetail("madvise(MADV_POPULATE_WRITE) failed while populating buffer-table bucket memory; the new range was not made visible to backends."),
+					 errhint("Check the server log for the underlying madvise() error and retry.")));
+		EmitResizeBytesRow(rsinfo, "Expand buftable buckets", (double) bt_bytes);
+	}
+
 	INJECTION_POINT("buf-resize-expand-before-publish", NULL);
 
 	/*
@@ -353,6 +438,15 @@ DoExpand(ReturnSetInfo *rsinfo, int old_size, int new_size)
 	 * re-scanning existing ones with usage_count == 0.
 	 */
 	StrategyReset(old_size, new_size);
+
+	/*
+	 * Publish the new water marks.  highNBuffers is advanced first so a
+	 * concurrent atomics reader never sees lowNBuffers > highNBuffers; the new
+	 * range is already backed by initialized memory (see above), so a clock
+	 * sweep observing the larger highNBuffers before lowNBuffers is safe.
+	 */
+	pg_atomic_write_u32(&DSBCtrl->highNBuffers, new_size);
+	pg_atomic_write_u32(&DSBCtrl->lowNBuffers, new_size);
 	LWLockRelease(&DSBCtrl->AccessNBuffersLock);
 
 	/*
diff --git a/src/backend/storage/buffer/buf_table.c b/src/backend/storage/buffer/buf_table.c
index 347bf267d73..f831be9b119 100644
--- a/src/backend/storage/buffer/buf_table.c
+++ b/src/backend/storage/buffer/buf_table.c
@@ -3,6 +3,37 @@
  * buf_table.c
  *	  routines for mapping BufferTags to buffer indexes.
  *
+ * The shared buffer mapping table is a flat, index-linked hash table (an
+ * open-chaining replacement for the former dynahash-based table).  It is made
+ * of two shared-memory arrays plus a small control block:
+ *
+ *	  buckets[num_buckets]      - one chain head per hash bucket
+ *	  entries[GetMaxNBuffers()] - one entry per buffer, indexed by buf_id
+ *
+ * Each buffer slot i permanently owns entry slot i, so no freelist is needed:
+ * bufmgr always removes a buffer's old mapping (BufTableDelete, called from
+ * InvalidateVictimBuffer) before inserting a new tag for that same buf_id (see
+ * GetVictimBuffer / BufferAlloc in bufmgr.c).  Empty entry slots are marked by
+ * tag.blockNum == P_NEW; chains are linked by int index and terminated by
+ * BUF_TABLE_CHAIN_END.
+ *
+ * Bucket count tracks the live pool.  The bucket array can be grown and shrunk
+ * online (dynahash-style linear hashing / bucket splitting) as shared_buffers
+ * is resized: calc_bucket() maps a hashcode to a bucket via the low/high split
+ * masks, and BufTableBucketsExpand/Shrink add or remove buckets one at a time,
+ * rehashing the affected split partner's chain each step.  num_buckets and
+ * low_mask therefore live in shared memory (BufTableControl), read lock-free on
+ * the lookup hot path and mutated only by the resize coordinator.
+ *
+ * num_buckets is floored at NUM_BUFFER_PARTITIONS and NUM_BUFFER_PARTITIONS is
+ * a power of two, so a bucket and its split partner always fall in the same
+ * buffer partition: partition = hashcode % NUM_BUFFER_PARTITIONS shares its low
+ * bits with the bucket index, so every tag that maps to a given bucket maps to
+ * a single partition, and the caller's BufMappingLock fully serializes each
+ * chain -- the same guarantee the dynahash table relied on.  A tag's partition
+ * is thus invariant across a resize, so lookups take that lock directly with no
+ * recheck.
+ *
  * Note: the routines in this file do no locking of their own.  The caller
  * must hold a suitable lock on the appropriate BufMappingLock, as specified
  * in the comments.  We can't do the locking inside these functions because
@@ -21,56 +52,240 @@
  */
 #include "postgres.h"
 
+#include "common/hashfn.h"
+#include "miscadmin.h"
+#include "port/atomics.h"
+#include "port/pg_bitutils.h"
 #include "storage/buf_internals.h"
+#include "storage/bufmgr.h"
+#include "storage/dynamic_shared_buffers.h"
+#include "storage/lwlock.h"
+#include "storage/shmem.h"
 #include "storage/subsystems.h"
 
+#define BUF_TABLE_CHAIN_END  (-1)
+
+/* bucket for buffer lookup hashtable */
+typedef struct
+{
+	int			head;			/* head of hash chain, or BUF_TABLE_CHAIN_END */
+} BufTableBucket;
+
 /* entry for buffer lookup hashtable */
 typedef struct
 {
-	BufferTag	key;			/* Tag of a disk page */
-	int			id;				/* Associated buffer ID */
-} BufferLookupEnt;
+	BufferTag	tag;			/* Tag of a disk page, or P_NEW if empty */
+	int			next;			/* next entry in hash chain */
+} BufTableEntry;
+
+/*
+ * Bucket-head and entry arrays for the buffer lookup hashtable (in shared
+ * memory).  Both are reserved at GetMaxNBuffers() so the live portion can grow
+ * up to the pool's upper bound; only the live prefix is populated with physical
+ * memory (see BufTableShmemInit / BufTableBucketsExpand / BufTableEntriesExpand).
+ */
+static BufTableBucket *buckets = NULL;
+static BufTableEntry *entries = NULL;
+
+/*
+ * num_buckets and low_mask are mutated at runtime by the resize coordinator
+ * (AddBucket / RemoveBucket) and read by every backend on the buffer-lookup hot
+ * path, so they must live in shared memory: a process-local copy would leave
+ * other backends (and backends forked later from the postmaster) with a stale
+ * bucket count after a resize.  Mirrors DSBCtrl's lowNBuffers/highNBuffers.
+ * high_mask is not stored -- it is always (low_mask << 1) | 1 (see calc_bucket).
+ */
+typedef struct BufTableControl
+{
+	pg_atomic_uint32 num_buckets;
+	pg_atomic_uint32 low_mask;
+} BufTableControl;
 
-static HTAB *SharedBufHash;
+static BufTableControl *BufTableCtl = NULL;
 
 static void BufTableShmemRequest(void *arg);
+static void BufTableShmemInit(void *arg);
 
 const ShmemCallbacks BufTableShmemCallbacks = {
 	.request_fn = BufTableShmemRequest,
-	/* no special initialization needed, the hash table will start empty */
+	.init_fn = BufTableShmemInit,
+	/* buckets/entries/control pointers are restored by the shmem framework */
 };
 
+#ifdef USE_ASSERT_CHECKING
 /*
- * Register shmem hash table for mapping buffers.
- *		size is the desired hash table size (possibly more than NBuffers)
+ * AssertMaskInvariant
+ *		Verify the split-mask invariant for a (low_mask, num_buckets) pair.
+ *
+ * The table uses dynahash-style linear hashing, so low_mask is NOT a pure
+ * function of num_buckets: when num_buckets is an exact power of two reached by
+ * incremental expansion, low_mask sits at the lower boundary (num_buckets/2 - 1)
+ * until the next split widens it.  Both forms route hashes identically.  The
+ * invariant that always holds is:
+ *   - low_mask + 1 is a power of two (low_mask is all-ones), and
+ *   - low_mask covers the lower half: low_mask < num_buckets <= 2*(low_mask+1).
+ *
+ * Must only be called by the single-writer resize coordinator (or at init),
+ * where low_mask and num_buckets are read consistently.  It is NOT safe to call
+ * from calc_bucket: that runs under the caller's own partition lock, but the
+ * resize coordinator may hold a *different* partition's lock while writing these
+ * two fields, so calc_bucket can observe a torn (low_mask, num_buckets) pair.
  */
-void
-BufTableShmemRequest(void *arg)
+static inline void
+AssertMaskInvariant(uint32 low_mask, uint32 num_buckets)
 {
-	int			size;
+	Assert(low_mask > 0);
+	Assert((low_mask & (low_mask + 1)) == 0);		/* low_mask+1 is 2^k */
+	Assert(num_buckets > low_mask);					/* lower-half mask */
+	Assert(num_buckets <= ((low_mask + 1) << 1));	/* within current half */
+}
+#endif
+
+static inline void
+InitializeBucket(int bucket_id)
+{
+	buckets[bucket_id].head = BUF_TABLE_CHAIN_END;
+}
 
+static inline void
+InitializeEntry(int entry_id)
+{
+	entries[entry_id].tag.blockNum = P_NEW;
+	entries[entry_id].next = BUF_TABLE_CHAIN_END;
+}
+
+/*
+ * calc_bucket
+ *		Map a hash code to a bucket index using dynahash-style linear hashing.
+ *
+ * Reads the live (low_mask, num_buckets) pair lock-free.  See dynahash.c's
+ * calc_bucket() for the underlying scheme.
+ */
+static uint32
+calc_bucket(uint32 hashcode)
+{
+	uint32		bucket_id;
+	uint32		lm = pg_atomic_read_u32(&BufTableCtl->low_mask);
+	uint32		nb = pg_atomic_read_u32(&BufTableCtl->num_buckets);
+
+	Assert(nb > 0 && nb <= (uint32) GetMaxNBuffers());
 	/*
-	 * Request the shared buffer lookup hashtable.
-	 *
-	 * Since we can't tolerate running out of lookup table entries, we must be
-	 * sure to specify an adequate table size here.  The maximum steady-state
-	 * usage is of course NBuffers entries, but BufferAlloc() tries to insert
-	 * a new entry before deleting the old.  In principle this could be
-	 * happening in each partition concurrently, so we could need as many as
-	 * NBuffers + NUM_BUFFER_PARTITIONS entries.
+	 * calc_bucket runs under the caller's partition lock, but a concurrent
+	 * resize of a *different* partition can be writing low_mask/num_buckets
+	 * under its own lock, so lm and nb may be a torn pair mid-resize; only
+	 * assert per-variable facts here.  The full (lm, nb) relationship is
+	 * checked at the write sites.
 	 */
-	size = NBuffers + NUM_BUFFER_PARTITIONS;
-
-	ShmemRequestHash(.name = "Shared Buffer Lookup Table",
-					 .nelems = size,
-					 .ptr = &SharedBufHash,
-					 .hash_info.keysize = sizeof(BufferTag),
-					 .hash_info.entrysize = sizeof(BufferLookupEnt),
-					 .hash_info.num_partitions = NUM_BUFFER_PARTITIONS,
-					 .hash_flags = HASH_ELEM | HASH_BLOBS | HASH_PARTITION | HASH_FIXED_SIZE,
+	Assert(lm > 0 && (lm & (lm + 1)) == 0);		/* low_mask is a 2^k-1 mask */
+
+	/* ((lm << 1) + 1) is the high mask */
+	bucket_id = hashcode & ((lm << 1) + 1);
+	if (bucket_id >= nb)
+		bucket_id = bucket_id & lm;
+	return bucket_id;
+}
+
+/*
+ * Number of buckets to set up at init.  Tracks the initial pool size
+ * (NBuffersGUC), floored at NUM_BUFFER_PARTITIONS so a bucket and its split
+ * partner always share a partition (see file header and ResizeBucket).
+ */
+static inline int
+BufTableInitialNumBuckets(void)
+{
+	return Max(NBuffersGUC, NUM_BUFFER_PARTITIONS);
+}
+
+/*
+ * Register shared memory for the buffer mapping table.
+ *
+ * The bucket and entry arrays are reserved at GetMaxNBuffers() so the live
+ * portion can grow up to the pool's upper bound without reallocating.  Both are
+ * page-aligned and given two extra pages of slack so the page-aligned madvise()
+ * ranges issued during resize stay within each array's own reservation (mirrors
+ * BufferManagerShmemRequest).
+ */
+static void
+BufTableShmemRequest(void *arg)
+{
+	int			max_nbuffers = GetMaxNBuffers();
+	Size		os_page_size = buffer_pool_madvise_alignment();
+
+	Assert(os_page_size != 0);
+
+	ShmemRequestStruct(.name = "Shared Buffer Lookup Table Control",
+					   .size = sizeof(BufTableControl),
+					   .ptr = (void **) &BufTableCtl,
+		);
+
+	ShmemRequestStruct(.name = "Shared Buffer Lookup Buckets",
+					   .size = add_size(mul_size(max_nbuffers, sizeof(BufTableBucket)),
+										mul_size(2, os_page_size)),
+					   .alignment = os_page_size,
+					   .ptr = (void **) &buckets,
+		);
+
+	ShmemRequestStruct(.name = "Shared Buffer Lookup Entries",
+					   .size = add_size(mul_size(max_nbuffers, sizeof(BufTableEntry)),
+										mul_size(2, os_page_size)),
+					   .alignment = os_page_size,
+					   .ptr = (void **) &entries,
 		);
 }
 
+/*
+ * Initialize the shared buffer lookup table.  Called once during shared-memory
+ * initialization (in the postmaster, or in a standalone backend).
+ *
+ * Only the initial live prefix is populated: buckets[0, initial_num_buckets)
+ * and entries[0, NBuffersGUC).  Shared memory is zeroed, but zero is a valid
+ * buf_id and block 0 is a valid block number, so we must explicitly mark every
+ * bucket empty (BUF_TABLE_CHAIN_END) and every entry empty (blockNum == P_NEW).
+ */
+static void
+BufTableShmemInit(void *arg)
+{
+	bool		success = true;
+	int			initial_num_buckets = BufTableInitialNumBuckets();
+
+	/*
+	 * A bucket and its split partner share a partition only when num_buckets is
+	 * floored at NUM_BUFFER_PARTITIONS, which requires the (MaxNBuffers-sized)
+	 * bucket array to be at least that large.  Enforce in release builds too: a
+	 * bare Assert compiles out, and running past here corrupts shared memory.
+	 */
+	Assert(GetMaxNBuffers() >= NUM_BUFFER_PARTITIONS);
+	if (GetMaxNBuffers() < NUM_BUFFER_PARTITIONS)
+		elog(PANIC,
+			 "shared buffer pool (%d buffers) must be at least NUM_BUFFER_PARTITIONS (%d)",
+			 GetMaxNBuffers(), NUM_BUFFER_PARTITIONS);
+
+	/* Publish the initial bucket count and split mask (see calc_bucket). */
+	pg_atomic_init_u32(&BufTableCtl->num_buckets, initial_num_buckets);
+	pg_atomic_init_u32(&BufTableCtl->low_mask,
+					   pg_prevpower2_32(initial_num_buckets) - 1);
+#ifdef USE_ASSERT_CHECKING
+	AssertMaskInvariant(pg_atomic_read_u32(&BufTableCtl->low_mask),
+						pg_atomic_read_u32(&BufTableCtl->num_buckets));
+#endif
+
+	/* Fault in and initialize the initial bucket heads. */
+	BufferPoolArrayPhysicalExpand(buckets, sizeof(BufTableBucket),
+								  0, initial_num_buckets, &success);
+	if (!success)
+		elog(PANIC, "could not populate initial buffer lookup buckets: madvise(MADV_POPULATE_WRITE) failed");
+	for (int i = 0; i < initial_num_buckets; i++)
+		InitializeBucket(i);
+
+	/* Fault in and initialize the initial entries (one per initial buffer). */
+	BufferPoolArrayPhysicalExpand(entries, sizeof(BufTableEntry),
+								  0, NBuffersGUC, &success);
+	if (!success)
+		elog(PANIC, "could not populate initial buffer lookup entries: madvise(MADV_POPULATE_WRITE) failed");
+	for (int i = 0; i < NBuffersGUC; i++)
+		InitializeEntry(i);
+}
+
 /*
  * BufTableHashCode
  *		Compute the hash code associated with a BufferTag
@@ -83,7 +298,7 @@ BufTableShmemRequest(void *arg)
 uint32
 BufTableHashCode(BufferTag *tagPtr)
 {
-	return get_hash_value(SharedBufHash, tagPtr);
+	return tag_hash(tagPtr, sizeof(BufferTag));
 }
 
 /*
@@ -95,19 +310,15 @@ BufTableHashCode(BufferTag *tagPtr)
 int
 BufTableLookup(BufferTag *tagPtr, uint32 hashcode)
 {
-	BufferLookupEnt *result;
-
-	result = (BufferLookupEnt *)
-		hash_search_with_hash_value(SharedBufHash,
-									tagPtr,
-									hashcode,
-									HASH_FIND,
-									NULL);
+	int			id = buckets[calc_bucket(hashcode)].head;
 
-	if (!result)
-		return -1;
-
-	return result->id;
+	while (id != BUF_TABLE_CHAIN_END)
+	{
+		if (BufferTagsEqual(&entries[id].tag, tagPtr))
+			return id;
+		id = entries[id].next;
+	}
+	return -1;
 }
 
 /*
@@ -123,23 +334,35 @@ BufTableLookup(BufferTag *tagPtr, uint32 hashcode)
 int
 BufTableInsert(BufferTag *tagPtr, uint32 hashcode, int buf_id)
 {
-	BufferLookupEnt *result;
-	bool		found;
+	int			bucket_id = calc_bucket(hashcode);
+	int			head = buckets[bucket_id].head;
+	int			id = head;
 
-	Assert(buf_id >= 0);		/* -1 is reserved for not-in-table */
+	Assert(buf_id >= 0 && buf_id < GetHighNBuffers());
 	Assert(tagPtr->blockNum != P_NEW);	/* invalid tag */
 
-	result = (BufferLookupEnt *)
-		hash_search_with_hash_value(SharedBufHash,
-									tagPtr,
-									hashcode,
-									HASH_ENTER,
-									&found);
+	/* If the tag is already in the chain, surface the existing buf_id. */
+	while (id != BUF_TABLE_CHAIN_END)
+	{
+		if (BufferTagsEqual(&entries[id].tag, tagPtr))
+			return id;
+		id = entries[id].next;
+	}
 
-	if (found)					/* found something already in the table */
-		return result->id;
+	/*
+	 * Not present.  entry[buf_id] must be empty: bufmgr always deletes a
+	 * buffer's old mapping before inserting a new tag for that buf_id.
+	 */
+	Assert(entries[buf_id].tag.blockNum == P_NEW);
 
-	result->id = buf_id;
+	/*
+	 * Link entry[buf_id] at the chain head, keeping the prior head as its
+	 * successor.  (Use the saved `head`, not `id`, which the loop above has
+	 * advanced to BUF_TABLE_CHAIN_END.)
+	 */
+	entries[buf_id].tag = *tagPtr;
+	entries[buf_id].next = head;
+	buckets[bucket_id].head = buf_id;
 
 	return -1;
 }
@@ -153,15 +376,466 @@ BufTableInsert(BufferTag *tagPtr, uint32 hashcode, int buf_id)
 void
 BufTableDelete(BufferTag *tagPtr, uint32 hashcode)
 {
-	BufferLookupEnt *result;
+	int			bucket_id = calc_bucket(hashcode);
+	int			prev = BUF_TABLE_CHAIN_END;
+	int			id = buckets[bucket_id].head;
+
+	while (id != BUF_TABLE_CHAIN_END)
+	{
+		if (BufferTagsEqual(&entries[id].tag, tagPtr))
+		{
+			/* unlink from the chain */
+			if (prev == BUF_TABLE_CHAIN_END)
+				buckets[bucket_id].head = entries[id].next;
+			else
+				entries[prev].next = entries[id].next;
+			/* mark the entry empty */
+			entries[id].tag.blockNum = P_NEW;
+			entries[id].next = BUF_TABLE_CHAIN_END;
+			return;
+		}
+		prev = id;
+		id = entries[id].next;
+	}
+
+	/*
+	 * Entry not in table.  Callers never double-delete (deletion is gated by
+	 * BM_TAG_VALID on the buffer header), so this indicates corruption.
+	 */
+	Assert(false);
+	elog(ERROR, "shared buffer hash table corrupted");
+}
+
+/*
+ * BufTableEntriesExpand
+ *		Grow the live portion of the entries array from old_size to new_size.
+ *
+ * Populates physical memory for entries[old_size, new_size) and initializes the
+ * freshly added entries as empty.  Mirrors the buffer pool's own
+ * BufferManagerShmemExpand() + BufferManagerShmemInitBuffers() sequence.
+ *
+ * Returns the number of bytes touched.  On madvise(MADV_POPULATE_WRITE) failure
+ * *success is set to false and the new entries are NOT initialized; the caller
+ * must not advance the water marks.
+ *
+ * Must be called by the resize coordinator before publishing the new
+ * highNBuffers, so [old_size, new_size) is invisible to other backends and no
+ * BufMappingLock is required.
+ */
+Size
+BufTableEntriesExpand(int old_size, int new_size, bool *success)
+{
+	Size		bytes;
+
+	*success = true;
+
+	if (entries == NULL)
+		return 0;
+
+	Assert(new_size > old_size);
+
+	bytes = BufferPoolArrayPhysicalExpand(entries, sizeof(BufTableEntry),
+										  old_size, new_size, success);
+	if (!*success)
+		return bytes;
+
+	for (int i = old_size; i < new_size; i++)
+		InitializeEntry(i);
+
+	return bytes;
+}
+
+/*
+ * BufTableEntriesShrink
+ *		Release the physical memory backing entries[new_size, old_size).
+ *
+ * Mirrors the buffer pool's BufferManagerShmemShrink().  Must be called after
+ * highNBuffers has been lowered to new_size (so the range is invisible to
+ * backends) and after entries[new_size, old_size) have been emptied.
+ *
+ * Returns the number of bytes released; on madvise(MADV_REMOVE) failure
+ * *success is set to false.
+ */
+Size
+BufTableEntriesShrink(int old_size, int new_size, bool *success)
+{
+	*success = true;
+
+	if (entries == NULL)
+		return 0;
+
+	Assert(new_size < old_size);
+
+	return BufferPoolArrayPhysicalShrink(entries, sizeof(BufTableEntry),
+										 new_size, old_size, success);
+}
+
+/*
+ * RehashEntries
+ *		Rehash the partner bucket's chain and move entries that now hash to the
+ *		freshly added bucket over to it.
+ *
+ * Caller (ResizeBucket) must hold the partition lock covering both buckets.
+ */
+static void
+RehashEntries(uint32 current_bucket_id, uint32 partner_bucket_id)
+{
+	int			prev_entry_id = BUF_TABLE_CHAIN_END;
+	int			curr_entry_id = buckets[partner_bucket_id].head;
+
+	while (curr_entry_id != BUF_TABLE_CHAIN_END)
+	{
+		uint32		hashcode = BufTableHashCode(&entries[curr_entry_id].tag);
+		uint32		rehashed_bucket_id = calc_bucket(hashcode);
+
+		Assert(rehashed_bucket_id == current_bucket_id ||
+			   rehashed_bucket_id == partner_bucket_id);
+
+		/* entry now belongs in the freshly added bucket: splice it over */
+		if (rehashed_bucket_id == current_bucket_id)
+		{
+			int			next_entry_id = entries[curr_entry_id].next;
+
+			/* unlink from the partner bucket */
+			if (prev_entry_id == BUF_TABLE_CHAIN_END)
+				buckets[partner_bucket_id].head = next_entry_id;
+			else
+				entries[prev_entry_id].next = next_entry_id;
+
+			/* push onto the head of the new bucket */
+			entries[curr_entry_id].next = buckets[current_bucket_id].head;
+			buckets[current_bucket_id].head = curr_entry_id;
+
+			curr_entry_id = next_entry_id;
+		}
+		else						/* entry stays in the partner bucket */
+		{
+			prev_entry_id = curr_entry_id;
+			curr_entry_id = entries[curr_entry_id].next;
+		}
+	}
+}
+
+/*
+ * AddBucket
+ *		Append a new bucket (memory already reserved), adjust num_buckets and
+ *		the split mask, and rehash the partner bucket's entries.
+ *
+ * Caller (ResizeBucket) must hold the partition lock covering both buckets.
+ */
+static void
+AddBucket(uint32 new_bucket_id, uint32 partner_bucket_id, uint32 low_mask)
+{
+	uint32		high_mask;
+
+	/*
+	 * Pre-split: the passed low_mask must be consistent with the current bucket
+	 * count.  The new bucket is appended at num_buckets, so new_bucket_id equals
+	 * the count before this append (see ResizeBucket).
+	 */
+#ifdef USE_ASSERT_CHECKING
+	AssertMaskInvariant(low_mask, new_bucket_id);
+	Assert(partner_bucket_id == (new_bucket_id & low_mask));
+#endif
+
+	/*
+	 * Publish the new bucket.  num_buckets must be bumped before RehashEntries,
+	 * because calc_bucket() -- which RehashEntries relies on to decide which
+	 * entries move -- only routes hashes to the new bucket once num_buckets
+	 * includes it.  Mask maintenance mirrors dynahash exactly: widen the masks
+	 * only once the new id outgrows high_mask, keeping high_mask = 2*low_mask+1.
+	 */
+	pg_atomic_write_u32(&BufTableCtl->num_buckets, new_bucket_id + 1);
+	high_mask = (low_mask << 1) + 1;
+	if (new_bucket_id > high_mask)
+		pg_atomic_write_u32(&BufTableCtl->low_mask, high_mask);
+
+#ifdef USE_ASSERT_CHECKING
+	AssertMaskInvariant(pg_atomic_read_u32(&BufTableCtl->low_mask),
+						pg_atomic_read_u32(&BufTableCtl->num_buckets));
+#endif
+
+	InitializeBucket(new_bucket_id);
+	RehashEntries(new_bucket_id, partner_bucket_id);
+}
+
+/*
+ * RemoveBucket
+ *		Remove the top bucket by moving its whole chain onto its partner, then
+ *		adjust num_buckets and the split mask.
+ *
+ * Caller (ResizeBucket) must hold the partition lock covering both buckets.
+ */
+static void
+RemoveBucket(uint32 bucket_id, uint32 partner_id)
+{
+	int			head = buckets[bucket_id].head;
+
+	/*
+	 * Move bucket_id's entire chain onto the partner bucket.  Every entry that
+	 * hashed to bucket_id will hash to partner_id after num_buckets drops, so
+	 * the whole chain moves unconditionally (inverse of RehashEntries).  Find
+	 * the tail of bucket_id's chain, then splice partner's chain after it.
+	 */
+	if (head != BUF_TABLE_CHAIN_END)
+	{
+		int			tail = head;
+
+		while (entries[tail].next != BUF_TABLE_CHAIN_END)
+			tail = entries[tail].next;
+
+		entries[tail].next = buckets[partner_id].head;
+		buckets[partner_id].head = head;
+		buckets[bucket_id].head = BUF_TABLE_CHAIN_END;
+	}
+
+	/*
+	 * Publish the new metadata.  These are plain atomic stores with no barrier
+	 * between them, so a lock-free calc_bucket caller on another partition may
+	 * observe them in either order -- the write order is NOT load-bearing.
+	 * Safety comes from the partition-lock protocol: partition = hashcode %
+	 * NUM_BUFFER_PARTITIONS and num_buckets is floored at NUM_BUFFER_PARTITIONS,
+	 * so a tag's partition is invariant across a resize.  A backend touching the
+	 * merged buckets holds the same partition lock the coordinator holds here
+	 * (LWLock acquire/release are full barriers), so it observes the
+	 * consolidated chain; a backend on any other partition never touches these
+	 * buckets and resolves to its own live bucket even under a torn (low_mask,
+	 * num_buckets) read.  The publish mask is recomputed here from bucket_id.
+	 */
+	Assert(bucket_id >= NUM_BUFFER_PARTITIONS);		/* floored, so bucket_id - 1 >= 1 */
+	pg_atomic_write_u32(&BufTableCtl->low_mask, pg_prevpower2_32(bucket_id - 1) - 1);
+	pg_atomic_write_u32(&BufTableCtl->num_buckets, bucket_id);
+
+#ifdef USE_ASSERT_CHECKING
+	AssertMaskInvariant(pg_atomic_read_u32(&BufTableCtl->low_mask),
+						pg_atomic_read_u32(&BufTableCtl->num_buckets));
+#endif
+}
+
+/*
+ * ResizeBucket
+ *		Add or remove a single bucket at the top of the table.
+ *
+ * Owns the shared bucket-split scaffolding -- deriving the target and partner
+ * bucket ids and their (shared) partition lock, acquiring it, and releasing it
+ * -- and dispatches the direction-specific critical section to AddBucket
+ * (expand) or RemoveBucket (shrink).
+ *
+ * Runs only in the resize coordinator (single writer), so a plain
+ * read-then-write of the shared metadata is safe.
+ */
+static void
+ResizeBucket(bool expand)
+{
+	uint32		num_buckets = pg_atomic_read_u32(&BufTableCtl->num_buckets);
+	uint32		target_bucket_id;
+	uint32		partner_bucket_id;
+	uint32		low_mask;
+	uint32		partition_id;
+	LWLock	   *partitionLock;
+
+	if (expand)
+	{
+		low_mask = pg_atomic_read_u32(&BufTableCtl->low_mask);
+		Assert(low_mask > 0);
+		target_bucket_id = num_buckets;		/* the new bucket being appended */
+	}
+	else
+	{
+		/* num_buckets is floored at NUM_BUFFER_PARTITIONS, so this never goes below it */
+		Assert(num_buckets > NUM_BUFFER_PARTITIONS);
+		target_bucket_id = num_buckets - 1; /* the top bucket being removed */
+
+		/*
+		 * Derive the split mask from the bucket id rather than the stored
+		 * low_mask: when num_buckets is an exact power of two the stored
+		 * low_mask equals the removed bucket id, which would make the removed
+		 * bucket its own partner.
+		 */
+		low_mask = pg_prevpower2_32(target_bucket_id) - 1;
+	}
+	partner_bucket_id = target_bucket_id & low_mask;
+	if (!expand)
+		Assert(partner_bucket_id < target_bucket_id);
+
+	/*
+	 * The target bucket and its split partner map to the same buffer partition:
+	 * NUM_BUFFER_PARTITIONS is a power of two and num_buckets is floored at it,
+	 * so calc_bucket never folds away the low log2(NUM_BUFFER_PARTITIONS) bits,
+	 * and target/partner differ only in higher bits.  A single partition lock
+	 * therefore covers the whole split, and a tag's partition never changes
+	 * across a resize.
+	 */
+	partition_id = target_bucket_id % NUM_BUFFER_PARTITIONS;
+	Assert(partition_id == partner_bucket_id % NUM_BUFFER_PARTITIONS);
+	partitionLock = BufMappingPartitionLockByIndex(partition_id);
+
+	/*
+	 * Acquire the affected partition lock BEFORE mutating/publishing anything,
+	 * so the whole num_buckets/mask change + entry move is atomic w.r.t. backends
+	 * operating on the same partition.  Backends on other partitions are
+	 * unaffected besides seeing potentially updated num_buckets/low_mask.
+	 */
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	if (expand)
+		AddBucket(target_bucket_id, partner_bucket_id, low_mask);
+	else
+		RemoveBucket(target_bucket_id, partner_bucket_id);
+
+	LWLockRelease(partitionLock);
+}
+
+/*
+ * BufTableBucketsExpand
+ *		Grow the bucket array to track a pool expanded to new_size.
+ *
+ * Faults in physical memory for the newly needed bucket heads, then adds one
+ * bucket at a time (dynahash-style splitting), rehashing each new bucket's split
+ * partner.  Adding incrementally -- num_buckets bumped, low_mask widened when
+ * needed, partner rehashed -- lets more-than-doubling redistribute existing
+ * entries evenly and keeps low_mask maintenance to low_mask = (low_mask<<1)+1.
+ *
+ * Returns the number of bytes touched.  Must be called by the resize
+ * coordinator.
+ */
+Size
+BufTableBucketsExpand(int new_size, bool *success)
+{
+	Size		bytes;
+	uint32		num_buckets;
+
+	*success = true;
+
+	if (buckets == NULL)
+		return 0;
+
+	Assert(new_size <= GetMaxNBuffers());
+
+	/*
+	 * Drive growth off the table's live num_buckets, not the pool's old_size.
+	 * The two are equal on the happy path, but a shrink interrupted past its
+	 * point of no return (highNBuffers lowered before BufTableBucketsShrink ran)
+	 * leaves num_buckets above old_size, and AddBucket appends at num_buckets.
+	 * Anchoring the fault bound and the loop to num_buckets keeps the count
+	 * correct and never re-touches the live [old_size, num_buckets) buckets:
+	 * those are already backed and hold real chain heads, and the memset()
+	 * fallback in the physical-expand helper would zero them.  When num_buckets
+	 * >= new_size the loop does not run, leaving a safe (over-provisioned) state.
+	 */
+	num_buckets = pg_atomic_read_u32(&BufTableCtl->num_buckets);
+
+	bytes = BufferPoolArrayPhysicalExpand(buckets, sizeof(BufTableBucket),
+										  num_buckets, new_size, success);
+	if (!*success)
+		return bytes;
+
+	for (uint32 i = num_buckets; i < (uint32) new_size; i++)
+		ResizeBucket(true);
+
+	return bytes;
+}
+
+/*
+ * BufTableBucketsShrink
+ *		Shrink the bucket array to track a pool shrunk to new_size.
+ *
+ * Removes one bucket at a time (inverse of expand): each step moves the top
+ * bucket's chain onto its partner, decrements num_buckets, and narrows low_mask
+ * when needed.  Removing incrementally preserves concurrency -- jumping
+ * num_buckets/low_mask straight to the final values would make calc_bucket route
+ * a live entry to a not-yet-removed bucket.
+ *
+ * num_buckets is floored at NUM_BUFFER_PARTITIONS, so the bucket array never
+ * shrinks below the partition count even though the pool and the entries array
+ * (shrunk separately by the caller) can reach MIN_SHARED_BUFFERS.  Keeping
+ * num_buckets >= NUM_BUFFER_PARTITIONS is what guarantees a bucket and its split
+ * partner always share a partition.
+ *
+ * Returns the number of bytes released.  Must be called by the resize
+ * coordinator.
+ */
+Size
+BufTableBucketsShrink(int new_size, bool *success)
+{
+	uint32		num_buckets;
+	uint32		bucket_floor;
+
+	*success = true;
+
+	if (buckets == NULL)
+		return 0;
+
+	Assert(new_size > 0);
+
+	/* the bucket array never shrinks below NUM_BUFFER_PARTITIONS */
+	bucket_floor = Max((uint32) new_size, NUM_BUFFER_PARTITIONS);
+
+	/*
+	 * Remove from the live num_buckets, not old_size: an interrupted prior
+	 * resize may have left num_buckets above old_size (see BufTableBucketsExpand).
+	 * RemoveBucket drops the top bucket (num_buckets - 1), so looping down to
+	 * bucket_floor lands num_buckets at exactly bucket_floor before the physical
+	 * tail [bucket_floor, MaxNBuffers) is released -- the release only ever
+	 * touches now-dead buckets.  When num_buckets is already at/below the floor
+	 * there is nothing to remove or release.
+	 */
+	num_buckets = pg_atomic_read_u32(&BufTableCtl->num_buckets);
+	if (num_buckets <= bucket_floor)
+		return 0;
+
+	for (uint32 i = num_buckets; i > bucket_floor; i--)
+		ResizeBucket(false);
+
+	return BufferPoolArrayPhysicalShrink(buckets, sizeof(BufTableBucket),
+										 bucket_floor, num_buckets, success);
+}
+
+#ifdef USE_ASSERT_CHECKING
+/*
+ * BufTableAssertShrinkInvariants
+ *		Confirm eviction left the table in a state where freeing the doomed
+ *		entries [new_size, old_size) cannot strand a dangling chain link.
+ *
+ * Verifies:
+ *   1. every entry in [new_size, old_size) is empty and unlinked; and
+ *   2. no surviving chain link -- neither a bucket head nor an entry's next
+ *      pointer in [0, new_size) -- references the doomed range.
+ *
+ * Backends keep allocating in [0, new_size) concurrently, so the [0, new_size)
+ * reads are not lock-synchronized; that is fine because any buf_id a backend
+ * ever stores is < new_size (= lowNBuffers), so the bound checks cannot
+ * spuriously fail.
+ */
+void
+BufTableAssertShrinkInvariants(int old_size, int new_size)
+{
+	if (entries == NULL)
+		return;
+
+	Assert(new_size < old_size);
+
+	/* (1) doomed range fully empty + unlinked */
+	for (int i = new_size; i < old_size; i++)
+	{
+		Assert(entries[i].tag.blockNum == P_NEW);
+		Assert(entries[i].next == BUF_TABLE_CHAIN_END);
+	}
+
+	/* (2a) surviving entries don't point into the doomed range */
+	for (int i = 0; i < new_size; i++)
+	{
+		int			next = entries[i].next;
+
+		Assert(next == BUF_TABLE_CHAIN_END || (next >= 0 && next < new_size));
+	}
 
-	result = (BufferLookupEnt *)
-		hash_search_with_hash_value(SharedBufHash,
-									tagPtr,
-									hashcode,
-									HASH_REMOVE,
-									NULL);
+	/* (2b) no bucket head points into the doomed range */
+	for (uint32 b = 0; b < pg_atomic_read_u32(&BufTableCtl->num_buckets); b++)
+	{
+		int			head = buckets[b].head;
 
-	if (!result)				/* shouldn't happen */
-		elog(ERROR, "shared buffer hash table corrupted");
+		Assert(head == BUF_TABLE_CHAIN_END || (head >= 0 && head < new_size));
+	}
 }
+#endif							/* USE_ASSERT_CHECKING */
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 6a7302917a1..7336ca0c05e 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -636,7 +636,6 @@ static void UnpinBuffer(BufferDesc *buf);
 static void UnpinBufferNoOwner(BufferDesc *buf);
 static bool EvictUnpinnedBufferInternal(BufferDesc *desc, bool *buffer_flushed);
 static void BufferSync(int flags, int localNBuffers);
-static uint32 WaitBufHdrUnlocked(BufferDesc *buf);
 static int	SyncOneBuffer(int buf_id, bool skip_recently_used,
 						  WritebackContext *wb_context);
 static void WaitIO(BufferDesc *buf);
@@ -2450,17 +2449,21 @@ retry:
 	oldFlags = buf_state & BUF_FLAG_MASK;
 	ClearBufferTag(&buf->tag);
 
-	UnlockBufHdrExt(buf, buf_state,
-					0,
-					BUF_FLAG_MASK | BUF_USAGECOUNT_MASK,
-					0);
-
 	/*
 	 * Remove the buffer from the lookup hashtable, if it was in there.
 	 */
 	if (oldFlags & BM_TAG_VALID)
 		BufTableDelete(&oldTag, oldHash);
 
+	/* Unlock buffer header after the entry is deleted to avoid a race condition:
+	 * If unlocked prior, a concurrent GetVictimBuffer() could insert a new entry
+	 * for the same buffer and overwrite the entry slot. Then, the BufTableDelete()
+	 * would be unable to find the entry and would corrupt the hashtable. */	
+	UnlockBufHdrExt(buf, buf_state,
+		0,
+		BUF_FLAG_MASK | BUF_USAGECOUNT_MASK,
+		0);
+
 	/*
 	 * Done with mapping lock.
 	 */
@@ -2725,11 +2728,7 @@ GetAdditionalPinLimit(void)
 	 */
 	estimated_pins_held = PrivateRefCountOverflowed + REFCOUNT_ARRAY_ENTRIES;
 
-	/*
-	 * Consult get_pin_limit_hook so the per-backend limit tracks the live
-	 * buffer pool size.
-	 */
-	limit = enable_dynamic_shared_buffers && get_pin_limit_hook ? get_pin_limit_hook() : MaxProportionalPins;
+	limit = MaxProportionalPins;
 
 	/* Is this backend already holding more than its fair share? */
 	if (estimated_pins_held > limit)
@@ -3950,6 +3949,20 @@ BgBufferSync(WritebackContext *wb_context)
 		strategy_delta = strategy_buf_id - prev_strategy_buf_id;
 		strategy_delta += (long) passes_delta * localNBuffers;
 
+		/*
+		 * A buffer-pool resize rewinds the clock-sweep cursor (see
+		 * StrategyReset), which can make the strategy point appear to move
+		 * backwards relative to the position we saved on the previous cycle.
+		 * That is not a steady-state condition, so rather than trip the
+		 * invariant below just discard the stale saved state and re-initialize
+		 * from the current strategy point this cycle.
+		 */
+		if (strategy_delta < 0)
+			saved_info_valid = false;
+	}
+
+	if (saved_info_valid)
+	{
 		Assert(strategy_delta >= 0);
 
 		if ((int32) (next_passes - strategy_passes) > 0)
@@ -4408,12 +4421,6 @@ AssertNotCatalogBufferLock(Buffer buffer, BufferLockMode mode)
 	if (mode != BUFFER_LOCK_EXCLUSIVE)
 		return;
 
-	if (!((BufferDescPadded *) lock > BufferDescriptors &&
-		  (BufferDescPadded *) lock < BufferDescriptors + GetMaxNBuffers()))
-		return;					/* not a buffer lock */
-
-	bufHdr = (BufferDesc *)
-		((char *) lock - offsetof(BufferDesc, content_lock));
 	tag = bufHdr->tag;
 
 	/*
@@ -5227,7 +5234,7 @@ FlushRelationBuffers(Relation rel)
 
 	BEGIN_NBUFFERS_ACCESS(localNBuffers);
 
-	if (RelationUsesLocalBuffers(rel) || am_wal_redo_postgres)
+	if (RelationUsesLocalBuffers(rel))
 	{
 		for (i = 0; i < NLocBuffer; i++)
 		{
@@ -5891,8 +5898,6 @@ MarkBufferDirtyHint(Buffer buffer, bool buffer_std)
 {
 	BufferDesc *bufHdr;
 
-	bufHdr = GetBufferDescriptor(buffer - 1);
-
 	if (!BufferIsValid(buffer))
 		elog(ERROR, "bad buffer ID: %d", buffer);
 
@@ -5902,6 +5907,8 @@ MarkBufferDirtyHint(Buffer buffer, bool buffer_std)
 		return;
 	}
 
+	bufHdr = GetBufferDescriptor(buffer - 1);
+
 	MarkSharedBufferDirtyHint(buffer, bufHdr,
 							  pg_atomic_read_u64(&bufHdr->state),
 							  buffer_std);
@@ -8252,13 +8259,15 @@ MarkDirtyRelUnpinnedBuffers(Relation rel,
 							int32 *buffers_already_dirty,
 							int32 *buffers_skipped)
 {
+	BEGIN_NBUFFERS_ACCESS(localNBuffers);
+
 	Assert(!RelationUsesLocalBuffers(rel));
 
 	*buffers_dirtied = 0;
 	*buffers_already_dirty = 0;
 	*buffers_skipped = 0;
 
-	for (int buf = 1; buf <= NBuffers; buf++)
+	for (int buf = 1; buf <= localNBuffers; buf++)
 	{
 		BufferDesc *desc = GetBufferDescriptor(buf - 1);
 		uint64		buf_state = pg_atomic_read_u64(&(desc->state));
@@ -8292,6 +8301,7 @@ MarkDirtyRelUnpinnedBuffers(Relation rel,
 		else
 			(*buffers_skipped)++;
 	}
+	END_NBUFFERS_ACCESS(localNBuffers);
 }
 
 /*
@@ -8308,11 +8318,13 @@ MarkDirtyAllUnpinnedBuffers(int32 *buffers_dirtied,
 							int32 *buffers_already_dirty,
 							int32 *buffers_skipped)
 {
+	BEGIN_NBUFFERS_ACCESS(localNBuffers);
+
 	*buffers_dirtied = 0;
 	*buffers_already_dirty = 0;
 	*buffers_skipped = 0;
 
-	for (int buf = 1; buf <= NBuffers; buf++)
+	for (int buf = 1; buf <= localNBuffers; buf++)
 	{
 		BufferDesc *desc = GetBufferDescriptor(buf - 1);
 		uint64		buf_state;
@@ -8336,6 +8348,7 @@ MarkDirtyAllUnpinnedBuffers(int32 *buffers_dirtied,
 		else
 			(*buffers_skipped)++;
 	}
+	END_NBUFFERS_ACCESS(localNBuffers);
 }
 
 /*
@@ -9056,7 +9069,7 @@ EvictExtraBuffers(int lowNBuffers, int highNBuffers)
 	for (int buf_id = lowNBuffers; buf_id < highNBuffers; buf_id++)
 	{
 		BufferDesc *desc = GetBufferDescriptor(buf_id);
-		uint32		buf_state;
+		uint64		buf_state;
 		bool		buffer_flushed;
 
 		/* Make sure we can pin the buffer (PinBuffer_Locked contract). */
@@ -9067,7 +9080,7 @@ EvictExtraBuffers(int lowNBuffers, int highNBuffers)
 
 		if (BUF_STATE_GET_REFCOUNT(buf_state) > 0)
 		{
-			UnlockBufHdr(desc, buf_state);
+			UnlockBufHdr(desc);
 			result = false;
 			continue;
 		}
@@ -9101,7 +9114,7 @@ EvictExtraBuffers(int lowNBuffers, int highNBuffers)
 			}
 			else
 			{
-				UnlockBufHdr(desc, buf_state);
+				UnlockBufHdr(desc);
 			}
 			continue;
 		}
diff --git a/src/backend/storage/buffer/dynamic_shared_buffers.c b/src/backend/storage/buffer/dynamic_shared_buffers.c
index efc535d365c..0ee53f04fc2 100644
--- a/src/backend/storage/buffer/dynamic_shared_buffers.c
+++ b/src/backend/storage/buffer/dynamic_shared_buffers.c
@@ -26,6 +26,14 @@
 
 DynamicSharedBuffersControl *DSBCtrl = NULL;
 
+static void DSBShmemRequest(void *arg);
+static void DSBShmemInit(void *arg);
+
+const ShmemCallbacks DSBShmemCallbacks = {
+	.request_fn = DSBShmemRequest,
+	.init_fn = DSBShmemInit,
+};
+
 /*
  * AcquireNBuffersLock
  *
@@ -95,31 +103,49 @@ ReleaseResizeCoordinator(void)
 }
 
 /*
- * DSBControlInit
+ * DSBShmemRequest
+ *
+ * Reserve shared memory for the DynamicSharedBuffersControl structure.  Only
+ * needed when dynamic_shared_buffers is enabled; when off, DSBCtrl stays NULL
+ * and every DSB code path is a no-op.
  *
- * Allocate and initialize the DynamicSharedBuffersControl structure in shared
- * memory. Must be called before BufferManagerShmemInit so that DSBCtrl is
- * available when the buffer pool is set up.
+ * Registered ahead of the buffer manager and buffer mapping table in
+ * subsystemlist.h so DSBCtrl is populated (and its water marks initialized)
+ * before those subsystems set up the pool and the lookup table.
  */
-void
-DSBControlInit(void)
+static void
+DSBShmemRequest(void *arg)
 {
-	bool		foundDSBCtrl;
+	if (!enable_dynamic_shared_buffers)
+		return;
 
-	DSBCtrl = (DynamicSharedBuffersControl *)
-		ShmemInitStruct("DSB Control", sizeof(DynamicSharedBuffersControl),
-						&foundDSBCtrl);
+	ShmemRequestStruct(.name = "DSB Control",
+					   .size = sizeof(DynamicSharedBuffersControl),
+					   .ptr = (void **) &DSBCtrl,
+		);
+}
 
-	if (!foundDSBCtrl)
-	{
-		pg_atomic_init_u32(&DSBCtrl->lowNBuffers, NBuffersGUC);
-		pg_atomic_init_u32(&DSBCtrl->highNBuffers, NBuffersGUC);
+/*
+ * DSBShmemInit
+ *
+ * Initialize the DynamicSharedBuffersControl structure.  The shmem framework
+ * has already set DSBCtrl to point at the reserved allocation.
+ */
+static void
+DSBShmemInit(void *arg)
+{
+	if (!enable_dynamic_shared_buffers)
+		return;
+
+	Assert(DSBCtrl != NULL);
 
-		SpinLockInit(&DSBCtrl->coordinator_lock);
-		DSBCtrl->resize_in_progress = false;
-		DSBCtrl->coordinator_pid = InvalidPid;
+	pg_atomic_init_u32(&DSBCtrl->lowNBuffers, NBuffersGUC);
+	pg_atomic_init_u32(&DSBCtrl->highNBuffers, NBuffersGUC);
 
-		LWLockInitialize(&DSBCtrl->AccessNBuffersLock,
-						 LWTRANCHE_ACCESS_NBUFFERS);
-	}
+	SpinLockInit(&DSBCtrl->coordinator_lock);
+	DSBCtrl->resize_in_progress = false;
+	DSBCtrl->coordinator_pid = InvalidPid;
+
+	LWLockInitialize(&DSBCtrl->AccessNBuffersLock,
+					 LWTRANCHE_ACCESS_NBUFFERS);
 }
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 15d0ed35c5b..e1656e1e47c 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -885,6 +885,12 @@
   boot_val => 'true',
 },
 
+{ name => 'enable_dynamic_shared_buffers', type => 'bool', context => 'PGC_POSTMASTER', group => 'RESOURCES_MEM',
+  short_desc => 'Enables dynamic resizing of the shared buffer pool.',
+  variable => 'enable_dynamic_shared_buffers',
+  boot_val => 'false',
+},
+
 { name => 'enable_eager_aggregate', type => 'bool', context => 'PGC_USERSET', group => 'QUERY_TUNING_METHOD',
   short_desc => 'Enables eager aggregation.',
   flags => 'GUC_EXPLAIN',
@@ -2123,6 +2129,16 @@
   max => 'MAX_BACKENDS /* XXX? */',
 },
 
+{ name => 'max_shared_buffers', type => 'int', context => 'PGC_POSTMASTER', group => 'RESOURCES_MEM',
+  short_desc => 'Sets the upper limit for the shared_buffers value.',
+  long_desc => 'If set above zero, it must be at least shared_buffers.',
+  flags => 'GUC_UNIT_BLOCKS',
+  variable => 'MaxNBuffers',
+  boot_val => '0',
+  min => '0',
+  max => 'INT_MAX / 2',
+},
+
 { name => 'max_slot_wal_keep_size', type => 'int', context => 'PGC_SIGHUP', group => 'REPLICATION_SENDING',
   short_desc => 'Sets the maximum WAL size that can be reserved by replication slots.',
   long_desc => 'Replication slots will be marked as failed, and segments released for deletion or recycling, if this much space is occupied by WAL on disk. -1 means no maximum.',
@@ -2709,23 +2725,6 @@
   min => 'MIN_SHARED_BUFFERS',
   max => 'INT_MAX / 2',
 },
-
-{ name => 'max_shared_buffers', type => 'int', context => 'PGC_POSTMASTER', group => 'RESOURCES_MEM',
-  short_desc => 'Sets the upper limit for the shared_buffers value.',
-  long_desc => 'If set above zero, it must be at least shared_buffers.',
-  flags => 'GUC_UNIT_BLOCKS',
-  variable => 'MaxNBuffers',
-  boot_val => '0',
-  min => '0',
-  max => 'INT_MAX / 2',
-},
-
-{ name => 'enable_dynamic_shared_buffers', type => 'bool', context => 'PGC_POSTMASTER', group => 'RESOURCES_MEM',
-  short_desc => 'Enables dynamic resizing of the shared buffer pool.',
-  variable => 'enable_dynamic_shared_buffers',
-  boot_val => 'false',
-},
-
 { name => 'shared_memory_size', type => 'int', context => 'PGC_INTERNAL', group => 'PRESET_OPTIONS',
   short_desc => 'Shows the size of the server\'s main shared memory area (rounded up to the nearest MB).',
   flags => 'GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_FILE | GUC_UNIT_MB | GUC_RUNTIME_COMPUTED',
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 904470a3f34..d205887cbd6 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12694,9 +12694,9 @@
   proargtypes => 'oid8 int8', prosrc => 'hashoid8extended' },
 
 # Online shared buffer pool resizing (see src/backend/storage/buffer/buf_resize.c)
-{ oid => '6500', descr => 'resize the shared buffer pool to a new size',
+{ oid => '9594', descr => 'resize the shared buffer pool to a new size',
   proname => 'pg_resize_shared_buffers', provolatile => 'v', proretset => 't',
-  prorettype => 'record', proargtypes => 'text',
+  prorows => '10', prorettype => 'record', proargtypes => 'text',
   proallargtypes => '{text,text,float8,text}',
   proargmodes => '{i,o,o,o}',
   proargnames => '{new_size,key,value,unit}',
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 1c5989c2472..89367225d52 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -596,6 +596,15 @@ extern int	BufTableLookup(BufferTag *tagPtr, uint32 hashcode);
 extern int	BufTableInsert(BufferTag *tagPtr, uint32 hashcode, int buf_id);
 extern void BufTableDelete(BufferTag *tagPtr, uint32 hashcode);
 
+/* buf_table.c -- online bucket/entry array resizing (see buf_resize.c) */
+extern Size BufTableEntriesExpand(int old_size, int new_size, bool *success);
+extern Size BufTableEntriesShrink(int old_size, int new_size, bool *success);
+extern Size BufTableBucketsExpand(int new_size, bool *success);
+extern Size BufTableBucketsShrink(int new_size, bool *success);
+#ifdef USE_ASSERT_CHECKING
+extern void BufTableAssertShrinkInvariants(int old_size, int new_size);
+#endif
+
 /* localbuf.c */
 extern bool PinLocalBuffer(BufferDesc *buf_hdr, bool adjust_usagecount);
 extern void UnpinLocalBuffer(Buffer buffer);
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index 3dbaf364133..ce35a68a78e 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -377,6 +377,19 @@ extern Size BufferManagerShmemExpand(int lowNBuffers, int highNBuffers, bool *su
 extern void BufferManagerShmemInitBuffers(int lowNBuffers, int highNBuffers);
 extern Size BufferManagerShmemShrink(int lowNBuffers, int highNBuffers, bool *success);
 
+/*
+ * Shared-memory array (de)population primitives, shared between the buffer pool
+ * (buf_init.c) and the buffer mapping table (buf_table.c) so both resize their
+ * MaxNBuffers-sized arrays with identical page-aligned madvise() handling.
+ */
+extern Size buffer_pool_madvise_alignment(void);
+extern Size BufferPoolArrayPhysicalExpand(void *baseptr, Size elem_size,
+										  int lowNBuffers, int highNBuffers,
+										  bool *success);
+extern Size BufferPoolArrayPhysicalShrink(void *baseptr, Size elem_size,
+										  int lowNBuffers, int highNBuffers,
+										  bool *success);
+
 /* in bufmgr.c */
 extern bool EvictExtraBuffers(int lowNBuffers, int highNBuffers);
 
diff --git a/src/include/storage/dynamic_shared_buffers.h b/src/include/storage/dynamic_shared_buffers.h
index 9d57b25e15d..b2dceff2472 100644
--- a/src/include/storage/dynamic_shared_buffers.h
+++ b/src/include/storage/dynamic_shared_buffers.h
@@ -90,8 +90,6 @@ GetLowNBuffers(void)
 	return pg_atomic_read_u32(&DSBCtrl->lowNBuffers);
 }
 
-extern void DSBControlInit(void);
-
 /*
  * Try to claim coordinator status for a buffer-pool resize. Returns true if
  * we became the coordinator (caller must eventually call
diff --git a/src/include/storage/subsystemlist.h b/src/include/storage/subsystemlist.h
index 9ad619080be..9842c501011 100644
--- a/src/include/storage/subsystemlist.h
+++ b/src/include/storage/subsystemlist.h
@@ -39,6 +39,7 @@ PG_SHMEM_SUBSYSTEM(CLOGShmemCallbacks)
 PG_SHMEM_SUBSYSTEM(CommitTsShmemCallbacks)
 PG_SHMEM_SUBSYSTEM(SUBTRANSShmemCallbacks)
 PG_SHMEM_SUBSYSTEM(MultiXactShmemCallbacks)
+PG_SHMEM_SUBSYSTEM(DSBShmemCallbacks)
 PG_SHMEM_SUBSYSTEM(BufferManagerShmemCallbacks)
 PG_SHMEM_SUBSYSTEM(StrategyCtlShmemCallbacks)
 PG_SHMEM_SUBSYSTEM(BufTableShmemCallbacks)