From 6be4e1a5f910ec53b240b082d35baf6905b39219 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Thu, 27 Feb 2025 13:41:35 -0800
Subject: [PATCH v12 5/5] Support parallelism for collecting dead items during
 lazy vacuum.

This feature allows the vacuum to leverage multiple CPUs in order to
collect dead items (i.e. the first pass over heap table) with parallel
workers. The parallel degree for parallel heap vacuuming is determined
based on the number of blocks to vacuum unless PARALLEL option of
VACUUM command is specified, and further limited by
max_parallel_maintenance_workers.

Author:
Reviewed-by:
Discussion: https://postgr.es/m/
Backpatch-through:
---
 doc/src/sgml/ref/vacuum.sgml             |   54 +-
 src/backend/access/heap/heapam_handler.c |    4 +
 src/backend/access/heap/vacuumlazy.c     | 1098 +++++++++++++++++++---
 src/backend/access/table/tableam.c       |   15 +
 src/backend/commands/vacuumparallel.c    |   20 +
 src/include/access/heapam.h              |   11 +
 src/include/access/tableam.h             |    4 +
 src/include/commands/vacuum.h            |    2 +
 src/test/regress/expected/vacuum.out     |    6 +
 src/test/regress/sql/vacuum.sql          |    7 +
 src/tools/pgindent/typedefs.list         |    4 +
 11 files changed, 1070 insertions(+), 155 deletions(-)

diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml
index 971b1237d47..9d73f6074de 100644
--- a/doc/src/sgml/ref/vacuum.sgml
+++ b/doc/src/sgml/ref/vacuum.sgml
@@ -279,25 +279,41 @@ VACUUM [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] [ <re
     <term><literal>PARALLEL</literal></term>
     <listitem>
      <para>
-      Perform index vacuum and index cleanup phases of <command>VACUUM</command>
-      in parallel using <replaceable class="parameter">integer</replaceable>
-      background workers (for the details of each vacuum phase, please
-      refer to <xref linkend="vacuum-phases"/>).  The number of workers used
-      to perform the operation is equal to the number of indexes on the
-      relation that support parallel vacuum which is limited by the number of
-      workers specified with <literal>PARALLEL</literal> option if any which is
-      further limited by <xref linkend="guc-max-parallel-maintenance-workers"/>.
-      An index can participate in parallel vacuum if and only if the size of the
-      index is more than <xref linkend="guc-min-parallel-index-scan-size"/>.
-      Please note that it is not guaranteed that the number of parallel workers
-      specified in <replaceable class="parameter">integer</replaceable> will be
-      used during execution.  It is possible for a vacuum to run with fewer
-      workers than specified, or even with no workers at all.  Only one worker
-      can be used per index.  So parallel workers are launched only when there
-      are at least <literal>2</literal> indexes in the table.  Workers for
-      vacuum are launched before the start of each phase and exit at the end of
-      the phase.  These behaviors might change in a future release.  This
-      option can't be used with the <literal>FULL</literal> option.
+      Perform scanning heap, index vacuum, and index cleanup phases of
+      <command>VACUUM</command> in parallel using
+      <replaceable class="parameter">integer</replaceable> background workers
+      (for the details of each vacuum phase, please refer to
+      <xref linkend="vacuum-phases"/>).
+     </para>
+     <para>
+      For heap tables, the number of workers used to perform the scanning
+      heap is determined based on the size of table. A table can participate in
+      parallel scanning heap if and only if the size of the table is more than
+      <xref linkend="guc-min-parallel-table-scan-size"/>. During scanning heap,
+      the heap table's blocks will be divided into ranges and shared among the
+      cooperating processes. Each worker process will complete the scanning of
+      its given range of blocks before requesting an additional range of blocks.
+     </para>
+     <para>
+      The number of workers used to perform parallel index vacuum and index
+      cleanup is equal to the number of indexes on the relation that support
+      parallel vacuum. An index can participate in parallel vacuum if and only
+      if the size of the index is more than <xref linkend="guc-min-parallel-index-scan-size"/>.
+      Only one worker can be used per index. So parallel workers for index vacuum
+      and index cleanup are launched only when there are at least <literal>2</literal>
+      indexes in the table.
+     </para>
+     <para>
+      Workers for vacuum are launched before the start of each phase and exit
+      at the end of the phase. The number of workers for each phase is limited by
+      the number of workers specified with <literal>PARALLEL</literal> option if
+      any which is futher limited by <xref linkend="guc-max-parallel-maintenance-workers"/>.
+      Please note that in any parallel vacuum phase, it is not guaanteed that the
+      number of parallel workers specified in <replaceable class="parameter">integer</replaceable>
+      will be used during execution. It is possible for a vacuum to run with fewer
+      workers than specified, or even with no workers at all. These behaviors might
+      change in a future release. This option can't be used with the <literal>FULL</literal>
+      option.
      </para>
     </listitem>
    </varlistentry>
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 1ab20edef94..0a3996f1ef1 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -2699,6 +2699,10 @@ static const TableAmRoutine heapam_methods = {
 	.scan_sample_next_tuple = heapam_scan_sample_next_tuple,
 
 	.parallel_vacuum_compute_workers = heap_parallel_vacuum_compute_workers,
+	.parallel_vacuum_estimate = heap_parallel_vacuum_estimate,
+	.parallel_vacuum_initialize = heap_parallel_vacuum_initialize,
+	.parallel_vacuum_initialize_worker = heap_parallel_vacuum_initialize_worker,
+	.parallel_vacuum_collect_dead_items = heap_parallel_vacuum_collect_dead_items,
 };
 
 
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index c54cffdc399..a595dd22134 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -99,6 +99,36 @@
  * After pruning and freezing, pages that are newly all-visible and all-frozen
  * are marked as such in the visibility map.
  *
+ * Parallel Lazy Heap Scanning:
+ *
+ * Lazy vacuum on heap tables supports parallel processing for phase I and
+ * phase II. Before starting phase I, we initialize parallel vacuum state,
+ * ParallelVacuumState, and allocate the TID store in a DSA area if we can
+ * use parallel mode for any of these two phases.
+ *
+ * We could require different number of parallel vacuum workers for each phase
+ * for various factors such as table size and number of indexes. Parallel
+ * workers are launched at the beginning of each phase and exit at the end of
+ * each phase.
+ *
+ * For scanning the heap table with parallel workers, we utilize the
+ * table_block_parallelscan_xxx facility which splits the table into several
+ * chunks and parallel workers allocate chunks to scan. If the TID store is
+ * close to overrunning the available space during phase I, parallel workers
+ * exit and leader process gathers the scan results. Then, it performs a index
+ * vacuuming that could also use the parallelism. After vacuuming both indexes
+ * and heap table, the leader process vacuums FSM to make newly-freed space
+ * visible. Then, it relaunches parallel workers to resume the scanning heap
+ * phase with parallel workers again. In order to be able to resume the parallel
+ * heap scan from the previous status, the workers' parallel scan descriptions
+ * are stored in the shared memory (DSM) space to share among parallel workers.
+ * If the leader launches fewer workers than the previous time to resume the
+ * parallel heap scan, some blocks are remained as un-scanned. The leader
+ * process serially deals with such blocks at the end of scanning heap phase
+ * (see complete_unfinihsed_lazy_scan_heap()). Whereas non-parallel heap scans
+ * use streaming read, parallel heap scans don't use it as the streaming read
+ * state cannot be stored easily in the shared memory space.
+ *
  * Dead TID Storage:
  *
  * The major space usage for vacuuming is storage for the dead tuple IDs that
@@ -147,6 +177,7 @@
 #include "common/pg_prng.h"
 #include "executor/instrument.h"
 #include "miscadmin.h"
+#include "optimizer/paths.h"
 #include "pgstat.h"
 #include "portability/instr_time.h"
 #include "postmaster/autovacuum.h"
@@ -214,11 +245,21 @@
  */
 #define PREFETCH_SIZE			((BlockNumber) 32)
 
+/*
+ * DSM keys for parallel heap vacuum. Unlike other parallel execution code, we
+ * we don't need to worry about DSM keys conflicting with plan_node_id, but need to
+ * avoid conflicting with DSM keys used in vacuumparallel.c.
+ */
+#define LV_PARALLEL_KEY_SHARED				0xFFFF0001
+#define LV_PARALLEL_KEY_SCANDESC			0xFFFF0002
+#define LV_PARALLEL_KEY_WORKER_SCANSTATE	0xFFFF0003
+
 /*
  * Macro to check if we are in a parallel vacuum.  If true, we are in the
  * parallel mode and the DSM segment is initialized.
  */
 #define ParallelVacuumIsActive(vacrel) ((vacrel)->pvs != NULL)
+#define ParallelHeapVacuumIsActive(vacrel) ((vacrel)->plvstate != NULL)
 
 /* Phases of vacuum during which we report error context. */
 typedef enum
@@ -306,6 +347,87 @@ typedef struct LVScanData
 	bool		skippedallvis;
 } LVScanData;
 
+/*
+ * Struct for information that needs to be shared among parallel workers
+ * for parallel heap vacuum.
+ */
+typedef struct PLVShared
+{
+	bool		aggressive;
+	bool		skipwithvm;
+
+	/* The current oldest extant XID/MXID shared by the leader process */
+	TransactionId NewRelfrozenXid;
+	MultiXactId NewRelminMxid;
+
+	/* VACUUM operation's cutoffs for freezing and pruning */
+	struct VacuumCutoffs cutoffs;
+	GlobalVisState vistest;
+
+	/* Per-worker scan data for parallel lazy heap scan */
+	LVScanData	worker_scandata[FLEXIBLE_ARRAY_MEMBER];
+} PLVShared;
+#define SizeOfPLVShared	(offsetof(PLVShared, worker_scandata))
+
+/* Per-worker scan state for parallel heap vacuum */
+typedef struct PLVScanWorkerState
+{
+	/* Has this worker data been initialized? */
+	bool		inited;
+
+	/* per-worker parallel table scan state */
+	ParallelBlockTableScanWorkerData pbscanwork;
+
+	/*
+	 * True if a parallel vacuum scan worker allocated blocks in state but
+	 * might have not scanned all of them. The leader process will take over
+	 * for scanning these remaining blocks.
+	 */
+	bool		maybe_have_unprocessed_blocks;
+
+	/* Last block number the worker scanned */
+	BlockNumber last_blkno;
+} PLVScanWorkerState;
+
+/*
+ * Struct to store the parallel lazy scan state.
+ */
+typedef struct PLVState
+{
+	/* Parallel scan description shared among parallel workers */
+	ParallelBlockTableScanDesc pbscan;
+
+	/* Shared information */
+	PLVShared  *shared;
+
+	/* Scan state for parallel heap vacuum */
+	PLVScanWorkerState *scanstate;
+} PLVState;
+
+/*
+ * Struct for leader in parallel heap vacuum.
+ */
+typedef struct PLVLeader
+{
+	/* Shared memory size for each shared object */
+	Size		pbscan_len;
+	Size		shared_len;
+	Size		scanstates_len;
+
+	int			nworkers_launched;
+
+	/*
+	 * Points to all per-worker scan states stored on DSM area.
+	 *
+	 * During parallel heap scan, each worker allocates some chunks of blocks
+	 * to scan in its scan state, and could exit while leaving some chunks
+	 * un-scanned if the size of dead_items TIDs is close to overrunning the
+	 * the available space. We store the scan states on shared memory area so
+	 * that workers can resume heap scans from the previous point.
+	 */
+	PLVScanWorkerState *scanstates;
+} PLVLeader;
+
 typedef struct LVRelState
 {
 	/* Target heap relation and its indexes */
@@ -313,6 +435,13 @@ typedef struct LVRelState
 	Relation   *indrels;
 	int			nindexes;
 
+	/*
+	 * For scans that stream reads. We use streaming read for non-parallel
+	 * heap scanning. For parallel heap scanning we use parallel block table
+	 * scan. See the comment atop of this file for details.
+	 */
+	ReadStream *stream;
+
 	/* Buffer access strategy and parallel vacuum state */
 	BufferAccessStrategy bstrategy;
 	ParallelVacuumState *pvs;
@@ -368,6 +497,9 @@ typedef struct LVRelState
 	/* Instrumentation counters */
 	int			num_index_scans;
 
+	/* Next block to check for FSM vacuum */
+	BlockNumber next_fsm_block_to_vacuum;
+
 	/* State maintained by heap_vac_scan_next_block() */
 	BlockNumber current_block;	/* last block returned */
 	BlockNumber next_unskippable_block; /* next unskippable block */
@@ -375,6 +507,16 @@ typedef struct LVRelState
 	bool		next_unskippable_eager_scanned; /* if it was eagerly scanned */
 	Buffer		next_unskippable_vmbuffer;	/* buffer containing its VM bit */
 
+	/* Fields used for parallel heap vacuum */
+
+	/* Parallel lazy vacuum working state */
+	PLVState   *plvstate;
+
+	/*
+	 * The leader state for parallel heap vacuum. NULL for parallel workers.
+	 */
+	PLVLeader  *leader;
+
 	/* State related to managing eager scanning of all-visible pages */
 
 	/*
@@ -434,12 +576,17 @@ typedef struct LVSavedErrInfo
 
 /* non-export function prototypes */
 static void lazy_scan_heap(LVRelState *vacrel);
+static bool do_lazy_scan_heap(LVRelState *vacrel);
 static void heap_vacuum_eager_scan_setup(LVRelState *vacrel,
 										 VacuumParams *params);
-static BlockNumber heap_vac_scan_next_block(ReadStream *stream,
-											void *callback_private_data,
-											void *per_buffer_data);
-static void find_next_unskippable_block(LVRelState *vacrel, bool *skipsallvis);
+static BlockNumber heap_vac_scan_next_block_read_stream(ReadStream *stream,
+														void *callback_private_data,
+														void *per_buffer_data);
+static inline Buffer heap_vac_get_next_buffer(LVRelState *vacrel, BlockNumber *blkno_out,
+											  uint8 *blk_info_out);
+static BlockNumber get_next_unskippable_block(LVRelState *vacrel, uint8 *blk_info);
+static bool find_next_unskippable_block(LVRelState *vacrel, bool *skipsallvis,
+										BlockNumber start_blk, BlockNumber end_blk);
 static bool lazy_scan_new_or_empty(LVRelState *vacrel, Buffer buf,
 								   BlockNumber blkno, Page page,
 								   bool sharelock, Buffer vmbuffer);
@@ -450,6 +597,12 @@ static void lazy_scan_prune(LVRelState *vacrel, Buffer buf,
 static bool lazy_scan_noprune(LVRelState *vacrel, Buffer buf,
 							  BlockNumber blkno, Page page,
 							  bool *has_lpdead_items);
+static void do_parallel_lazy_scan_heap(LVRelState *vacrel);
+static BlockNumber parallel_lazy_scan_compute_min_scan_block(LVRelState *vacrel);
+static void complete_unfinihsed_lazy_scan_heap(LVRelState *vacrel);
+static void parallel_lazy_scan_heap_begin(LVRelState *vacrel);
+static void parallel_lazy_scan_heap_end(LVRelState *vacrel);
+static void parallel_lazy_scan_gather_scan_results(LVRelState *vacrel);
 static void lazy_vacuum(LVRelState *vacrel);
 static bool lazy_vacuum_all_indexes(LVRelState *vacrel);
 static void lazy_vacuum_heap_rel(LVRelState *vacrel);
@@ -529,6 +682,22 @@ heap_vacuum_eager_scan_setup(LVRelState *vacrel, VacuumParams *params)
 	if (vacrel->aggressive)
 		return;
 
+	/*
+	 * Disable eager scanning if parallel heap vacuum is enabled.
+	 *
+	 * One might think that it would make sense to use the eager scanning even
+	 * during parallel heap scanning, but parallel vacuum is available only in
+	 * VACUUM command and would not be something that happens frequently,
+	 * which seems not fit to the purpose of the eager scanning. Also, it
+	 * would require making the code complex. So it would make sense to
+	 * disable it for now.
+	 *
+	 * XXX: this limitation might be eliminated  in the future for example
+	 * when we use parallel vacuum also in autovacuum.
+	 */
+	if (ParallelHeapVacuumIsActive(vacrel))
+		return;
+
 	/*
 	 * Aggressively vacuuming a small relation shouldn't take long, so it
 	 * isn't worth amortizing. We use two times the region size as the size
@@ -771,6 +940,7 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 
 	/* Initialize remaining counters (be tidy) */
 	vacrel->num_index_scans = 0;
+	vacrel->next_fsm_block_to_vacuum = 0;
 
 	/* dead_items_alloc allocates vacrel->dead_items later on */
 
@@ -815,13 +985,6 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 
 	vacrel->skipwithvm = skipwithvm;
 
-	/*
-	 * Set up eager scan tracking state. This must happen after determining
-	 * whether or not the vacuum must be aggressive, because only normal
-	 * vacuums use the eager scan algorithm.
-	 */
-	heap_vacuum_eager_scan_setup(vacrel, params);
-
 	if (verbose)
 	{
 		if (vacrel->aggressive)
@@ -846,6 +1009,13 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 	lazy_check_wraparound_failsafe(vacrel);
 	dead_items_alloc(vacrel, params->nworkers);
 
+	/*
+	 * Set up eager scan tracking state. This must happen after determining
+	 * whether or not the vacuum must be aggressive, because only normal
+	 * vacuums use the eager scan algorithm.
+	 */
+	heap_vacuum_eager_scan_setup(vacrel, params);
+
 	/*
 	 * Call lazy_scan_heap to perform all required heap pruning, index
 	 * vacuuming, and heap vacuuming (plus related processing)
@@ -1215,13 +1385,7 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 static void
 lazy_scan_heap(LVRelState *vacrel)
 {
-	ReadStream *stream;
-	BlockNumber rel_pages = vacrel->scan_data->rel_pages,
-				blkno = 0,
-				next_fsm_block_to_vacuum = 0;
-	BlockNumber orig_eager_scan_success_limit =
-		vacrel->eager_scan_remaining_successes; /* for logging */
-	Buffer		vmbuffer = InvalidBuffer;
+	BlockNumber rel_pages = vacrel->scan_data->rel_pages;
 	const int	initprog_index[] = {
 		PROGRESS_VACUUM_PHASE,
 		PROGRESS_VACUUM_TOTAL_HEAP_BLKS,
@@ -1242,14 +1406,88 @@ lazy_scan_heap(LVRelState *vacrel)
 	vacrel->next_unskippable_eager_scanned = false;
 	vacrel->next_unskippable_vmbuffer = InvalidBuffer;
 
-	/* Set up the read stream for vacuum's first pass through the heap */
-	stream = read_stream_begin_relation(READ_STREAM_MAINTENANCE,
-										vacrel->bstrategy,
-										vacrel->rel,
-										MAIN_FORKNUM,
-										heap_vac_scan_next_block,
-										vacrel,
-										sizeof(uint8));
+	/* Do the actual work */
+	if (ParallelHeapVacuumIsActive(vacrel))
+		do_parallel_lazy_scan_heap(vacrel);
+	else
+	{
+		/* Set up the read stream for vacuum's first pass through the heap */
+		vacrel->stream = read_stream_begin_relation(READ_STREAM_MAINTENANCE,
+													vacrel->bstrategy,
+													vacrel->rel,
+													MAIN_FORKNUM,
+													heap_vac_scan_next_block_read_stream,
+													vacrel,
+													sizeof(uint8));
+		do_lazy_scan_heap(vacrel);
+
+		read_stream_end(vacrel->stream);
+	}
+
+	/*
+	 * Report that everything is now scanned. We never skip scanning the last
+	 * block in the relation, so we can pass rel_pages here.
+	 */
+	pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED,
+								 rel_pages);
+
+	/* now we can compute the new value for pg_class.reltuples */
+	vacrel->new_live_tuples = vac_estimate_reltuples(vacrel->rel, rel_pages,
+													 vacrel->scan_data->scanned_pages,
+													 vacrel->scan_data->live_tuples);
+
+	/*
+	 * Also compute the total number of surviving heap entries.  In the
+	 * (unlikely) scenario that new_live_tuples is -1, take it as zero.
+	 */
+	vacrel->new_rel_tuples =
+		Max(vacrel->new_live_tuples, 0) + vacrel->scan_data->recently_dead_tuples +
+		vacrel->scan_data->missed_dead_tuples;
+
+	/*
+	 * Do index vacuuming (call each index's ambulkdelete routine), then do
+	 * related heap vacuuming
+	 */
+	if (vacrel->dead_items_info->num_items > 0)
+		lazy_vacuum(vacrel);
+
+	/*
+	 * Vacuum the remainder of the Free Space Map.  We must do this whether or
+	 * not there were indexes, and whether or not we bypassed index vacuuming.
+	 * We can pass rel_pages here because we never skip scanning the last
+	 * block of the relation.
+	 */
+	if (rel_pages > vacrel->next_fsm_block_to_vacuum)
+		FreeSpaceMapVacuumRange(vacrel->rel, vacrel->next_fsm_block_to_vacuum, rel_pages);
+
+	/* report all blocks vacuumed */
+	pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, rel_pages);
+
+	/* Do final index cleanup (call each index's amvacuumcleanup routine) */
+	if (vacrel->nindexes > 0 && vacrel->do_index_cleanup)
+		lazy_cleanup_all_indexes(vacrel);
+}
+
+/*
+ * Workhorse for lazy_scan_heap().
+ *
+ * Return true if the scan reaches the end of the table, otherwise false. In single
+ * process vacuum, since we loop the cycle of heap scanning, vacuuming and heap
+ * vacuuming until we reach to the end of table, it always returns true. On the other
+ * hand, in parallel vacuum case, if the dead items space is overrunning the available
+ * space, we exit from this function without invoking a cycle of index and heap vacuuming.
+ * In this case, we return false.
+ */
+static bool
+do_lazy_scan_heap(LVRelState *vacrel)
+{
+	BlockNumber blkno = InvalidBlockNumber;
+	BlockNumber orig_eager_scan_success_limit =
+		vacrel->eager_scan_remaining_successes; /* for logging */
+	Buffer		vmbuffer = InvalidBuffer;
+	bool		reach_eot = true;
+
+	Assert((vacrel->stream != NULL) || ParallelHeapVacuumIsActive(vacrel));
 
 	while (true)
 	{
@@ -1257,7 +1495,6 @@ lazy_scan_heap(LVRelState *vacrel)
 		Page		page;
 		uint8		blk_info = 0;
 		bool		has_lpdead_items;
-		void	   *per_buffer_data = NULL;
 		bool		vm_page_frozen = false;
 		bool		got_cleanup_lock = false;
 
@@ -1271,8 +1508,11 @@ lazy_scan_heap(LVRelState *vacrel)
 		 * that point.  This check also provides failsafe coverage for the
 		 * one-pass strategy, and the two-pass strategy with the index_cleanup
 		 * param set to 'off'.
+		 *
+		 * The failsafe check should be done only by the leader process.
 		 */
-		if (vacrel->scan_data->scanned_pages > 0 &&
+		if (!IsParallelWorker() &&
+			vacrel->scan_data->scanned_pages > 0 &&
 			vacrel->scan_data->scanned_pages % FAILSAFE_EVERY_PAGES == 0)
 			lazy_check_wraparound_failsafe(vacrel);
 
@@ -1299,6 +1539,19 @@ lazy_scan_heap(LVRelState *vacrel)
 				vmbuffer = InvalidBuffer;
 			}
 
+			/*
+			 * In parallel heap vacuum, we return false to the caller without
+			 * index and heap vacuuming. The parallel vacuum workers will exit
+			 * and the leader process will perform both index and heap
+			 * vacuuming.
+			 */
+			if (ParallelHeapVacuumIsActive(vacrel))
+			{
+				vacrel->plvstate->scanstate->last_blkno = blkno;
+				reach_eot = false;
+				break;
+			}
+
 			/* Perform a round of index and heap vacuuming */
 			vacrel->consider_bypass_optimization = false;
 			lazy_vacuum(vacrel);
@@ -1308,25 +1561,23 @@ lazy_scan_heap(LVRelState *vacrel)
 			 * upper-level FSM pages. Note that blkno is the previously
 			 * processed block.
 			 */
-			FreeSpaceMapVacuumRange(vacrel->rel, next_fsm_block_to_vacuum,
+			FreeSpaceMapVacuumRange(vacrel->rel, vacrel->next_fsm_block_to_vacuum,
 									blkno + 1);
-			next_fsm_block_to_vacuum = blkno;
+			vacrel->next_fsm_block_to_vacuum = blkno;
 
 			/* Report that we are once again scanning the heap */
 			pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
 										 PROGRESS_VACUUM_PHASE_SCAN_HEAP);
 		}
 
-		buf = read_stream_next_buffer(stream, &per_buffer_data);
+		/* Read the next block to process */
+		buf = heap_vac_get_next_buffer(vacrel, &blkno, &blk_info);
 
 		/* The relation is exhausted. */
 		if (!BufferIsValid(buf))
 			break;
 
-		blk_info = *((uint8 *) per_buffer_data);
-		CheckBufferIsPinnedOnce(buf);
 		page = BufferGetPage(buf);
-		blkno = BufferGetBlockNumber(buf);
 
 		vacrel->scan_data->scanned_pages++;
 		if (blk_info & VAC_BLK_WAS_EAGER_SCANNED)
@@ -1471,10 +1722,13 @@ lazy_scan_heap(LVRelState *vacrel)
 		 * also be no opportunity to update the FSM later, because we'll never
 		 * revisit this page. Since updating the FSM is desirable but not
 		 * absolutely required, that's OK.
+		 *
+		 * FSM vacuum should be done only by the leader process.
 		 */
-		if (vacrel->nindexes == 0
-			|| !vacrel->do_index_vacuuming
-			|| !has_lpdead_items)
+		if (!IsParallelWorker() &&
+			(vacrel->nindexes == 0
+			 || !vacrel->do_index_vacuuming
+			 || !has_lpdead_items))
 		{
 			Size		freespace = PageGetHeapFreeSpace(page);
 
@@ -1488,11 +1742,17 @@ lazy_scan_heap(LVRelState *vacrel)
 			 * held the cleanup lock and lazy_scan_prune() was called.
 			 */
 			if (got_cleanup_lock && vacrel->nindexes == 0 && has_lpdead_items &&
-				blkno - next_fsm_block_to_vacuum >= VACUUM_FSM_EVERY_PAGES)
+				blkno - vacrel->next_fsm_block_to_vacuum >= VACUUM_FSM_EVERY_PAGES)
 			{
-				FreeSpaceMapVacuumRange(vacrel->rel, next_fsm_block_to_vacuum,
+				/*
+				 * XXX: Since the following logic doesn't consider the
+				 * progress of workers' scan processes, there might be
+				 * unprocessed pages between next_fsm_block_to_vacuum and
+				 * blkno.
+				 */
+				FreeSpaceMapVacuumRange(vacrel->rel, vacrel->next_fsm_block_to_vacuum,
 										blkno);
-				next_fsm_block_to_vacuum = blkno;
+				vacrel->next_fsm_block_to_vacuum = blkno;
 			}
 		}
 		else
@@ -1504,98 +1764,129 @@ lazy_scan_heap(LVRelState *vacrel)
 		ReleaseBuffer(vmbuffer);
 
 	/*
-	 * Report that everything is now scanned. We never skip scanning the last
-	 * block in the relation, so we can pass rel_pages here.
+	 * We could return from this function while not reaching to the end of
+	 * table only in parallel heap vacuum cases.
 	 */
-	pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED,
-								 rel_pages);
+	Assert(reach_eot || ParallelHeapVacuumIsActive(vacrel));
+	return reach_eot;
+}
 
-	/* now we can compute the new value for pg_class.reltuples */
-	vacrel->new_live_tuples = vac_estimate_reltuples(vacrel->rel, rel_pages,
-													 vacrel->scan_data->scanned_pages,
-													 vacrel->scan_data->live_tuples);
+/*
+ * Return the buffer for the block to process. Both blkno and blk_info are output
+ * parameters; the block number and VAC_BLK_XXX flags are set, respectively. Return
+ * InvalidBuffer if there are no remaining blocks.
+ */
+static inline Buffer
+heap_vac_get_next_buffer(LVRelState *vacrel, BlockNumber *blkno_out, uint8 *blk_info_out)
+{
+	Buffer		buf;
 
-	/*
-	 * Also compute the total number of surviving heap entries.  In the
-	 * (unlikely) scenario that new_live_tuples is -1, take it as zero.
-	 */
-	vacrel->new_rel_tuples =
-		Max(vacrel->new_live_tuples, 0) + vacrel->scan_data->recently_dead_tuples +
-		vacrel->scan_data->missed_dead_tuples;
+	if (ParallelHeapVacuumIsActive(vacrel))
+	{
+		*blkno_out = get_next_unskippable_block(vacrel, blk_info_out);
 
-	read_stream_end(stream);
+		/* Have we reached the end of the relation? */
+		if (!BlockNumberIsValid(*blkno_out))
+			goto reach_eot;
 
-	/*
-	 * Do index vacuuming (call each index's ambulkdelete routine), then do
-	 * related heap vacuuming
-	 */
-	if (vacrel->dead_items_info->num_items > 0)
-		lazy_vacuum(vacrel);
+		buf = ReadBufferExtended(vacrel->rel, MAIN_FORKNUM, *blkno_out, RBM_NORMAL,
+								 vacrel->bstrategy);
+	}
+	else
+	{
+		void	   *per_buffer_data = NULL;
 
-	/*
-	 * Vacuum the remainder of the Free Space Map.  We must do this whether or
-	 * not there were indexes, and whether or not we bypassed index vacuuming.
-	 * We can pass rel_pages here because we never skip scanning the last
-	 * block of the relation.
-	 */
-	if (rel_pages > next_fsm_block_to_vacuum)
-		FreeSpaceMapVacuumRange(vacrel->rel, next_fsm_block_to_vacuum, rel_pages);
+		/* Use streaming read for non-parallel heap scans */
+		buf = read_stream_next_buffer(vacrel->stream, &per_buffer_data);
 
-	/* report all blocks vacuumed */
-	pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, rel_pages);
+		/* Have we reached the end of the relation? */
+		if (!BufferIsValid(buf))
+			goto reach_eot;
 
-	/* Do final index cleanup (call each index's amvacuumcleanup routine) */
-	if (vacrel->nindexes > 0 && vacrel->do_index_cleanup)
-		lazy_cleanup_all_indexes(vacrel);
+		*blk_info_out = *((uint8 *) per_buffer_data);
+		*blkno_out = BufferGetBlockNumber(buf);
+	}
+
+	Assert(BufferIsValid(buf));
+	CheckBufferIsPinnedOnce(buf);
+
+	return buf;
+
+reach_eot:
+	if (BufferIsValid(vacrel->next_unskippable_vmbuffer))
+	{
+		ReleaseBuffer(vacrel->next_unskippable_vmbuffer);
+		vacrel->next_unskippable_vmbuffer = InvalidBuffer;
+	}
+
+	return InvalidBuffer;
 }
 
 /*
- *	heap_vac_scan_next_block() -- read stream callback to get the next block
- *	for vacuum to process
- *
- * Every time lazy_scan_heap() needs a new block to process during its first
- * phase, it invokes read_stream_next_buffer() with a stream set up to call
- * heap_vac_scan_next_block() to get the next block.
- *
- * heap_vac_scan_next_block() uses the visibility map, vacuum options, and
- * various thresholds to skip blocks which do not need to be processed and
- * returns the next block to process or InvalidBlockNumber if there are no
- * remaining blocks.
- *
- * The visibility status of the next block to process and whether or not it
- * was eager scanned is set in the per_buffer_data.
- *
- * callback_private_data contains a reference to the LVRelState, passed to the
- * read stream API during stream setup. The LVRelState is an in/out parameter
- * here (locally named `vacrel`). Vacuum options and information about the
- * relation are read from it. vacrel->scan_data->skippedallvis is set if we skip a block
- * that's all-visible but not all-frozen (to ensure that we don't update
- * relfrozenxid in that case). vacrel also holds information about the next
- * unskippable block -- as bookkeeping for this function.
+ * heap_vac_scan_next_block_read_stream() is the read stream callback. Every
+ * time lazy_scan_heap() needs a new block to process during its first phase, it
+ * invokes read_stream_next_buffer() with a stream set up to call
+ * heap_vac_scan_next_block_read_stream() to get the next block.
+ * callback_private_data contains a reference to the LVRelState, passed to the read
+ * stream API during stream setup.
  */
 static BlockNumber
-heap_vac_scan_next_block(ReadStream *stream,
-						 void *callback_private_data,
-						 void *per_buffer_data)
+heap_vac_scan_next_block_read_stream(ReadStream *stream,
+									 void *callback_private_data,
+									 void *per_buffer_data)
 {
-	BlockNumber next_block;
 	LVRelState *vacrel = callback_private_data;
-	uint8		blk_info = 0;
 
-	/* relies on InvalidBlockNumber + 1 overflowing to 0 on first call */
-	next_block = vacrel->current_block + 1;
+	Assert(!ParallelHeapVacuumIsActive(vacrel));
+	return get_next_unskippable_block(vacrel, (uint8 *) per_buffer_data);
+}
+
+/*
+ * Get the next unskippable block for vacuum process found. Return InvalidBlockNumber
+ * if there are no remaining blocks to process.
+ *
+ * get_next_unskippable_block() uses the visibility map, vacuum options, and various
+ * thresholds to skip blocks which do not need to be processed and returns the next
+ * block to process or InvalidBlockNumber if there are no remaining blocks. The
+ * visibility status of the next block to process and whether or not it was eager
+ * scanned are set to the blk_info.
+ *
+ * The LVRelState is an in/out parameter here (locally named `vacrel`). Vacuum options
+ * and information about the relation are read from it.
+ * vacrel->scan_data->skippedallvis is set if we skip a block that's all-visible but
+ * not all-frozen (to ensure that we don't update relfrozenxid in that case). vacrel
+ * also holds information about the next unskippable block -- as bookkeeping for this
+ * function.
+ */
+static BlockNumber
+get_next_unskippable_block(LVRelState *vacrel, uint8 *blk_info)
+{
+	BlockNumber next_block;
+
+	*blk_info = 0;
 
-	/* Have we reached the end of the relation? */
-	if (next_block >= vacrel->scan_data->rel_pages)
+retry:
+	/* Get the next block to process */
+	if (ParallelHeapVacuumIsActive(vacrel))
 	{
-		if (BufferIsValid(vacrel->next_unskippable_vmbuffer))
-		{
-			ReleaseBuffer(vacrel->next_unskippable_vmbuffer);
-			vacrel->next_unskippable_vmbuffer = InvalidBuffer;
-		}
-		return InvalidBlockNumber;
+		next_block = table_block_parallelscan_nextpage(vacrel->rel,
+													   &(vacrel->plvstate->scanstate->pbscanwork),
+													   vacrel->plvstate->pbscan);
+	}
+	else
+	{
+		/* relies on InvalidBlockNumber + 1 overflowing to 0 on first call */
+		next_block = vacrel->current_block + 1;
 	}
 
+	/*
+	 * Have we reached the end of the relation? Note that
+	 * table_block_parallelscan_nextpage() returns InvalidBlockNumber if there
+	 * are no remaining blocks.
+	 */
+	if (!BlockNumberIsValid(next_block) || next_block >= vacrel->scan_data->rel_pages)
+		return InvalidBlockNumber;
+
 	/*
 	 * We must be in one of the three following states:
 	 */
@@ -1608,8 +1899,54 @@ heap_vac_scan_next_block(ReadStream *stream,
 		 * visibility map.
 		 */
 		bool		skipsallvis;
+		bool		found;
+		BlockNumber end_block;
+		BlockNumber nblocks_skip;
+
+		if (ParallelHeapVacuumIsActive(vacrel))
+		{
+			/* We look for the next unskippable block within the chunk */
+			end_block = next_block +
+				vacrel->plvstate->scanstate->pbscanwork.phsw_chunk_remaining + 1;
+		}
+		else
+			end_block = vacrel->scan_data->rel_pages;
 
-		find_next_unskippable_block(vacrel, &skipsallvis);
+		found = find_next_unskippable_block(vacrel, &skipsallvis, next_block,
+											end_block);
+
+		/*
+		 * Handle the case where we could not find the unskippable block
+		 * within the specified range. This case must happen only in parallel
+		 * heap vacuum cases.
+		 *
+		 * In non-parallel heap vacuums, we already did the boundary check
+		 * above and always specify the range of next_block and rel_pages. And
+		 * we must scan the last page.
+		 */
+		if (!found)
+		{
+			Assert(ParallelHeapVacuumIsActive(vacrel));
+
+			if (skipsallvis)
+				vacrel->scan_data->skippedallvis = true;
+
+			/*
+			 * Tell the parallel scans to consume all remaining blocks in the
+			 * current chunk.
+			 */
+			table_block_parallelscan_skip_pages_in_chunk(vacrel->rel,
+														 &(vacrel->plvstate->scanstate->pbscanwork),
+														 vacrel->plvstate->pbscan,
+														 vacrel->plvstate->scanstate->pbscanwork.phsw_chunk_remaining);
+
+			/*
+			 * Reset the next_unskippable_blocks so we can try to find an
+			 * unskippable block in the next chunk.
+			 */
+			vacrel->next_unskippable_block = InvalidBlockNumber;
+			goto retry;
+		}
 
 		/*
 		 * We now know the next block that we must process.  It can be the
@@ -1626,11 +1963,20 @@ heap_vac_scan_next_block(ReadStream *stream,
 		 * pages then skipping makes updating relfrozenxid unsafe, which is a
 		 * real downside.
 		 */
-		if (vacrel->next_unskippable_block - next_block >= SKIP_PAGES_THRESHOLD)
+		nblocks_skip = vacrel->next_unskippable_block - next_block;
+		if (nblocks_skip >= SKIP_PAGES_THRESHOLD)
 		{
-			next_block = vacrel->next_unskippable_block;
 			if (skipsallvis)
 				vacrel->scan_data->skippedallvis = true;
+
+			/* Tell the parallel scans to skip blocks */
+			if (ParallelHeapVacuumIsActive(vacrel))
+				table_block_parallelscan_skip_pages_in_chunk(vacrel->rel,
+															 &(vacrel->plvstate->scanstate->pbscanwork),
+															 vacrel->plvstate->pbscan,
+															 nblocks_skip);
+
+			next_block = vacrel->next_unskippable_block;
 		}
 	}
 
@@ -1643,8 +1989,7 @@ heap_vac_scan_next_block(ReadStream *stream,
 		 * otherwise they would've been unskippable.
 		 */
 		vacrel->current_block = next_block;
-		blk_info |= VAC_BLK_ALL_VISIBLE_ACCORDING_TO_VM;
-		*((uint8 *) per_buffer_data) = blk_info;
+		*blk_info |= VAC_BLK_ALL_VISIBLE_ACCORDING_TO_VM;
 		return vacrel->current_block;
 	}
 	else
@@ -1657,16 +2002,18 @@ heap_vac_scan_next_block(ReadStream *stream,
 
 		vacrel->current_block = next_block;
 		if (vacrel->next_unskippable_allvis)
-			blk_info |= VAC_BLK_ALL_VISIBLE_ACCORDING_TO_VM;
+			*blk_info |= VAC_BLK_ALL_VISIBLE_ACCORDING_TO_VM;
 		if (vacrel->next_unskippable_eager_scanned)
-			blk_info |= VAC_BLK_WAS_EAGER_SCANNED;
-		*((uint8 *) per_buffer_data) = blk_info;
+			*blk_info |= VAC_BLK_WAS_EAGER_SCANNED;
 		return vacrel->current_block;
 	}
 }
 
 /*
- * Find the next unskippable block in a vacuum scan using the visibility map.
+ * Find the next unskippable block in a vacuum scan using the visibility map,
+ * in a range of start_blk (inclusive) and end_blk (exclusive). Return true
+ * if we successfully found the block within the range.
+ *
  * The next unskippable block and its visibility information is updated in
  * vacrel.
  *
@@ -1678,23 +2025,35 @@ heap_vac_scan_next_block(ReadStream *stream,
  * older XIDs/MXIDs.  The *skippedallvis flag will be set here when the choice
  * to skip such a range is actually made, making everything safe.)
  */
-static void
-find_next_unskippable_block(LVRelState *vacrel, bool *skipsallvis)
+static bool
+find_next_unskippable_block(LVRelState *vacrel, bool *skipsallvis,
+							BlockNumber start_blk, BlockNumber end_blk)
 {
 	BlockNumber rel_pages = vacrel->scan_data->rel_pages;
-	BlockNumber next_unskippable_block = vacrel->next_unskippable_block + 1;
+	BlockNumber next_unskippable_block = start_blk;
 	Buffer		next_unskippable_vmbuffer = vacrel->next_unskippable_vmbuffer;
 	bool		next_unskippable_eager_scanned = false;
 	bool		next_unskippable_allvis;
+	bool		found = true;
 
 	*skipsallvis = false;
 
 	for (;; next_unskippable_block++)
 	{
-		uint8		mapbits = visibilitymap_get_status(vacrel->rel,
-													   next_unskippable_block,
-													   &next_unskippable_vmbuffer);
+		uint8		mapbits;
+
+		/*
+		 * Stop the search as we reach the end of the range. Let the caller
+		 * know we could not find the unskippable page within the range.
+		 */
+		if (next_unskippable_block >= end_blk)
+		{
+			found = false;
+			break;
+		}
 
+		mapbits = visibilitymap_get_status(vacrel->rel, next_unskippable_block,
+										   &next_unskippable_vmbuffer);
 		next_unskippable_allvis = (mapbits & VISIBILITYMAP_ALL_VISIBLE) != 0;
 
 		/*
@@ -1774,6 +2133,241 @@ find_next_unskippable_block(LVRelState *vacrel, bool *skipsallvis)
 	vacrel->next_unskippable_allvis = next_unskippable_allvis;
 	vacrel->next_unskippable_eager_scanned = next_unskippable_eager_scanned;
 	vacrel->next_unskippable_vmbuffer = next_unskippable_vmbuffer;
+
+	return found;
+}
+
+/*
+ * A parallel variant of do_lazy_scan_heap(). The leader process launches
+ * parallel workers to scan the heap in parallel.
+*/
+static void
+do_parallel_lazy_scan_heap(LVRelState *vacrel)
+{
+	PLVScanWorkerState *scanstate;
+
+	Assert(ParallelHeapVacuumIsActive(vacrel));
+	Assert(!IsParallelWorker());
+	Assert(vacrel->stream == NULL);
+
+	/* Launch parallel workers */
+	parallel_lazy_scan_heap_begin(vacrel);
+
+	/*
+	 * Setup the parallel scan description for the leader to join as a worker.
+	 */
+	scanstate = palloc0(sizeof(PLVScanWorkerState));
+	scanstate->last_blkno = InvalidBlockNumber;
+	table_block_parallelscan_startblock_init(vacrel->rel,
+											 &(scanstate->pbscanwork),
+											 vacrel->plvstate->pbscan);
+	vacrel->plvstate->scanstate = scanstate;
+
+	for (;;)
+	{
+		bool		reach_eot;
+		BlockNumber min_scan_blk;
+
+		/*
+		 * Scan the table until either we are close to overrunning the
+		 * available space for dead_items TIDs or we reach the end of the
+		 * relation.
+		 */
+		reach_eot = do_lazy_scan_heap(vacrel);
+
+		/*
+		 * Parallel lazy heap scan finished. Wait for parallel workers to
+		 * finish and gather scan results.
+		 */
+		parallel_lazy_scan_heap_end(vacrel);
+
+		/* We reach the end of the table */
+		if (reach_eot)
+			break;
+
+		/* Perform a round of index and heap vacuuming */
+		vacrel->consider_bypass_optimization = false;
+		lazy_vacuum(vacrel);
+
+		min_scan_blk = parallel_lazy_scan_compute_min_scan_block(vacrel);
+
+		/*
+		 * Vacuum the Free Space Map to make newly-freed space visible on
+		 * upper-level FSM pages.
+		 */
+		if (min_scan_blk > vacrel->next_fsm_block_to_vacuum)
+		{
+			/*
+			 * min_scanned_blkno was updated when gathering the workers' scan
+			 * results.
+			 */
+			FreeSpaceMapVacuumRange(vacrel->rel, vacrel->next_fsm_block_to_vacuum,
+									min_scan_blk + 1);
+			vacrel->next_fsm_block_to_vacuum = min_scan_blk;
+		}
+
+		/* Report that we are once again scanning the heap */
+		pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
+									 PROGRESS_VACUUM_PHASE_SCAN_HEAP);
+
+		/* Re-launch workers to restart parallel heap scan */
+		parallel_lazy_scan_heap_begin(vacrel);
+	}
+
+	/*
+	 * The parallel heap scan finished, but it's possible that some workers
+	 * have allocated blocks but not processed them yet. This can happen for
+	 * example when workers exit because they are full of dead_items TIDs and
+	 * the leader process launched fewer workers in the next cycle.
+	 */
+	complete_unfinihsed_lazy_scan_heap(vacrel);
+}
+
+/*
+ * Return the minimum block number the leader and workers have scanned so far.
+ */
+static BlockNumber
+parallel_lazy_scan_compute_min_scan_block(LVRelState *vacrel)
+{
+	BlockNumber min_blk;
+
+	Assert(ParallelHeapVacuumIsActive(vacrel));
+
+	min_blk = vacrel->plvstate->scanstate->last_blkno;
+
+	/*
+	 * We check all worker scan states here to compute the minimum block
+	 * number among all scan states.
+	 */
+	for (int i = 0; i < vacrel->leader->nworkers_launched; i++)
+	{
+		PLVScanWorkerState *scanstate = &(vacrel->leader->scanstates[i]);
+
+		/* Skip if no worker has been initialized the scan state */
+		if (!scanstate->inited)
+			continue;
+
+		if (!BlockNumberIsValid(min_blk) || scanstate->last_blkno < min_blk)
+			min_blk = scanstate->last_blkno;
+	}
+
+	Assert(BlockNumberIsValid(min_blk));
+	return min_blk;
+}
+
+/*
+ * Complete parallel heaps scans that have remaining blocks in their
+ * chunks.
+ */
+static void
+complete_unfinihsed_lazy_scan_heap(LVRelState *vacrel)
+{
+	int			nworkers;
+
+	Assert(!IsParallelWorker());
+
+	nworkers = parallel_vacuum_get_nworkers_table(vacrel->pvs);
+
+	for (int i = 0; i < nworkers; i++)
+	{
+		PLVScanWorkerState *scanstate = &(vacrel->leader->scanstates[i]);
+
+		/*
+		 * Skip if this worker's scan has not been used or doesn't have
+		 * unprocessed block in chunks.
+		 */
+		if (!scanstate->inited || !scanstate->maybe_have_unprocessed_blocks)
+			continue;
+
+		/* Attach the worker's scan state */
+		vacrel->plvstate->scanstate = scanstate;
+
+		/* Complete the unfinished scan */
+		do_lazy_scan_heap(vacrel);
+	}
+
+	/*
+	 * We don't need to gather the scan results here because the leader's scan
+	 * state got updated directly.
+	 */
+}
+
+/*
+ * Helper routine to launch parallel workers for parallel lazy heap scan.
+ */
+static void
+parallel_lazy_scan_heap_begin(LVRelState *vacrel)
+{
+	Assert(ParallelHeapVacuumIsActive(vacrel));
+	Assert(!IsParallelWorker());
+
+	/* launcher workers */
+	vacrel->leader->nworkers_launched = parallel_vacuum_collect_dead_items_begin(vacrel->pvs);
+
+	ereport(vacrel->verbose ? INFO : DEBUG2,
+			(errmsg(ngettext("launched %d parallel vacuum worker for collecting dead tuples (planned: %d)",
+							 "launched %d parallel vacuum workers for collecting dead tuples (planned: %d)",
+							 vacrel->leader->nworkers_launched),
+					vacrel->leader->nworkers_launched,
+					parallel_vacuum_get_nworkers_table(vacrel->pvs))));
+}
+
+/*
+ * Helper routine to finish the parallel lazy heap scan.
+ */
+static void
+parallel_lazy_scan_heap_end(LVRelState *vacrel)
+{
+	/* Wait for all parallel workers to finish */
+	parallel_vacuum_scan_end(vacrel->pvs);
+
+	/* Gather the workers' scan results */
+	parallel_lazy_scan_gather_scan_results(vacrel);
+}
+
+/* Accumulate each worker's scan results into the leader's */
+static void
+parallel_lazy_scan_gather_scan_results(LVRelState *vacrel)
+{
+	Assert(ParallelHeapVacuumIsActive(vacrel));
+	Assert(!IsParallelWorker());
+
+	/* Gather the workers' scan results */
+	for (int i = 0; i < vacrel->leader->nworkers_launched; i++)
+	{
+		LVScanData *data = &(vacrel->plvstate->shared->worker_scandata[i]);
+
+#define ACCUM_COUNT(item) vacrel->scan_data->item += data->item
+		ACCUM_COUNT(scanned_pages);
+		ACCUM_COUNT(removed_pages);
+		ACCUM_COUNT(new_frozen_tuple_pages);
+		ACCUM_COUNT(vm_new_visible_pages);
+		ACCUM_COUNT(vm_new_visible_frozen_pages);
+		ACCUM_COUNT(vm_new_frozen_pages);
+		ACCUM_COUNT(lpdead_item_pages);
+		ACCUM_COUNT(missed_dead_pages);
+		ACCUM_COUNT(tuples_deleted);
+		ACCUM_COUNT(tuples_frozen);
+		ACCUM_COUNT(lpdead_items);
+		ACCUM_COUNT(live_tuples);
+		ACCUM_COUNT(recently_dead_tuples);
+		ACCUM_COUNT(missed_dead_tuples);
+#undef ACCUM_COUNT
+
+		Assert(TransactionIdIsValid(data->NewRelfrozenXid));
+		Assert(MultiXactIdIsValid(data->NewRelminMxid));
+
+		if (TransactionIdPrecedes(data->NewRelfrozenXid, vacrel->scan_data->NewRelfrozenXid))
+			vacrel->scan_data->NewRelfrozenXid = data->NewRelfrozenXid;
+
+		if (MultiXactIdPrecedesOrEquals(data->NewRelminMxid, vacrel->scan_data->NewRelminMxid))
+			vacrel->scan_data->NewRelminMxid = data->NewRelminMxid;
+
+		if (data->nonempty_pages < vacrel->scan_data->nonempty_pages)
+			vacrel->scan_data->nonempty_pages = data->nonempty_pages;
+
+		vacrel->scan_data->skippedallvis |= data->skippedallvis;
+	}
 }
 
 /*
@@ -3489,12 +4083,8 @@ dead_items_alloc(LVRelState *vacrel, int nworkers)
 		autovacuum_work_mem != -1 ?
 		autovacuum_work_mem : maintenance_work_mem;
 
-	/*
-	 * Initialize state for a parallel vacuum.  As of now, only one worker can
-	 * be used for an index, so we invoke parallelism only if there are at
-	 * least two indexes on a table.
-	 */
-	if (nworkers >= 0 && vacrel->nindexes > 1 && vacrel->do_index_vacuuming)
+	/* Initialize state for a parallel vacuum */
+	if (nworkers >= 0)
 	{
 		/*
 		 * Since parallel workers cannot access data in temporary tables, we
@@ -3512,11 +4102,17 @@ dead_items_alloc(LVRelState *vacrel, int nworkers)
 								vacrel->relname)));
 		}
 		else
+		{
+			/*
+			 * We initialize the parallel vacuum state for either lazy heap
+			 * scan or index vacuuming, or both.
+			 */
 			vacrel->pvs = parallel_vacuum_init(vacrel->rel, vacrel->indrels,
 											   vacrel->nindexes, nworkers,
 											   vac_work_mem,
 											   vacrel->verbose ? INFO : DEBUG2,
 											   vacrel->bstrategy, (void *) vacrel);
+		}
 
 		/*
 		 * If parallel mode started, dead_items and dead_items_info spaces are
@@ -3556,9 +4152,15 @@ dead_items_add(LVRelState *vacrel, BlockNumber blkno, OffsetNumber *offsets,
 	};
 	int64		prog_val[2];
 
+	if (ParallelHeapVacuumIsActive(vacrel))
+		TidStoreLockExclusive(vacrel->dead_items);
+
 	TidStoreSetBlockOffsets(vacrel->dead_items, blkno, offsets, num_offsets);
 	vacrel->dead_items_info->num_items += num_offsets;
 
+	if (ParallelHeapVacuumIsActive(vacrel))
+		TidStoreUnlock(vacrel->dead_items);
+
 	/* update the progress information */
 	prog_val[0] = vacrel->dead_items_info->num_items;
 	prog_val[1] = TidStoreMemoryUsage(vacrel->dead_items);
@@ -3760,14 +4362,238 @@ update_relstats_all_indexes(LVRelState *vacrel)
 
 /*
  * Compute the number of workers for parallel heap vacuum.
- *
- * Return 0 to disable parallel vacuum.
  */
 int
 heap_parallel_vacuum_compute_workers(Relation rel, int nworkers_requested,
 									 void *state)
 {
-	return 0;
+	int			parallel_workers = 0;
+
+	if (nworkers_requested == 0)
+	{
+		LVRelState *vacrel = (LVRelState *) state;
+		int			heap_parallel_threshold;
+		int			heap_pages;
+		BlockNumber allvisible;
+		BlockNumber allfrozen;
+
+		/*
+		 * Estimate the number of blocks that we're going to scan during
+		 * lazy_scan_heap().
+		 */
+		visibilitymap_count(rel, &allvisible, &allfrozen);
+		heap_pages = RelationGetNumberOfBlocks(rel) -
+			(vacrel->aggressive ? allfrozen : allvisible);
+
+		Assert(heap_pages >= 0);
+
+		/*
+		 * Select the number of workers based on the log of the number of
+		 * pages to scan. Note that the upper limit of the
+		 * min_parallel_table_scan_size GUC is chosen to prevent overflow
+		 * here.
+		 */
+		heap_parallel_threshold = Max(min_parallel_table_scan_size, 1);
+		while (heap_pages >= (BlockNumber) (heap_parallel_threshold * 3))
+		{
+			parallel_workers++;
+			heap_parallel_threshold *= 3;
+			if (heap_parallel_threshold > INT_MAX / 3)
+				break;
+		}
+	}
+	else
+		parallel_workers = nworkers_requested;
+
+	return parallel_workers;
+}
+
+/*
+ * Estimate shared memory size required for parallel heap vacuum.
+ */
+void
+heap_parallel_vacuum_estimate(Relation rel, ParallelContext *pcxt, int nworkers,
+							  void *state)
+{
+	LVRelState *vacrel = (LVRelState *) state;
+	Size		size = 0;
+
+	vacrel->leader = palloc(sizeof(PLVLeader));
+
+	/* Estimate space for PLVShared */
+	size = add_size(size, SizeOfPLVShared);
+	size = add_size(size, mul_size(sizeof(LVScanData), nworkers));
+	vacrel->leader->shared_len = size;
+	shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->shared_len);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+	/* Estimate space for ParallelBlockTableScanDesc */
+	vacrel->leader->pbscan_len = table_block_parallelscan_estimate(rel);
+	shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->pbscan_len);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+	/* Estimate space for an array of PLVScanWorkerState */
+	vacrel->leader->scanstates_len = mul_size(sizeof(PLVScanWorkerState), nworkers);
+	shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->scanstates_len);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+}
+
+/*
+ * Set up shared memory for parallel heap vacuum.
+ */
+void
+heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt, int nworkers,
+								void *state)
+{
+	LVRelState *vacrel = (LVRelState *) state;
+	PLVShared  *shared;
+	ParallelBlockTableScanDesc pbscan;
+	PLVScanWorkerState *scanstates;
+
+	vacrel->plvstate = palloc0(sizeof(PLVState));
+
+	/* Initialize PLVShared */
+	shared = shm_toc_allocate(pcxt->toc, vacrel->leader->shared_len);
+	MemSet(shared, 0, vacrel->leader->shared_len);
+	shared->aggressive = vacrel->aggressive;
+	shared->skipwithvm = vacrel->skipwithvm;
+	shared->cutoffs = vacrel->cutoffs;
+	shared->NewRelfrozenXid = vacrel->scan_data->NewRelfrozenXid;
+	shared->NewRelminMxid = vacrel->scan_data->NewRelminMxid;
+	shared->vistest = *vacrel->vistest;
+	shm_toc_insert(pcxt->toc, LV_PARALLEL_KEY_SHARED, shared);
+	vacrel->plvstate->shared = shared;
+
+	/* Initialize ParallelBlockTableScanDesc */
+	pbscan = shm_toc_allocate(pcxt->toc, vacrel->leader->pbscan_len);
+	table_block_parallelscan_initialize(rel, (ParallelTableScanDesc) pbscan);
+	pbscan->base.phs_syncscan = false;	/* always start from the first block */
+	shm_toc_insert(pcxt->toc, LV_PARALLEL_KEY_SCANDESC, pbscan);
+	vacrel->plvstate->pbscan = pbscan;
+
+	/* Initialize the array of PLVScanWorkerState */
+	scanstates = shm_toc_allocate(pcxt->toc, vacrel->leader->scanstates_len);
+	MemSet(scanstates, 0, vacrel->leader->scanstates_len);
+	shm_toc_insert(pcxt->toc, LV_PARALLEL_KEY_WORKER_SCANSTATE, scanstates);
+	vacrel->leader->scanstates = scanstates;
+}
+
+/*
+ * Initialize lazy vacuum state with the information retrieved from
+ * shared memory.
+ */
+void
+heap_parallel_vacuum_initialize_worker(Relation rel, ParallelVacuumState *pvs,
+									   ParallelWorkerContext *pwcxt,
+									   void **state_out)
+{
+	LVRelState *vacrel;
+	PLVState   *plvstate;
+	PLVShared  *shared;
+	PLVScanWorkerState *scanstates;
+	ParallelBlockTableScanDesc pbscan;
+
+	/* Initialize PLVState and prepare the related objects */
+
+	plvstate = palloc0(sizeof(PLVState));
+
+	/* Prepare PLVShared */
+	shared = (PLVShared *) shm_toc_lookup(pwcxt->toc, LV_PARALLEL_KEY_SHARED, false);
+	plvstate->shared = shared;
+
+	/* Prepare ParallelBlockTableScanWorkerData */
+	pbscan = shm_toc_lookup(pwcxt->toc, LV_PARALLEL_KEY_SCANDESC, false);
+	plvstate->pbscan = pbscan;
+
+	/* Prepare PLVScanWorkerState */
+	scanstates = shm_toc_lookup(pwcxt->toc, LV_PARALLEL_KEY_WORKER_SCANSTATE, false);
+	plvstate->scanstate = &(scanstates[ParallelWorkerNumber]);
+
+	/* Initialize LVRelState and prepare fields required by lazy scan heap */
+	vacrel = palloc0(sizeof(LVRelState));
+	vacrel->rel = rel;
+	vacrel->indrels = parallel_vacuum_get_table_indexes(pvs,
+														&vacrel->nindexes);
+	vacrel->pvs = pvs;
+	vacrel->aggressive = shared->aggressive;
+	vacrel->skipwithvm = shared->skipwithvm;
+	vacrel->cutoffs = shared->cutoffs;
+	vacrel->vistest = &(shared->vistest);
+	vacrel->dead_items = parallel_vacuum_get_dead_items(pvs,
+														&vacrel->dead_items_info);
+	vacrel->plvstate = plvstate;
+	vacrel->scan_data = &(shared->worker_scandata[ParallelWorkerNumber]);
+	MemSet(vacrel->scan_data, 0, sizeof(LVScanData));
+	vacrel->scan_data->NewRelfrozenXid = shared->NewRelfrozenXid;
+	vacrel->scan_data->NewRelminMxid = shared->NewRelminMxid;
+	vacrel->scan_data->skippedallvis = false;
+	vacrel->scan_data->rel_pages = RelationGetNumberOfBlocks(rel);
+
+	/*
+	 * Initialize the scan state if not yet. The chunk of blocks will be
+	 * allocated when to get the scan block for the first time.
+	 */
+	if (!vacrel->plvstate->scanstate->inited)
+	{
+		vacrel->plvstate->scanstate->inited = true;
+		table_block_parallelscan_startblock_init(rel,
+												 &(vacrel->plvstate->scanstate->pbscanwork),
+												 vacrel->plvstate->pbscan);
+		vacrel->plvstate->scanstate->maybe_have_unprocessed_blocks = false;
+	}
+
+	*state_out = (void *) vacrel;
+}
+
+/*
+ * Parallel heap vacuum callback for collecting dead items (i.e., lazy heap scan).
+ */
+void
+heap_parallel_vacuum_collect_dead_items(Relation rel, ParallelVacuumState *pvs,
+										void *state)
+{
+	LVRelState *vacrel = (LVRelState *) state;
+	ErrorContextCallback errcallback;
+	bool		reach_eot;
+
+	Assert(ParallelHeapVacuumIsActive(vacrel));
+
+	/*
+	 * Setup error traceback support for ereport() for parallel table vacuum
+	 * workers
+	 */
+	vacrel->dbname = get_database_name(MyDatabaseId);
+	vacrel->relnamespace = get_database_name(RelationGetNamespace(rel));
+	vacrel->relname = pstrdup(RelationGetRelationName(rel));
+	vacrel->indname = NULL;
+	vacrel->phase = VACUUM_ERRCB_PHASE_SCAN_HEAP;
+	errcallback.callback = vacuum_error_callback;
+	errcallback.arg = &vacrel;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
+	/* Join the parallel heap vacuum */
+	reach_eot = do_lazy_scan_heap(vacrel);
+
+	/*
+	 * Release vmbuffer if we hold as we don't take over the state maintained
+	 * by heap_vac_scan_next_block() to the next time.
+	 */
+	if (BufferIsValid(vacrel->next_unskippable_vmbuffer))
+		ReleaseBuffer(vacrel->next_unskippable_vmbuffer);
+
+	/*
+	 * If the leader or a worker finishes the heap scan because dead_items
+	 * TIDs is close to the limit, lazy heap scan stops while it might have
+	 * some unscanned blocks in the allocated chunk. Since this scan state
+	 * could not be used in the next heap scan, we remember that it might have
+	 * some unconsumed blocks so that the leader complete the scans after the
+	 * heap scan phase finishes.
+	 */
+	vacrel->plvstate->scanstate->maybe_have_unprocessed_blocks = !reach_eot;
+
+	/* Pop the error context stack */
+	error_context_stack = errcallback.previous;
 }
 
 /*
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index a56c5eceb14..a0a92dc8be5 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -600,6 +600,21 @@ table_block_parallelscan_nextpage(Relation rel,
 	return page;
 }
 
+/*
+ * skip some blocks to scan.
+ *
+ * Consume the given number of blocks in the current chunk. It doesn't skip blocks
+ * beyond the current chunk.
+ */
+void
+table_block_parallelscan_skip_pages_in_chunk(Relation rel,
+											 ParallelBlockTableScanWorker pbscanwork,
+											 ParallelBlockTableScanDesc pbscan,
+											 BlockNumber nblocks_skip)
+{
+	pbscanwork->phsw_chunk_remaining -= Min(nblocks_skip, pbscanwork->phsw_chunk_remaining);
+}
+
 /* ----------------------------------------------------------------------------
  * Helper functions to implement relation sizing for block oriented AMs.
  * ----------------------------------------------------------------------------
diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c
index bb0d690aed8..8473971ac4f 100644
--- a/src/backend/commands/vacuumparallel.c
+++ b/src/backend/commands/vacuumparallel.c
@@ -501,6 +501,26 @@ parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats)
 	pfree(pvs);
 }
 
+/*
+ * Return the number of parallel workers initialized for parallel table vacuum.
+ */
+int
+parallel_vacuum_get_nworkers_table(ParallelVacuumState *pvs)
+{
+	return pvs->nworkers_for_table;
+}
+
+/*
+ * Return the array of indexes associated to the given table to be vacuumed.
+ */
+Relation *
+parallel_vacuum_get_table_indexes(ParallelVacuumState *pvs, int *nindexes)
+{
+	*nindexes = pvs->nindexes;
+
+	return pvs->indrels;
+}
+
 /*
  * Returns the dead items space and dead items information.
  */
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index 6a1ca5d5ca7..b94d783c31e 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -15,6 +15,7 @@
 #define HEAPAM_H
 
 #include "access/heapam_xlog.h"
+#include "access/parallel.h"
 #include "access/relation.h"	/* for backward compatibility */
 #include "access/relscan.h"
 #include "access/sdir.h"
@@ -407,10 +408,20 @@ extern void log_heap_prune_and_freeze(Relation relation, Buffer buffer,
 
 /* in heap/vacuumlazy.c */
 struct VacuumParams;
+struct ParallelVacuumState;
 extern void heap_vacuum_rel(Relation rel,
 							struct VacuumParams *params, BufferAccessStrategy bstrategy);
 extern int	heap_parallel_vacuum_compute_workers(Relation rel, int nworkers_requested,
 												 void *state);
+extern void heap_parallel_vacuum_estimate(Relation rel, ParallelContext *pcxt, int nworkers,
+										  void *state);
+extern void heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt,
+											int nworkers, void *state);
+extern void heap_parallel_vacuum_initialize_worker(Relation rel, struct ParallelVacuumState *pvs,
+												   ParallelWorkerContext *pwcxt,
+												   void **state_out);
+extern void heap_parallel_vacuum_collect_dead_items(Relation rel, struct ParallelVacuumState *pvs,
+													void *state);
 
 /* in heap/heapam_visibility.c */
 extern bool HeapTupleSatisfiesVisibility(HeapTuple htup, Snapshot snapshot,
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index c61b1700953..6b72883353e 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -2170,6 +2170,10 @@ extern BlockNumber table_block_parallelscan_nextpage(Relation rel,
 extern void table_block_parallelscan_startblock_init(Relation rel,
 													 ParallelBlockTableScanWorker pbscanwork,
 													 ParallelBlockTableScanDesc pbscan);
+extern void table_block_parallelscan_skip_pages_in_chunk(Relation rel,
+														 ParallelBlockTableScanWorker pbscanwork,
+														 ParallelBlockTableScanDesc pbscan,
+														 BlockNumber nblocks_skip);
 
 
 /* ----------------------------------------------------------------------------
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 596e901c207..20eecad4ec4 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -384,6 +384,8 @@ extern ParallelVacuumState *parallel_vacuum_init(Relation rel, Relation *indrels
 												 BufferAccessStrategy bstrategy,
 												 void *state);
 extern void parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats);
+extern int	parallel_vacuum_get_nworkers_table(ParallelVacuumState *pvs);
+extern Relation *parallel_vacuum_get_table_indexes(ParallelVacuumState *pvs, int *nindexes);
 extern TidStore *parallel_vacuum_get_dead_items(ParallelVacuumState *pvs,
 												VacDeadItemsInfo **dead_items_info_p);
 extern void parallel_vacuum_reset_dead_items(ParallelVacuumState *pvs);
diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index 3f91b69b324..0baf592ae5d 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -160,6 +160,11 @@ UPDATE pvactst SET i = i WHERE i < 1000;
 VACUUM (PARALLEL 2) pvactst;
 UPDATE pvactst SET i = i WHERE i < 1000;
 VACUUM (PARALLEL 0) pvactst; -- disable parallel vacuum
+-- VACUUM invokes parallel heap vacuum.
+SET min_parallel_table_scan_size to 0;
+VACUUM (PARALLEL 2, FREEZE) pvactst2;
+UPDATE pvactst2 SET i = i WHERE i < 1000;
+VACUUM (PARALLEL 1) pvactst2;
 VACUUM (PARALLEL -1) pvactst; -- error
 ERROR:  parallel workers for vacuum must be between 0 and 1024
 LINE 1: VACUUM (PARALLEL -1) pvactst;
@@ -185,6 +190,7 @@ VACUUM (PARALLEL 1, FULL FALSE) tmp; -- parallel vacuum disabled for temp tables
 WARNING:  disabling parallel option of vacuum on "tmp" --- cannot vacuum temporary tables in parallel
 VACUUM (PARALLEL 0, FULL TRUE) tmp; -- can specify parallel disabled (even though that's implied by FULL)
 RESET min_parallel_index_scan_size;
+RESET min_parallel_table_scan_size;
 DROP TABLE pvactst;
 DROP TABLE pvactst2;
 -- INDEX_CLEANUP option
diff --git a/src/test/regress/sql/vacuum.sql b/src/test/regress/sql/vacuum.sql
index 058add027f1..3f156147a70 100644
--- a/src/test/regress/sql/vacuum.sql
+++ b/src/test/regress/sql/vacuum.sql
@@ -129,6 +129,12 @@ VACUUM (PARALLEL 2) pvactst;
 UPDATE pvactst SET i = i WHERE i < 1000;
 VACUUM (PARALLEL 0) pvactst; -- disable parallel vacuum
 
+-- VACUUM invokes parallel heap vacuum.
+SET min_parallel_table_scan_size to 0;
+VACUUM (PARALLEL 2, FREEZE) pvactst2;
+UPDATE pvactst2 SET i = i WHERE i < 1000;
+VACUUM (PARALLEL 1) pvactst2;
+
 VACUUM (PARALLEL -1) pvactst; -- error
 VACUUM (PARALLEL 2, INDEX_CLEANUP FALSE) pvactst;
 VACUUM (PARALLEL 2, FULL TRUE) pvactst; -- error, cannot use both PARALLEL and FULL
@@ -148,6 +154,7 @@ CREATE INDEX tmp_idx1 ON tmp (a);
 VACUUM (PARALLEL 1, FULL FALSE) tmp; -- parallel vacuum disabled for temp tables
 VACUUM (PARALLEL 0, FULL TRUE) tmp; -- can specify parallel disabled (even though that's implied by FULL)
 RESET min_parallel_index_scan_size;
+RESET min_parallel_table_scan_size;
 DROP TABLE pvactst;
 DROP TABLE pvactst2;
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index b8bc8bd3c22..b7b0ae8d9b8 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1935,6 +1935,10 @@ PLpgSQL_type
 PLpgSQL_type_type
 PLpgSQL_var
 PLpgSQL_variable
+PLVLeader
+PLVScanWorkerState
+PLVShared
+PLVState
 PLwdatum
 PLword
 PLyArrayToOb
-- 
2.43.5

