From 51843cbf0ca6459c4dc9f1f0936f88a038fadf20 Mon Sep 17 00:00:00 2001
From: Pavan Deolasee <pavan.deolasee@gmail.com>
Date: Wed, 29 Mar 2017 11:16:29 +0530
Subject: [PATCH 4/4] Provide control knobs to decide when to do heap and index
 WARM cleanup.

We provide two knobs to control maintenance activity on WARM. A guc
autovacuum_warm_cleanup_scale_factor can be set to trigger WARM cleanup.
Similarly, a GUC autovacuum_warm_cleanup_index_scale_factor can be set to
determine when to do index cleanup. Note that in the current design VACUUM
needs two index scans to remove a WARM index pointer. Hence we want to do that
work only when it makes sense (i.e. the index has significant number of WARM
pointers)

Similarly, VACUUM command is enhanced to accept another parameter, WARMCLEAN,
and if specified then only WARM cleanup will be carried out.
---
 src/backend/access/common/reloptions.c |  22 +++
 src/backend/catalog/system_views.sql   |   1 +
 src/backend/commands/analyze.c         |  60 +++++--
 src/backend/commands/vacuum.c          |   2 +
 src/backend/commands/vacuumlazy.c      | 320 +++++++++++++++++++++++++--------
 src/backend/parser/gram.y              |  26 ++-
 src/backend/postmaster/autovacuum.c    |  58 +++++-
 src/backend/postmaster/pgstat.c        |  50 +++++-
 src/backend/utils/adt/pgstatfuncs.c    |  15 ++
 src/backend/utils/init/globals.c       |   3 +
 src/backend/utils/misc/guc.c           |  30 ++++
 src/include/catalog/pg_proc.h          |   2 +
 src/include/commands/vacuum.h          |   2 +
 src/include/foreign/fdwapi.h           |   3 +-
 src/include/miscadmin.h                |   1 +
 src/include/nodes/parsenodes.h         |   3 +-
 src/include/parser/kwlist.h            |   1 +
 src/include/pgstat.h                   |  11 +-
 src/include/postmaster/autovacuum.h    |   2 +
 src/include/utils/guc_tables.h         |   1 +
 src/include/utils/rel.h                |   2 +
 src/test/regress/expected/rules.out    |   3 +
 src/test/regress/expected/warm.out     |  59 ++++++
 src/test/regress/sql/warm.sql          |  47 +++++
 24 files changed, 614 insertions(+), 110 deletions(-)

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index ce7d4da..c8c7bba 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -361,6 +361,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,
@@ -1362,6 +1380,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 4ef964f..363fdf0 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 c2d5705..6cf942a 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -156,18 +156,23 @@ typedef struct LVRelStats
 	double		tuples_deleted;
 	BlockNumber nonempty_pages; /* actually, last nonempty page + 1 */
 
+	int			maxtuples;		/* maxtuples computed while allocating space */
+	Size		work_area_size;	/* working area size */
+	char		*work_area;		/* working area for storing dead tuples and
+								 * warm chains */
 	/* List of candidate WARM chains that can be converted into HOT chains */
-	/* NB: this list is ordered by TID of the root pointers */
+	/* 
+	 * NB: this list grows from bottom to top and is ordered by TID of the root
+	 * pointers, with the lowest entry at the bottom
+	 */
 	int				num_warm_chains;	/* current # of entries */
-	int				max_warm_chains;	/* # slots allocated in array */
 	LVWarmChain 	*warm_chains;		/* array of LVWarmChain */
 	double			num_non_convertible_warm_chains;
-
 	/* List of TIDs of tuples we intend to delete */
 	/* NB: this list is ordered by TID address */
 	int			num_dead_tuples;	/* current # of entries */
-	int			max_dead_tuples;	/* # slots allocated in array */
 	ItemPointer dead_tuples;	/* array of ItemPointerData */
+
 	int			num_index_scans;
 	TransactionId latestRemovedXid;
 	bool		lock_waiter_detected;
@@ -187,11 +192,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 +213,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 +290,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 +319,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 +407,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 +519,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 +557,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,13 +580,13 @@ 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 */
 	initprog_val[0] = PROGRESS_VACUUM_PHASE_SCAN_HEAP;
 	initprog_val[1] = nblocks;
-	initprog_val[2] = vacrelstats->max_dead_tuples;
+	initprog_val[2] = vacrelstats->maxtuples;
 	pgstat_progress_update_multi_param(3, initprog_index, initprog_val);
 
 	/*
@@ -656,6 +678,11 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		bool		all_frozen = true;	/* provided all_visible is also true */
 		bool		has_dead_tuples;
 		TransactionId visibility_cutoff_xid = InvalidTransactionId;
+		char		*end_deads;
+		char		*end_warms;
+		Size		free_work_area;
+		int			avail_dead_tuples;
+		int			avail_warm_chains;
 
 		/* see note above about forcing scanning of last page */
 #define FORCE_CHECK_PAGE() \
@@ -740,13 +767,39 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		vacuum_delay_point();
 
 		/*
+		 * The dead tuples are stored starting from the start of the work
+		 * area and growing downwards. The candidate warm chains are stored
+		 * starting from the bottom on the work area and growing upwards. Once
+		 * the difference between these two segments is too small to accomodate
+		 * potentially all tuples in the current page, we stop and do one round
+		 * of index cleanup.
+		 */
+		end_deads = (char *)(vacrelstats->dead_tuples + vacrelstats->num_dead_tuples);
+
+		/*
+		 * If we are not doing WARM cleanup, then the entire work area is used
+		 * by the dead tuples.
+		 */
+		if (vacrelstats->warm_chains)
+		{
+			end_warms = (char *)(vacrelstats->warm_chains - vacrelstats->num_warm_chains);
+			free_work_area = end_warms - end_deads;
+			avail_warm_chains = (free_work_area / sizeof (LVWarmChain));
+		}
+		else
+		{
+			free_work_area = vacrelstats->work_area +
+				vacrelstats->work_area_size - end_deads;
+			avail_warm_chains = 0;
+		}
+		avail_dead_tuples = (free_work_area / sizeof (ItemPointerData));
+
+		/*
 		 * If we are close to overrunning the available space for dead-tuple
 		 * TIDs, pause and do a cycle of vacuuming before we tackle this page.
 		 */
-		if (((vacrelstats->max_dead_tuples - vacrelstats->num_dead_tuples) < MaxHeapTuplesPerPage &&
-			vacrelstats->num_dead_tuples > 0) ||
-			((vacrelstats->max_warm_chains - vacrelstats->num_warm_chains) < MaxHeapTuplesPerPage &&
-			 vacrelstats->num_warm_chains > 0))
+		if ((avail_dead_tuples < MaxHeapTuplesPerPage && vacrelstats->num_dead_tuples > 0) ||
+			(avail_warm_chains < MaxHeapTuplesPerPage && vacrelstats->num_warm_chains > 0))
 		{
 			const int	hvp_index[] = {
 				PROGRESS_VACUUM_PHASE,
@@ -776,7 +829,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);
 
@@ -800,8 +854,7 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 			 */
 			vacrelstats->num_dead_tuples = 0;
 			vacrelstats->num_warm_chains = 0;
-			memset(vacrelstats->warm_chains, 0,
-					vacrelstats->max_warm_chains * sizeof (LVWarmChain));
+			memset(vacrelstats->work_area, 0, vacrelstats->work_area_size);
 			vacrelstats->num_index_scans++;
 
 			/* Report that we are once again scanning the heap */
@@ -1413,7 +1466,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);
 
@@ -1518,9 +1572,12 @@ lazy_vacuum_heap(Relation onerel, LVRelStats *vacrelstats)
 		vacuum_delay_point();
 
 		tblk = chainblk = InvalidBlockNumber;
-		if (chainindex < vacrelstats->num_warm_chains)
-			chainblk =
-				ItemPointerGetBlockNumber(&(vacrelstats->warm_chains[chainindex].chain_tid));
+		if (vacrelstats->warm_chains &&
+			chainindex < vacrelstats->num_warm_chains)
+		{
+			LVWarmChain *chain = vacrelstats->warm_chains - (chainindex + 1);
+			chainblk = ItemPointerGetBlockNumber(&chain->chain_tid);
+		}
 
 		if (tupindex < vacrelstats->num_dead_tuples)
 			tblk = ItemPointerGetBlockNumber(&vacrelstats->dead_tuples[tupindex]);
@@ -1618,7 +1675,8 @@ lazy_warmclear_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 		BlockNumber tblk;
 		LVWarmChain	*chain;
 
-		chain = &vacrelstats->warm_chains[chainindex];
+		/* The warm chains are indexed from bottom */
+		chain = vacrelstats->warm_chains - (chainindex + 1);
 
 		tblk = ItemPointerGetBlockNumber(&chain->chain_tid);
 		if (tblk != blkno)
@@ -1852,9 +1910,11 @@ static void
 lazy_reset_warm_pointer_count(LVRelStats *vacrelstats)
 {
 	int i;
-	for (i = 0; i < vacrelstats->num_warm_chains; i++)
+
+	/* Start from the bottom and move upwards */
+	for (i = 1; i <= vacrelstats->num_warm_chains; i++)
 	{
-		LVWarmChain *chain = &vacrelstats->warm_chains[i];
+		LVWarmChain *chain = (vacrelstats->warm_chains - i);
 		chain->num_clear_pointers = chain->num_warm_pointers = 0;
 	}
 }
@@ -1868,6 +1928,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)
 {
@@ -1932,25 +1993,57 @@ 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.
+			 *
+			 * Start from bottom and move upwards.
+			 */
+			for (ii = 1; 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
 	{
@@ -2328,7 +2421,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() &&
@@ -2337,11 +2431,16 @@ 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)));
+				dowarmcleanup ? sizeof(LVWarmChain) : 0));
 
 		/* curious coding here to ensure the multiplication can't overflow */
 		if ((BlockNumber) (maxtuples / LAZY_ALLOC_TUPLES) > relblocks)
@@ -2355,21 +2454,29 @@ lazy_space_alloc(LVRelStats *vacrelstats, BlockNumber relblocks)
 		maxtuples = MaxHeapTuplesPerPage;
 	}
 
-	vacrelstats->num_dead_tuples = 0;
-	vacrelstats->max_dead_tuples = (int) maxtuples;
-	vacrelstats->dead_tuples = (ItemPointer)
-		palloc(maxtuples * sizeof(ItemPointerData));
-
-	/*
-	 * 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.
+	/* Allocate work area of the desired size and setup dead_tuples and
+	 * warm_chains to the start and the end of the area respectively. They grow
+	 * in opposite directions as dead tuples and warm chains are added. Note
+	 * that if we are not doing WARM cleanup then the entire area will only be
+	 * used for tracking dead tuples.
 	 */
-	vacrelstats->num_warm_chains = 0;
-	vacrelstats->max_warm_chains = (int) maxtuples;
-	vacrelstats->warm_chains = (LVWarmChain *)
-		palloc0(maxtuples * sizeof(LVWarmChain));
+	vacrelstats->work_area_size = maxtuples * (sizeof(ItemPointerData) +
+				dowarmcleanup ? sizeof(LVWarmChain) : 0);
+	vacrelstats->work_area = (char *) palloc0(vacrelstats->work_area_size);
+	vacrelstats->num_dead_tuples = 0;
+	vacrelstats->dead_tuples = (ItemPointer)vacrelstats->work_area;
+	vacrelstats->maxtuples = maxtuples;
 
+	if (dowarmcleanup)
+	{
+		vacrelstats->num_warm_chains = 0;
+		vacrelstats->warm_chains = (LVWarmChain *)
+			(vacrelstats->work_area + vacrelstats->work_area_size);
+	}
+	else
+	{
+		vacrelstats->warm_chains = NULL;
+	}
 }
 
 /*
@@ -2379,17 +2486,38 @@ static void
 lazy_record_clear_chain(LVRelStats *vacrelstats,
 					   ItemPointer itemptr)
 {
+	char *end_deads, *end_warms;
+	Size free_work_area;
+
+	if (vacrelstats->warm_chains == NULL)
+	{
+		vacrelstats->num_non_convertible_warm_chains++;
+		return;
+	}
+
+	end_deads = (char *) (vacrelstats->dead_tuples +
+					vacrelstats->num_dead_tuples);
+	end_warms = (char *) (vacrelstats->warm_chains -
+					vacrelstats->num_warm_chains);
+	free_work_area = (end_warms - end_deads);
+
+	Assert(free_work_area >= 0);
 	/*
 	 * The array shouldn't overflow under normal behavior, but perhaps it
 	 * could if we are given a really small maintenance_work_mem. In that
 	 * case, just forget the last few tuples (we'll get 'em next time).
 	 */
-	if (vacrelstats->num_warm_chains < vacrelstats->max_warm_chains)
+	if (free_work_area >= sizeof (LVWarmChain))
 	{
-		vacrelstats->warm_chains[vacrelstats->num_warm_chains].chain_tid = *itemptr;
-		vacrelstats->warm_chains[vacrelstats->num_warm_chains].is_postwarm_chain = 0;
+		LVWarmChain *chain;
+
 		vacrelstats->num_warm_chains++;
+	   	chain = vacrelstats->warm_chains - vacrelstats->num_warm_chains;
+		chain->chain_tid = *itemptr;
+		chain->is_postwarm_chain = 0;
 	}
+	else
+		vacrelstats->num_non_convertible_warm_chains++;
 }
 
 /*
@@ -2399,17 +2527,39 @@ static void
 lazy_record_warm_chain(LVRelStats *vacrelstats,
 					   ItemPointer itemptr)
 {
+	char *end_deads, *end_warms;
+	Size free_work_area;
+
+	if (vacrelstats->warm_chains == NULL)
+	{
+		vacrelstats->num_non_convertible_warm_chains++;
+		return;
+	}
+
+	end_deads = (char *) (vacrelstats->dead_tuples +
+					vacrelstats->num_dead_tuples);
+	end_warms = (char *) (vacrelstats->warm_chains -
+					vacrelstats->num_warm_chains);
+	free_work_area = (end_warms - end_deads);
+
+	Assert(free_work_area >= 0);
+
 	/*
 	 * The array shouldn't overflow under normal behavior, but perhaps it
 	 * could if we are given a really small maintenance_work_mem. In that
 	 * case, just forget the last few tuples (we'll get 'em next time).
 	 */
-	if (vacrelstats->num_warm_chains < vacrelstats->max_warm_chains)
+	if (free_work_area >= sizeof (LVWarmChain))
 	{
-		vacrelstats->warm_chains[vacrelstats->num_warm_chains].chain_tid = *itemptr;
-		vacrelstats->warm_chains[vacrelstats->num_warm_chains].is_postwarm_chain = 1;
+		LVWarmChain *chain;
+
 		vacrelstats->num_warm_chains++;
+	   	chain = vacrelstats->warm_chains - vacrelstats->num_warm_chains;
+		chain->chain_tid = *itemptr;
+		chain->is_postwarm_chain = 1;
 	}
+	else
+		vacrelstats->num_non_convertible_warm_chains++;
 }
 
 /*
@@ -2419,12 +2569,20 @@ static void
 lazy_record_dead_tuple(LVRelStats *vacrelstats,
 					   ItemPointer itemptr)
 {
+	char *end_deads = (char *) (vacrelstats->dead_tuples +
+		 	vacrelstats->num_dead_tuples);
+	char *end_warms = (char *) (vacrelstats->warm_chains -
+			vacrelstats->num_warm_chains);
+	Size freespace = (end_warms - end_deads);
+
+	Assert(freespace >= 0);
+	
 	/*
 	 * The array shouldn't overflow under normal behavior, but perhaps it
 	 * could if we are given a really small maintenance_work_mem. In that
 	 * case, just forget the last few tuples (we'll get 'em next time).
 	 */
-	if (vacrelstats->num_dead_tuples < vacrelstats->max_dead_tuples)
+	if (freespace >= sizeof (ItemPointer))
 	{
 		vacrelstats->dead_tuples[vacrelstats->num_dead_tuples] = *itemptr;
 		vacrelstats->num_dead_tuples++;
@@ -2477,10 +2635,10 @@ lazy_indexvac_phase1(ItemPointer itemptr, bool is_warm, void *state)
 		return IBDCR_DELETE;
 
 	chain = (LVWarmChain *) bsearch((void *) itemptr,
-								(void *) vacrelstats->warm_chains,
-								vacrelstats->num_warm_chains,
-								sizeof(LVWarmChain),
-								vac_cmp_warm_chain);
+				(void *) (vacrelstats->warm_chains - vacrelstats->num_warm_chains),
+				vacrelstats->num_warm_chains,
+				sizeof(LVWarmChain),
+				vac_cmp_warm_chain);
 	if (chain != NULL)
 	{
 		if (is_warm)
@@ -2500,13 +2658,13 @@ static IndexBulkDeleteCallbackResult
 lazy_indexvac_phase2(ItemPointer itemptr, bool is_warm, void *state)
 {
 	LVRelStats		*vacrelstats = (LVRelStats *) state;
-	LVWarmChain	*chain;
+	LVWarmChain		*chain;
 
 	chain = (LVWarmChain *) bsearch((void *) itemptr,
-								(void *) vacrelstats->warm_chains,
-								vacrelstats->num_warm_chains,
-								sizeof(LVWarmChain),
-								vac_cmp_warm_chain);
+				(void *) (vacrelstats->warm_chains - vacrelstats->num_warm_chains),
+				vacrelstats->num_warm_chains,
+				sizeof(LVWarmChain),
+				vac_cmp_warm_chain);
 
 	if (chain != NULL && (chain->keep_warm_chain != 1))
 	{
@@ -2619,6 +2777,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;
@@ -2627,6 +2786,9 @@ lazy_indexvac_phase2(ItemPointer itemptr, bool is_warm, void *state)
 /*
  * Comparator routines for use with qsort() and bsearch(). Similar to
  * vac_cmp_itemptr, but right hand argument is LVWarmChain struct pointer.
+ *
+ * The warm_chains array is sorted in descending order hence the return values
+ * are flipped.
  */
 static int
 vac_cmp_warm_chain(const void *left, const void *right)
@@ -2640,17 +2802,17 @@ vac_cmp_warm_chain(const void *left, const void *right)
 	rblk = ItemPointerGetBlockNumber(&((LVWarmChain *) right)->chain_tid);
 
 	if (lblk < rblk)
-		return -1;
-	if (lblk > rblk)
 		return 1;
+	if (lblk > rblk)
+		return -1;
 
 	loff = ItemPointerGetOffsetNumber((ItemPointer) left);
 	roff = ItemPointerGetOffsetNumber(&((LVWarmChain *) right)->chain_tid);
 
 	if (loff < roff)
-		return -1;
-	if (loff > roff)
 		return 1;
+	if (loff > roff)
+		return -1;
 
 	return 0;
 }
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 9d53a29..1592220 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
@@ -10059,7 +10059,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;
@@ -10069,11 +10069,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;
@@ -10083,13 +10085,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;
@@ -10097,6 +10101,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 ')'
@@ -10129,6 +10135,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)
@@ -10182,6 +10189,10 @@ opt_freeze: FREEZE									{ $$ = TRUE; }
 			| /*EMPTY*/								{ $$ = FALSE; }
 		;
 
+opt_warmclean: WARMCLEAN							{ $$ = TRUE; }
+			| /*EMPTY*/								{ $$ = FALSE; }
+		;
+
 opt_name_list:
 			'(' name_list ')'						{ $$ = $2; }
 			| /*EMPTY*/								{ $$ = NIL; }
@@ -14886,6 +14897,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 89dd3b3..a157c05 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -117,6 +117,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;
 
@@ -338,7 +340,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);
@@ -2076,6 +2079,7 @@ do_autovacuum(void)
 		bool		dovacuum;
 		bool		doanalyze;
 		bool		wraparound;
+		bool		dowarmcleanup;
 
 		if (classForm->relkind != RELKIND_RELATION &&
 			classForm->relkind != RELKIND_MATVIEW)
@@ -2115,10 +2119,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)
@@ -2171,6 +2179,7 @@ do_autovacuum(void)
 		bool		dovacuum;
 		bool		doanalyze;
 		bool		wraparound;
+		bool		dowarmcleanup;
 
 		/*
 		 * We cannot safely process other backends' temp tables, so skip 'em.
@@ -2201,7 +2210,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)
@@ -2792,6 +2802,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;
@@ -2833,7 +2844,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)
@@ -2849,6 +2861,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
@@ -2895,19 +2908,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;
@@ -2974,7 +2994,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;
@@ -2986,6 +3007,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;
@@ -2994,6 +3018,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;
@@ -3026,6 +3053,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;
@@ -3073,6 +3105,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;
@@ -3089,6 +3122,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 52fe4ba..f38ce8a 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -226,9 +226,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? */
@@ -1367,7 +1369,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;
 
@@ -1381,6 +1384,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));
 }
 
@@ -1396,7 +1400,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;
 
@@ -1421,12 +1425,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);
@@ -1437,6 +1443,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));
 }
 
@@ -1907,7 +1914,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++;
+		}
 	}
 }
 
@@ -2070,6 +2080,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 +
@@ -2080,6 +2096,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;
@@ -2136,12 +2158,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);
@@ -2177,9 +2203,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);
 			}
@@ -2221,9 +2251,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;
@@ -2298,11 +2330,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;
@@ -2330,12 +2365,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;
 }
 
 
@@ -4526,6 +4565,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;
@@ -5636,6 +5676,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;
@@ -5667,9 +5708,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;
@@ -5679,6 +5722,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.
@@ -5904,6 +5948,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)
 	{
@@ -5938,6 +5983,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 227a87d..8804908 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -193,6 +193,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 8b5f064..ecf8028 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3017,6 +3017,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 3f1a142..61a4e23 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -2795,6 +2795,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 = 3403 (  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 b2afd50..f5fc001 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3039,7 +3039,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 cd21a78..7d9818b 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -433,6 +433,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 99bdc8b..883cbd4 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;
@@ -1156,10 +1162,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 d383fd3..19fb0a2 100644
--- a/src/include/postmaster/autovacuum.h
+++ b/src/include/postmaster/autovacuum.h
@@ -39,6 +39,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 2b86054..f0dd350 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -278,6 +278,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 f7dc4a4..d34aa68 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1759,6 +1759,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,
@@ -1907,6 +1908,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,
@@ -1951,6 +1953,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 1f07272..34cdbe5 100644
--- a/src/test/regress/expected/warm.out
+++ b/src/test/regress/expected/warm.out
@@ -745,6 +745,65 @@ 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, e int);
+CREATE INDEX test_vacuum_warm_index1 ON test_vacuum_warm(b);
+CREATE INDEX test_vacuum_warm_index2 ON test_vacuum_warm(c);
+CREATE INDEX test_vacuum_warm_index3 ON test_vacuum_warm(d);
+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;
 -- Toasted heap attributes
 CREATE TABLE toasttest(descr text , cnt int DEFAULT 0, f1 text, f2 text);
 CREATE INDEX testindx1 ON toasttest(descr);
diff --git a/src/test/regress/sql/warm.sql b/src/test/regress/sql/warm.sql
index fc80c0f..ae9db9a 100644
--- a/src/test/regress/sql/warm.sql
+++ b/src/test/regress/sql/warm.sql
@@ -285,6 +285,53 @@ 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, e int);
+CREATE INDEX test_vacuum_warm_index1 ON test_vacuum_warm(b);
+CREATE INDEX test_vacuum_warm_index2 ON test_vacuum_warm(c);
+CREATE INDEX test_vacuum_warm_index3 ON test_vacuum_warm(d);
+
+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;
+
 -- Toasted heap attributes
 CREATE TABLE toasttest(descr text , cnt int DEFAULT 0, f1 text, f2 text);
 CREATE INDEX testindx1 ON toasttest(descr);
-- 
2.9.3 (Apple Git-75)

