v3-0001-Add-DISABLE_INDEX_CLEANUP-option-to-VACUUM-comman.patch

From 0f0432f1be115411602dc97400937cff42434f19 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Wed, 9 Jan 2019 09:22:20 +0900
Subject: [PATCH v3] Add DISABLE_INDEX_CLEANUP option to VACUUM command.

---
 doc/src/sgml/ref/vacuum.sgml         |  19 +++++-
 src/backend/commands/vacuum.c        |   8 ++-
 src/backend/commands/vacuumlazy.c    | 128 ++++++++++++++++++++++++++++++-----
 src/backend/parser/gram.y            |   2 +
 src/include/nodes/parsenodes.h       |   4 +-
 src/test/regress/expected/vacuum.out |   2 +
 src/test/regress/sql/vacuum.sql      |   2 +
 7 files changed, 145 insertions(+), 20 deletions(-)

diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml
index fd911f5..4ae0473 100644
--- a/doc/src/sgml/ref/vacuum.sgml
+++ b/doc/src/sgml/ref/vacuum.sgml
@@ -31,6 +31,7 @@ VACUUM [ FULL ] [ FREEZE ] [ VERBOSE ] [ ANALYZE ] [ <replaceable class="paramet
     VERBOSE
     ANALYZE
     DISABLE_PAGE_SKIPPING
+    DISABLE_INDEX_CLEANUP
     SKIP_LOCKED
 
 <phrase>and <replaceable class="parameter">table_and_columns</replaceable> is:</phrase>
@@ -161,7 +162,23 @@ VACUUM [ FULL ] [ FREEZE ] [ VERBOSE ] [ ANALYZE ] [ <replaceable class="paramet
     </listitem>
    </varlistentry>
 
-   <varlistentry>
+    <varlistentry>
+    <term><literal>DISABLE_INDEX_CLEANUP</literal></term>
+    <listitem>
+     <para>
+      <command>VACUUM</command> removes dead tuples and prunes HOT-updated
+      tuples chain for live tuples on table. If the table has any dead tuple
+      it removes them from both table and indexes for re-use. With this
+      option <command>VACUUM</command> marks tuples as dead (i.e., it doesn't
+      remove tuple storage) and disables removing dead tuples from indexes.
+      This is suitable for avoiding transaction ID wraparound but not
+      sufficient for avoiding index bloat. This option cannot be used in
+      conjunction with <literal>FULL</literal> option.
+     </para>
+    </listitem>
+   </varlistentry>
+
+    <varlistentry>
     <term><literal>SKIP_LOCKED</literal></term>
     <listitem>
      <para>
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index ff1e178..5f0e7a4 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -204,7 +204,8 @@ vacuum(int options, List *relations, VacuumParams *params,
 						stmttype)));
 
 	/*
-	 * Sanity check DISABLE_PAGE_SKIPPING option.
+	 * Sanity check DISABLE_PAGE_SKIPPING option and DISABLE_INDEX_CLEANUP
+	 * option.
 	 */
 	if ((options & VACOPT_FULL) != 0 &&
 		(options & VACOPT_DISABLE_PAGE_SKIPPING) != 0)
@@ -212,6 +213,11 @@ vacuum(int options, List *relations, VacuumParams *params,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("VACUUM option DISABLE_PAGE_SKIPPING cannot be used with FULL")));
 
+	if ((options & VACOPT_FULL) != 0 &&
+		(options & VACOPT_DISABLE_INDEX_CLEANUP) != 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("VACUUM option DISABLE_INDEX_CLEANUP cannot be used with FULL")));
 	/*
 	 * Send info about dead objects to the statistics collector, unless we are
 	 * in autovacuum --- autovacuum.c does this for itself.
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index b67267f..994d4cb 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -164,11 +164,14 @@ static void lazy_cleanup_index(Relation indrel,
 				   LVRelStats *vacrelstats);
 static int lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 				 int tupindex, LVRelStats *vacrelstats, Buffer *vmbuffer);
+static void lazy_set_tuples_dead(Relation onerel, BlockNumber blkno, Buffer buffer,
+								 LVRelStats *vacrelstats);
 static bool should_attempt_truncation(LVRelStats *vacrelstats);
 static void lazy_truncate_heap(Relation onerel, LVRelStats *vacrelstats);
 static BlockNumber count_nondeletable_pages(Relation onerel,
 						 LVRelStats *vacrelstats);
-static void lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks);
+static void lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks,
+							 bool disable_removing_tuples);
 static void lazy_record_dead_tuple(LVRelStats *vacrelstats,
 					   ItemPointer itemptr);
 static bool lazy_tid_reaped(ItemPointer itemptr, void *state);
@@ -284,9 +287,11 @@ lazy_vacuum_rel(Relation onerel, int options, VacuumParams *params,
 		scanned_all_unfrozen = true;
 
 	/*
-	 * Optionally truncate the relation.
+	 * Optionally truncate the relation. If DISABLE_INDEX_CLEANUP is specified,
+	 * since we skipped to remove dead tuples there is no need to truncate heap.
 	 */
-	if (should_attempt_truncation(vacrelstats))
+	if (should_attempt_truncation(vacrelstats) &&
+		(options & VACOPT_DISABLE_INDEX_CLEANUP) == 0)
 		lazy_truncate_heap(onerel, vacrelstats);
 
 	/* Report that we are now doing final cleanup */
@@ -493,6 +498,7 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 	Buffer		vmbuffer = InvalidBuffer;
 	BlockNumber next_unskippable_block;
 	bool		skipping_blocks;
+	bool		disable_index_cleanup = (options & VACOPT_DISABLE_INDEX_CLEANUP) != 0;
 	xl_heap_freeze_tuple *frozen;
 	StringInfoData buf;
 	const int	initprog_index[] = {
@@ -530,7 +536,7 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 	vacrelstats->nonempty_pages = 0;
 	vacrelstats->latestRemovedXid = InvalidTransactionId;
 
-	lazy_space_alloc(vacrelstats, nblocks);
+	lazy_space_alloc(vacrelstats, nblocks, disable_index_cleanup);
 	frozen = palloc(sizeof(xl_heap_freeze_tuple) * MaxHeapTuplesPerPage);
 
 	/* Report that we're scanning the heap, advertising total # of blocks */
@@ -723,6 +729,8 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 			};
 			int64		hvp_val[2];
 
+			Assert(!disable_index_cleanup);
+
 			/*
 			 * Before beginning index vacuuming, we release any pin we may
 			 * hold on the visibility map page.  This isn't necessary for
@@ -1201,14 +1209,23 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		}
 
 		/*
-		 * If there are no indexes then we can vacuum the page right now
-		 * instead of doing a second scan.
+		 * If either there are no indexes or index cleanup is disabled then
+		 * we can vacuum the page right now instead of doing a second scan.
 		 */
-		if (nindexes == 0 &&
+		if ((nindexes == 0 || disable_index_cleanup) &&
 			vacrelstats->num_dead_tuples > 0)
 		{
-			/* Remove tuples from heap */
-			lazy_vacuum_page(onerel, blkno, buf, 0, vacrelstats, &vmbuffer);
+			/*
+			 * If index cleanup is disabled, we set the recorded tuples as
+			 * dead, which removes its tuple storage but leaves ItemIDs.
+			 * By leaving dead ItemIDs an index lookup can properly see the
+			 * tuple is already dead. And the remained dead ItemIds will be
+			 * removed by the next vacuum enabled the index cleanup.
+			 */
+			if (nindexes > 0 && disable_index_cleanup)
+				lazy_set_tuples_dead(onerel, blkno, buf, vacrelstats);
+			else
+				lazy_vacuum_page(onerel, blkno, buf, 0, vacrelstats, &vmbuffer);
 			has_dead_tuples = false;
 
 			/*
@@ -1374,6 +1391,8 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		};
 		int64		hvp_val[2];
 
+		Assert(!disable_index_cleanup);
+
 		/* Log cleanup info before we touch indexes */
 		vacuum_log_cleanup_info(onerel, vacrelstats);
 
@@ -1412,15 +1431,24 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 								 PROGRESS_VACUUM_PHASE_INDEX_CLEANUP);
 
 	/* Do post-vacuum cleanup and statistics update for each index */
-	for (i = 0; i < nindexes; i++)
-		lazy_cleanup_index(Irel[i], indstats[i], vacrelstats);
+	if (!disable_index_cleanup)
+		for (i = 0; i < nindexes; i++)
+			lazy_cleanup_index(Irel[i], indstats[i], vacrelstats);
 
 	/* If no indexes, make log report that lazy_vacuum_heap would've made */
 	if (vacuumed_pages)
-		ereport(elevel,
-				(errmsg("\"%s\": removed %.0f row versions in %u pages",
-						RelationGetRelationName(onerel),
-						tups_vacuumed, vacuumed_pages)));
+	{
+		if (disable_index_cleanup)
+			ereport(elevel,
+					(errmsg("\"%s\": marked %.0f row versions in %u pages as dead",
+							RelationGetRelationName(onerel),
+							tups_vacuumed, vacuumed_pages)));
+		else
+			ereport(elevel,
+					(errmsg("\"%s\": removed %.0f row versions in %u pages",
+							RelationGetRelationName(onerel),
+							tups_vacuumed, vacuumed_pages)));
+	}
 
 	/*
 	 * This is pretty messy, but we split it up so that we can skip emitting
@@ -1623,6 +1651,71 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 	return tupindex;
 }
 
+
+/*
+ *	lazy_set_tuples_dead() -- set tuples as dead on a page
+ *							and repair its fragmentation.
+ *
+ * Unlike lazy_vacuum_page, this function sets the recorded tuples as dead
+ * instead of as unused, and doesn't update the page visibility.
+ * Also this function assumes that vacrelstats->dead_tuples has only dead
+ * tuples on the 'blkno' block.
+ *
+ * Caller must hold pin and buffer cleanup lock on the buffer.
+ */
+static void
+lazy_set_tuples_dead(Relation onerel, BlockNumber blkno, Buffer buffer,
+					LVRelStats *vacrelstats)
+{
+	Page		page = BufferGetPage(buffer);
+	int			deadcnt = 0;
+	OffsetNumber dead[MaxOffsetNumber];
+	int			tupindex;
+
+	pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, blkno);
+
+	START_CRIT_SECTION();
+
+	for (tupindex = 0; tupindex < vacrelstats->num_dead_tuples; tupindex++)
+	{
+		OffsetNumber toff;
+		ItemId		itemid;
+
+		Assert(blkno == ItemPointerGetBlockNumber(&vacrelstats->dead_tuples[tupindex]));
+
+		toff = ItemPointerGetOffsetNumber(&vacrelstats->dead_tuples[tupindex]);
+		itemid = PageGetItemId(page, toff);
+
+		if (ItemIdIsDead(itemid))
+			continue;
+
+		ItemIdSetDead(itemid);
+		dead[deadcnt++] = toff;
+	}
+
+	PageRepairFragmentation(page);
+
+	/*
+	 * Mark buffer dirty before we write WAL.
+	 */
+	MarkBufferDirty(buffer);
+
+	/* XLOG stuff */
+	if (RelationNeedsWAL(onerel))
+	{
+		XLogRecPtr	recptr;
+
+		recptr = log_heap_clean(onerel, buffer,
+								NULL, 0,
+								dead, deadcnt,
+								NULL, 0,
+								vacrelstats->latestRemovedXid);
+		PageSetLSN(page, recptr);
+	}
+
+	END_CRIT_SECTION();
+}
+
 /*
  *	lazy_check_needs_freeze() -- scan page to see if any tuples
  *					 need to be cleaned to avoid wraparound
@@ -2079,14 +2172,15 @@ count_nondeletable_pages(Relation onerel, LVRelStats *vacrelstats)
  * See the comments at the head of this file for rationale.
  */
 static void
-lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks)
+lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks,
+				 bool disable_index_cleanup)
 {
 	long		maxtuples;
 	int			vac_work_mem = IsAutoVacuumWorkerProcess() &&
 	autovacuum_work_mem != -1 ?
 	autovacuum_work_mem : maintenance_work_mem;
 
-	if (vacrelstats->hasindex)
+	if (vacrelstats->hasindex && !disable_index_cleanup)
 	{
 		maxtuples = (vac_work_mem * 1024L) / sizeof(ItemPointerData);
 		maxtuples = Min(maxtuples, INT_MAX);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index c086235..e688c04 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -10511,6 +10511,8 @@ vacuum_option_elem:
 				{
 					if (strcmp($1, "disable_page_skipping") == 0)
 						$$ = VACOPT_DISABLE_PAGE_SKIPPING;
+					else if (strcmp($1, "disable_index_cleanup") == 0)
+						$$ = VACOPT_DISABLE_INDEX_CLEANUP;
 					else if (strcmp($1, "skip_locked") == 0)
 						$$ = VACOPT_SKIP_LOCKED;
 					else
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 27782fe..8266eaf 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3152,7 +3152,9 @@ typedef enum VacuumOption
 	VACOPT_FULL = 1 << 4,		/* FULL (non-concurrent) vacuum */
 	VACOPT_SKIP_LOCKED = 1 << 5,	/* skip if cannot get lock */
 	VACOPT_SKIPTOAST = 1 << 6,	/* don't process the TOAST table, if any */
-	VACOPT_DISABLE_PAGE_SKIPPING = 1 << 7	/* don't skip any pages */
+	VACOPT_DISABLE_PAGE_SKIPPING = 1 << 7,	/* don't skip any pages */
+	VACOPT_DISABLE_INDEX_CLEANUP = 1 << 8	/* don't remove dead tuple and
+											 * cleanup indexes */
 } VacuumOption;
 
 /*
diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index fa9d663..bf3e50e 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -80,6 +80,8 @@ CONTEXT:  SQL function "do_analyze" statement 1
 SQL function "wrap_do_analyze" statement 1
 VACUUM FULL vactst;
 VACUUM (DISABLE_PAGE_SKIPPING) vaccluster;
+VACUUM (DISABLE_INDEX_CLEANUP) vaccluster, vactst;
+VACUUM (DISABLE_INDEX_CLEANUP, FREEZE) vaccluster, vactst;
 -- partitioned table
 CREATE TABLE vacparted (a int, b char) PARTITION BY LIST (a);
 CREATE TABLE vacparted1 PARTITION OF vacparted FOR VALUES IN (1);
diff --git a/src/test/regress/sql/vacuum.sql b/src/test/regress/sql/vacuum.sql
index 9defa0d..a1beafc 100644
--- a/src/test/regress/sql/vacuum.sql
+++ b/src/test/regress/sql/vacuum.sql
@@ -61,6 +61,8 @@ VACUUM FULL vaccluster;
 VACUUM FULL vactst;
 
 VACUUM (DISABLE_PAGE_SKIPPING) vaccluster;
+VACUUM (DISABLE_INDEX_CLEANUP) vaccluster, vactst;
+VACUUM (DISABLE_INDEX_CLEANUP, FREEZE) vaccluster, vactst;
 
 -- partitioned table
 CREATE TABLE vacparted (a int, b char) PARTITION BY LIST (a);
-- 
2.10.5