From 7f998a084205ff08e1e95a432a5258515c0ca693 Mon Sep 17 00:00:00 2001
From: Amit Langote <amitlan@postgresql.org>
Date: Mon, 22 Jun 2026 15:40:35 +0000
Subject: [PATCH v8 1/3] heapam: store full HeapTupleData in rs_vistuples[] for
 pagemode scans

page_collect_tuples() builds full HeapTupleData headers for every
visible tuple on a page -- t_data, t_len, t_self, t_tableOid -- but
previously discarded them immediately after writing just the
OffsetNumber of each survivor into rs_vistuples[].  heapgettup_pagemode()
then re-derived those same values on every call by reading the saved
OffsetNumber, calling PageGetItemId() and PageGetItem() to locate the
tuple on the page, and populating rs_ctup field by field.

Change rs_vistuples[] element type from OffsetNumber to HeapTupleData
and populate it inside page_collect_tuples() while lpp, lineoff, page,
and block are already in scope, so no additional page reads are needed.
A new rs_ctup_p field on HeapScanDesc points directly into
rs_vistuples[] for pagemode scans (or into rs_ctup for the non-pagemode
path).  Callers read through rs_ctup_p and check rs_ctup_p == NULL for
scan exhaustion instead of rs_ctup.t_data == NULL.

The same simplification applies to heapam_scan_bitmap_next_tuple() and
BitmapHeapScanNextBlock(); SampleHeapTupleVisible() extracts the offset
from rs_vistuples[].t_self.ip_posid.

Having pre-built HeapTupleData headers available at the scan descriptor
level also lays groundwork for a batched tuple interface, where an AM
can serve multiple tuples per call without repeating the line-pointer
traversal.

Suggested-by: Andres Freund <andres@anarazel.de>
---
 src/backend/access/heap/heapam.c            | 126 ++++++++++----------
 src/backend/access/heap/heapam_handler.c    |  22 +---
 src/backend/access/heap/heapam_visibility.c |  19 ++-
 src/include/access/heapam.h                 |   7 +-
 4 files changed, 79 insertions(+), 95 deletions(-)

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index abfd8e8970a..2b85e6805b8 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -462,6 +462,7 @@ initscan(HeapScanDesc scan, ScanKey key, bool keep_startblock)
 	scan->rs_numblocks = InvalidBlockNumber;
 	scan->rs_inited = false;
 	scan->rs_ctup.t_data = NULL;
+	scan->rs_ctup_p = NULL;
 	ItemPointerSetInvalid(&scan->rs_ctup.t_self);
 	scan->rs_cbuf = InvalidBuffer;
 	scan->rs_cblock = InvalidBlockNumber;
@@ -526,7 +527,6 @@ page_collect_tuples(HeapScanDesc scan, Snapshot snapshot,
 					BlockNumber block, int lines,
 					bool all_visible, bool check_serializable)
 {
-	Oid			relid = RelationGetRelid(scan->rs_base.rs_rd);
 	int			ntup = 0;
 	int			nvis = 0;
 	BatchMVCCState batchmvcc;
@@ -534,43 +534,27 @@ page_collect_tuples(HeapScanDesc scan, Snapshot snapshot,
 	/* page at a time should have been disabled otherwise */
 	Assert(IsMVCCSnapshot(snapshot));
 
-	/* first find all tuples on the page */
+	/*
+	 * Collect all normal tuples on the page into rs_vistuples[].
+	 * Every entry gets a fully populated HeapTupleData header with
+	 * t_data pointing into the pinned page.  t_tableOid was set
+	 * once at beginscan time.
+	 */
 	for (OffsetNumber lineoff = FirstOffsetNumber; lineoff <= lines; lineoff++)
 	{
 		ItemId		lpp = PageGetItemId(page, lineoff);
-		HeapTuple	tup;
+		HeapTuple   tup = &scan->rs_vistuples[ntup];
 
 		if (unlikely(!ItemIdIsNormal(lpp)))
 			continue;
 
-		/*
-		 * If the page is not all-visible or we need to check serializability,
-		 * maintain enough state to be able to refind the tuple efficiently,
-		 * without again first needing to fetch the item and then via that the
-		 * tuple.
-		 */
-		if (!all_visible || check_serializable)
-		{
-			tup = &batchmvcc.tuples[ntup];
+		tup->t_data = (HeapTupleHeader) PageGetItem(page, lpp);
+		tup->t_len = ItemIdGetLength(lpp);
+		Assert(tup->t_tableOid == RelationGetRelid(scan->rs_base.rs_rd));
+		ItemPointerSet(&(tup->t_self), block, lineoff);
 
-			tup->t_data = (HeapTupleHeader) PageGetItem(page, lpp);
-			tup->t_len = ItemIdGetLength(lpp);
-			tup->t_tableOid = relid;
-			ItemPointerSet(&(tup->t_self), block, lineoff);
-		}
-
-		/*
-		 * If the page is all visible, these fields otherwise won't be
-		 * populated in loop below.
-		 */
-		if (all_visible)
-		{
-			if (check_serializable)
-			{
-				batchmvcc.visible[ntup] = true;
-			}
-			scan->rs_vistuples[ntup] = lineoff;
-		}
+		if (all_visible && check_serializable)
+			batchmvcc.visible[ntup] = true;
 
 		ntup++;
 	}
@@ -600,11 +584,30 @@ page_collect_tuples(HeapScanDesc scan, Snapshot snapshot,
 		{
 			HeapCheckForSerializableConflictOut(batchmvcc.visible[i],
 												scan->rs_base.rs_rd,
-												&batchmvcc.tuples[i],
+												&scan->rs_vistuples[i],
 												buffer, snapshot);
 		}
 	}
 
+	/*
+	 * Compact rs_vistuples[] to contain only visible survivors in
+	 * [0..nvis-1].  This is done here rather than inside
+	 * HeapTupleSatisfiesMVCCBatch() because the serializable conflict
+	 * check above needs visible[i] and rs_vistuples[i] to correspond
+	 * before compaction.  Callers iterate rs_vistuples[] directly
+	 * without rechecking visibility.
+	 */
+	if (!all_visible)
+	{
+		int dst = 0;
+		for (int i = 0; i < ntup; i++)
+		{
+			if (batchmvcc.visible[i])
+				scan->rs_vistuples[dst++] = scan->rs_vistuples[i];
+		}
+		Assert(dst == nvis);
+	}
+
 	return nvis;
 }
 
@@ -1035,6 +1038,7 @@ continue_page:
 
 			LockBuffer(scan->rs_cbuf, BUFFER_LOCK_UNLOCK);
 			scan->rs_coffset = lineoff;
+			scan->rs_ctup_p = &scan->rs_ctup;
 			return;
 		}
 
@@ -1052,7 +1056,7 @@ continue_page:
 	scan->rs_cbuf = InvalidBuffer;
 	scan->rs_cblock = InvalidBlockNumber;
 	scan->rs_prefetch_block = InvalidBlockNumber;
-	tuple->t_data = NULL;
+	scan->rs_ctup_p = NULL;
 	scan->rs_inited = false;
 }
 
@@ -1075,15 +1079,13 @@ heapgettup_pagemode(HeapScanDesc scan,
 					int nkeys,
 					ScanKey key)
 {
-	HeapTuple	tuple = &(scan->rs_ctup);
-	Page		page;
 	uint32		lineindex;
 	uint32		linesleft;
 
 	if (likely(scan->rs_inited))
 	{
 		/* continue from previously returned page/tuple */
-		page = BufferGetPage(scan->rs_cbuf);
+		Assert(BufferIsValid(scan->rs_cbuf));
 
 		lineindex = scan->rs_cindex + dir;
 		if (ScanDirectionIsForward(dir))
@@ -1111,37 +1113,32 @@ heapgettup_pagemode(HeapScanDesc scan,
 
 		/* prune the page and determine visible tuple offsets */
 		heap_prepare_pagescan((TableScanDesc) scan);
-		page = BufferGetPage(scan->rs_cbuf);
 		linesleft = scan->rs_ntuples;
 		lineindex = ScanDirectionIsForward(dir) ? 0 : linesleft - 1;
 
-		/* block is the same for all tuples, set it once outside the loop */
-		ItemPointerSetBlockNumber(&tuple->t_self, scan->rs_cblock);
-
 		/* lineindex now references the next or previous visible tid */
 continue_page:
 
 		for (; linesleft > 0; linesleft--, lineindex += dir)
 		{
-			ItemId		lpp;
-			OffsetNumber lineoff;
-
-			Assert(lineindex < scan->rs_ntuples);
-			lineoff = scan->rs_vistuples[lineindex];
-			lpp = PageGetItemId(page, lineoff);
-			Assert(ItemIdIsNormal(lpp));
-
-			tuple->t_data = (HeapTupleHeader) PageGetItem(page, lpp);
-			tuple->t_len = ItemIdGetLength(lpp);
-			ItemPointerSetOffsetNumber(&tuple->t_self, lineoff);
-
 			/* skip any tuples that don't match the scan key */
-			if (key != NULL &&
-				!HeapKeyTest(tuple, RelationGetDescr(scan->rs_base.rs_rd),
-							 nkeys, key))
+			if (key != NULL)
+			{
+				/*
+				 * Headers were pre-built by page_collect_tuples() into
+				 * rs_vistuples[].  Copy the entry; t_data still points into
+				 * the pinned page, which is safe for the lifetime of the
+				 * current page scan.
+				 */
+				HeapTuple	tuple = &scan->rs_vistuples[lineindex];
+
+				if (!HeapKeyTest(tuple, RelationGetDescr(scan->rs_base.rs_rd),
+								 nkeys, key))
 				continue;
+			}
 
 			scan->rs_cindex = lineindex;
+			scan->rs_ctup_p = &scan->rs_vistuples[lineindex];
 			return;
 		}
 	}
@@ -1152,7 +1149,7 @@ continue_page:
 	scan->rs_cbuf = InvalidBuffer;
 	scan->rs_cblock = InvalidBlockNumber;
 	scan->rs_prefetch_block = InvalidBlockNumber;
-	tuple->t_data = NULL;
+	scan->rs_ctup_p = NULL;
 	scan->rs_inited = false;
 }
 
@@ -1248,6 +1245,8 @@ heap_beginscan(Relation relation, Snapshot snapshot,
 
 	/* we only need to set this up once */
 	scan->rs_ctup.t_tableOid = RelationGetRelid(relation);
+	for (int i = 0; i < MaxHeapTuplesPerPage; i++)
+		scan->rs_vistuples[i].t_tableOid = RelationGetRelid(relation);
 
 	/*
 	 * Allocate memory to keep track of page allocation for parallel workers
@@ -1457,7 +1456,7 @@ heap_getnext(TableScanDesc sscan, ScanDirection direction)
 		heapgettup(scan, direction,
 				   scan->rs_base.rs_nkeys, scan->rs_base.rs_key);
 
-	if (scan->rs_ctup.t_data == NULL)
+	if (scan->rs_ctup_p == NULL)
 		return NULL;
 
 	/*
@@ -1467,7 +1466,7 @@ heap_getnext(TableScanDesc sscan, ScanDirection direction)
 
 	pgstat_count_heap_getnext(scan->rs_base.rs_rd);
 
-	return &scan->rs_ctup;
+	return scan->rs_ctup_p;
 }
 
 bool
@@ -1482,7 +1481,7 @@ heap_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *s
 	else
 		heapgettup(scan, direction, sscan->rs_nkeys, sscan->rs_key);
 
-	if (scan->rs_ctup.t_data == NULL)
+	if (scan->rs_ctup_p == NULL)
 	{
 		ExecClearTuple(slot);
 		return false;
@@ -1495,8 +1494,7 @@ heap_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *s
 
 	pgstat_count_heap_getnext(scan->rs_base.rs_rd);
 
-	ExecStoreBufferHeapTuple(&scan->rs_ctup, slot,
-							 scan->rs_cbuf);
+	ExecStoreBufferHeapTuple(scan->rs_ctup_p, slot, scan->rs_cbuf);
 	return true;
 }
 
@@ -1589,7 +1587,7 @@ heap_getnextslot_tidrange(TableScanDesc sscan, ScanDirection direction,
 		else
 			heapgettup(scan, direction, sscan->rs_nkeys, sscan->rs_key);
 
-		if (scan->rs_ctup.t_data == NULL)
+		if (scan->rs_ctup_p == NULL)
 		{
 			ExecClearTuple(slot);
 			return false;
@@ -1601,7 +1599,7 @@ heap_getnextslot_tidrange(TableScanDesc sscan, ScanDirection direction,
 		 * we're scanning for.  Here we must filter out any tuples from these
 		 * pages that are outside of that range.
 		 */
-		if (ItemPointerCompare(&scan->rs_ctup.t_self, mintid) < 0)
+		if (ItemPointerCompare(&scan->rs_ctup_p->t_self, mintid) < 0)
 		{
 			ExecClearTuple(slot);
 
@@ -1620,7 +1618,7 @@ heap_getnextslot_tidrange(TableScanDesc sscan, ScanDirection direction,
 		 * Likewise for the final page, we must filter out TIDs greater than
 		 * maxtid.
 		 */
-		if (ItemPointerCompare(&scan->rs_ctup.t_self, maxtid) > 0)
+		if (ItemPointerCompare(&scan->rs_ctup_p->t_self, maxtid) > 0)
 		{
 			ExecClearTuple(slot);
 
@@ -1643,7 +1641,7 @@ heap_getnextslot_tidrange(TableScanDesc sscan, ScanDirection direction,
 	 */
 	pgstat_count_heap_getnext(scan->rs_base.rs_rd);
 
-	ExecStoreBufferHeapTuple(&scan->rs_ctup, slot, scan->rs_cbuf);
+	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 2268cc277bc..d4c2c67f564 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -2099,9 +2099,6 @@ heapam_scan_bitmap_next_tuple(TableScanDesc scan,
 {
 	BitmapHeapScanDesc bscan = (BitmapHeapScanDesc) scan;
 	HeapScanDesc hscan = (HeapScanDesc) bscan;
-	OffsetNumber targoffset;
-	Page		page;
-	ItemId		lp;
 
 	/*
 	 * Out of range?  If so, nothing more to look at on this page
@@ -2115,16 +2112,7 @@ heapam_scan_bitmap_next_tuple(TableScanDesc scan,
 		if (!BitmapHeapScanNextBlock(scan, recheck, lossy_pages, exact_pages))
 			return false;
 	}
-
-	targoffset = hscan->rs_vistuples[hscan->rs_cindex];
-	page = BufferGetPage(hscan->rs_cbuf);
-	lp = PageGetItemId(page, targoffset);
-	Assert(ItemIdIsNormal(lp));
-
-	hscan->rs_ctup.t_data = (HeapTupleHeader) PageGetItem(page, lp);
-	hscan->rs_ctup.t_len = ItemIdGetLength(lp);
-	hscan->rs_ctup.t_tableOid = scan->rs_rd->rd_id;
-	ItemPointerSet(&hscan->rs_ctup.t_self, hscan->rs_cblock, targoffset);
+	hscan->rs_ctup_p = &hscan->rs_vistuples[hscan->rs_cindex];
 
 	pgstat_count_heap_fetch(scan->rs_rd);
 
@@ -2132,7 +2120,7 @@ heapam_scan_bitmap_next_tuple(TableScanDesc scan,
 	 * Set up the result slot to point to this tuple.  Note that the slot
 	 * acquires a pin on the buffer.
 	 */
-	ExecStoreBufferHeapTuple(&hscan->rs_ctup,
+	ExecStoreBufferHeapTuple(hscan->rs_ctup_p,
 							 slot,
 							 hscan->rs_cbuf);
 
@@ -2479,7 +2467,7 @@ SampleHeapTupleVisible(TableScanDesc scan, Buffer buffer,
 		while (start < end)
 		{
 			uint32		mid = start + (end - start) / 2;
-			OffsetNumber curoffset = hscan->rs_vistuples[mid];
+			OffsetNumber curoffset = hscan->rs_vistuples[mid].t_self.ip_posid;
 
 			if (tupoffset == curoffset)
 				return true;
@@ -2599,7 +2587,7 @@ BitmapHeapScanNextBlock(TableScanDesc scan,
 			ItemPointerSet(&tid, block, offnum);
 			if (heap_hot_search_buffer(&tid, scan->rs_rd, buffer, snapshot,
 									   &heapTuple, NULL, true))
-				hscan->rs_vistuples[ntup++] = ItemPointerGetOffsetNumber(&tid);
+				hscan->rs_vistuples[ntup++] = heapTuple;
 		}
 	}
 	else
@@ -2628,7 +2616,7 @@ BitmapHeapScanNextBlock(TableScanDesc scan,
 			valid = HeapTupleSatisfiesVisibility(&loctup, snapshot, buffer);
 			if (valid)
 			{
-				hscan->rs_vistuples[ntup++] = offnum;
+				hscan->rs_vistuples[ntup++] = loctup;
 				PredicateLockTID(scan->rs_rd, &loctup.t_self, snapshot,
 								 HeapTupleHeaderGetXmin(loctup.t_data));
 			}
diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c
index 361b76e5065..235493f0e99 100644
--- a/src/backend/access/heap/heapam_visibility.c
+++ b/src/backend/access/heap/heapam_visibility.c
@@ -1674,13 +1674,13 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
  * Perform HeapTupleSatisfiesMVCC() on each passed in tuple. This is more
  * efficient than doing HeapTupleSatisfiesMVCC() one-by-one.
  *
- * To be checked tuples are passed via BatchMVCCState->tuples. Each tuple's
- * visibility is stored in batchmvcc->visible[]. In addition,
- * ->vistuples_dense is set to contain the offsets of visible tuples.
+ * Each tuple's visibility is stored in batchmvcc->visible[].  The caller
+ * is responsible for compacting the tuples array to contain only visible
+ * survivors after this function returns.
  *
- * The reason this is more efficient than HeapTupleSatisfiesMVCC() is that it
- * avoids a cross-translation-unit function call for each tuple, allows the
- * compiler to optimize across calls to HeapTupleSatisfiesMVCC and allows
+ * The reason this is more efficient than HeapTupleSatisfiesMVCC() is that
+ * it avoids a cross-translation-unit function call for each tuple, allows
+ * the compiler to optimize across calls to HeapTupleSatisfiesMVCC and allows
  * setting hint bits more efficiently (see the one BufferFinishSetHintBits()
  * call below).
  *
@@ -1690,7 +1690,7 @@ int
 HeapTupleSatisfiesMVCCBatch(Snapshot snapshot, Buffer buffer,
 							int ntups,
 							BatchMVCCState *batchmvcc,
-							OffsetNumber *vistuples_dense)
+							HeapTupleData *tuples)
 {
 	int			nvis = 0;
 	SetHintBitsState state = SHB_INITIAL;
@@ -1700,16 +1700,13 @@ HeapTupleSatisfiesMVCCBatch(Snapshot snapshot, Buffer buffer,
 	for (int i = 0; i < ntups; i++)
 	{
 		bool		valid;
-		HeapTuple	tup = &batchmvcc->tuples[i];
+		HeapTuple	tup = &tuples[i];
 
 		valid = HeapTupleSatisfiesMVCC(tup, snapshot, buffer, &state);
 		batchmvcc->visible[i] = valid;
 
 		if (likely(valid))
-		{
-			vistuples_dense[nvis] = tup->t_self.ip_posid;
 			nvis++;
-		}
 	}
 
 	if (state == SHB_ENABLED)
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index 5176478c295..c17076455bd 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -102,7 +102,9 @@ typedef struct HeapScanDescData
 	/* these fields only used in page-at-a-time mode and for bitmap scans */
 	uint32		rs_cindex;		/* current tuple's index in vistuples */
 	uint32		rs_ntuples;		/* number of visible tuples on page */
-	OffsetNumber rs_vistuples[MaxHeapTuplesPerPage];	/* their offsets */
+	HeapTupleData rs_vistuples[MaxHeapTuplesPerPage];	/* tuples */
+	HeapTuple	rs_ctup_p;	/* points to current tuple in rs_vistuples[]
+							 * or &rs_ctup depending on scan mode */
 } HeapScanDescData;
 typedef struct HeapScanDescData *HeapScanDesc;
 
@@ -498,14 +500,13 @@ extern bool HeapTupleIsSurelyDead(HeapTuple htup,
  */
 typedef struct BatchMVCCState
 {
-	HeapTupleData tuples[MaxHeapTuplesPerPage];
 	bool		visible[MaxHeapTuplesPerPage];
 } BatchMVCCState;
 
 extern int	HeapTupleSatisfiesMVCCBatch(Snapshot snapshot, Buffer buffer,
 										int ntups,
 										BatchMVCCState *batchmvcc,
-										OffsetNumber *vistuples_dense);
+										HeapTupleData *tuples);
 
 /*
  * To avoid leaking too much knowledge about reorderbuffer implementation
-- 
2.47.3

