Prefetch cleanup: (#246)

- Prefetch the pages in index vacuum's sequential scans
   Implemented in NBTREE, GIST and SP-GIST.
   BRIN does not have a 2nd phase of vacuum, and both GIN and HASH clean up
   their indexes in a non-seqscan fashion: GIN scans the btree from left to
   right, and HASH only scans the initial buckets sequentially.

Author: MMeent

contrib/pg_prewarm/pg_prewarm.c

@@ -198,7 +198,8 @@ pg_prewarm(PG_FUNCTION_ARGS)
 		for (block = first_block; block <= last_block; ++block)
 		{
 			Buffer buf;
-			int prefetch_stop = block + Min(last_block - block + 1, io_concurrency);
+			BlockNumber prefetch_stop = block + Min(last_block - block + 1,
+													io_concurrency);
 			CHECK_FOR_INTERRUPTS();
 			while (prefetch_block < prefetch_stop)
 			{

src/backend/access/gist/gistvacuum.c

@@ -23,6 +23,7 @@
 #include "storage/indexfsm.h"
 #include "storage/lmgr.h"
 #include "utils/memutils.h"
+#include "utils/spccache.h"
 
 /* Working state needed by gistbulkdelete */
 typedef struct
@@ -130,8 +131,14 @@ gistvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	BlockNumber num_pages;
 	bool		needLock;
 	BlockNumber blkno;
+	BlockNumber prefetch_blkno;
+	int			io_concurrency;
 	MemoryContext oldctx;
 
+	io_concurrency = get_tablespace_maintenance_io_concurrency(
+		rel->rd_rel->reltablespace
+	);
+
 	/*
 	 * Reset fields that track information about the entire index now.  This
 	 * avoids double-counting in the case where a single VACUUM command
@@ -209,6 +216,7 @@ gistvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	needLock = !RELATION_IS_LOCAL(rel);
 
 	blkno = GIST_ROOT_BLKNO;
+	prefetch_blkno = blkno;
 	for (;;)
 	{
 		/* Get the current relation length */
@@ -221,9 +229,21 @@ gistvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 		/* Quit if we've scanned the whole relation */
 		if (blkno >= num_pages)
 			break;
+
+		if (prefetch_blkno < blkno)
+			prefetch_blkno = blkno;
+		for (; prefetch_blkno < num_pages &&
+			   prefetch_blkno < blkno + io_concurrency; prefetch_blkno++)
+			PrefetchBuffer(rel, MAIN_FORKNUM, prefetch_blkno);
+
 		/* Iterate over pages, then loop back to recheck length */
 		for (; blkno < num_pages; blkno++)
+		{
+			if (io_concurrency > 0 && prefetch_blkno < num_pages)
+				PrefetchBuffer(rel, MAIN_FORKNUM, prefetch_blkno++);
+
 			gistvacuumpage(&vstate, blkno, blkno);
+		}
 	}
 
 	/*

src/backend/access/hash/hash.c

@@ -32,6 +32,7 @@
 #include "utils/builtins.h"
 #include "utils/index_selfuncs.h"
 #include "utils/rel.h"
+#include "utils/spccache.h"
 
 /* Working state for hashbuild and its callback */
 typedef struct
@@ -466,13 +467,17 @@ hashbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	Bucket		orig_maxbucket;
 	Bucket		cur_maxbucket;
 	Bucket		cur_bucket;
+	Bucket		prf_bucket;
 	Buffer		metabuf = InvalidBuffer;
 	HashMetaPage metap;
 	HashMetaPage cachedmetap;
+	int			io_concurrency;
 
 	tuples_removed = 0;
 	num_index_tuples = 0;
 
+	io_concurrency = get_tablespace_maintenance_io_concurrency(rel->rd_rel->reltablespace);
+
 	/*
 	 * We need a copy of the metapage so that we can use its hashm_spares[]
 	 * values to compute bucket page addresses, but a cached copy should be
@@ -487,9 +492,14 @@ hashbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 
 	/* Scan the buckets that we know exist */
 	cur_bucket = 0;
+	prf_bucket = cur_bucket;
 	cur_maxbucket = orig_maxbucket;
 
 loop_top:
+	for (; prf_bucket <= cur_maxbucket &&
+		   prf_bucket < cur_bucket + io_concurrency; prf_bucket++)
+		PrefetchBuffer(rel, MAIN_FORKNUM, BUCKET_TO_BLKNO(cachedmetap, prf_bucket));
+
 	while (cur_bucket <= cur_maxbucket)
 	{
 		BlockNumber bucket_blkno;
@@ -500,6 +510,12 @@ hashbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 		Page		page;
 		bool		split_cleanup = false;
 
+		if (io_concurrency > 0 && prf_bucket <= cur_maxbucket)
+		{
+			PrefetchBuffer(rel, MAIN_FORKNUM, BUCKET_TO_BLKNO(cachedmetap, prf_bucket));
+			prf_bucket++;
+		}
+
 		/* Get address of bucket's start page */
 		bucket_blkno = BUCKET_TO_BLKNO(cachedmetap, cur_bucket);
 

src/backend/access/heap/vacuumlazy.c

@@ -981,7 +981,8 @@ lazy_scan_heap(LVRelState *vacrel)
 		 */
 		visibilitymap_pin(vacrel->rel, blkno, &vmbuffer);
 
-		if (vacrel->io_concurrency > 0) {
+		if (vacrel->io_concurrency > 0)
+		{
 			/*
 			 * Prefetch io_concurrency blocks ahead
 			 */
@@ -2464,32 +2465,40 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 
 		tblk = ItemPointerGetBlockNumber(&vacrel->dead_items->items[index]);
 
-		if (vacrel->io_concurrency > 0) {
+		if (vacrel->io_concurrency > 0)
+		{
 			/*
 			 * If we're just starting out, prefetch N consecutive blocks.
 			 * If not, only the next 1 block
 			 */
-			if (pindex == 0) {
+			if (pindex == 0)
+			{
 				int prefetch_budget = Min(vacrel->dead_items->num_items,
 										  Min(vacrel->rel_pages,
 											  vacrel->io_concurrency));
 				BlockNumber prev_prefetch = ItemPointerGetBlockNumber(&vacrel->dead_items->items[pindex]);
 				PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, prev_prefetch);
 
 				while (++pindex < vacrel->dead_items->num_items &&
-					   prefetch_budget > 0) {
+					   prefetch_budget > 0)
+				{
 					ItemPointer ptr = &vacrel->dead_items->items[pindex];
-					if (ItemPointerGetBlockNumber(ptr) != prev_prefetch) {
+					if (ItemPointerGetBlockNumber(ptr) != prev_prefetch)
+					{
 						prev_prefetch = ItemPointerGetBlockNumber(ptr);
 						prefetch_budget -= 1;
 						PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, prev_prefetch);
 					}
 				}
-			} else if (pindex < vacrel->dead_items->num_items) {
+			}
+			else if (pindex < vacrel->dead_items->num_items)
+			{
 				BlockNumber previous = ItemPointerGetBlockNumber(&vacrel->dead_items->items[pindex]);
-				while (++pindex < vacrel->dead_items->num_items) {
+				while (++pindex < vacrel->dead_items->num_items)
+				{
 					BlockNumber toPrefetch = ItemPointerGetBlockNumber(&vacrel->dead_items->items[pindex]);
-					if (previous != toPrefetch) {
+					if (previous != toPrefetch)
+					{
 						PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, toPrefetch);
 						break;
 					}

src/backend/access/nbtree/nbtree.c

@@ -37,6 +37,7 @@
 #include "utils/builtins.h"
 #include "utils/index_selfuncs.h"
 #include "utils/memutils.h"
+#include "utils/spccache.h"
 
 
 /*
@@ -909,6 +910,8 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	BTVacState	vstate;
 	BlockNumber num_pages;
 	BlockNumber scanblkno;
+	BlockNumber prefetch_blkno;
+	int			io_concurrency;
 	bool		needLock;
 
 	/*
@@ -948,6 +951,9 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	vstate.maxbufsize = 0;
 	vstate.pendingpages = NULL;
 	vstate.npendingpages = 0;
+
+	io_concurrency = get_tablespace_maintenance_io_concurrency(rel->rd_rel->reltablespace);
+
 	/* Consider applying _bt_pendingfsm_finalize optimization */
 	_bt_pendingfsm_init(rel, &vstate, (callback == NULL));
 
@@ -976,6 +982,8 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	needLock = !RELATION_IS_LOCAL(rel);
 
 	scanblkno = BTREE_METAPAGE + 1;
+	prefetch_blkno = scanblkno;
+
 	for (;;)
 	{
 		/* Get the current relation length */
@@ -992,9 +1000,19 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 		/* Quit if we've scanned the whole relation */
 		if (scanblkno >= num_pages)
 			break;
+
+		if (prefetch_blkno < scanblkno)
+			prefetch_blkno = scanblkno;
+		for (; prefetch_blkno < num_pages &&
+			   prefetch_blkno < scanblkno + io_concurrency; prefetch_blkno++)
+			PrefetchBuffer(rel, MAIN_FORKNUM, prefetch_blkno);
+
 		/* Iterate over pages, then loop back to recheck length */
 		for (; scanblkno < num_pages; scanblkno++)
 		{
+			if (io_concurrency > 0 && prefetch_blkno < num_pages)
+				PrefetchBuffer(rel, MAIN_FORKNUM, prefetch_blkno++);
+
 			btvacuumpage(&vstate, scanblkno);
 			if (info->report_progress)
 				pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_DONE,

src/backend/access/spgist/spgvacuum.c

@@ -27,6 +27,7 @@
 #include "storage/indexfsm.h"
 #include "storage/lmgr.h"
 #include "utils/snapmgr.h"
+#include "utils/spccache.h"
 
 
 /* Entry in pending-list of TIDs we need to revisit */
@@ -796,7 +797,14 @@ spgvacuumscan(spgBulkDeleteState *bds)
 	Relation	index = bds->info->index;
 	bool		needLock;
 	BlockNumber num_pages,
-				blkno;
+				blkno,
+				prefetch_blkno;
+	int			io_concurrency;
+
+	/* initiate concurrency */
+	io_concurrency = get_tablespace_maintenance_io_concurrency(
+		index->rd_rel->reltablespace
+	);
 
 	/* Finish setting up spgBulkDeleteState */
 	initSpGistState(&bds->spgstate, index);
@@ -824,6 +832,8 @@ spgvacuumscan(spgBulkDeleteState *bds)
 	 * in btvacuumscan().
 	 */
 	blkno = SPGIST_METAPAGE_BLKNO + 1;
+	prefetch_blkno = blkno;
+
 	for (;;)
 	{
 		/* Get the current relation length */
@@ -836,9 +846,19 @@ spgvacuumscan(spgBulkDeleteState *bds)
 		/* Quit if we've scanned the whole relation */
 		if (blkno >= num_pages)
 			break;
+
+		if (prefetch_blkno < blkno)
+			prefetch_blkno = blkno;
+		for (; prefetch_blkno < num_pages &&
+			   prefetch_blkno < blkno + io_concurrency; prefetch_blkno++)
+			PrefetchBuffer(index, MAIN_FORKNUM, prefetch_blkno);
+
 		/* Iterate over pages, then loop back to recheck length */
 		for (; blkno < num_pages; blkno++)
 		{
+			if (io_concurrency > 0 && prefetch_blkno < num_pages)
+				PrefetchBuffer(index, MAIN_FORKNUM, prefetch_blkno++);
+
 			spgvacuumpage(bds, blkno);
 			/* empty the pending-list after each page */
 			if (bds->pendingList != NULL)