0007_vacuum_enhancements_v21.patch

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 72e1253..b856503 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -338,6 +338,24 @@ static relopt_real realRelOpts[] =
 	},
 	{
 		{
+			"autovacuum_warmcleanup_scale_factor",
+			"Number of WARM chains prior to WARM cleanup as a fraction of reltuples",
+			RELOPT_KIND_HEAP | RELOPT_KIND_TOAST,
+			ShareUpdateExclusiveLock
+		},
+		-1, 0.0, 100.0
+	},
+	{
+		{
+			"autovacuum_warmcleanup_index_scale_factor",
+			"Number of WARM pointers in an index prior to WARM cleanup as a fraction of total WARM chains",
+			RELOPT_KIND_HEAP | RELOPT_KIND_TOAST,
+			ShareUpdateExclusiveLock
+		},
+		-1, 0.0, 100.0
+	},
+	{
+		{
 			"autovacuum_analyze_scale_factor",
 			"Number of tuple inserts, updates or deletes prior to analyze as a fraction of reltuples",
 			RELOPT_KIND_HEAP,
@@ -1341,6 +1359,10 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, vacuum_scale_factor)},
 		{"autovacuum_analyze_scale_factor", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)},
+		{"autovacuum_warmcleanup_scale_factor", RELOPT_TYPE_REAL,
+		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, warmcleanup_scale_factor)},
+		{"autovacuum_warmcleanup_index_scale_factor", RELOPT_TYPE_REAL,
+		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, warmcleanup_index_scale)},
 		{"user_catalog_table", RELOPT_TYPE_BOOL,
 		offsetof(StdRdOptions, user_catalog_table)},
 		{"parallel_workers", RELOPT_TYPE_INT,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index ca44e03..8d06c93 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -533,6 +533,7 @@ CREATE VIEW pg_stat_all_tables AS
             pg_stat_get_tuples_warm_updated(C.oid) AS n_tup_warm_upd,
             pg_stat_get_live_tuples(C.oid) AS n_live_tup,
             pg_stat_get_dead_tuples(C.oid) AS n_dead_tup,
+            pg_stat_get_warm_chains(C.oid) AS n_warm_chains,
             pg_stat_get_mod_since_analyze(C.oid) AS n_mod_since_analyze,
             pg_stat_get_last_vacuum_time(C.oid) as last_vacuum,
             pg_stat_get_last_autovacuum_time(C.oid) as last_autovacuum,
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 404acb2..6c4fc4e 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -93,7 +93,8 @@ static VacAttrStats *examine_attribute(Relation onerel, int attnum,
 				  Node *index_expr);
 static int acquire_sample_rows(Relation onerel, int elevel,
 					HeapTuple *rows, int targrows,
-					double *totalrows, double *totaldeadrows);
+					double *totalrows, double *totaldeadrows,
+					double *totalwarmchains);
 static int	compare_rows(const void *a, const void *b);
 static int acquire_inherited_sample_rows(Relation onerel, int elevel,
 							  HeapTuple *rows, int targrows,
@@ -320,7 +321,8 @@ do_analyze_rel(Relation onerel, int options, VacuumParams *params,
 	int			targrows,
 				numrows;
 	double		totalrows,
-				totaldeadrows;
+				totaldeadrows,
+				totalwarmchains;
 	HeapTuple  *rows;
 	PGRUsage	ru0;
 	TimestampTz starttime = 0;
@@ -501,7 +503,8 @@ do_analyze_rel(Relation onerel, int options, VacuumParams *params,
 	else
 		numrows = (*acquirefunc) (onerel, elevel,
 								  rows, targrows,
-								  &totalrows, &totaldeadrows);
+								  &totalrows, &totaldeadrows,
+								  &totalwarmchains);
 
 	/*
 	 * Compute the statistics.  Temporary results during the calculations for
@@ -631,7 +634,7 @@ do_analyze_rel(Relation onerel, int options, VacuumParams *params,
 	 */
 	if (!inh)
 		pgstat_report_analyze(onerel, totalrows, totaldeadrows,
-							  (va_cols == NIL));
+							  totalwarmchains, (va_cols == NIL));
 
 	/* If this isn't part of VACUUM ANALYZE, let index AMs do cleanup */
 	if (!(options & VACOPT_VACUUM))
@@ -991,12 +994,14 @@ examine_attribute(Relation onerel, int attnum, Node *index_expr)
 static int
 acquire_sample_rows(Relation onerel, int elevel,
 					HeapTuple *rows, int targrows,
-					double *totalrows, double *totaldeadrows)
+					double *totalrows, double *totaldeadrows,
+					double *totalwarmchains)
 {
 	int			numrows = 0;	/* # rows now in reservoir */
 	double		samplerows = 0; /* total # rows collected */
 	double		liverows = 0;	/* # live rows seen */
 	double		deadrows = 0;	/* # dead rows seen */
+	double		warmchains = 0;
 	double		rowstoskip = -1;	/* -1 means not set yet */
 	BlockNumber totalblocks;
 	TransactionId OldestXmin;
@@ -1023,9 +1028,14 @@ acquire_sample_rows(Relation onerel, int elevel,
 		Page		targpage;
 		OffsetNumber targoffset,
 					maxoffset;
+		bool		marked[MaxHeapTuplesPerPage];
+		OffsetNumber root_offsets[MaxHeapTuplesPerPage];
 
 		vacuum_delay_point();
 
+		/* Track which root line pointers are already counted. */
+		memset(marked, 0, sizeof (marked));
+
 		/*
 		 * We must maintain a pin on the target page's buffer to ensure that
 		 * the maxoffset value stays good (else concurrent VACUUM might delete
@@ -1041,6 +1051,9 @@ acquire_sample_rows(Relation onerel, int elevel,
 		targpage = BufferGetPage(targbuffer);
 		maxoffset = PageGetMaxOffsetNumber(targpage);
 
+		/* Get all root line pointers first */
+		heap_get_root_tuples(targpage, root_offsets);
+
 		/* Inner loop over all tuples on the selected page */
 		for (targoffset = FirstOffsetNumber; targoffset <= maxoffset; targoffset++)
 		{
@@ -1069,6 +1082,22 @@ acquire_sample_rows(Relation onerel, int elevel,
 			targtuple.t_data = (HeapTupleHeader) PageGetItem(targpage, itemid);
 			targtuple.t_len = ItemIdGetLength(itemid);
 
+			/*
+			 * If this is a WARM-updated tuple, check if we have already seen
+			 * the root line pointer. If not, count this as a WARM chain. This
+			 * ensures that we count every WARM-chain just once, irrespective
+			 * of how many tuples exist in the chain.
+			 */
+			if (HeapTupleHeaderIsWarmUpdated(targtuple.t_data))
+			{
+				OffsetNumber root_offnum = root_offsets[targoffset];
+				if (!marked[root_offnum])
+				{
+					warmchains += 1;
+					marked[root_offnum] = true;
+				}
+			}
+
 			switch (HeapTupleSatisfiesVacuum(&targtuple,
 											 OldestXmin,
 											 targbuffer))
@@ -1200,18 +1229,24 @@ acquire_sample_rows(Relation onerel, int elevel,
 
 	/*
 	 * Estimate total numbers of rows in relation.  For live rows, use
-	 * vac_estimate_reltuples; for dead rows, we have no source of old
-	 * information, so we have to assume the density is the same in unseen
-	 * pages as in the pages we scanned.
+	 * vac_estimate_reltuples; for dead rows and WARM chains, we have no source
+	 * of old information, so we have to assume the density is the same in
+	 * unseen pages as in the pages we scanned.
 	 */
 	*totalrows = vac_estimate_reltuples(onerel, true,
 										totalblocks,
 										bs.m,
 										liverows);
 	if (bs.m > 0)
+	{
 		*totaldeadrows = floor((deadrows / bs.m) * totalblocks + 0.5);
+		*totalwarmchains = floor((warmchains / bs.m) * totalblocks + 0.5);
+	}
 	else
+	{
 		*totaldeadrows = 0.0;
+		*totalwarmchains = 0.0;
+	}
 
 	/*
 	 * Emit some interesting relation info
@@ -1219,11 +1254,13 @@ acquire_sample_rows(Relation onerel, int elevel,
 	ereport(elevel,
 			(errmsg("\"%s\": scanned %d of %u pages, "
 					"containing %.0f live rows and %.0f dead rows; "
-					"%d rows in sample, %.0f estimated total rows",
+					"%d rows in sample, %.0f estimated total rows; "
+					"%.0f warm chains",
 					RelationGetRelationName(onerel),
 					bs.m, totalblocks,
 					liverows, deadrows,
-					numrows, *totalrows)));
+					numrows, *totalrows,
+					*totalwarmchains)));
 
 	return numrows;
 }
@@ -1428,11 +1465,12 @@ acquire_inherited_sample_rows(Relation onerel, int elevel,
 				int			childrows;
 				double		trows,
 							tdrows;
+				double		twarmchains;
 
 				/* Fetch a random sample of the child's rows */
 				childrows = (*acquirefunc) (childrel, elevel,
 											rows + numrows, childtargrows,
-											&trows, &tdrows);
+											&trows, &tdrows, &twarmchains);
 
 				/* We may need to convert from child's rowtype to parent's */
 				if (childrows > 0 &&
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 9fbb0eb..52a7838 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -103,6 +103,7 @@ ExecVacuum(VacuumStmt *vacstmt, bool isTopLevel)
 		params.freeze_table_age = 0;
 		params.multixact_freeze_min_age = 0;
 		params.multixact_freeze_table_age = 0;
+		params.warmcleanup_index_scale = -1;
 	}
 	else
 	{
@@ -110,6 +111,7 @@ ExecVacuum(VacuumStmt *vacstmt, bool isTopLevel)
 		params.freeze_table_age = -1;
 		params.multixact_freeze_min_age = -1;
 		params.multixact_freeze_table_age = -1;
+		params.warmcleanup_index_scale = -1;
 	}
 
 	/* user-invoked vacuum is never "for wraparound" */
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index f52490f..87510ea 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -187,11 +187,12 @@ static BufferAccessStrategy vac_strategy;
 /* non-export function prototypes */
 static void lazy_scan_heap(Relation onerel, int options,
 			   LVRelStats *vacrelstats, Relation *Irel, int nindexes,
-			   bool aggressive);
+			   bool aggressive, double warmcleanup_index_scale);
 static void lazy_vacuum_heap(Relation onerel, LVRelStats *vacrelstats);
 static bool lazy_check_needs_freeze(Buffer buf, bool *hastup);
 static void lazy_vacuum_index(Relation indrel,
 				  bool clear_warm,
+				  double warmcleanup_index_scale,
 				  IndexBulkDeleteResult **stats,
 				  LVRelStats *vacrelstats);
 static void lazy_cleanup_index(Relation indrel,
@@ -207,7 +208,8 @@ 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 dowarmcleanup);
 static void lazy_record_dead_tuple(LVRelStats *vacrelstats,
 					   ItemPointer itemptr);
 static void lazy_record_warm_chain(LVRelStats *vacrelstats,
@@ -283,6 +285,9 @@ lazy_vacuum_rel(Relation onerel, int options, VacuumParams *params,
 						  &OldestXmin, &FreezeLimit, &xidFullScanLimit,
 						  &MultiXactCutoff, &mxactFullScanLimit);
 
+	/* Use default if the caller hasn't specified any value */
+	if (params->warmcleanup_index_scale == -1)
+		params->warmcleanup_index_scale = VacuumWarmCleanupIndexScale;
 	/*
 	 * We request an aggressive scan if the table's frozen Xid is now older
 	 * than or equal to the requested Xid full-table scan limit; or if the
@@ -309,7 +314,8 @@ lazy_vacuum_rel(Relation onerel, int options, VacuumParams *params,
 	vacrelstats->hasindex = (nindexes > 0);
 
 	/* Do the vacuuming */
-	lazy_scan_heap(onerel, options, vacrelstats, Irel, nindexes, aggressive);
+	lazy_scan_heap(onerel, options, vacrelstats, Irel, nindexes, aggressive,
+			params->warmcleanup_index_scale);
 
 	/* Done with indexes */
 	vac_close_indexes(nindexes, Irel, NoLock);
@@ -396,7 +402,8 @@ lazy_vacuum_rel(Relation onerel, int options, VacuumParams *params,
 	pgstat_report_vacuum(RelationGetRelid(onerel),
 						 onerel->rd_rel->relisshared,
 						 new_live_tuples,
-						 vacrelstats->new_dead_tuples);
+						 vacrelstats->new_dead_tuples,
+						 vacrelstats->num_non_convertible_warm_chains);
 	pgstat_progress_end_command();
 
 	/* and log the action if appropriate */
@@ -507,10 +514,19 @@ vacuum_log_cleanup_info(Relation rel, LVRelStats *vacrelstats)
  *		If there are no indexes then we can reclaim line pointers on the fly;
  *		dead line pointers need only be retained until all index pointers that
  *		reference them have been killed.
+ *
+ *		warmcleanup_index_scale specifies the number of WARM pointers in an
+ *		index as a fraction of total candidate WARM chains. If we find less
+ *		WARM pointers in an index than the specified fraction, then we don't
+ *		invoke cleanup that index. If WARM cleanup is skipped for any one
+ *		index, the WARM chain can't be cleared in the heap and no further WARM
+ *		updates are possible to such chains. Such chains are also not
+ *		considered for WARM cleanup in other indexes.
  */
 static void
 lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
-			   Relation *Irel, int nindexes, bool aggressive)
+			   Relation *Irel, int nindexes, bool aggressive,
+			   double warmcleanup_index_scale)
 {
 	BlockNumber nblocks,
 				blkno;
@@ -536,6 +552,7 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		PROGRESS_VACUUM_MAX_DEAD_TUPLES
 	};
 	int64		initprog_val[3];
+	bool		dowarmcleanup = ((options & VACOPT_WARM_CLEANUP) != 0);
 
 	pg_rusage_init(&ru0);
 
@@ -558,7 +575,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, dowarmcleanup);
 	frozen = palloc(sizeof(xl_heap_freeze_tuple) * MaxHeapTuplesPerPage);
 
 	/* Report that we're scanning the heap, advertising total # of blocks */
@@ -776,7 +793,8 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 			/* Remove index entries */
 			for (i = 0; i < nindexes; i++)
 				lazy_vacuum_index(Irel[i],
-								  (vacrelstats->num_warm_chains > 0),
+								  dowarmcleanup && (vacrelstats->num_warm_chains > 0),
+								  warmcleanup_index_scale,
 								  &indstats[i],
 								  vacrelstats);
 
@@ -1408,7 +1426,8 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		/* Remove index entries */
 		for (i = 0; i < nindexes; i++)
 			lazy_vacuum_index(Irel[i],
-							  (vacrelstats->num_warm_chains > 0),
+							  dowarmcleanup && (vacrelstats->num_warm_chains > 0),
+							  warmcleanup_index_scale,
 							  &indstats[i],
 							  vacrelstats);
 
@@ -1863,6 +1882,7 @@ lazy_reset_warm_pointer_count(LVRelStats *vacrelstats)
 static void
 lazy_vacuum_index(Relation indrel,
 				  bool clear_warm,
+				  double warmcleanup_index_scale,
 				  IndexBulkDeleteResult **stats,
 				  LVRelStats *vacrelstats)
 {
@@ -1927,25 +1947,55 @@ lazy_vacuum_index(Relation indrel,
 						(*stats)->warm_pointers_removed,
 						(*stats)->clear_pointers_removed)));
 
-		(*stats)->num_warm_pointers = 0;
-		(*stats)->num_clear_pointers = 0;
-		(*stats)->warm_pointers_removed = 0;
-		(*stats)->clear_pointers_removed = 0;
-		(*stats)->pointers_cleared = 0;
+		/*
+		 * If the number of WARM pointers found in the index are more than the
+		 * configured fraction of total candidate WARM chains, then do the
+		 * second index scan to clean up WARM chains.
+		 *
+		 * Otherwise we must set these WARM chains as non-convertible chains.
+		 */
+		if ((*stats)->num_warm_pointers >
+				((double)vacrelstats->num_warm_chains * warmcleanup_index_scale))
+		{
+			(*stats)->num_warm_pointers = 0;
+			(*stats)->num_clear_pointers = 0;
+			(*stats)->warm_pointers_removed = 0;
+			(*stats)->clear_pointers_removed = 0;
+			(*stats)->pointers_cleared = 0;
+
+			*stats = index_bulk_delete(&ivinfo, *stats,
+					lazy_indexvac_phase2, (void *) vacrelstats);
+			ereport(elevel,
+					(errmsg("scanned index \"%s\" to convert WARM pointers, found "
+							"%0.f WARM pointers, %0.f CLEAR pointers, removed "
+							"%0.f WARM pointers, removed %0.f CLEAR pointers, "
+							"cleared %0.f WARM pointers",
+							RelationGetRelationName(indrel),
+							(*stats)->num_warm_pointers,
+							(*stats)->num_clear_pointers,
+							(*stats)->warm_pointers_removed,
+							(*stats)->clear_pointers_removed,
+							(*stats)->pointers_cleared)));
+		}
+		else
+		{
+			int ii;
 
-		*stats = index_bulk_delete(&ivinfo, *stats,
-				lazy_indexvac_phase2, (void *) vacrelstats);
-		ereport(elevel,
-				(errmsg("scanned index \"%s\" to convert WARM pointers, found "
-						"%0.f WARM pointers, %0.f CLEAR pointers, removed "
-						"%0.f WARM pointers, removed %0.f CLEAR pointers, "
-						"cleared %0.f WARM pointers",
-						RelationGetRelationName(indrel),
-						(*stats)->num_warm_pointers,
-						(*stats)->num_clear_pointers,
-						(*stats)->warm_pointers_removed,
-						(*stats)->clear_pointers_removed,
-						(*stats)->pointers_cleared)));
+			/*
+			 * All chains skipped by this index are marked non-convertible.
+			 */
+			for (ii = 0; ii < vacrelstats->num_warm_chains; ii++)
+			{
+				LVWarmChain *chain = &vacrelstats->warm_chains[ii];
+				if (chain->num_warm_pointers > 0 ||
+					chain->num_clear_pointers > 1)
+				{
+					chain->keep_warm_chain = 1;
+					vacrelstats->num_non_convertible_warm_chains++;
+				}
+			}
+
+		}
 	}
 	else
 	{
@@ -2323,7 +2373,8 @@ 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 dowarmcleanup)
 {
 	long		maxtuples;
 	int			vac_work_mem = IsAutoVacuumWorkerProcess() &&
@@ -2332,8 +2383,13 @@ lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks)
 
 	if (vacrelstats->hasindex)
 	{
+		/*
+		 * If we're not doing WARM cleanup then the entire memory is available
+		 * for tracking dead tuples. Otherwise it gets split between tracking
+		 * dead tuples and tracking WARM chains.
+		 */
 		maxtuples = (vac_work_mem * 1024L) / (sizeof(ItemPointerData) +
-				sizeof(LVWarmChain));
+				dowarmcleanup ? sizeof(LVWarmChain) : 0);
 		maxtuples = Min(maxtuples, INT_MAX);
 		maxtuples = Min(maxtuples, MaxAllocSize / (sizeof(ItemPointerData) +
 					sizeof(LVWarmChain)));
@@ -2359,11 +2415,18 @@ lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks)
 	 * XXX Cheat for now and allocate the same size array for tracking warm
 	 * chains. maxtuples must have been already adjusted above to ensure we
 	 * don't cross vac_work_mem.
+	 *
+	 * XXX A better strategy seems to consume the available memory from two
+	 * ends and do a round of index cleanup if all available memory is
+	 * exhausted.
 	 */
-	vacrelstats->num_warm_chains = 0;
-	vacrelstats->max_warm_chains = (int) maxtuples;
-	vacrelstats->warm_chains = (LVWarmChain *)
-		palloc0(maxtuples * sizeof(LVWarmChain));
+	if (dowarmcleanup)
+	{
+		vacrelstats->num_warm_chains = 0;
+		vacrelstats->max_warm_chains = (int) maxtuples;
+		vacrelstats->warm_chains = (LVWarmChain *)
+			palloc0(maxtuples * sizeof(LVWarmChain));
+	}
 
 }
 
@@ -2385,6 +2448,8 @@ lazy_record_clear_chain(LVRelStats *vacrelstats,
 		vacrelstats->warm_chains[vacrelstats->num_warm_chains].is_postwarm_chain = 0;
 		vacrelstats->num_warm_chains++;
 	}
+	else
+		vacrelstats->num_non_convertible_warm_chains++;
 }
 
 /*
@@ -2405,6 +2470,8 @@ lazy_record_warm_chain(LVRelStats *vacrelstats,
 		vacrelstats->warm_chains[vacrelstats->num_warm_chains].is_postwarm_chain = 1;
 		vacrelstats->num_warm_chains++;
 	}
+	else
+		vacrelstats->num_non_convertible_warm_chains++;
 }
 
 /*
@@ -2600,6 +2667,7 @@ lazy_indexvac_phase2(ItemPointer itemptr, bool is_warm, void *state)
 		 * index pointers.
 		 */
 		chain->keep_warm_chain = 1;
+		vacrelstats->num_non_convertible_warm_chains++;
 		return IBDCR_KEEP;
 	}
 	return IBDCR_KEEP;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 19dd77d..25b2d69 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -433,7 +433,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type <node>	overlay_placing substr_from substr_for
 
 %type <boolean> opt_instead
-%type <boolean> opt_unique opt_concurrently opt_verbose opt_full
+%type <boolean> opt_unique opt_concurrently opt_verbose opt_full opt_warmclean
 %type <boolean> opt_freeze opt_default opt_recheck
 %type <defelt>	opt_binary opt_oids copy_delimiter
 
@@ -684,7 +684,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 	VACUUM VALID VALIDATE VALIDATOR VALUE_P VALUES VARCHAR VARIADIC VARYING
 	VERBOSE VERSION_P VIEW VIEWS VOLATILE
 
-	WHEN WHERE WHITESPACE_P WINDOW WITH WITHIN WITHOUT WORK WRAPPER WRITE
+	WARMCLEAN WHEN WHERE WHITESPACE_P WINDOW WITH WITHIN WITHOUT WORK WRAPPER WRITE
 
 	XML_P XMLATTRIBUTES XMLCONCAT XMLELEMENT XMLEXISTS XMLFOREST XMLNAMESPACES
 	XMLPARSE XMLPI XMLROOT XMLSERIALIZE XMLTABLE
@@ -10058,7 +10058,7 @@ cluster_index_specification:
  *
  *****************************************************************************/
 
-VacuumStmt: VACUUM opt_full opt_freeze opt_verbose
+VacuumStmt: VACUUM opt_full opt_freeze opt_verbose opt_warmclean
 				{
 					VacuumStmt *n = makeNode(VacuumStmt);
 					n->options = VACOPT_VACUUM;
@@ -10068,11 +10068,13 @@ VacuumStmt: VACUUM opt_full opt_freeze opt_verbose
 						n->options |= VACOPT_FREEZE;
 					if ($4)
 						n->options |= VACOPT_VERBOSE;
+					if ($5)
+						n->options |= VACOPT_WARM_CLEANUP;
 					n->relation = NULL;
 					n->va_cols = NIL;
 					$$ = (Node *)n;
 				}
-			| VACUUM opt_full opt_freeze opt_verbose qualified_name
+			| VACUUM opt_full opt_freeze opt_verbose opt_warmclean qualified_name
 				{
 					VacuumStmt *n = makeNode(VacuumStmt);
 					n->options = VACOPT_VACUUM;
@@ -10082,13 +10084,15 @@ VacuumStmt: VACUUM opt_full opt_freeze opt_verbose
 						n->options |= VACOPT_FREEZE;
 					if ($4)
 						n->options |= VACOPT_VERBOSE;
-					n->relation = $5;
+					if ($5)
+						n->options |= VACOPT_WARM_CLEANUP;
+					n->relation = $6;
 					n->va_cols = NIL;
 					$$ = (Node *)n;
 				}
-			| VACUUM opt_full opt_freeze opt_verbose AnalyzeStmt
+			| VACUUM opt_full opt_freeze opt_verbose opt_warmclean AnalyzeStmt
 				{
-					VacuumStmt *n = (VacuumStmt *) $5;
+					VacuumStmt *n = (VacuumStmt *) $6;
 					n->options |= VACOPT_VACUUM;
 					if ($2)
 						n->options |= VACOPT_FULL;
@@ -10096,6 +10100,8 @@ VacuumStmt: VACUUM opt_full opt_freeze opt_verbose
 						n->options |= VACOPT_FREEZE;
 					if ($4)
 						n->options |= VACOPT_VERBOSE;
+					if ($5)
+						n->options |= VACOPT_WARM_CLEANUP;
 					$$ = (Node *)n;
 				}
 			| VACUUM '(' vacuum_option_list ')'
@@ -10128,6 +10134,7 @@ vacuum_option_elem:
 			| VERBOSE			{ $$ = VACOPT_VERBOSE; }
 			| FREEZE			{ $$ = VACOPT_FREEZE; }
 			| FULL				{ $$ = VACOPT_FULL; }
+			| WARMCLEAN			{ $$ = VACOPT_WARM_CLEANUP; }
 			| IDENT
 				{
 					if (strcmp($1, "disable_page_skipping") == 0)
@@ -10181,6 +10188,10 @@ opt_freeze: FREEZE									{ $$ = TRUE; }
 			| /*EMPTY*/								{ $$ = FALSE; }
 		;
 
+opt_warmclean: WARMCLEAN							{ $$ = TRUE; }
+			| /*EMPTY*/								{ $$ = FALSE; }
+		;
+
 opt_name_list:
 			'(' name_list ')'						{ $$ = $2; }
 			| /*EMPTY*/								{ $$ = NIL; }
@@ -14884,6 +14895,7 @@ type_func_name_keyword:
 			| SIMILAR
 			| TABLESAMPLE
 			| VERBOSE
+			| WARMCLEAN
 		;
 
 /* Reserved keyword --- these keywords are usable only as a ColLabel.
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 33ca749..91793e4 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -115,6 +115,8 @@ int			autovacuum_vac_thresh;
 double		autovacuum_vac_scale;
 int			autovacuum_anl_thresh;
 double		autovacuum_anl_scale;
+double		autovacuum_warmcleanup_scale;
+double		autovacuum_warmcleanup_index_scale;
 int			autovacuum_freeze_max_age;
 int			autovacuum_multixact_freeze_max_age;
 
@@ -307,7 +309,8 @@ static void relation_needs_vacanalyze(Oid relid, AutoVacOpts *relopts,
 						  Form_pg_class classForm,
 						  PgStat_StatTabEntry *tabentry,
 						  int effective_multixact_freeze_max_age,
-						  bool *dovacuum, bool *doanalyze, bool *wraparound);
+						  bool *dovacuum, bool *doanalyze, bool *wraparound,
+						  bool *dowarmcleanup);
 
 static void autovacuum_do_vac_analyze(autovac_table *tab,
 						  BufferAccessStrategy bstrategy);
@@ -2010,6 +2013,7 @@ do_autovacuum(void)
 		bool		dovacuum;
 		bool		doanalyze;
 		bool		wraparound;
+		bool		dowarmcleanup;
 
 		if (classForm->relkind != RELKIND_RELATION &&
 			classForm->relkind != RELKIND_MATVIEW)
@@ -2049,10 +2053,14 @@ do_autovacuum(void)
 		tabentry = get_pgstat_tabentry_relid(relid, classForm->relisshared,
 											 shared, dbentry);
 
-		/* Check if it needs vacuum or analyze */
+		/* 
+		 * Check if it needs vacuum or analyze. For vacuum, also check if it
+		 * needs WARM cleanup.
+		 */
 		relation_needs_vacanalyze(relid, relopts, classForm, tabentry,
 								  effective_multixact_freeze_max_age,
-								  &dovacuum, &doanalyze, &wraparound);
+								  &dovacuum, &doanalyze, &wraparound,
+								  &dowarmcleanup);
 
 		/* Relations that need work are added to table_oids */
 		if (dovacuum || doanalyze)
@@ -2105,6 +2113,7 @@ do_autovacuum(void)
 		bool		dovacuum;
 		bool		doanalyze;
 		bool		wraparound;
+		bool		dowarmcleanup;
 
 		/*
 		 * We cannot safely process other backends' temp tables, so skip 'em.
@@ -2135,7 +2144,8 @@ do_autovacuum(void)
 
 		relation_needs_vacanalyze(relid, relopts, classForm, tabentry,
 								  effective_multixact_freeze_max_age,
-								  &dovacuum, &doanalyze, &wraparound);
+								  &dovacuum, &doanalyze, &wraparound,
+								  &dowarmcleanup);
 
 		/* ignore analyze for toast tables */
 		if (dovacuum)
@@ -2566,6 +2576,7 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 	HeapTuple	classTup;
 	bool		dovacuum;
 	bool		doanalyze;
+	bool		dowarmcleanup;
 	autovac_table *tab = NULL;
 	PgStat_StatTabEntry *tabentry;
 	PgStat_StatDBEntry *shared;
@@ -2607,7 +2618,8 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 
 	relation_needs_vacanalyze(relid, avopts, classForm, tabentry,
 							  effective_multixact_freeze_max_age,
-							  &dovacuum, &doanalyze, &wraparound);
+							  &dovacuum, &doanalyze, &wraparound,
+							  &dowarmcleanup);
 
 	/* ignore ANALYZE for toast tables */
 	if (classForm->relkind == RELKIND_TOASTVALUE)
@@ -2623,6 +2635,7 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 		int			vac_cost_limit;
 		int			vac_cost_delay;
 		int			log_min_duration;
+		double		warmcleanup_index_scale;
 
 		/*
 		 * Calculate the vacuum cost parameters and the freeze ages.  If there
@@ -2669,19 +2682,26 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 			? avopts->multixact_freeze_table_age
 			: default_multixact_freeze_table_age;
 
+		warmcleanup_index_scale = (avopts &&
+								   avopts->warmcleanup_index_scale >= 0)
+			? avopts->warmcleanup_index_scale
+			: autovacuum_warmcleanup_index_scale;
+
 		tab = palloc(sizeof(autovac_table));
 		tab->at_relid = relid;
 		tab->at_sharedrel = classForm->relisshared;
 		tab->at_vacoptions = VACOPT_SKIPTOAST |
 			(dovacuum ? VACOPT_VACUUM : 0) |
 			(doanalyze ? VACOPT_ANALYZE : 0) |
-			(!wraparound ? VACOPT_NOWAIT : 0);
+			(!wraparound ? VACOPT_NOWAIT : 0) |
+			(dowarmcleanup ? VACOPT_WARM_CLEANUP : 0);
 		tab->at_params.freeze_min_age = freeze_min_age;
 		tab->at_params.freeze_table_age = freeze_table_age;
 		tab->at_params.multixact_freeze_min_age = multixact_freeze_min_age;
 		tab->at_params.multixact_freeze_table_age = multixact_freeze_table_age;
 		tab->at_params.is_wraparound = wraparound;
 		tab->at_params.log_min_duration = log_min_duration;
+		tab->at_params.warmcleanup_index_scale = warmcleanup_index_scale;
 		tab->at_vacuum_cost_limit = vac_cost_limit;
 		tab->at_vacuum_cost_delay = vac_cost_delay;
 		tab->at_relname = NULL;
@@ -2748,7 +2768,8 @@ relation_needs_vacanalyze(Oid relid,
  /* output params below */
 						  bool *dovacuum,
 						  bool *doanalyze,
-						  bool *wraparound)
+						  bool *wraparound,
+						  bool *dowarmcleanup)
 {
 	bool		force_vacuum;
 	bool		av_enabled;
@@ -2760,6 +2781,9 @@ relation_needs_vacanalyze(Oid relid,
 	float4		vac_scale_factor,
 				anl_scale_factor;
 
+	/* constant from reloptions or GUC valriable */
+	float4		warmcleanup_scale_factor;
+
 	/* thresholds calculated from above constants */
 	float4		vacthresh,
 				anlthresh;
@@ -2768,6 +2792,9 @@ relation_needs_vacanalyze(Oid relid,
 	float4		vactuples,
 				anltuples;
 
+	/* number of WARM chains in the table */
+	float4		warmchains;
+
 	/* freeze parameters */
 	int			freeze_max_age;
 	int			multixact_freeze_max_age;
@@ -2800,6 +2827,11 @@ relation_needs_vacanalyze(Oid relid,
 		? relopts->analyze_threshold
 		: autovacuum_anl_thresh;
 
+	/* Use table specific value or the GUC value */
+	warmcleanup_scale_factor = (relopts && relopts->warmcleanup_scale_factor >= 0)
+		? relopts->warmcleanup_scale_factor
+		: autovacuum_warmcleanup_scale;
+
 	freeze_max_age = (relopts && relopts->freeze_max_age >= 0)
 		? Min(relopts->freeze_max_age, autovacuum_freeze_max_age)
 		: autovacuum_freeze_max_age;
@@ -2847,6 +2879,7 @@ relation_needs_vacanalyze(Oid relid,
 		reltuples = classForm->reltuples;
 		vactuples = tabentry->n_dead_tuples;
 		anltuples = tabentry->changes_since_analyze;
+		warmchains = tabentry->n_warm_chains;
 
 		vacthresh = (float4) vac_base_thresh + vac_scale_factor * reltuples;
 		anlthresh = (float4) anl_base_thresh + anl_scale_factor * reltuples;
@@ -2863,6 +2896,17 @@ relation_needs_vacanalyze(Oid relid,
 		/* Determine if this table needs vacuum or analyze. */
 		*dovacuum = force_vacuum || (vactuples > vacthresh);
 		*doanalyze = (anltuples > anlthresh);
+
+		/*
+		 * If the number of WARM chains in the is more than the configured
+		 * fraction, then we also do a WARM cleanup. This only triggers at the
+		 * table level, but we then look at each index and do cleanup for the
+		 * index only if the WARM pointers in the index are more than
+		 * configured index-level scale factor. lazy_vacuum_index() later deals
+		 * with that.
+		 */
+		if (*dovacuum && (warmcleanup_scale_factor * reltuples < warmchains))
+			*dowarmcleanup = true;
 	}
 	else
 	{
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index cdfd76e..017ac20 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -199,9 +199,11 @@ typedef struct TwoPhasePgStatRecord
 	PgStat_Counter tuples_inserted;		/* tuples inserted in xact */
 	PgStat_Counter tuples_updated;		/* tuples updated in xact */
 	PgStat_Counter tuples_deleted;		/* tuples deleted in xact */
+	PgStat_Counter tuples_warm_updated;	/* tuples warm updated in xact */
 	PgStat_Counter inserted_pre_trunc;	/* tuples inserted prior to truncate */
 	PgStat_Counter updated_pre_trunc;	/* tuples updated prior to truncate */
 	PgStat_Counter deleted_pre_trunc;	/* tuples deleted prior to truncate */
+	PgStat_Counter warm_updated_pre_trunc;	/* tuples warm updated prior to truncate */
 	Oid			t_id;			/* table's OID */
 	bool		t_shared;		/* is it a shared catalog? */
 	bool		t_truncated;	/* was the relation truncated? */
@@ -1328,7 +1330,8 @@ pgstat_report_autovac(Oid dboid)
  */
 void
 pgstat_report_vacuum(Oid tableoid, bool shared,
-					 PgStat_Counter livetuples, PgStat_Counter deadtuples)
+					 PgStat_Counter livetuples, PgStat_Counter deadtuples,
+					 PgStat_Counter warmchains)
 {
 	PgStat_MsgVacuum msg;
 
@@ -1342,6 +1345,7 @@ pgstat_report_vacuum(Oid tableoid, bool shared,
 	msg.m_vacuumtime = GetCurrentTimestamp();
 	msg.m_live_tuples = livetuples;
 	msg.m_dead_tuples = deadtuples;
+	msg.m_warm_chains = warmchains;
 	pgstat_send(&msg, sizeof(msg));
 }
 
@@ -1357,7 +1361,7 @@ pgstat_report_vacuum(Oid tableoid, bool shared,
 void
 pgstat_report_analyze(Relation rel,
 					  PgStat_Counter livetuples, PgStat_Counter deadtuples,
-					  bool resetcounter)
+					  PgStat_Counter warmchains, bool resetcounter)
 {
 	PgStat_MsgAnalyze msg;
 
@@ -1382,12 +1386,14 @@ pgstat_report_analyze(Relation rel,
 		{
 			livetuples -= trans->tuples_inserted - trans->tuples_deleted;
 			deadtuples -= trans->tuples_updated + trans->tuples_deleted;
+			warmchains -= trans->tuples_warm_updated;
 		}
 		/* count stuff inserted by already-aborted subxacts, too */
 		deadtuples -= rel->pgstat_info->t_counts.t_delta_dead_tuples;
 		/* Since ANALYZE's counts are estimates, we could have underflowed */
 		livetuples = Max(livetuples, 0);
 		deadtuples = Max(deadtuples, 0);
+		warmchains = Max(warmchains, 0);
 	}
 
 	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_ANALYZE);
@@ -1398,6 +1404,7 @@ pgstat_report_analyze(Relation rel,
 	msg.m_analyzetime = GetCurrentTimestamp();
 	msg.m_live_tuples = livetuples;
 	msg.m_dead_tuples = deadtuples;
+	msg.m_warm_chains = warmchains;
 	pgstat_send(&msg, sizeof(msg));
 }
 
@@ -1843,7 +1850,10 @@ pgstat_count_heap_update(Relation rel, bool hot, bool warm)
 		if (hot)
 			pgstat_info->t_counts.t_tuples_hot_updated++;
 		else if (warm)
+		{
+			pgstat_info->trans->tuples_warm_updated++;
 			pgstat_info->t_counts.t_tuples_warm_updated++;
+		}
 	}
 }
 
@@ -2006,6 +2016,12 @@ AtEOXact_PgStat(bool isCommit)
 				/* update and delete each create a dead tuple */
 				tabstat->t_counts.t_delta_dead_tuples +=
 					trans->tuples_updated + trans->tuples_deleted;
+				/* 
+				 * commit or abort, a WARM update generates a WARM chain which
+				 * needs cleanup.
+				 */
+				tabstat->t_counts.t_delta_warm_chains +=
+					trans->tuples_warm_updated;
 				/* insert, update, delete each count as one change event */
 				tabstat->t_counts.t_changed_tuples +=
 					trans->tuples_inserted + trans->tuples_updated +
@@ -2016,6 +2032,12 @@ AtEOXact_PgStat(bool isCommit)
 				/* inserted tuples are dead, deleted tuples are unaffected */
 				tabstat->t_counts.t_delta_dead_tuples +=
 					trans->tuples_inserted + trans->tuples_updated;
+				/* 
+				 * commit or abort, a WARM update generates a WARM chain which
+				 * needs cleanup.
+				 */
+				tabstat->t_counts.t_delta_warm_chains +=
+					trans->tuples_warm_updated;
 				/* an aborted xact generates no changed_tuple events */
 			}
 			tabstat->trans = NULL;
@@ -2072,12 +2094,16 @@ AtEOSubXact_PgStat(bool isCommit, int nestDepth)
 						trans->upper->tuples_inserted = trans->tuples_inserted;
 						trans->upper->tuples_updated = trans->tuples_updated;
 						trans->upper->tuples_deleted = trans->tuples_deleted;
+						trans->upper->tuples_warm_updated =
+							trans->tuples_warm_updated;
 					}
 					else
 					{
 						trans->upper->tuples_inserted += trans->tuples_inserted;
 						trans->upper->tuples_updated += trans->tuples_updated;
 						trans->upper->tuples_deleted += trans->tuples_deleted;
+						trans->upper->tuples_warm_updated +=
+							trans->tuples_warm_updated;
 					}
 					tabstat->trans = trans->upper;
 					pfree(trans);
@@ -2113,9 +2139,13 @@ AtEOSubXact_PgStat(bool isCommit, int nestDepth)
 				tabstat->t_counts.t_tuples_inserted += trans->tuples_inserted;
 				tabstat->t_counts.t_tuples_updated += trans->tuples_updated;
 				tabstat->t_counts.t_tuples_deleted += trans->tuples_deleted;
+				tabstat->t_counts.t_tuples_warm_updated +=
+					trans->tuples_warm_updated;
 				/* inserted tuples are dead, deleted tuples are unaffected */
 				tabstat->t_counts.t_delta_dead_tuples +=
 					trans->tuples_inserted + trans->tuples_updated;
+				tabstat->t_counts.t_delta_warm_chains +=
+					trans->tuples_warm_updated;
 				tabstat->trans = trans->upper;
 				pfree(trans);
 			}
@@ -2157,9 +2187,11 @@ AtPrepare_PgStat(void)
 			record.tuples_inserted = trans->tuples_inserted;
 			record.tuples_updated = trans->tuples_updated;
 			record.tuples_deleted = trans->tuples_deleted;
+			record.tuples_warm_updated = trans->tuples_warm_updated;
 			record.inserted_pre_trunc = trans->inserted_pre_trunc;
 			record.updated_pre_trunc = trans->updated_pre_trunc;
 			record.deleted_pre_trunc = trans->deleted_pre_trunc;
+			record.warm_updated_pre_trunc = trans->warm_updated_pre_trunc;
 			record.t_id = tabstat->t_id;
 			record.t_shared = tabstat->t_shared;
 			record.t_truncated = trans->truncated;
@@ -2234,11 +2266,14 @@ pgstat_twophase_postcommit(TransactionId xid, uint16 info,
 		/* forget live/dead stats seen by backend thus far */
 		pgstat_info->t_counts.t_delta_live_tuples = 0;
 		pgstat_info->t_counts.t_delta_dead_tuples = 0;
+		pgstat_info->t_counts.t_delta_warm_chains = 0;
 	}
 	pgstat_info->t_counts.t_delta_live_tuples +=
 		rec->tuples_inserted - rec->tuples_deleted;
 	pgstat_info->t_counts.t_delta_dead_tuples +=
 		rec->tuples_updated + rec->tuples_deleted;
+	pgstat_info->t_counts.t_delta_warm_chains +=
+		rec->tuples_warm_updated;
 	pgstat_info->t_counts.t_changed_tuples +=
 		rec->tuples_inserted + rec->tuples_updated +
 		rec->tuples_deleted;
@@ -2266,12 +2301,16 @@ pgstat_twophase_postabort(TransactionId xid, uint16 info,
 		rec->tuples_inserted = rec->inserted_pre_trunc;
 		rec->tuples_updated = rec->updated_pre_trunc;
 		rec->tuples_deleted = rec->deleted_pre_trunc;
+		rec->tuples_warm_updated = rec->warm_updated_pre_trunc;
 	}
 	pgstat_info->t_counts.t_tuples_inserted += rec->tuples_inserted;
 	pgstat_info->t_counts.t_tuples_updated += rec->tuples_updated;
 	pgstat_info->t_counts.t_tuples_deleted += rec->tuples_deleted;
+	pgstat_info->t_counts.t_tuples_warm_updated += rec->tuples_warm_updated;
 	pgstat_info->t_counts.t_delta_dead_tuples +=
 		rec->tuples_inserted + rec->tuples_updated;
+	pgstat_info->t_counts.t_delta_warm_chains +=
+		rec->tuples_warm_updated;
 }
 
 
@@ -4335,6 +4374,7 @@ pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create)
 		result->tuples_warm_updated = 0;
 		result->n_live_tuples = 0;
 		result->n_dead_tuples = 0;
+		result->n_warm_chains = 0;
 		result->changes_since_analyze = 0;
 		result->blocks_fetched = 0;
 		result->blocks_hit = 0;
@@ -5445,6 +5485,7 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 			tabentry->tuples_warm_updated = tabmsg->t_counts.t_tuples_warm_updated;
 			tabentry->n_live_tuples = tabmsg->t_counts.t_delta_live_tuples;
 			tabentry->n_dead_tuples = tabmsg->t_counts.t_delta_dead_tuples;
+			tabentry->n_warm_chains = tabmsg->t_counts.t_delta_warm_chains;
 			tabentry->changes_since_analyze = tabmsg->t_counts.t_changed_tuples;
 			tabentry->blocks_fetched = tabmsg->t_counts.t_blocks_fetched;
 			tabentry->blocks_hit = tabmsg->t_counts.t_blocks_hit;
@@ -5476,9 +5517,11 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 			{
 				tabentry->n_live_tuples = 0;
 				tabentry->n_dead_tuples = 0;
+				tabentry->n_warm_chains = 0;
 			}
 			tabentry->n_live_tuples += tabmsg->t_counts.t_delta_live_tuples;
 			tabentry->n_dead_tuples += tabmsg->t_counts.t_delta_dead_tuples;
+			tabentry->n_warm_chains += tabmsg->t_counts.t_delta_warm_chains;
 			tabentry->changes_since_analyze += tabmsg->t_counts.t_changed_tuples;
 			tabentry->blocks_fetched += tabmsg->t_counts.t_blocks_fetched;
 			tabentry->blocks_hit += tabmsg->t_counts.t_blocks_hit;
@@ -5488,6 +5531,7 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 		tabentry->n_live_tuples = Max(tabentry->n_live_tuples, 0);
 		/* Likewise for n_dead_tuples */
 		tabentry->n_dead_tuples = Max(tabentry->n_dead_tuples, 0);
+		tabentry->n_warm_chains = Max(tabentry->n_warm_chains, 0);
 
 		/*
 		 * Add per-table stats to the per-database entry, too.
@@ -5713,6 +5757,7 @@ pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
 
 	tabentry->n_live_tuples = msg->m_live_tuples;
 	tabentry->n_dead_tuples = msg->m_dead_tuples;
+	tabentry->n_warm_chains = msg->m_warm_chains;
 
 	if (msg->m_autovacuum)
 	{
@@ -5747,6 +5792,7 @@ pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len)
 
 	tabentry->n_live_tuples = msg->m_live_tuples;
 	tabentry->n_dead_tuples = msg->m_dead_tuples;
+	tabentry->n_warm_chains = msg->m_warm_chains;
 
 	/*
 	 * If commanded, reset changes_since_analyze to zero.  This forgets any
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index b8677f3..814d071 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -191,6 +191,21 @@ pg_stat_get_dead_tuples(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(result);
 }
 
+Datum
+pg_stat_get_warm_chains(PG_FUNCTION_ARGS)
+{
+	Oid			relid = PG_GETARG_OID(0);
+	int64		result;
+	PgStat_StatTabEntry *tabentry;
+
+	if ((tabentry = pgstat_fetch_stat_tabentry(relid)) == NULL)
+		result = 0;
+	else
+		result = (int64) (tabentry->n_warm_chains);
+
+	PG_RETURN_INT64(result);
+}
+
 
 Datum
 pg_stat_get_mod_since_analyze(PG_FUNCTION_ARGS)
diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c
index 08b6030..81fec03 100644
--- a/src/backend/utils/init/globals.c
+++ b/src/backend/utils/init/globals.c
@@ -130,6 +130,7 @@ int			VacuumCostPageMiss = 10;
 int			VacuumCostPageDirty = 20;
 int			VacuumCostLimit = 200;
 int			VacuumCostDelay = 0;
+double		VacuumWarmCleanupScale;
 
 int			VacuumPageHit = 0;
 int			VacuumPageMiss = 0;
@@ -137,3 +138,5 @@ int			VacuumPageDirty = 0;
 
 int			VacuumCostBalance = 0;		/* working state for vacuum */
 bool		VacuumCostActive = false;
+
+double		VacuumWarmCleanupIndexScale = 1;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 291bf76..b4daa2c 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3016,6 +3016,36 @@ static struct config_real ConfigureNamesReal[] =
 	},
 
 	{
+		{"autovacuum_warmcleanup_scale_factor", PGC_SIGHUP, AUTOVACUUM,
+			gettext_noop("Number of WARM chains prior to cleanup as a fraction of reltuples."),
+			NULL
+		},
+		&autovacuum_warmcleanup_scale,
+		0.1, 0.0, 100.0,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"autovacuum_warmcleanup_index_scale_factor", PGC_SIGHUP, AUTOVACUUM,
+			gettext_noop("Number of WARM pointers prior to cleanup as a fraction of total WARM chains."),
+			NULL
+		},
+		&autovacuum_warmcleanup_index_scale,
+		0.2, 0.0, 100.0,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"vacuum_warmcleanup_index_scale_factor", PGC_USERSET, WARM_CLEANUP,
+			gettext_noop("Number of WARM pointers in the index prior to cleanup as a fraction of total WARM chains."),
+			NULL
+		},
+		&VacuumWarmCleanupIndexScale,
+		0.2, 0.0, 100.0,
+		NULL, NULL, NULL
+	},
+
+	{
 		{"checkpoint_completion_target", PGC_SIGHUP, WAL_CHECKPOINTS,
 			gettext_noop("Time spent flushing dirty buffers during checkpoint, as fraction of checkpoint interval."),
 			NULL
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 509adda..70025ba 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -2789,6 +2789,8 @@ DATA(insert OID = 2878 (  pg_stat_get_live_tuples	PGNSP PGUID 12 1 0 0 0 f f f f
 DESCR("statistics: number of live tuples");
 DATA(insert OID = 2879 (  pg_stat_get_dead_tuples	PGNSP PGUID 12 1 0 0 0 f f f f t f s r 1 0 20 "26" _null_ _null_ _null_ _null_ _null_ pg_stat_get_dead_tuples _null_ _null_ _null_ ));
 DESCR("statistics: number of dead tuples");
+DATA(insert OID = 3374 (  pg_stat_get_warm_chains	PGNSP PGUID 12 1 0 0 0 f f f f t f s r 1 0 20 "26" _null_ _null_ _null_ _null_ _null_ pg_stat_get_warm_chains _null_ _null_ _null_ ));
+DESCR("statistics: number of warm chains");
 DATA(insert OID = 3177 (  pg_stat_get_mod_since_analyze PGNSP PGUID 12 1 0 0 0 f f f f t f s r 1 0 20 "26" _null_ _null_ _null_ _null_ _null_ pg_stat_get_mod_since_analyze _null_ _null_ _null_ ));
 DESCR("statistics: number of tuples changed since last analyze");
 DATA(insert OID = 1934 (  pg_stat_get_blocks_fetched	PGNSP PGUID 12 1 0 0 0 f f f f t f s r 1 0 20 "26" _null_ _null_ _null_ _null_ _null_ pg_stat_get_blocks_fetched _null_ _null_ _null_ ));
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 541c2fa..9914143 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -145,6 +145,8 @@ typedef struct VacuumParams
 	int			log_min_duration;		/* minimum execution threshold in ms
 										 * at which  verbose logs are
 										 * activated, -1 to use default */
+	double		warmcleanup_index_scale; /* Fraction of WARM pointers to cause
+										  * index WARM cleanup */
 } VacuumParams;
 
 /* GUC parameters */
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 6ca44f7..2993b1a 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -134,7 +134,8 @@ typedef void (*ExplainDirectModify_function) (ForeignScanState *node,
 typedef int (*AcquireSampleRowsFunc) (Relation relation, int elevel,
 											   HeapTuple *rows, int targrows,
 												  double *totalrows,
-												  double *totaldeadrows);
+												  double *totaldeadrows,
+												  double *totalwarmchains);
 
 typedef bool (*AnalyzeForeignTable_function) (Relation relation,
 												 AcquireSampleRowsFunc *func,
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index 4c607b2..901960a 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -255,6 +255,7 @@ extern int	VacuumPageDirty;
 extern int	VacuumCostBalance;
 extern bool VacuumCostActive;
 
+extern double VacuumWarmCleanupIndexScale;
 
 /* in tcop/postgres.c */
 
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 3a71dd5..f842374 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3035,7 +3035,8 @@ typedef enum VacuumOption
 	VACOPT_FULL = 1 << 4,		/* FULL (non-concurrent) vacuum */
 	VACOPT_NOWAIT = 1 << 5,		/* don't wait to get lock (autovacuum only) */
 	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_WARM_CLEANUP = 1 << 8	/* do WARM cleanup */
 } VacuumOption;
 
 typedef struct VacuumStmt
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 6cd36c7..632283b 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -432,6 +432,7 @@ PG_KEYWORD("version", VERSION_P, UNRESERVED_KEYWORD)
 PG_KEYWORD("view", VIEW, UNRESERVED_KEYWORD)
 PG_KEYWORD("views", VIEWS, UNRESERVED_KEYWORD)
 PG_KEYWORD("volatile", VOLATILE, UNRESERVED_KEYWORD)
+PG_KEYWORD("warmclean", WARMCLEAN, TYPE_FUNC_NAME_KEYWORD)
 PG_KEYWORD("when", WHEN, RESERVED_KEYWORD)
 PG_KEYWORD("where", WHERE, RESERVED_KEYWORD)
 PG_KEYWORD("whitespace", WHITESPACE_P, UNRESERVED_KEYWORD)
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 4b7d671..8901c9d 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -110,6 +110,7 @@ typedef struct PgStat_TableCounts
 
 	PgStat_Counter t_delta_live_tuples;
 	PgStat_Counter t_delta_dead_tuples;
+	PgStat_Counter t_delta_warm_chains;
 	PgStat_Counter t_changed_tuples;
 
 	PgStat_Counter t_blocks_fetched;
@@ -167,11 +168,13 @@ typedef struct PgStat_TableXactStatus
 {
 	PgStat_Counter tuples_inserted;		/* tuples inserted in (sub)xact */
 	PgStat_Counter tuples_updated;		/* tuples updated in (sub)xact */
+	PgStat_Counter tuples_warm_updated;	/* tuples warm-updated in (sub)xact */
 	PgStat_Counter tuples_deleted;		/* tuples deleted in (sub)xact */
 	bool		truncated;		/* relation truncated in this (sub)xact */
 	PgStat_Counter inserted_pre_trunc;	/* tuples inserted prior to truncate */
 	PgStat_Counter updated_pre_trunc;	/* tuples updated prior to truncate */
 	PgStat_Counter deleted_pre_trunc;	/* tuples deleted prior to truncate */
+	PgStat_Counter warm_updated_pre_trunc;	/* tuples warm updated prior to truncate */
 	int			nest_level;		/* subtransaction nest level */
 	/* links to other structs for same relation: */
 	struct PgStat_TableXactStatus *upper;		/* next higher subxact if any */
@@ -370,6 +373,7 @@ typedef struct PgStat_MsgVacuum
 	TimestampTz m_vacuumtime;
 	PgStat_Counter m_live_tuples;
 	PgStat_Counter m_dead_tuples;
+	PgStat_Counter m_warm_chains;
 } PgStat_MsgVacuum;
 
 
@@ -388,6 +392,7 @@ typedef struct PgStat_MsgAnalyze
 	TimestampTz m_analyzetime;
 	PgStat_Counter m_live_tuples;
 	PgStat_Counter m_dead_tuples;
+	PgStat_Counter m_warm_chains;
 } PgStat_MsgAnalyze;
 
 
@@ -630,6 +635,7 @@ typedef struct PgStat_StatTabEntry
 
 	PgStat_Counter n_live_tuples;
 	PgStat_Counter n_dead_tuples;
+	PgStat_Counter n_warm_chains;
 	PgStat_Counter changes_since_analyze;
 
 	PgStat_Counter blocks_fetched;
@@ -1131,10 +1137,11 @@ extern void pgstat_reset_single_counter(Oid objectid, PgStat_Single_Reset_Type t
 
 extern void pgstat_report_autovac(Oid dboid);
 extern void pgstat_report_vacuum(Oid tableoid, bool shared,
-					 PgStat_Counter livetuples, PgStat_Counter deadtuples);
+					 PgStat_Counter livetuples, PgStat_Counter deadtuples,
+					 PgStat_Counter warmchains);
 extern void pgstat_report_analyze(Relation rel,
 					  PgStat_Counter livetuples, PgStat_Counter deadtuples,
-					  bool resetcounter);
+					  PgStat_Counter warmchains, bool resetcounter);
 
 extern void pgstat_report_recovery_conflict(int reason);
 extern void pgstat_report_deadlock(void);
diff --git a/src/include/postmaster/autovacuum.h b/src/include/postmaster/autovacuum.h
index 99d7f09..5ac9c8f 100644
--- a/src/include/postmaster/autovacuum.h
+++ b/src/include/postmaster/autovacuum.h
@@ -28,6 +28,8 @@ extern int	autovacuum_freeze_max_age;
 extern int	autovacuum_multixact_freeze_max_age;
 extern int	autovacuum_vac_cost_delay;
 extern int	autovacuum_vac_cost_limit;
+extern double autovacuum_warmcleanup_scale;
+extern double autovacuum_warmcleanup_index_scale;
 
 /* autovacuum launcher PID, only valid when worker is shutting down */
 extern int	AutovacuumLauncherPid;
diff --git a/src/include/utils/guc_tables.h b/src/include/utils/guc_tables.h
index 2da9115..cd4532b 100644
--- a/src/include/utils/guc_tables.h
+++ b/src/include/utils/guc_tables.h
@@ -68,6 +68,7 @@ enum config_group
 	WAL_SETTINGS,
 	WAL_CHECKPOINTS,
 	WAL_ARCHIVING,
+	WARM_CLEANUP,
 	REPLICATION,
 	REPLICATION_SENDING,
 	REPLICATION_MASTER,
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index cd1976a..9164f60 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -276,6 +276,8 @@ typedef struct AutoVacOpts
 	int			log_min_duration;
 	float8		vacuum_scale_factor;
 	float8		analyze_scale_factor;
+	float8		warmcleanup_scale_factor;
+	float8		warmcleanup_index_scale;
 } AutoVacOpts;
 
 typedef struct StdRdOptions
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index a37d443..1b2a69a 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1758,6 +1758,7 @@ pg_stat_all_tables| SELECT c.oid AS relid,
     pg_stat_get_tuples_warm_updated(c.oid) AS n_tup_warm_upd,
     pg_stat_get_live_tuples(c.oid) AS n_live_tup,
     pg_stat_get_dead_tuples(c.oid) AS n_dead_tup,
+    pg_stat_get_warm_chains(c.oid) AS n_warm_chains,
     pg_stat_get_mod_since_analyze(c.oid) AS n_mod_since_analyze,
     pg_stat_get_last_vacuum_time(c.oid) AS last_vacuum,
     pg_stat_get_last_autovacuum_time(c.oid) AS last_autovacuum,
@@ -1906,6 +1907,7 @@ pg_stat_sys_tables| SELECT pg_stat_all_tables.relid,
     pg_stat_all_tables.n_tup_warm_upd,
     pg_stat_all_tables.n_live_tup,
     pg_stat_all_tables.n_dead_tup,
+    pg_stat_all_tables.n_warm_chains,
     pg_stat_all_tables.n_mod_since_analyze,
     pg_stat_all_tables.last_vacuum,
     pg_stat_all_tables.last_autovacuum,
@@ -1950,6 +1952,7 @@ pg_stat_user_tables| SELECT pg_stat_all_tables.relid,
     pg_stat_all_tables.n_tup_warm_upd,
     pg_stat_all_tables.n_live_tup,
     pg_stat_all_tables.n_dead_tup,
+    pg_stat_all_tables.n_warm_chains,
     pg_stat_all_tables.n_mod_since_analyze,
     pg_stat_all_tables.last_vacuum,
     pg_stat_all_tables.last_autovacuum,
diff --git a/src/test/regress/expected/warm.out b/src/test/regress/expected/warm.out
index 8aa1505..1346bb1 100644
--- a/src/test/regress/expected/warm.out
+++ b/src/test/regress/expected/warm.out
@@ -745,3 +745,61 @@ SELECT a, b FROM test_toast_warm WHERE b = 104.20;
 (1 row)
 
 DROP TABLE test_toast_warm;
+-- Test VACUUM
+CREATE TABLE test_vacuum_warm (a int unique, b text, c int, d int);
+CREATE INDEX test_vacuum_warm_index1 ON test_vacuum_warm(b);
+CREATE INDEX test_vacuum_warm_index2 ON test_vacuum_warm(c);
+INSERT INTO test_vacuum_warm VALUES (1, 'a', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (2, 'b', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (3, 'c', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (4, 'd', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (5, 'e', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (6, 'f', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (7, 'g', 100, 200);
+UPDATE test_vacuum_warm SET b = 'u', c = 300 WHERE a = 1;
+UPDATE test_vacuum_warm SET b = 'v', c = 300 WHERE a = 2;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 3;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 4;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 5;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 6;
+-- a plain vacuum cannot clear WARM chains.
+SET enable_seqscan = false;
+SET enable_bitmapscan = false;
+SET seq_page_cost = 10000;
+VACUUM test_vacuum_warm;
+-- We expect non-zero heap-fetches here
+EXPLAIN (analyze, costs off, timing off, summary off) SELECT b FROM test_vacuum_warm WHERE b = 'u';
+                                        QUERY PLAN                                         
+-------------------------------------------------------------------------------------------
+ Index Only Scan using test_vacuum_warm_index1 on test_vacuum_warm (actual rows=1 loops=1)
+   Index Cond: (b = 'u'::text)
+   Heap Fetches: 1
+(3 rows)
+
+-- Now set vacuum_warmcleanup_index_scale_factor such that only
+-- test_vacuum_warm_index2 can be cleaned up.
+SET vacuum_warmcleanup_index_scale_factor=0.5;
+VACUUM WARMCLEAN test_vacuum_warm;
+-- We expect non-zero heap-fetches here
+EXPLAIN (analyze, costs off, timing off, summary off) SELECT b FROM test_vacuum_warm WHERE b = 'u';
+                                        QUERY PLAN                                         
+-------------------------------------------------------------------------------------------
+ Index Only Scan using test_vacuum_warm_index1 on test_vacuum_warm (actual rows=1 loops=1)
+   Index Cond: (b = 'u'::text)
+   Heap Fetches: 1
+(3 rows)
+
+-- All WARM chains cleaned up, so index-only scan should be used now without
+-- any heap fetches
+SET vacuum_warmcleanup_index_scale_factor=0;
+VACUUM WARMCLEAN test_vacuum_warm;
+-- We expect zero heap-fetches now
+EXPLAIN (analyze, costs off, timing off, summary off) SELECT b FROM test_vacuum_warm WHERE b = 'u';
+                                        QUERY PLAN                                         
+-------------------------------------------------------------------------------------------
+ Index Only Scan using test_vacuum_warm_index1 on test_vacuum_warm (actual rows=1 loops=1)
+   Index Cond: (b = 'u'::text)
+   Heap Fetches: 0
+(3 rows)
+
+DROP TABLE test_vacuum_warm;
diff --git a/src/test/regress/sql/warm.sql b/src/test/regress/sql/warm.sql
index ab61cfb..0d751e2 100644
--- a/src/test/regress/sql/warm.sql
+++ b/src/test/regress/sql/warm.sql
@@ -284,3 +284,50 @@ SELECT a, b FROM test_toast_warm WHERE b = 104.201;
 SELECT a, b FROM test_toast_warm WHERE b = 104.20;
 
 DROP TABLE test_toast_warm;
+
+
+-- Test VACUUM
+
+CREATE TABLE test_vacuum_warm (a int unique, b text, c int, d int);
+CREATE INDEX test_vacuum_warm_index1 ON test_vacuum_warm(b);
+CREATE INDEX test_vacuum_warm_index2 ON test_vacuum_warm(c);
+
+INSERT INTO test_vacuum_warm VALUES (1, 'a', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (2, 'b', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (3, 'c', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (4, 'd', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (5, 'e', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (6, 'f', 100, 200);
+INSERT INTO test_vacuum_warm VALUES (7, 'g', 100, 200);
+
+UPDATE test_vacuum_warm SET b = 'u', c = 300 WHERE a = 1;
+UPDATE test_vacuum_warm SET b = 'v', c = 300 WHERE a = 2;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 3;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 4;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 5;
+UPDATE test_vacuum_warm SET c = 300 WHERE a = 6;
+
+-- a plain vacuum cannot clear WARM chains.
+SET enable_seqscan = false;
+SET enable_bitmapscan = false;
+SET seq_page_cost = 10000;
+VACUUM test_vacuum_warm;
+-- We expect non-zero heap-fetches here
+EXPLAIN (analyze, costs off, timing off, summary off) SELECT b FROM test_vacuum_warm WHERE b = 'u';
+
+-- Now set vacuum_warmcleanup_index_scale_factor such that only
+-- test_vacuum_warm_index2 can be cleaned up.
+SET vacuum_warmcleanup_index_scale_factor=0.5;
+VACUUM WARMCLEAN test_vacuum_warm;
+-- We expect non-zero heap-fetches here
+EXPLAIN (analyze, costs off, timing off, summary off) SELECT b FROM test_vacuum_warm WHERE b = 'u';
+
+
+-- All WARM chains cleaned up, so index-only scan should be used now without
+-- any heap fetches
+SET vacuum_warmcleanup_index_scale_factor=0;
+VACUUM WARMCLEAN test_vacuum_warm;
+-- We expect zero heap-fetches now
+EXPLAIN (analyze, costs off, timing off, summary off) SELECT b FROM test_vacuum_warm WHERE b = 'u';
+
+DROP TABLE test_vacuum_warm;