From a5664f6fa42f615d1fb46250ac61aa9ac8a9d32d Mon Sep 17 00:00:00 2001
From: "Andrey V. Lepikhov" <a.lepikhov@postgrespro.ru>
Date: Thu, 20 Sep 2018 13:04:26 +0500
Subject: [PATCH] Quick Vacuum Strategy

---
 src/backend/access/heap/heapam.c    |  31 +++++++
 src/backend/access/heap/pruneheap.c |  41 ++++++++--
 src/backend/catalog/index.c         | 122 ++++++++++++++++++++++++++++
 src/backend/commands/vacuumlazy.c   |  32 ++++++--
 src/include/access/heapam.h         |   2 +
 src/include/catalog/index.h         |   4 +
 6 files changed, 220 insertions(+), 12 deletions(-)

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 56f1d82f96..0f0949bef3 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -9412,6 +9412,36 @@ heap_sync(Relation rel)
 	}
 }
 
+/*
+ * Mask DEAD tuples at a HEAP page
+ *
+ * We introduce this function for wal_consistency_checking satisfaction at
+ * Hot Standby node.
+ *
+ * DEAD tuple on a master node has t_cid and t_infomask, which can not be
+ * obtained by WAL records applying on a Hot Standby node.
+ */
+static void
+mask_dead_tuples(Page page)
+{
+	OffsetNumber	offnum;
+
+	for (offnum = FirstOffsetNumber; offnum <= PageGetMaxOffsetNumber(page); offnum = OffsetNumberNext(offnum))
+	{
+		ItemId	lp = PageGetItemId(page, offnum);
+		char*	htup_begin;
+
+		if (!ItemIdIsDead(lp))
+			continue;
+
+		if (!ItemIdHasStorage(lp))
+			continue;
+
+		htup_begin = (char *) PageGetItem(page, lp);
+		memset(htup_begin, MASK_MARKER, ItemIdGetLength(lp));
+	}
+}
+
 /*
  * Mask a heap page before performing consistency checks on it.
  */
@@ -9425,6 +9455,7 @@ heap_mask(char *pagedata, BlockNumber blkno)
 
 	mask_page_hint_bits(page);
 	mask_unused_space(page);
+	mask_dead_tuples(page);
 
 	for (off = 1; off <= PageGetMaxOffsetNumber(page); off++)
 	{
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index c2f5343dac..f66e5d0260 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -43,6 +43,9 @@ typedef struct
 	bool		marked[MaxHeapTuplesPerPage + 1];
 } PruneState;
 
+/* Parameter for target deletion strategy in lazy vacuum */
+double target_index_deletion_factor = 0.01;
+
 /* Local functions */
 static int heap_prune_chain(Relation relation, Buffer buffer,
 				 OffsetNumber rootoffnum,
@@ -405,10 +408,9 @@ heap_prune_chain(Relation relation, Buffer buffer, OffsetNumber rootoffnum,
 			if (HeapTupleSatisfiesVacuum(&tup, OldestXmin, buffer)
 				== HEAPTUPLE_DEAD && !HeapTupleHeaderIsHotUpdated(htup))
 			{
-				heap_prune_record_unused(prstate, rootoffnum);
+				heap_prune_record_dead(prstate, rootoffnum);
 				HeapTupleHeaderAdvanceLatestRemovedXid(htup,
 													   &prstate->latestRemovedXid);
-				ndeleted++;
 			}
 
 			/* Nothing more to do */
@@ -580,8 +582,10 @@ heap_prune_chain(Relation relation, Buffer buffer, OffsetNumber rootoffnum,
 		 */
 		for (i = 1; (i < nchain) && (chainitems[i - 1] != latestdead); i++)
 		{
-			heap_prune_record_unused(prstate, chainitems[i]);
+			if (chainitems[i] == latestdead)
+				continue;
 			ndeleted++;
+			heap_prune_record_unused(prstate, chainitems[i]);
 		}
 
 		/*
@@ -598,9 +602,28 @@ heap_prune_chain(Relation relation, Buffer buffer, OffsetNumber rootoffnum,
 		 * redirect the root to the correct chain member.
 		 */
 		if (i >= nchain)
+		{
+			if (rootoffnum != latestdead)
+			{
+				if (ItemIdIsNormal(rootlp))
+					heap_prune_record_unused(prstate, latestdead);
+				else
+				{
+					/*
+					 * We allow overlapping of redirected and dead items
+						 */
+					heap_prune_record_redirect(prstate, rootoffnum, latestdead);
+					heap_prune_record_dead(prstate, latestdead);
+				}
+			}
 			heap_prune_record_dead(prstate, rootoffnum);
+		}
 		else
+		{
+			if (rootoffnum != latestdead)
+				heap_prune_record_unused(prstate, latestdead);
 			heap_prune_record_redirect(prstate, rootoffnum, chainitems[i]);
+		}
 	}
 	else if (nchain < 2 && ItemIdIsRedirected(rootlp))
 	{
@@ -653,7 +676,12 @@ heap_prune_record_dead(PruneState *prstate, OffsetNumber offnum)
 	Assert(prstate->ndead < MaxHeapTuplesPerPage);
 	prstate->nowdead[prstate->ndead] = offnum;
 	prstate->ndead++;
-	Assert(!prstate->marked[offnum]);
+
+	/*
+	 * We suppress checking prstate->marked[offnum]. It is not the best idea,
+	 * but this is most simplistic way to enable Dead Redirecting by
+	 * overlapping Dead and Redirected states.
+	 */
 	prstate->marked[offnum] = true;
 }
 
@@ -706,7 +734,10 @@ heap_page_prune_execute(Buffer buffer,
 		OffsetNumber off = *offnum++;
 		ItemId		lp = PageGetItemId(page, off);
 
-		ItemIdSetDead(lp);
+		if (target_index_deletion_factor > 0)
+			ItemIdMarkDead(lp);
+		else
+			ItemIdSetDead(lp);
 	}
 
 	/* Update all now-unused line pointers */
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 7eb3e35166..15648c49c5 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -4160,3 +4160,125 @@ RestoreReindexState(void *reindexstate)
 						sistate->pendingReindexedIndexes[c]);
 	MemoryContextSwitchTo(oldcontext);
 }
+
+
+/*
+ * htid2itup - get datum for index tuple assembly.
+ *
+ * Returns false, if Index Datum can't be generated by current TID
+ */
+bool
+htid2IndexDatum(Relation hrel, Relation irel, ItemPointer htid, Datum *values, bool *isnull)
+{
+	IndexInfo 		*indexInfo = BuildIndexInfo(irel);
+	EState			*estate = CreateExecutorState();
+	ExprContext		*econtext = GetPerTupleExprContext(estate);
+	ExprState		*predicate = ExecPrepareQual(indexInfo->ii_Predicate, estate);
+	TupleTableSlot	*slot = MakeSingleTupleTableSlot(RelationGetDescr(hrel));
+	HeapTuple		tuple;
+	Buffer			buffer;
+	Page			page;
+	OffsetNumber	offnum = ItemPointerGetOffsetNumber(htid);
+	ItemId			lp;
+	BlockNumber		npages;
+
+	npages = RelationGetNumberOfBlocks(hrel);
+
+	if (ItemPointerGetBlockNumber(htid) > npages)
+		return false;
+
+	buffer = ReadBuffer(hrel, ItemPointerGetBlockNumber(htid));
+	page = (Page) BufferGetPage(buffer);
+	lp = PageGetItemId(page, offnum);
+
+	Assert(ItemIdIsUsed(lp));
+
+	/* Walk along the chain */
+	while (!ItemIdHasStorage(lp))
+	{
+		offnum = ItemIdGetRedirect(lp);
+		lp = PageGetItemId(page, offnum);
+		Assert(ItemIdIsUsed(lp));
+	}
+
+	/* Form a tuple */
+	tuple = palloc(sizeof(HeapTupleData));
+	ItemPointerSet(&(tuple->t_self), BufferGetBlockNumber(buffer), offnum);
+	tuple->t_tableOid = RelationGetRelid(hrel);
+	tuple->t_data = (HeapTupleHeader) PageGetItem(page, lp);
+	tuple->t_len = ItemIdGetLength(lp);
+	ReleaseBuffer(buffer);
+
+	econtext->ecxt_scantuple = slot;
+	MemoryContextReset(econtext->ecxt_per_tuple_memory);
+	ExecStoreTuple(tuple, slot, InvalidBuffer, false);
+
+	/*
+	 * In a partial index, ignore tuples that don't satisfy the
+	 * predicate.
+	 */
+	if ((predicate != NULL) && (!ExecQual(predicate, econtext)))
+	{
+		pfree(tuple);
+		ExecDropSingleTupleTableSlot(slot);
+		FreeExecutorState(estate);
+		return false;
+	}
+
+	FormIndexDatum(indexInfo, slot, estate, values, isnull);
+
+	pfree(tuple);
+	ExecDropSingleTupleTableSlot(slot);
+	FreeExecutorState(estate);
+	return true;
+}
+
+/*
+ *	quick_vacuum_index() -- quick vacuum one index relation.
+ *
+ *		Delete all the index entries pointing to tuples listed in
+ *		dead_tuples.
+ */
+void
+quick_vacuum_index(Relation irel, Relation hrel,
+				   ItemPointer dead_tuples,
+				   int num_dead_tuples)
+{
+	int						tnum;
+	bool					*found = palloc0(num_dead_tuples*sizeof(bool));
+	IndexTargetDeleteResult	stats;
+	IndexTargetDeleteInfo	ivinfo;
+
+	Assert(found != NULL);
+
+	stats.tuples_removed = 0;
+	ivinfo.indexRelation = irel;
+	ivinfo.heapRelation = hrel;
+
+	/* Get tuple from heap */
+	for (tnum = num_dead_tuples-1; tnum >= 0; tnum--)
+	{
+		Datum		values[INDEX_MAX_KEYS];
+		bool		isnull[INDEX_MAX_KEYS];
+
+		/* Index entry for the TID was deleted early */
+		if (found[tnum])
+			continue;
+
+		/* Get an index tuple building data by heap tid */
+		if (!htid2IndexDatum(hrel, irel, &(dead_tuples[tnum]), values, isnull))
+			continue;
+
+		/*
+		 * Make attempt to delete some index entries by one tree descent.
+		 * We use only a part of TID list, which contains not found TID's.
+		 */
+		ivinfo.dead_tuples = dead_tuples;
+		ivinfo.last_dead_tuple = tnum;
+		ivinfo.found_dead_tuples = found;
+
+		index_target_delete(&ivinfo, &stats, values, isnull);
+	}
+
+	pfree(found);
+}
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index 8996d366e9..f0b0a2ffb9 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -41,9 +41,11 @@
 #include "access/heapam_xlog.h"
 #include "access/htup_details.h"
 #include "access/multixact.h"
+#include "access/nbtree.h"
 #include "access/transam.h"
 #include "access/visibilitymap.h"
 #include "access/xlog.h"
+#include "catalog/index.h"
 #include "catalog/storage.h"
 #include "commands/dbcommands.h"
 #include "commands/progress.h"
@@ -744,9 +746,18 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 
 			/* Remove index entries */
 			for (i = 0; i < nindexes; i++)
-				lazy_vacuum_index(Irel[i],
-								  &indstats[i],
-								  vacrelstats);
+			{
+				bool use_quick_strategy = (vacrelstats->num_dead_tuples/vacrelstats->old_live_tuples < target_index_deletion_factor);
+
+				if (use_quick_strategy && (Irel[i]->rd_amroutine->amtargetdelete != NULL))
+					quick_vacuum_index(Irel[i], onerel,
+									   vacrelstats->dead_tuples,
+									   vacrelstats->num_dead_tuples);
+				else
+					lazy_vacuum_index(Irel[i],
+									  &indstats[i],
+									  vacrelstats);
+			}
 
 			/*
 			 * Report that we are now vacuuming the heap.  We also increase
@@ -1389,10 +1400,18 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 
 		/* Remove index entries */
 		for (i = 0; i < nindexes; i++)
-			lazy_vacuum_index(Irel[i],
-							  &indstats[i],
-							  vacrelstats);
+		{
+			bool use_quick_strategy = (vacrelstats->num_dead_tuples/vacrelstats->old_live_tuples < target_index_deletion_factor);
 
+			if (use_quick_strategy && (Irel[i]->rd_amroutine->amtargetdelete != NULL))
+				quick_vacuum_index(Irel[i], onerel,
+								   vacrelstats->dead_tuples,
+								   vacrelstats->num_dead_tuples);
+			else
+				lazy_vacuum_index(Irel[i],
+								  &indstats[i],
+								  vacrelstats);
+		}
 		/* Report that we are now vacuuming the heap */
 		hvp_val[0] = PROGRESS_VACUUM_PHASE_VACUUM_HEAP;
 		hvp_val[1] = vacrelstats->num_index_scans + 1;
@@ -1681,7 +1700,6 @@ lazy_check_needs_freeze(Buffer buf, bool *hastup)
 	return false;
 }
 
-
 /*
  *	lazy_vacuum_index() -- vacuum one index relation.
  *
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index ca5cad7497..cba8130422 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -100,6 +100,8 @@ extern Relation heap_openrv_extended(const RangeVar *relation,
 typedef struct HeapScanDescData *HeapScanDesc;
 typedef struct ParallelHeapScanDescData *ParallelHeapScanDesc;
 
+extern double target_index_deletion_factor;
+
 /*
  * HeapScanIsValid
  *		True iff the heap scan is valid.
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index f20c5f789b..6b1d726338 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -153,5 +153,9 @@ extern void SerializeReindexState(Size maxsize, char *start_address);
 extern void RestoreReindexState(void *reindexstate);
 
 extern void IndexSetParentIndex(Relation idx, Oid parentOid);
+extern bool htid2IndexDatum(Relation hrel, Relation irel, ItemPointer htid, Datum *values, bool *isnull);
+extern void quick_vacuum_index(Relation irel, Relation hrel,
+				   ItemPointer dead_tuples,
+				   int num_dead_tuples);
 
 #endif							/* INDEX_H */
-- 
2.17.1

