Have the heap check if current page needs scan-time pruning

Add scan-time heap page pruning functionality and a mechanism to track page
modifications during scans and re-evaluate pruning eligibility.  This addresses
the issue where multiple updates during a scan don't trigger heap page pruning
until the next scan.

Author: gburd

src/backend/access/heap/heapam.c

@@ -446,6 +446,11 @@ initscan(HeapScanDesc scan, ScanKey key, bool keep_startblock)
 	scan->rs_ntuples = 0;
 	scan->rs_cindex = 0;
 
+	/* Initialize scan-time pruning tracking */
+	scan->rs_page_updates = 0;
+	scan->rs_page_pruned = false;
+	scan->rs_last_pruned_block = InvalidBlockNumber;
+
 	/*
 	 * Initialize to ForwardScanDirection because it is most common and
 	 * because heap scans go forward before going backward (e.g. CURSORs).
@@ -925,6 +930,10 @@ heapgettup(HeapScanDesc scan,
 
 		Assert(BufferGetBlockNumber(scan->rs_cbuf) == scan->rs_cblock);
 
+		/* Reset page tracking for new page */
+		scan->rs_page_updates = 0;
+		scan->rs_page_pruned = false;
+
 		LockBuffer(scan->rs_cbuf, BUFFER_LOCK_SHARE);
 		page = heapgettup_start_page(scan, dir, &linesleft, &lineoff);
 continue_page:
@@ -942,7 +951,12 @@ heapgettup(HeapScanDesc scan,
 			ItemId		lpp = PageGetItemId(page, lineoff);
 
 			if (!ItemIdIsNormal(lpp))
+			{
+				/* Track dead line pointers as potential modifications */
+				if (ItemIdIsDead(lpp))
+					scan->rs_page_updates++;
 				continue;
+			}
 
 			tuple->t_data = (HeapTupleHeader) PageGetItem(page, lpp);
 			tuple->t_len = ItemIdGetLength(lpp);
@@ -972,10 +986,27 @@ heapgettup(HeapScanDesc scan,
 		}
 
 		/*
-		 * if we get here, it means we've exhausted the items on this page and
-		 * it's time to move to the next.
+		 * Before moving to next page, check if current page needs scan-time
+		 * pruning. This addresses the issue where multiple updates during a
+		 * scan don't trigger pruning until the next scan.
 		 */
 		LockBuffer(scan->rs_cbuf, BUFFER_LOCK_UNLOCK);
+		if (scan->rs_page_updates > 0 &&
+			scan->rs_cblock != scan->rs_last_pruned_block &&
+			PageNeedsScanPruning(page, scan->rs_page_updates))
+		{
+			/* Attempt opportunistic pruning */
+			heap_page_prune_opt(scan->rs_base.rs_rd, scan->rs_cbuf, 0);
+
+			/* Mark this block as pruned to avoid repeated attempts */
+			scan->rs_last_pruned_block = scan->rs_cblock;
+			scan->rs_page_pruned = true;
+		}
+
+		/*
+		 * If we get here, it means we've exhausted the items on this page and
+		 * it's time to move to the next.
+		 */
 	}
 
 	/* end of scan */
@@ -1042,6 +1073,10 @@ heapgettup_pagemode(HeapScanDesc scan,
 
 		Assert(BufferGetBlockNumber(scan->rs_cbuf) == scan->rs_cblock);
 
+		/* Reset page tracking for new page */
+		scan->rs_page_updates = 0;
+		scan->rs_page_pruned = false;
+
 		/* prune the page and determine visible tuple offsets */
 		heap_prepare_pagescan((TableScanDesc) scan);
 		page = BufferGetPage(scan->rs_cbuf);
@@ -1079,6 +1114,15 @@ heapgettup_pagemode(HeapScanDesc scan,
 		}
 	}
 
+	/* Before ending scan, check if current page needs scan-time pruning */
+	if (BufferIsValid(scan->rs_cbuf) && scan->rs_page_updates > 0 &&
+		scan->rs_cblock != scan->rs_last_pruned_block &&
+		PageNeedsScanPruning(BufferGetPage(scan->rs_cbuf), scan->rs_page_updates))
+	{
+		heap_page_prune_opt(scan->rs_base.rs_rd, scan->rs_cbuf, 0);
+		scan->rs_last_pruned_block = scan->rs_cblock;
+	}
+
 	/* end of scan */
 	if (BufferIsValid(scan->rs_cbuf))
 		ReleaseBuffer(scan->rs_cbuf);

src/include/access/heapam.h

@@ -97,6 +97,12 @@ typedef struct HeapScanDescData
 	uint32		rs_cindex;		/* current tuple's index in vistuples */
 	uint32		rs_ntuples;		/* number of visible tuples on page */
 	OffsetNumber rs_vistuples[MaxHeapTuplesPerPage];	/* their offsets */
+
+	/* scan-time pruning tracking */
+	int			rs_page_updates;	/* count of updates/deletes on current
+									 * page */
+	bool		rs_page_pruned; /* whether current page was already pruned */
+	BlockNumber rs_last_pruned_block;	/* last block we attempted pruning on */
 } HeapScanDescData;
 typedef struct HeapScanDescData *HeapScanDesc;
 

src/include/storage/bufpage.h

@@ -424,6 +424,12 @@ PageHasPrunable(const PageData *page)
 	return PageHasFreeLinePointers(page) && PageGetMaxOffsetNumber(page) >= 4;
 }
 
+static inline bool
+PageNeedsScanPruning(const PageData *page, int modifications)
+{
+	return modifications >= 3 && PageHasPrunable(page);
+}
+
 static inline void
 PageSetFull(Page page)
 {