Optimize prefetch patterns in both heap seqscan and vacuum scans. (#228)

Previously, we called PrefetchBuffer [NBlkScanned * seqscan_prefetch_buffers]
times in each of those situations, but now only NBlkScanned.

In addition, the prefetch mechanism for the vacuum scans is now based on
blocks instead of tuples - improving the efficiency.

Author: MMeent

src/backend/access/heap/heapam.c

@@ -399,19 +399,32 @@ heapgetpage(TableScanDesc sscan, BlockNumber page)
 	 */
 	CHECK_FOR_INTERRUPTS();
 
-	/* Prefetch next block */
-	if (enable_seqscan_prefetch)
+	/* Prefetch up to seqscan_prefetch_buffers blocks ahead */
+	if (enable_seqscan_prefetch && seqscan_prefetch_buffers > 0)
 	{
-		int prefetch_limit = seqscan_prefetch_buffers;
+		uint32 prefetch_limit = seqscan_prefetch_buffers;
+		BlockNumber	prefetch_start = page;
 		ParallelBlockTableScanWorker pbscanwork = scan->rs_parallelworkerdata;
+
 		if (pbscanwork != NULL && pbscanwork->phsw_chunk_remaining < prefetch_limit)
 			prefetch_limit = pbscanwork->phsw_chunk_remaining;
-		if (page + prefetch_limit >= scan->rs_nblocks)
-			prefetch_limit = scan->rs_nblocks - page - 1;
 
-		smgr_reset_prefetch(RelationGetSmgr(scan->rs_base.rs_rd));
+		/*
+		 * If this is the first page, initiate prefetch of pages page..page + n.
+		 * On each subsequent call, prefetch the next page that we haven't
+		 * prefetched yet, at page + n.
+		 */
+		if (scan->rs_startblock != page)
+		{
+			prefetch_start = (page + prefetch_limit - 1) % scan->rs_nblocks;
+			prefetch_limit = 1;
+		}
+		else
+			prefetch_start = page;
+
 		for (int i = 1; i <= prefetch_limit; i++)
-			PrefetchBuffer(scan->rs_base.rs_rd, MAIN_FORKNUM, page+i);
+			PrefetchBuffer(scan->rs_base.rs_rd, MAIN_FORKNUM,
+						   (prefetch_start+i) % scan->rs_nblocks);
 	}
 
 	/* read page using selected strategy */

src/backend/access/heap/vacuumlazy.c

@@ -971,12 +971,27 @@ lazy_scan_heap(LVRelState *vacrel)
 		 */
 		visibilitymap_pin(vacrel->rel, blkno, &vmbuffer);
 
-		if (enable_seqscan_prefetch)
+		if (enable_seqscan_prefetch && seqscan_prefetch_buffers > 0)
 		{
-			int prefetch_limit = Min(rel_pages - blkno - 1, seqscan_prefetch_buffers);
-			smgr_reset_prefetch(RelationGetSmgr(vacrel->rel));
-			for (int i = 1; i <= prefetch_limit; i++)
-				PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, blkno+i);
+			/*
+			 * If we're starting the scan, we need to prefetch the first N pages.
+			 * If not, we need to only prefetch page blkno+n.
+			 */
+			if (blkno == 0)
+			{
+				int prefetch_limit = Min(rel_pages - blkno - 1,
+										 seqscan_prefetch_buffers);
+
+				for (int i = 1; i <= prefetch_limit; i++)
+					PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, blkno+i);
+			}
+			else
+			{
+				/* No need to prefetch past the end of the relation */
+				if (blkno + seqscan_prefetch_buffers < rel_pages)
+					PrefetchBuffer(vacrel->rel, MAIN_FORKNUM,
+								   blkno + seqscan_prefetch_buffers);
+			}
 		}
 
 		/* Finished preparatory checks.  Actually scan the page. */
@@ -2402,7 +2417,8 @@ lazy_vacuum_all_indexes(LVRelState *vacrel)
 static void
 lazy_vacuum_heap_rel(LVRelState *vacrel)
 {
-	int			index;
+	int			index,
+				pindex;
 	BlockNumber vacuumed_pages;
 	Buffer		vmbuffer = InvalidBuffer;
 	LVSavedErrInfo saved_err_info;
@@ -2423,6 +2439,7 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 	vacuumed_pages = 0;
 
 	index = 0;
+	pindex = 0;
 	while (index < vacrel->dead_items->num_items)
 	{
 		BlockNumber tblk;
@@ -2433,13 +2450,44 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 		vacuum_delay_point();
 
 		tblk = ItemPointerGetBlockNumber(&vacrel->dead_items->items[index]);
-		if (enable_seqscan_prefetch)
+
+		if (enable_seqscan_prefetch && seqscan_prefetch_buffers > 0)
 		{
-			int prefetch_limit = Min(vacrel->dead_items->num_items - index - 1, seqscan_prefetch_buffers);
-			smgr_reset_prefetch(RelationGetSmgr(vacrel->rel));
-			for (int i = 1; i <= prefetch_limit; i++)
-				PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, ItemPointerGetBlockNumber(&vacrel->dead_items->items[index + i]));
+			/*
+			 * If we're just starting out, prefetch N consecutive blocks.
+			 * If not, only the next 1 block
+			 */
+			if (index == 0)
+			{
+				int prefetch_limit = Min(vacrel->dead_items->num_items - 1,
+										 Min(vacrel->rel_pages,
+											 seqscan_prefetch_buffers));
+				BlockNumber prev_prefetch = 0;
+
+				while (++pindex < vacrel->dead_items->num_items &&
+					   prefetch_limit > 0)
+				{
+					ItemPointer ptr = &vacrel->dead_items->items[pindex];
+					if (ItemPointerGetBlockNumber(ptr) != prev_prefetch)
+					{
+						prev_prefetch = ItemPointerGetBlockNumber(ptr);
+						prefetch_limit -= 1;
+						PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, prev_prefetch);
+					}
+				}
+			}
+			else
+			{
+				BlockNumber toPrefetch = ItemPointerGetBlockNumber(&vacrel->dead_items->items[pindex]);
+				while (pindex < vacrel->dead_items->num_items)
+				{
+					if (toPrefetch != ItemPointerGetBlockNumber(&vacrel->dead_items->items[pindex]))
+						break;
+				}
+				PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, toPrefetch);
+			}
 		}
+
 		vacrel->blkno = tblk;
 		buf = ReadBufferExtended(vacrel->rel, MAIN_FORKNUM, tblk, RBM_NORMAL,
 								 vacrel->bstrategy);

src/backend/storage/smgr/smgr.c

@@ -38,7 +38,6 @@ static const f_smgr smgr_md = {
 		.smgr_unlink = mdunlink,
 		.smgr_extend = mdextend,
 		.smgr_prefetch = mdprefetch,
-		.smgr_reset_prefetch = md_reset_prefetch,
 		.smgr_read = mdread,
 		.smgr_write = mdwrite,
 		.smgr_writeback = mdwriteback,
@@ -533,15 +532,6 @@ smgrprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 	return (*reln->smgr).smgr_prefetch(reln, forknum, blocknum);
 }
 
-/*
- *	smgr_reset_prefetch() -- Cancel all previos prefetch requests
- */
-void
-smgr_reset_prefetch(SMgrRelation reln)
-{
-	(*reln->smgr).smgr_reset_prefetch(reln);
-}
-
 /*
  *	smgrread() -- read a particular block from a relation into the supplied
  *				  buffer.

src/include/storage/smgr.h

@@ -159,7 +159,6 @@ extern void smgrextend(SMgrRelation reln, ForkNumber forknum,
 					   BlockNumber blocknum, char *buffer, bool skipFsync);
 extern bool smgrprefetch(SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum);
-extern void smgr_reset_prefetch(SMgrRelation reln);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
 					 BlockNumber blocknum, char *buffer);
 extern void smgrwrite(SMgrRelation reln, ForkNumber forknum,