From 5aa95474f6808ff2167b7613f77d712adc05ca1f Mon Sep 17 00:00:00 2001
From: Amit Langote <amitlan@postgresql.org>
Date: Fri, 3 Jul 2026 12:35:04 +0900
Subject: [PATCH v8 3/3] Implement batched sequential scans for heap

Implement the batch scan interface for heapam and use it from the
sequential scan node, so a heap SeqScan fetches the visible tuples of a
page per AM call and serves them to the executor one tuple at a time.

Heap
----
heap_getnextbatch() advances to the next block via heap_fetch_next_buffer()
and runs heap_prepare_pagescan() to populate rs_vistuples[], then exposes
the page's tuples through the scan slot with ExecStoreBatchBufferHeapTuples().

BatchBufferHeapTupleTableSlot (TTSOpsBatchBufferHeapTuple) extends
BufferHeapTupleTableSlot with a pointer into rs_vistuples[], a tuple count,
and a cursor.  It holds its own pin on the page buffer so the tuples stay
valid for the slot's lifetime.  It deforms exactly as TTSOpsBufferHeapTuple
does and shares getsomeattrs, getsysattr and the copy callbacks with it;
only clear, materialize and batch_next differ.  materialize copies the
current tuple and retains the buffer pin, since the remaining batch entries
still reference the page.

batch_next moves to the tuple at cindex + dir, the same resume rule
heapgettup_pagemode() uses with rs_cindex, giving one path for forward and
backward iteration and for a direction change within a batch.
ExecStoreBatchBufferHeapTuples positions the cursor before the first tuple
of the requested direction.

TTS_IS_BUFFERTUPLE() recognizes the batch slot; pgstat_count_heap_getnext()
takes a tuple count so a batch is accounted in one call.

Sequential scan
---------------
A heap SeqScan uses the batch slot as its scan slot: ExecInitSeqScan
obtains it from table_slot_batch_callbacks().  SeqNext returns the next
tuple of the current batch with slot_batch_next() and fetches the next
batch with table_scan_getnextbatch() once the current one is drained,
passing es_direction so a single path covers both scan directions.  The
ExecSeqScan variants drive the batch slot as they do any scan slot and are
unchanged.

Required callbacks
------------------
scan_getnextbatch and batch_slot_callbacks are now required of any table
AM: the sequential scan node uses them with no fallback, so an AM that
omits them cannot be sequentially scanned.  GetTableAmRoutine() asserts
their presence alongside the other required callbacks.

Expression evaluation
---------------------
A scan slot's ops are recorded as a fixed deform target (fetch.kind) when
scan expressions are compiled, and a node may evaluate the same expression
against more than one slot type: an EvalPlanQual recheck reuses the scan's
expressions against the EPQ slot, a TTSOpsBufferHeapTuple.  The batch slot
and TTSOpsBufferHeapTuple share their deform code, so
CheckOpSlotCompatibility() treats them, and TTSOpsHeapTuple, as equivalent
fixed targets.
---
 src/backend/access/heap/heapam.c         |  73 ++++++++-
 src/backend/access/heap/heapam_handler.c |   9 +-
 src/backend/access/table/tableamapi.c    |   2 +
 src/backend/executor/execExprInterp.c    |  23 +++
 src/backend/executor/execTuples.c        | 198 ++++++++++++++++++++++-
 src/backend/executor/nodeSeqscan.c       |  24 ++-
 src/include/access/heapam.h              |   2 +
 src/include/executor/tuptable.h          |  37 ++++-
 src/include/pgstat.h                     |   4 +-
 9 files changed, 355 insertions(+), 17 deletions(-)

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 2b85e6805b8..d8fb8d998c7 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -1464,7 +1464,7 @@ heap_getnext(TableScanDesc sscan, ScanDirection direction)
 	 * the proper return buffer and return the tuple.
 	 */
 
-	pgstat_count_heap_getnext(scan->rs_base.rs_rd);
+	pgstat_count_heap_getnext(scan->rs_base.rs_rd, 1);
 
 	return scan->rs_ctup_p;
 }
@@ -1492,12 +1492,79 @@ heap_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *s
 	 * the proper return buffer and return the tuple.
 	 */
 
-	pgstat_count_heap_getnext(scan->rs_base.rs_rd);
+	pgstat_count_heap_getnext(scan->rs_base.rs_rd, 1);
 
 	ExecStoreBufferHeapTuple(scan->rs_ctup_p, slot, scan->rs_cbuf);
 	return true;
 }
 
+/*
+ * heap_getnextbatch
+ *
+ * Modeled on heapgettup_pagemode() but without the per-tuple
+ * iteration loop.  Advances to the next page with visible tuples
+ * via heap_fetch_next_buffer(), calls heap_prepare_pagescan() to
+ * populate rs_vistuples[], then hands the whole array to the slot.
+ *
+ * The caller must fully consume the previous batch (via batch_next)
+ * before calling again.
+ *
+ * Returns true if a non-empty batch was produced, false when the
+ * scan is exhausted.
+ */
+bool
+heap_getnextbatch(TableScanDesc sscan,
+				  ScanDirection direction,
+				  TupleTableSlot *slot)
+{
+	HeapScanDesc scan = (HeapScanDesc) sscan;
+
+	/*
+	 * Each call returns a full page worth of tuples, so there is no
+	 * resume-mid-page path like heapgettup_pagemode's rs_inited +
+	 * goto continue_page.  We always advance to the next block.
+	 */
+	while (true)
+	{
+		heap_fetch_next_buffer(scan, direction);
+
+		/* did we run out of blocks to scan? */
+		if (!BufferIsValid(scan->rs_cbuf))
+			break;
+
+		Assert(BufferGetBlockNumber(scan->rs_cbuf) == scan->rs_cblock);
+
+		/* prune the page and determine visible tuple offsets */
+		heap_prepare_pagescan(sscan);
+
+		if (scan->rs_ntuples == 0)
+			continue;		/* empty page, try next */
+
+		/*
+		 * Hand the whole page to the slot.
+		 */
+		ExecStoreBatchBufferHeapTuples(scan->rs_vistuples,
+									   scan->rs_ntuples,
+									   scan->rs_cbuf,
+									   direction,
+									   slot);
+
+		pgstat_count_heap_getnext(scan->rs_base.rs_rd, scan->rs_ntuples);
+		scan->rs_inited = true;
+		return true;
+	}
+
+	/* end of scan */
+	if (BufferIsValid(scan->rs_cbuf))
+		ReleaseBuffer(scan->rs_cbuf);
+	scan->rs_cbuf = InvalidBuffer;
+	scan->rs_cblock = InvalidBlockNumber;
+	scan->rs_prefetch_block = InvalidBlockNumber;
+	scan->rs_inited = false;
+
+	return false;
+}
+
 void
 heap_set_tidrange(TableScanDesc sscan, ItemPointer mintid,
 				  ItemPointer maxtid)
@@ -1639,7 +1706,7 @@ heap_getnextslot_tidrange(TableScanDesc sscan, ScanDirection direction,
 	 * if we get here it means we have a new current scan tuple, so point to
 	 * the proper return buffer and return the tuple.
 	 */
-	pgstat_count_heap_getnext(scan->rs_base.rs_rd);
+	pgstat_count_heap_getnext(scan->rs_base.rs_rd, 1);
 
 	ExecStoreBufferHeapTuple(scan->rs_ctup_p, slot, scan->rs_cbuf);
 	return true;
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index d4c2c67f564..9a6e5e4707a 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -79,6 +79,11 @@ heapam_slot_callbacks(Relation relation)
 	return &TTSOpsBufferHeapTuple;
 }
 
+static const TupleTableSlotOps *
+heap_batch_slot_callbacks(Relation rel)
+{
+	return &TTSOpsBatchBufferHeapTuple;
+}
 
 /* ------------------------------------------------------------------------
  * Callbacks for non-modifying operations on individual tuples for heap AM
@@ -2293,7 +2298,7 @@ heapam_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate,
 			ExecStoreBufferHeapTuple(tuple, slot, hscan->rs_cbuf);
 
 			/* Count successfully-fetched tuples as heap fetches */
-			pgstat_count_heap_getnext(scan->rs_rd);
+			pgstat_count_heap_getnext(scan->rs_rd, 1);
 
 			return true;
 		}
@@ -2654,11 +2659,13 @@ static const TableAmRoutine heapam_methods = {
 	.type = T_TableAmRoutine,
 
 	.slot_callbacks = heapam_slot_callbacks,
+	.batch_slot_callbacks = heap_batch_slot_callbacks,
 
 	.scan_begin = heap_beginscan,
 	.scan_end = heap_endscan,
 	.scan_rescan = heap_rescan,
 	.scan_getnextslot = heap_getnextslot,
+	.scan_getnextbatch = heap_getnextbatch,
 
 	.scan_set_tidrange = heap_set_tidrange,
 	.scan_getnextslot_tidrange = heap_getnextslot_tidrange,
diff --git a/src/backend/access/table/tableamapi.c b/src/backend/access/table/tableamapi.c
index 5450a27faeb..9ef6b5cca65 100644
--- a/src/backend/access/table/tableamapi.c
+++ b/src/backend/access/table/tableamapi.c
@@ -45,6 +45,8 @@ GetTableAmRoutine(Oid amhandler)
 	Assert(routine->scan_end != NULL);
 	Assert(routine->scan_rescan != NULL);
 	Assert(routine->scan_getnextslot != NULL);
+	Assert(routine->scan_getnextbatch != NULL);
+	Assert(routine->batch_slot_callbacks != NULL);
 
 	Assert(routine->parallelscan_estimate != NULL);
 	Assert(routine->parallelscan_initialize != NULL);
diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c
index 0634af964a9..c61c0872f35 100644
--- a/src/backend/executor/execExprInterp.c
+++ b/src/backend/executor/execExprInterp.c
@@ -2458,6 +2458,29 @@ CheckOpSlotCompatibility(ExprEvalStep *op, TupleTableSlot *slot)
 		op->d.fetch.kind == &TTSOpsBufferHeapTuple)
 		return;
 
+	/*
+	 * The batch buffer-heap slot shares getsomeattrs/getsysattr with
+	 * TTSOpsBufferHeapTuple, so it deforms identically to it, and buffer-heap
+	 * and heap are already interchangeable above.  Extend that same
+	 * equivalence to the batch slot, in either direction.
+	 */
+
+	/*
+	 * Compiled for the batch slot, run against a buffer-heap/heap slot.  This
+	 * is the case that arises today: a scan expression compiled for the batch
+	 * slot is re-run under EPQ against the plain buffer-heap EPQ slot.
+	 */
+	if (op->d.fetch.kind == &TTSOpsBatchBufferHeapTuple &&
+		(slot->tts_ops == &TTSOpsBufferHeapTuple ||
+		 slot->tts_ops == &TTSOpsHeapTuple))
+		return;
+
+	/* the reverse: compiled for buffer-heap/heap, run against the batch slot */
+	if (slot->tts_ops == &TTSOpsBatchBufferHeapTuple &&
+		(op->d.fetch.kind == &TTSOpsBufferHeapTuple ||
+		 op->d.fetch.kind == &TTSOpsHeapTuple))
+		return;
+
 	/*
 	 * At the moment we consider it OK if a virtual slot is used instead of a
 	 * specific type of slot, as a virtual slot never needs to be deformed.
diff --git a/src/backend/executor/execTuples.c b/src/backend/executor/execTuples.c
index 7f4ebf95432..d565ed017d6 100644
--- a/src/backend/executor/execTuples.c
+++ b/src/backend/executor/execTuples.c
@@ -992,6 +992,133 @@ tts_buffer_heap_store_tuple(TupleTableSlot *slot, HeapTuple tuple,
 	}
 }
 
+/*
+ * clear -- release buffer pin and reset batch state.
+ */
+static void
+tts_batch_buffer_heap_clear(TupleTableSlot *slot)
+{
+	BatchBufferHeapTupleTableSlot *bslot =
+		(BatchBufferHeapTupleTableSlot *) slot;
+	BufferHeapTupleTableSlot *bhslot = &bslot->base;
+
+
+	if (TTS_SHOULDFREE(slot))
+	{
+		/* batch slots may still have a valid buffer here */
+		heap_freetuple(bhslot->base.tuple);
+		slot->tts_flags &= ~TTS_FLAG_SHOULDFREE;
+	}
+
+	if (BufferIsValid(bhslot->buffer))
+	{
+		ReleaseBuffer(bhslot->buffer);
+		bhslot->buffer = InvalidBuffer;
+	}
+	bhslot->base.tuple = NULL;
+	bhslot->base.off = 0;
+
+	bslot->tuples = NULL;
+	bslot->ntuples = 0;
+	bslot->cindex = -1;
+
+	slot->tts_nvalid = 0;
+	slot->tts_flags |= TTS_FLAG_EMPTY;
+	ItemPointerSetInvalid(&slot->tts_tid);
+}
+
+static void
+tts_batch_buffer_heap_materialize(TupleTableSlot *slot)
+{
+	BatchBufferHeapTupleTableSlot *bslot =
+		(BatchBufferHeapTupleTableSlot *) slot;
+	BufferHeapTupleTableSlot *bhslot = &bslot->base;
+	HeapTupleTableSlot *hslot = &bhslot->base;
+	MemoryContext oldContext;
+
+	Assert(!TTS_EMPTY(slot));
+
+	/* already materialized */
+	if (TTS_SHOULDFREE(slot))
+		return;
+
+	oldContext = MemoryContextSwitchTo(slot->tts_mcxt);
+
+	hslot->off = 0;
+	slot->tts_nvalid = 0;
+
+	if (!hslot->tuple)
+	{
+		hslot->tuple = heap_form_tuple(slot->tts_tupleDescriptor,
+									   slot->tts_values,
+									   slot->tts_isnull);
+	}
+	else
+	{
+		hslot->tuple = heap_copytuple(hslot->tuple);
+
+		/*
+		 * Unlike the non-batch case, do NOT release the buffer pin.
+		 * The remaining tuples in the batch still have t_data pointing
+		 * into the pinned page.  The pin is released when the slot
+		 * is cleared after the batch is fully consumed.
+		 */
+	}
+
+	slot->tts_flags |= TTS_FLAG_SHOULDFREE;
+	MemoryContextSwitchTo(oldContext);
+}
+
+/*
+ * tts_batch_buffer_heap_next -- advance to the next tuple in the batch.
+ *
+ * Points the underlying BufferHeapTupleTableSlot at tuples[cursor]
+ * and bumps the cursor.  Returns false when the batch is exhausted.
+ */
+static bool
+tts_batch_buffer_heap_next(TupleTableSlot *slot, ScanDirection direction)
+{
+	BatchBufferHeapTupleTableSlot *bslot =
+		(BatchBufferHeapTupleTableSlot *) slot;
+	HeapTupleTableSlot *hslot = &bslot->base.base;
+	int			dir = (int) direction;		/* +1 forward, -1 backward */
+	int			next;
+
+	Assert(direction != NoMovementScanDirection);
+
+	/*
+	 * Resume relative to the last-returned tuple, mirroring
+	 * heapgettup_pagemode()'s rs_cindex + dir logic.  This keeps the
+	 * iteration correct across a scan-direction change within a batch:
+	 * after returning index i, a forward step yields i+1 and a backward
+	 * step yields i-1, regardless of how the batch was first entered.
+	 */
+	next = bslot->cindex + dir;
+
+	if (next < 0 || next >= bslot->ntuples)
+		return false;
+
+	/*
+	 * If the slot was materialized (e.g. by a parent node), free the
+	 * palloc'd copy before overwriting the pointer.
+	 */
+	if (TTS_SHOULDFREE(slot))
+	{
+		heap_freetuple(hslot->tuple);
+		slot->tts_flags &= ~TTS_FLAG_SHOULDFREE;
+	}
+
+	hslot->tuple = &bslot->tuples[next];
+	hslot->off = 0;
+	bslot->cindex = next;
+
+	slot->tts_nvalid = 0;
+	slot->tts_flags &= ~TTS_FLAG_EMPTY;
+	slot->tts_tid = hslot->tuple->t_self;
+
+	return true;
+}
+
 /*
  * slot_deform_heap_tuple
  *		Given a TupleTableSlot, extract data from the slot's physical tuple
@@ -1281,7 +1408,8 @@ const TupleTableSlotOps TTSOpsVirtual = {
 	.get_heap_tuple = NULL,
 	.get_minimal_tuple = NULL,
 	.copy_heap_tuple = tts_virtual_copy_heap_tuple,
-	.copy_minimal_tuple = tts_virtual_copy_minimal_tuple
+	.copy_minimal_tuple = tts_virtual_copy_minimal_tuple,
+	.batch_next = NULL
 };
 
 const TupleTableSlotOps TTSOpsHeapTuple = {
@@ -1299,7 +1427,8 @@ const TupleTableSlotOps TTSOpsHeapTuple = {
 	/* A heap tuple table slot can not "own" a minimal tuple. */
 	.get_minimal_tuple = NULL,
 	.copy_heap_tuple = tts_heap_copy_heap_tuple,
-	.copy_minimal_tuple = tts_heap_copy_minimal_tuple
+	.copy_minimal_tuple = tts_heap_copy_minimal_tuple,
+	.batch_next = NULL
 };
 
 const TupleTableSlotOps TTSOpsMinimalTuple = {
@@ -1317,7 +1446,8 @@ const TupleTableSlotOps TTSOpsMinimalTuple = {
 	.get_heap_tuple = NULL,
 	.get_minimal_tuple = tts_minimal_get_minimal_tuple,
 	.copy_heap_tuple = tts_minimal_copy_heap_tuple,
-	.copy_minimal_tuple = tts_minimal_copy_minimal_tuple
+	.copy_minimal_tuple = tts_minimal_copy_minimal_tuple,
+	.batch_next = NULL
 };
 
 const TupleTableSlotOps TTSOpsBufferHeapTuple = {
@@ -1335,9 +1465,29 @@ const TupleTableSlotOps TTSOpsBufferHeapTuple = {
 	/* A buffer heap tuple table slot can not "own" a minimal tuple. */
 	.get_minimal_tuple = NULL,
 	.copy_heap_tuple = tts_buffer_heap_copy_heap_tuple,
-	.copy_minimal_tuple = tts_buffer_heap_copy_minimal_tuple
+	.copy_minimal_tuple = tts_buffer_heap_copy_minimal_tuple,
+	.batch_next = NULL
 };
 
+/*
+ * Everything except clear(), materialize(), batch_next() shared with
+ * BufferHeapTuple.
+ */
+const TupleTableSlotOps TTSOpsBatchBufferHeapTuple = {
+	.base_slot_size = sizeof(BatchBufferHeapTupleTableSlot),
+	.init = tts_buffer_heap_init,
+	.release = tts_buffer_heap_release,
+	.clear = tts_batch_buffer_heap_clear,
+	.getsomeattrs = tts_buffer_heap_getsomeattrs,
+	.getsysattr = tts_buffer_heap_getsysattr,
+	.materialize = tts_batch_buffer_heap_materialize,
+	.copyslot = tts_buffer_heap_copyslot,
+	.get_heap_tuple = tts_buffer_heap_get_heap_tuple,
+	.get_minimal_tuple = NULL,
+	.copy_heap_tuple = tts_buffer_heap_copy_heap_tuple,
+	.copy_minimal_tuple = tts_buffer_heap_copy_minimal_tuple,
+	.batch_next = tts_batch_buffer_heap_next
+};
 
 /* ----------------------------------------------------------------
  *				  tuple table create/delete functions
@@ -1708,6 +1858,46 @@ ExecStorePinnedBufferHeapTuple(HeapTuple tuple,
 	return slot;
 }
 
+/*
+ * ExecStoreBatchBufferHeapTuples -- load a batch into a
+ * BatchBufferHeapTupleTableSlot.
+ *
+ * tuples[] must remain valid (buffer pinned) for the slot's lifetime.
+ * No copy -- the slot takes a pointer and a count.
+ */
+void
+ExecStoreBatchBufferHeapTuples(HeapTupleData *tuples,
+							   int ntuples,
+							   Buffer buffer,
+							   ScanDirection direction,
+							   TupleTableSlot *slot)
+{
+	BatchBufferHeapTupleTableSlot *bslot =
+		(BatchBufferHeapTupleTableSlot *) slot;
+	BufferHeapTupleTableSlot *bhslot = &bslot->base;
+
+	Assert(ntuples > 0);
+
+	bslot->tuples = tuples;
+	bslot->ntuples = ntuples;
+
+	/*
+	 * Position the cursor just before the first tuple to return, so that
+	 * the first batch_next (which advances by the scan direction) lands on
+	 * the correct edge: index 0 for a forward scan, ntuples-1 for backward.
+	 */
+	if (ScanDirectionIsBackward(direction))
+		bslot->cindex = ntuples;	/* ntuples + (-1) = ntuples - 1 */
+	else
+		bslot->cindex = -1;			/* -1 + 1 = 0 */
+
+	if (BufferIsValid(bhslot->buffer))
+		ReleaseBuffer(bhslot->buffer);
+
+	bhslot->buffer = buffer;
+	IncrBufferRefCount(buffer);
+}
+
 /*
  * Store a minimal tuple into TTSOpsMinimalTuple type slot.
  *
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index 5bcb0a861d7..943944ededa 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -85,11 +85,22 @@ SeqNext(SeqScanState *node)
 	}
 
 	/*
-	 * get the next tuple from the table
+	 * Serve the next tuple from the current batch held in the scan slot.
+	 * When the batch is exhausted, ask the AM for the next one.  The AM
+	 * builds the batch honoring scan direction and sets the slot's batch
+	 * cursor accordingly, so this single path serves forward and backward
+	 * scans alike.
 	 */
-	if (table_scan_getnextslot(scandesc, direction, slot))
+	if (slot_batch_next(slot, direction))
 		return slot;
-	return NULL;
+
+	ExecClearTuple(slot);
+
+	if (!table_scan_getnextbatch(scandesc, direction, slot))
+		return slot;
+
+	slot_batch_next(slot, direction);
+	return slot;
 }
 
 /*
@@ -220,6 +231,7 @@ SeqScanState *
 ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 {
 	SeqScanState *scanstate;
+	Relation	rel;
 
 	/*
 	 * Once upon a time it was possible to have an outerPlan of a SeqScan, but
@@ -245,15 +257,15 @@ ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 	/*
 	 * open the scan relation
 	 */
-	scanstate->ss.ss_currentRelation =
+	scanstate->ss.ss_currentRelation = rel =
 		ExecOpenScanRelation(estate,
 							 node->scan.scanrelid,
 							 eflags);
 
 	/* and create slot with the appropriate rowtype */
 	ExecInitScanTupleSlot(estate, &scanstate->ss,
-						  RelationGetDescr(scanstate->ss.ss_currentRelation),
-						  table_slot_callbacks(scanstate->ss.ss_currentRelation),
+						  RelationGetDescr(rel),
+						  table_slot_batch_callbacks(rel),
 						  TTS_FLAG_OBEYS_NOT_NULL_CONSTRAINTS);
 
 	/*
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index c17076455bd..c3340f8e29a 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -362,6 +362,8 @@ extern void heap_endscan(TableScanDesc sscan);
 extern HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction);
 extern bool heap_getnextslot(TableScanDesc sscan,
 							 ScanDirection direction, TupleTableSlot *slot);
+extern bool heap_getnextbatch(TableScanDesc sscan,
+							  ScanDirection direction, TupleTableSlot *slot);
 extern void heap_set_tidrange(TableScanDesc sscan, ItemPointer mintid,
 							  ItemPointer maxtid);
 extern bool heap_getnextslot_tidrange(TableScanDesc sscan,
diff --git a/src/include/executor/tuptable.h b/src/include/executor/tuptable.h
index 890115314b0..524a3b78772 100644
--- a/src/include/executor/tuptable.h
+++ b/src/include/executor/tuptable.h
@@ -260,11 +260,13 @@ extern PGDLLIMPORT const TupleTableSlotOps TTSOpsVirtual;
 extern PGDLLIMPORT const TupleTableSlotOps TTSOpsHeapTuple;
 extern PGDLLIMPORT const TupleTableSlotOps TTSOpsMinimalTuple;
 extern PGDLLIMPORT const TupleTableSlotOps TTSOpsBufferHeapTuple;
+extern PGDLLIMPORT const TupleTableSlotOps TTSOpsBatchBufferHeapTuple;
 
 #define TTS_IS_VIRTUAL(slot) ((slot)->tts_ops == &TTSOpsVirtual)
 #define TTS_IS_HEAPTUPLE(slot) ((slot)->tts_ops == &TTSOpsHeapTuple)
 #define TTS_IS_MINIMALTUPLE(slot) ((slot)->tts_ops == &TTSOpsMinimalTuple)
-#define TTS_IS_BUFFERTUPLE(slot) ((slot)->tts_ops == &TTSOpsBufferHeapTuple)
+#define TTS_IS_BUFFERTUPLE(slot) ((slot)->tts_ops == &TTSOpsBufferHeapTuple || \
+								  (slot)->tts_ops == &TTSOpsBatchBufferHeapTuple)
 
 
 /*
@@ -309,6 +311,34 @@ typedef struct BufferHeapTupleTableSlot
 	Buffer		buffer;			/* tuple's buffer, or InvalidBuffer */
 } BufferHeapTupleTableSlot;
 
+/*
+ * BatchBufferHeapTupleTableSlot -- a batch of heap tuples residing in a
+ * pinned buffer page.
+ *
+ * Extends BufferHeapTupleTableSlot to hold a pointer to an array of
+ * pre-built HeapTupleData entries living in HeapScanDesc.rs_vistuples[].
+ * The tuple headers' t_data pointers reference the pinned buffer page
+ * and remain valid for the lifetime of the pin.  No data is copied.
+ *
+ * When consumed tuple-at-a-time by the existing executor machinery,
+ * the batch_next slot ops callback points the underlying
+ * BufferHeapTupleTableSlot at successive tuples in the array.
+ * Standard slot_getsomeattrs() then deforms the current tuple
+ * through the normal heap deform path.
+ *
+ * Other table AMs that support batching would define their own batch
+ * slot type with AM-specific internals and TupleTableSlotOps.
+ */
+typedef struct BatchBufferHeapTupleTableSlot
+{
+	BufferHeapTupleTableSlot base;
+
+	HeapTupleData  *tuples;		/* points into HeapScanDesc->rs_vistuples[] */
+	int				ntuples;	/* number of tuples in this batch */
+	int				cindex;		/* index of last-returned tuple, or -1 if
+								 * none returned from this batch yet */
+} BatchBufferHeapTupleTableSlot;
+
 typedef struct MinimalTupleTableSlot
 {
 	pg_node_attr(abstract)
@@ -360,6 +390,11 @@ extern TupleTableSlot *ExecStoreBufferHeapTuple(HeapTuple tuple,
 extern TupleTableSlot *ExecStorePinnedBufferHeapTuple(HeapTuple tuple,
 													  TupleTableSlot *slot,
 													  Buffer buffer);
+extern void ExecStoreBatchBufferHeapTuples(HeapTupleData *tuples,
+										   int ntuples,
+										   Buffer buffer,
+										   ScanDirection direction,
+										   TupleTableSlot *slot);
 extern TupleTableSlot *ExecStoreMinimalTuple(MinimalTuple mtup,
 											 TupleTableSlot *slot,
 											 bool shouldFree);
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 58a44857f13..bc3f56a6724 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -732,10 +732,10 @@ extern void pgstat_report_analyze(Relation rel,
 		if (pgstat_should_count_relation(rel))						\
 			(rel)->pgstat_info->counts.numscans++;					\
 	} while (0)
-#define pgstat_count_heap_getnext(rel)								\
+#define pgstat_count_heap_getnext(rel, n)							\
 	do {															\
 		if (pgstat_should_count_relation(rel))						\
-			(rel)->pgstat_info->counts.tuples_returned++;			\
+			(rel)->pgstat_info->counts.tuples_returned += (n);		\
 	} while (0)
 #define pgstat_count_heap_fetch(rel)								\
 	do {															\
-- 
2.47.3

