From 99c9bc340a730a85328549165ae8023a0716e147 Mon Sep 17 00:00:00 2001
From: "Andrey V. Lepikhov" <a.lepikhov@postgrespro.ru>
Date: Thu, 20 Sep 2018 07:56:18 +0500
Subject: [PATCH] Retail-IndexTuple-Deletion-Access-Method

---
 contrib/bloom/blutils.c              |   1 +
 src/backend/access/brin/brin.c       |   1 +
 src/backend/access/gin/ginutil.c     |   1 +
 src/backend/access/gist/gist.c       |   1 +
 src/backend/access/hash/hash.c       |   1 +
 src/backend/access/index/indexam.c   |  15 +++
 src/backend/access/nbtree/nbtree.c   | 160 +++++++++++++++++++++++++++
 src/backend/access/spgist/spgutils.c |   1 +
 src/include/access/amapi.h           |   6 +
 src/include/access/genam.h           |  18 +++
 src/include/access/nbtree.h          |   4 +
 11 files changed, 209 insertions(+)

diff --git a/contrib/bloom/blutils.c b/contrib/bloom/blutils.c
index 6b2b9e3742..96f1d47c70 100644
--- a/contrib/bloom/blutils.c
+++ b/contrib/bloom/blutils.c
@@ -126,6 +126,7 @@ blhandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = blbuild;
 	amroutine->ambuildempty = blbuildempty;
 	amroutine->aminsert = blinsert;
+	amroutine->amtargetdelete = NULL;
 	amroutine->ambulkdelete = blbulkdelete;
 	amroutine->amvacuumcleanup = blvacuumcleanup;
 	amroutine->amcanreturn = NULL;
diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index e95fbbcea7..a0e06bd868 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -103,6 +103,7 @@ brinhandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = brinbuild;
 	amroutine->ambuildempty = brinbuildempty;
 	amroutine->aminsert = brininsert;
+	amroutine->amtargetdelete = NULL;
 	amroutine->ambulkdelete = brinbulkdelete;
 	amroutine->amvacuumcleanup = brinvacuumcleanup;
 	amroutine->amcanreturn = NULL;
diff --git a/src/backend/access/gin/ginutil.c b/src/backend/access/gin/ginutil.c
index 0a32182dd7..acf14e7bec 100644
--- a/src/backend/access/gin/ginutil.c
+++ b/src/backend/access/gin/ginutil.c
@@ -58,6 +58,7 @@ ginhandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = ginbuild;
 	amroutine->ambuildempty = ginbuildempty;
 	amroutine->aminsert = gininsert;
+	amroutine->amtargetdelete = NULL;
 	amroutine->ambulkdelete = ginbulkdelete;
 	amroutine->amvacuumcleanup = ginvacuumcleanup;
 	amroutine->amcanreturn = NULL;
diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 8a42effdf7..d7a53d2ca9 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -80,6 +80,7 @@ gisthandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = gistbuild;
 	amroutine->ambuildempty = gistbuildempty;
 	amroutine->aminsert = gistinsert;
+	amroutine->amtargetdelete = NULL;
 	amroutine->ambulkdelete = gistbulkdelete;
 	amroutine->amvacuumcleanup = gistvacuumcleanup;
 	amroutine->amcanreturn = gistcanreturn;
diff --git a/src/backend/access/hash/hash.c b/src/backend/access/hash/hash.c
index 0002df30c0..5fb32d6eba 100644
--- a/src/backend/access/hash/hash.c
+++ b/src/backend/access/hash/hash.c
@@ -76,6 +76,7 @@ hashhandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = hashbuild;
 	amroutine->ambuildempty = hashbuildempty;
 	amroutine->aminsert = hashinsert;
+	amroutine->amtargetdelete = NULL;
 	amroutine->ambulkdelete = hashbulkdelete;
 	amroutine->amvacuumcleanup = hashvacuumcleanup;
 	amroutine->amcanreturn = NULL;
diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index eade540ef5..20d9acddad 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -731,6 +731,21 @@ index_getbitmap(IndexScanDesc scan, TIDBitmap *bitmap)
 	return ntids;
 }
 
+IndexTargetDeleteResult *
+index_target_delete(IndexTargetDeleteInfo *info,
+					IndexTargetDeleteResult *stats,
+					Datum *values,
+					bool *isnull)
+{
+	Relation indexRelation = info->indexRelation;
+
+	RELATION_CHECKS;
+
+	CHECK_REL_PROCEDURE(amtargetdelete);
+
+	return indexRelation->rd_amroutine->amtargetdelete(info, stats, values, isnull);
+}
+
 /* ----------------
  *		index_bulk_delete - do mass deletion of index entries
  *
diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index e8725fbbe1..a5d8ad4aa2 100644
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -127,6 +127,7 @@ bthandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = btbuild;
 	amroutine->ambuildempty = btbuildempty;
 	amroutine->aminsert = btinsert;
+	amroutine->amtargetdelete = bttargetdelete;
 	amroutine->ambulkdelete = btbulkdelete;
 	amroutine->amvacuumcleanup = btvacuumcleanup;
 	amroutine->amcanreturn = btcanreturn;
@@ -886,6 +887,165 @@ btbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
 	return stats;
 }
 
+/*
+ * Deletion of index entries corresponding to heap TIDs.
+ *
+ * Constraints:
+ * 1. TID list info->dead_tuples arranged in ASC order.
+ * 2. Logical duplicates of index tuples stored in DESC order.
+ *
+ * The function generates an insertion scan key and descent by btree for the first
+ * index tuple what satisfies scan key and last TID in info->dead_tuples list.
+ * For the scan results it deletes all index entries, matched to the TID list.
+ *
+ * Result: a palloc'd struct containing statistical info.
+ */
+IndexTargetDeleteResult*
+bttargetdelete(IndexTargetDeleteInfo *info, IndexTargetDeleteResult *stats,
+			   Datum *values, bool *isnull)
+{
+	Relation		irel = info->indexRelation;
+	Relation		hrel = info->heapRelation;
+	ScanKey			skey;
+	int				keysCount = IndexRelationGetNumberOfKeyAttributes(irel);
+	BTStack			stack;
+	Buffer			buf;
+	Page			page;
+	BTPageOpaque	opaque;
+	OffsetNumber	offnum;
+	int				ndeletable = 0;
+	OffsetNumber	deletable[MaxOffsetNumber];
+	int				pos = info->last_dead_tuple;
+	IndexTuple		itup;
+
+	if (stats == NULL)
+		stats = (IndexTargetDeleteResult *) palloc0(sizeof(IndexTargetDeleteResult));
+
+	/* Assemble scankey */
+	itup = index_form_tuple(RelationGetDescr(irel), values, isnull);
+	skey = _bt_mkscankey(irel, itup);
+
+	/* Descend the tree and position ourselves on the target leaf page. */
+	stack = _bt_search(irel, keysCount, skey, &info->dead_tuples[pos], false, &buf, BT_WRITE, NULL);
+
+	/* trade in our read lock for a write lock */
+	LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+	LockBuffer(buf, BT_WRITE);
+
+	buf = _bt_moveright(irel, buf, keysCount, skey, &info->dead_tuples[pos],
+													false, true, stack, BT_WRITE, NULL);
+
+	/* To prepare tuple entries search across index pages */
+	Assert(BufferIsValid(buf));
+	page = BufferGetPage(buf);
+	_bt_checkpage(irel, buf);
+	opaque = (BTPageOpaque) PageGetSpecialPointer(page);
+	offnum = _bt_binsrch(irel, buf, keysCount, skey, &info->dead_tuples[pos], P_FIRSTDATAKEY(opaque), false);
+
+	for (;;)
+	{
+		int32		cmpval;
+		ItemId		itemid;
+		IndexTuple	itup;
+
+		/* Switch to the next page */
+		if (offnum > PageGetMaxOffsetNumber(page))
+		{
+			/*
+			 * Before unlocking index page we need to delete
+			 * all currently found tuples
+			 */
+			if (ndeletable > 0)
+			{
+				_bt_delitems_delete(irel, buf, deletable, ndeletable, hrel);
+				stats->tuples_removed += ndeletable;
+				ndeletable = 0;
+			}
+
+			/*
+			 * Check for end-of-index
+			 */
+			if (P_RIGHTMOST(opaque))
+				/* it is rightmost leaf */
+				break;
+
+			/*
+			 * Traverse to a next reliable index page if current page do not
+			 * includes the (scan key; TID) value at the range.
+			 */
+			buf = _bt_moveright(irel, buf, keysCount, skey, &info->dead_tuples[pos],
+												false, true, stack, BT_WRITE, NULL);
+			page = BufferGetPage(buf);
+			_bt_checkpage(irel, buf);
+			opaque = (BTPageOpaque) PageGetSpecialPointer(page);
+			Assert(!P_IGNORE(opaque));
+
+			/* Set offnum to first potentially interesting item */
+			offnum = _bt_binsrch(irel, buf, keysCount, skey, &info->dead_tuples[pos], P_FIRSTDATAKEY(opaque), false);
+
+			if (offnum > PageGetMaxOffsetNumber(page))
+				/* Index relation do not contain TID of this DEAD tuple. */
+				break;
+		}
+
+		/* This index entry satisfied to the scan key? */
+		cmpval = _bt_compare(irel, keysCount, skey, NULL, page, offnum);
+
+		if (cmpval != 0)
+			/* End of the logical duplicates chain */
+			break;
+
+		/*
+		 * To load index tuple and look for matches in the TID list of
+		 * dead heap tuples
+		 */
+		itemid = PageGetItemId(page, offnum);
+		itup = (IndexTuple) PageGetItem(page, itemid);
+
+		/*
+		 * Search for next TID from presorted btree result comparable
+		 * to TID from presorted dead_tuples tid list
+		 */
+		while (pos >= 0)
+		{
+			int res = ItemPointerCompare(&(itup->t_tid), &info->dead_tuples[pos]);
+			if ((res == 0) && (!info->found_dead_tuples[pos]))
+			{
+				/* index entry for TID of dead tuple is found */
+				deletable[ndeletable++] = offnum;
+				info->found_dead_tuples[pos] = true;
+			}
+			else if (res > 0)
+				break;
+
+			pos--;
+		}
+
+		if (pos < 0)
+			/* End of DEAD TIDs list was reached */
+			break;
+
+		offnum = OffsetNumberNext(offnum);
+	}
+
+	/*
+	 * Delete rest of dead index entries
+	 */
+	if (ndeletable > 0)
+	{
+		_bt_delitems_delete(irel, buf, deletable, ndeletable, hrel);
+		stats->tuples_removed += ndeletable;
+	}
+
+	/* Release stack, scan key, unpin and unlock buffer */
+	if (stack)
+		_bt_freestack(stack);
+	_bt_freeskey(skey);
+	_bt_relbuf(irel, buf);
+
+	return stats;
+}
+
 /*
  * Post-VACUUM cleanup.
  *
diff --git a/src/backend/access/spgist/spgutils.c b/src/backend/access/spgist/spgutils.c
index 9919e6f0d7..bdf1fdc472 100644
--- a/src/backend/access/spgist/spgutils.c
+++ b/src/backend/access/spgist/spgutils.c
@@ -65,6 +65,7 @@ spghandler(PG_FUNCTION_ARGS)
 	amroutine->ambuild = spgbuild;
 	amroutine->ambuildempty = spgbuildempty;
 	amroutine->aminsert = spginsert;
+	amroutine->amtargetdelete = NULL;
 	amroutine->ambulkdelete = spgbulkdelete;
 	amroutine->amvacuumcleanup = spgvacuumcleanup;
 	amroutine->amcanreturn = spgcanreturn;
diff --git a/src/include/access/amapi.h b/src/include/access/amapi.h
index 14526a6bb2..497b54e8a8 100644
--- a/src/include/access/amapi.h
+++ b/src/include/access/amapi.h
@@ -76,6 +76,11 @@ typedef bool (*aminsert_function) (Relation indexRelation,
 								   IndexUniqueCheck checkUnique,
 								   struct IndexInfo *indexInfo);
 
+/* target delete */
+typedef IndexTargetDeleteResult *(*amtargetdelete_function) (IndexTargetDeleteInfo *info,
+															 IndexTargetDeleteResult *stats,
+															 Datum *values,
+															 bool *isnull);
 /* bulk delete */
 typedef IndexBulkDeleteResult *(*ambulkdelete_function) (IndexVacuumInfo *info,
 														 IndexBulkDeleteResult *stats,
@@ -207,6 +212,7 @@ typedef struct IndexAmRoutine
 	ambuild_function ambuild;
 	ambuildempty_function ambuildempty;
 	aminsert_function aminsert;
+	amtargetdelete_function amtargetdelete;
 	ambulkdelete_function ambulkdelete;
 	amvacuumcleanup_function amvacuumcleanup;
 	amcanreturn_function amcanreturn;	/* can be NULL */
diff --git a/src/include/access/genam.h b/src/include/access/genam.h
index 534fac7bf2..ee005891d2 100644
--- a/src/include/access/genam.h
+++ b/src/include/access/genam.h
@@ -51,6 +51,15 @@ typedef struct IndexVacuumInfo
 	BufferAccessStrategy strategy;	/* access strategy for reads */
 } IndexVacuumInfo;
 
+typedef struct IndexTargetDeleteInfo
+{
+	Relation	heapRelation;
+	Relation	indexRelation;			/* the index being vacuumed */
+	int			last_dead_tuple;
+	ItemPointer	dead_tuples;
+	bool*		found_dead_tuples;
+} IndexTargetDeleteInfo;
+
 /*
  * Struct for statistics returned by ambulkdelete and amvacuumcleanup
  *
@@ -79,6 +88,11 @@ typedef struct IndexBulkDeleteResult
 	BlockNumber pages_free;		/* # pages available for reuse */
 } IndexBulkDeleteResult;
 
+typedef struct IndexTargetDeleteResult
+{
+	int		tuples_removed; /* # removed during vacuum operation */
+} IndexTargetDeleteResult;
+
 /* Typedef for callback function to determine if a tuple is bulk-deletable */
 typedef bool (*IndexBulkDeleteCallback) (ItemPointer itemptr, void *state);
 
@@ -163,6 +177,10 @@ extern HeapTuple index_fetch_heap(IndexScanDesc scan);
 extern HeapTuple index_getnext(IndexScanDesc scan, ScanDirection direction);
 extern int64 index_getbitmap(IndexScanDesc scan, TIDBitmap *bitmap);
 
+extern IndexTargetDeleteResult *index_target_delete(IndexTargetDeleteInfo *info,
+													IndexTargetDeleteResult *stats,
+													Datum *values,
+													bool *isnull);
 extern IndexBulkDeleteResult *index_bulk_delete(IndexVacuumInfo *info,
 				  IndexBulkDeleteResult *stats,
 				  IndexBulkDeleteCallback callback,
diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h
index 12f57773e7..59f8456e0b 100644
--- a/src/include/access/nbtree.h
+++ b/src/include/access/nbtree.h
@@ -557,6 +557,10 @@ extern void btparallelrescan(IndexScanDesc scan);
 extern void btendscan(IndexScanDesc scan);
 extern void btmarkpos(IndexScanDesc scan);
 extern void btrestrpos(IndexScanDesc scan);
+extern IndexTargetDeleteResult *bttargetdelete(IndexTargetDeleteInfo *info,
+											   IndexTargetDeleteResult *stats,
+											   Datum *values,
+											   bool *isnull);
 extern IndexBulkDeleteResult *btbulkdelete(IndexVacuumInfo *info,
 			 IndexBulkDeleteResult *stats,
 			 IndexBulkDeleteCallback callback,
-- 
2.17.1

