From 8a0e97565e6bcc1b952de0d2b3034a7dce35a62d Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Thu, 24 Oct 2024 17:37:45 -0700
Subject: [PATCH v5 8/8] Support parallel heap vacuum during lazy vacuum.

This commit further extends parallel vacuum to perform the heap vacuum
phase with parallel workers. It leverages the shared TidStore iteration.

Author:
Reviewed-by:
Discussion: https://postgr.es/m/
Backpatch-through:
---
 doc/src/sgml/ref/vacuum.sgml          |  17 +-
 src/backend/access/heap/vacuumlazy.c  | 280 +++++++++++++++++++-------
 src/backend/commands/vacuumparallel.c |  10 +-
 src/include/commands/vacuum.h         |   2 +-
 4 files changed, 223 insertions(+), 86 deletions(-)

diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml
index aae0bbcd577..104157b5a56 100644
--- a/doc/src/sgml/ref/vacuum.sgml
+++ b/doc/src/sgml/ref/vacuum.sgml
@@ -278,20 +278,21 @@ VACUUM [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] [ <re
     <term><literal>PARALLEL</literal></term>
     <listitem>
       <para>
-       Perform scanning heap, index vacuum, and index cleanup phases of
-       <command>VACUUM</command> in parallel using
+       Perform scanning heap, vacuuming 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.
+       heap and vacuuming 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
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 2e70bc68d2c..67516391d89 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -20,6 +20,41 @@
  * that there only needs to be one call to lazy_vacuum, after the initial pass
  * completes.
  *
+ * Parallel Vacuum
+ * ----------------
+ * Lazy vacuum on heap tables supports parallel processing for three vacuum
+ * phases: scanning heap, vacuuming indexes, and vacuuming heap. Before the
+ * scanning heap phase, 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 three phases.
+ *
+ * We could require different number of parallel vacuum workers for each phase
+ * for various factors such as table size, number of indexes, and the number
+ * of pages having dead tuples. 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 dead_items TIDs is
+ * close to overrunning the available space during parallel heap scan, parallel
+ * workers exit and leader process gathers the scan results. Then, it performs
+ * a round of index and heap 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 could launch fewer workers 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
+ * parallel_heap_complete_unfinished_scan()).
+ *
+ * At the beginning of the vacuuming heap phase, the leader launches parallel
+ * workers and initiates the shared iteration on the shared TID store. At the
+ * end of the phase, the leader process waits for all workers to finish and gather
+ * the workers' results.
+ *
+ *
  * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
@@ -172,6 +207,7 @@ typedef struct LVRelScanState
 	BlockNumber lpdead_item_pages;	/* # pages with LP_DEAD items */
 	BlockNumber missed_dead_pages;	/* # pages with missed dead tuples */
 	BlockNumber nonempty_pages; /* actually, last nonempty page + 1 */
+	BlockNumber vacuumed_pages; /* # pages vacuumed in one second pass */
 
 	/* Counters that follow are only for scanned_pages */
 	int64		tuples_deleted; /* # deleted from table */
@@ -205,11 +241,15 @@ typedef struct PHVShared
 	 * The final value is OR of worker's skippedallvis.
 	 */
 	bool		skippedallvis;
+	bool		do_index_vacuuming;
 
 	/* VACUUM operation's cutoffs for freezing and pruning */
 	struct VacuumCutoffs cutoffs;
 	GlobalVisState vistest;
 
+	dsa_pointer shared_iter_handle;
+	bool		do_heap_vacuum;
+
 	/* per-worker scan stats for parallel heap vacuum scan */
 	LVRelScanState worker_scan_state[FLEXIBLE_ARRAY_MEMBER];
 } PHVShared;
@@ -257,6 +297,14 @@ typedef struct PHVState
 	/* Assigned per-worker scan state */
 	PHVScanWorkerState *myscanstate;
 
+	/*
+	 * The number of parallel workers to launch for parallel heap scanning.
+	 * Note that the number of parallel workers for parallel heap vacuuming
+	 * could vary but is less than num_heapscan_workers. So this works also as
+	 * the maximum number of workers for parallel heap scanning and vacuuming.
+	 */
+	int			num_heapscan_workers;
+
 	/*
 	 * All blocks up to this value has been scanned, i.e. the minimum of all
 	 * PHVScanWorkerState->last_blkno. This field is updated by
@@ -374,6 +422,7 @@ static bool lazy_scan_noprune(LVRelState *vacrel, Buffer buf,
 static void lazy_vacuum(LVRelState *vacrel);
 static bool lazy_vacuum_all_indexes(LVRelState *vacrel);
 static void lazy_vacuum_heap_rel(LVRelState *vacrel);
+static void do_lazy_vacuum_heap_rel(LVRelState *vacrel, TidStoreIter *iter);
 static void lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno,
 								  Buffer buffer, OffsetNumber *deadoffsets,
 								  int num_offsets, Buffer vmbuffer);
@@ -404,6 +453,7 @@ static void do_parallel_lazy_scan_heap(LVRelState *vacrel);
 static void parallel_heap_vacuum_compute_min_scanned_blkno(LVRelState *vacrel);
 static void parallel_heap_vacuum_gather_scan_results(LVRelState *vacrel);
 static void parallel_heap_complete_unfinished_scan(LVRelState *vacrel);
+static int	compute_heap_vacuum_parallel_workers(Relation rel, BlockNumber nblocks);
 
 static void vacuum_error_callback(void *arg);
 static void update_vacuum_error_info(LVRelState *vacrel,
@@ -551,6 +601,7 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 	scan_state->lpdead_item_pages = 0;
 	scan_state->missed_dead_pages = 0;
 	scan_state->nonempty_pages = 0;
+	scan_state->vacuumed_pages = 0;
 	scan_state->tuples_deleted = 0;
 	scan_state->tuples_frozen = 0;
 	scan_state->lpdead_items = 0;
@@ -2456,46 +2507,14 @@ lazy_vacuum_all_indexes(LVRelState *vacrel)
 	return allindexes;
 }
 
-/*
- *	lazy_vacuum_heap_rel() -- second pass over the heap for two pass strategy
- *
- * This routine marks LP_DEAD items in vacrel->dead_items as LP_UNUSED. Pages
- * that never had lazy_scan_prune record LP_DEAD items are not visited at all.
- *
- * We may also be able to truncate the line pointer array of the heap pages we
- * visit.  If there is a contiguous group of LP_UNUSED items at the end of the
- * array, it can be reclaimed as free space.  These LP_UNUSED items usually
- * start out as LP_DEAD items recorded by lazy_scan_prune (we set items from
- * each page to LP_UNUSED, and then consider if it's possible to truncate the
- * page's line pointer array).
- *
- * Note: the reason for doing this as a second pass is we cannot remove the
- * tuples until we've removed their index entries, and we want to process
- * index entry removal in batches as large as possible.
- */
 static void
-lazy_vacuum_heap_rel(LVRelState *vacrel)
+do_lazy_vacuum_heap_rel(LVRelState *vacrel, TidStoreIter *iter)
 {
-	BlockNumber vacuumed_pages = 0;
 	Buffer		vmbuffer = InvalidBuffer;
-	LVSavedErrInfo saved_err_info;
-	TidStoreIter *iter;
-	TidStoreIterResult *iter_result;
-
-	Assert(vacrel->do_index_vacuuming);
-	Assert(vacrel->do_index_cleanup);
-	Assert(vacrel->num_index_scans > 0);
-
-	/* Report that we are now vacuuming the heap */
-	pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
-								 PROGRESS_VACUUM_PHASE_VACUUM_HEAP);
 
-	/* Update error traceback information */
-	update_vacuum_error_info(vacrel, &saved_err_info,
-							 VACUUM_ERRCB_PHASE_VACUUM_HEAP,
-							 InvalidBlockNumber, InvalidOffsetNumber);
+	/* LVSavedErrInfo saved_err_info; */
+	TidStoreIterResult *iter_result;
 
-	iter = TidStoreBeginIterate(vacrel->dead_items);
 	while ((iter_result = TidStoreIterateNext(iter)) != NULL)
 	{
 		BlockNumber blkno;
@@ -2533,26 +2552,106 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 
 		UnlockReleaseBuffer(buf);
 		RecordPageWithFreeSpace(vacrel->rel, blkno, freespace);
-		vacuumed_pages++;
+		vacrel->scan_state->vacuumed_pages++;
 	}
-	TidStoreEndIterate(iter);
 
 	vacrel->blkno = InvalidBlockNumber;
 	if (BufferIsValid(vmbuffer))
 		ReleaseBuffer(vmbuffer);
 
+}
+
+/*
+ *	lazy_vacuum_heap_rel() -- second pass over the heap for two pass strategy
+ *
+ * This routine marks LP_DEAD items in vacrel->dead_items as LP_UNUSED. Pages
+ * that never had lazy_scan_prune record LP_DEAD items are not visited at all.
+ *
+ * We may also be able to truncate the line pointer array of the heap pages we
+ * visit.  If there is a contiguous group of LP_UNUSED items at the end of the
+ * array, it can be reclaimed as free space.  These LP_UNUSED items usually
+ * start out as LP_DEAD items recorded by lazy_scan_prune (we set items from
+ * each page to LP_UNUSED, and then consider if it's possible to truncate the
+ * page's line pointer array).
+ *
+ * Note: the reason for doing this as a second pass is we cannot remove the
+ * tuples until we've removed their index entries, and we want to process
+ * index entry removal in batches as large as possible.
+ */
+static void
+lazy_vacuum_heap_rel(LVRelState *vacrel)
+{
+	LVSavedErrInfo saved_err_info;
+	TidStoreIter *iter;
+	int			nworkers = 0;
+
+	Assert(vacrel->do_index_vacuuming);
+	Assert(vacrel->do_index_cleanup);
+	Assert(vacrel->num_index_scans > 0);
+
+	/* Report that we are now vacuuming the heap */
+	pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
+								 PROGRESS_VACUUM_PHASE_VACUUM_HEAP);
+
+	/* Update error traceback information */
+	update_vacuum_error_info(vacrel, &saved_err_info,
+							 VACUUM_ERRCB_PHASE_VACUUM_HEAP,
+							 InvalidBlockNumber, InvalidOffsetNumber);
+
+	vacrel->scan_state->vacuumed_pages = 0;
+
+	/* Compute parallel workers required to scan blocks to vacuum */
+	if (ParallelHeapVacuumIsActive(vacrel))
+		nworkers = compute_heap_vacuum_parallel_workers(vacrel->rel,
+														TidStoreNumBlocks(vacrel->dead_items));
+
+	if (nworkers > 0)
+	{
+		PHVState   *phvstate = vacrel->phvstate;
+
+		iter = TidStoreBeginIterateShared(vacrel->dead_items);
+
+		/* launch workers */
+		phvstate->shared->do_heap_vacuum = true;
+		phvstate->shared->shared_iter_handle = TidStoreGetSharedIterHandle(iter);
+		phvstate->nworkers_launched = parallel_vacuum_table_scan_begin(vacrel->pvs,
+																	   nworkers);
+	}
+	else
+		iter = TidStoreBeginIterate(vacrel->dead_items);
+
+	/* do the real work */
+	do_lazy_vacuum_heap_rel(vacrel, iter);
+
+	if (ParallelHeapVacuumIsActive(vacrel) && nworkers > 0)
+	{
+		PHVState   *phvstate = vacrel->phvstate;
+
+		parallel_vacuum_table_scan_end(vacrel->pvs);
+
+		/* Gather the heap vacuum statistics that workers collected */
+		for (int i = 0; i < phvstate->nworkers_launched; i++)
+		{
+			LVRelScanState *ss = &(phvstate->shared->worker_scan_state[i]);
+
+			vacrel->scan_state->vacuumed_pages += ss->vacuumed_pages;
+		}
+	}
+
+	TidStoreEndIterate(iter);
+
 	/*
 	 * We set all LP_DEAD items from the first heap pass to LP_UNUSED during
 	 * the second heap pass.  No more, no less.
 	 */
 	Assert(vacrel->num_index_scans > 1 ||
 		   (vacrel->dead_items_info->num_items == vacrel->scan_state->lpdead_items &&
-			vacuumed_pages == vacrel->scan_state->lpdead_item_pages));
+			vacrel->scan_state->vacuumed_pages == vacrel->scan_state->lpdead_item_pages));
 
 	ereport(DEBUG2,
 			(errmsg("table \"%s\": removed %lld dead item identifiers in %u pages",
 					vacrel->relname, (long long) vacrel->dead_items_info->num_items,
-					vacuumed_pages)));
+					vacrel->scan_state->vacuumed_pages)));
 
 	/* Revert to the previous phase information for error traceback */
 	restore_vacuum_error_info(vacrel, &saved_err_info);
@@ -3261,6 +3360,11 @@ dead_items_alloc(LVRelState *vacrel, int nworkers)
 		{
 			vacrel->dead_items = parallel_vacuum_get_dead_items(vacrel->pvs,
 																&vacrel->dead_items_info);
+
+			if (ParallelHeapVacuumIsActive(vacrel))
+				vacrel->phvstate->num_heapscan_workers =
+					parallel_vacuum_get_nworkers_table(vacrel->pvs);
+
 			return;
 		}
 	}
@@ -3508,37 +3612,41 @@ update_relstats_all_indexes(LVRelState *vacrel)
  *
  * The calculation logic is borrowed from compute_parallel_worker().
  */
-int
-heap_parallel_vacuum_compute_workers(Relation rel, int nrequested)
+static int
+compute_heap_vacuum_parallel_workers(Relation rel, BlockNumber nblocks)
 {
 	int			parallel_workers = 0;
 	int			heap_parallel_threshold;
 	int			heap_pages;
 
-	if (nrequested == 0)
+	/*
+	 * Select the number of workers based on the log of the size of the
+	 * relation.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);
+	heap_pages = BlockNumberIsValid(nblocks) ?
+		nblocks : RelationGetNumberOfBlocks(rel);
+	while (heap_pages >= (BlockNumber) (heap_parallel_threshold * 3))
 	{
-		/*
-		 * Select the number of workers based on the log of the size of the
-		 * relation. 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);
-		heap_pages = RelationGetNumberOfBlocks(rel);
-		while (heap_pages >= (BlockNumber) (heap_parallel_threshold * 3))
-		{
-			parallel_workers++;
-			heap_parallel_threshold *= 3;
-			if (heap_parallel_threshold > INT_MAX / 3)
-				break;
-		}
+		parallel_workers++;
+		heap_parallel_threshold *= 3;
+		if (heap_parallel_threshold > INT_MAX / 3)
+			break;
 	}
-	else
-		parallel_workers = nrequested;
 
 	return parallel_workers;
 }
 
+int
+heap_parallel_vacuum_compute_workers(Relation rel, int nrequested)
+{
+	if (nrequested == 0)
+		return compute_heap_vacuum_parallel_workers(rel, InvalidBlockNumber);
+	else
+		return nrequested;
+}
+
 /* Estimate shared memory sizes required for parallel heap vacuum */
 static inline void
 heap_parallel_estimate_shared_memory_size(Relation rel, int nworkers, Size *pscan_len,
@@ -3620,6 +3728,7 @@ heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt,
 	shared->NewRelfrozenXid = vacrel->scan_state->NewRelfrozenXid;
 	shared->NewRelminMxid = vacrel->scan_state->NewRelminMxid;
 	shared->skippedallvis = vacrel->scan_state->skippedallvis;
+	shared->do_index_vacuuming = vacrel->do_index_vacuuming;
 
 	/*
 	 * XXX: we copy the contents of vistest to the shared area, but in order
@@ -3672,7 +3781,6 @@ heap_parallel_vacuum_worker(Relation rel, ParallelVacuumState *pvs,
 	PHVScanWorkerState *scanstate;
 	LVRelScanState *scan_state;
 	ErrorContextCallback errcallback;
-	bool		scan_done;
 
 	phvstate = palloc(sizeof(PHVState));
 
@@ -3709,10 +3817,11 @@ heap_parallel_vacuum_worker(Relation rel, ParallelVacuumState *pvs,
 	/* initialize per-worker relation statistics */
 	MemSet(scan_state, 0, sizeof(LVRelScanState));
 
-	/* Set fields necessary for heap scan */
+	/* Set fields necessary for heap scan and vacuum */
 	vacrel.scan_state->NewRelfrozenXid = shared->NewRelfrozenXid;
 	vacrel.scan_state->NewRelminMxid = shared->NewRelminMxid;
 	vacrel.scan_state->skippedallvis = shared->skippedallvis;
+	vacrel.do_index_vacuuming = shared->do_index_vacuuming;
 
 	/* Initialize the per-worker scan state if not yet */
 	if (!phvstate->myscanstate->initialized)
@@ -3734,25 +3843,44 @@ heap_parallel_vacuum_worker(Relation rel, ParallelVacuumState *pvs,
 	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;
 
-	scan_done = do_lazy_scan_heap(&vacrel);
+	if (shared->do_heap_vacuum)
+	{
+		TidStoreIter *iter;
+
+		iter = TidStoreAttachIterateShared(vacrel.dead_items, shared->shared_iter_handle);
+
+		/* Join parallel heap vacuum */
+		vacrel.phase = VACUUM_ERRCB_PHASE_VACUUM_HEAP;
+		do_lazy_vacuum_heap_rel(&vacrel, iter);
+
+		TidStoreEndIterate(iter);
+	}
+	else
+	{
+		bool		scan_done;
+
+		/* Join parallel heap scan */
+		vacrel.phase = VACUUM_ERRCB_PHASE_SCAN_HEAP;
+		scan_done = do_lazy_scan_heap(&vacrel);
+
+		/*
+		 * If the leader or a worker finishes the heap scan because dead_items
+		 * TIDs is close to the limit, it might have some allocated blocks in
+		 * its scan state. Since this scan state might 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.
+		 */
+		phvstate->myscanstate->maybe_have_blocks = !scan_done;
+	}
 
 	/* Pop the error context stack */
 	error_context_stack = errcallback.previous;
-
-	/*
-	 * If the leader or a worker finishes the heap scan because dead_items
-	 * TIDs is close to the limit, it might have some allocated blocks in its
-	 * scan state. Since this scan state might 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.
-	 */
-	phvstate->myscanstate->maybe_have_blocks = !scan_done;
 }
 
 /*
@@ -3874,7 +4002,10 @@ do_parallel_lazy_scan_heap(LVRelState *vacrel)
 	Assert(!IsParallelWorker());
 
 	/* launcher workers */
-	vacrel->phvstate->nworkers_launched = parallel_vacuum_table_scan_begin(vacrel->pvs);
+	vacrel->phvstate->shared->do_heap_vacuum = false;
+	vacrel->phvstate->nworkers_launched =
+		parallel_vacuum_table_scan_begin(vacrel->pvs,
+										 vacrel->phvstate->num_heapscan_workers);
 
 	/* initialize parallel scan description to join as a worker */
 	scanstate = palloc0(sizeof(PHVScanWorkerState));
@@ -3933,7 +4064,8 @@ do_parallel_lazy_scan_heap(LVRelState *vacrel)
 
 		/* Re-launch workers to restart parallel heap scan */
 		vacrel->phvstate->nworkers_launched =
-			parallel_vacuum_table_scan_begin(vacrel->pvs);
+			parallel_vacuum_table_scan_begin(vacrel->pvs,
+											 vacrel->phvstate->num_heapscan_workers);
 	}
 
 	/*
diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c
index 3001be84ddf..fd897ddadf3 100644
--- a/src/backend/commands/vacuumparallel.c
+++ b/src/backend/commands/vacuumparallel.c
@@ -1054,8 +1054,10 @@ parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans,
  * table vacuum.
  */
 int
-parallel_vacuum_table_scan_begin(ParallelVacuumState *pvs)
+parallel_vacuum_table_scan_begin(ParallelVacuumState *pvs, int nworkers_request)
 {
+	int			nworkers;
+
 	Assert(!IsParallelWorker());
 
 	if (pvs->shared->nworkers_for_table == 0)
@@ -1069,11 +1071,13 @@ parallel_vacuum_table_scan_begin(ParallelVacuumState *pvs)
 	if (pvs->num_table_scans > 0)
 		ReinitializeParallelDSM(pvs->pcxt);
 
+	nworkers = Min(nworkers_request, pvs->shared->nworkers_for_table);
+
 	/*
 	 * The number of workers might vary between table vacuum and index
 	 * processing
 	 */
-	ReinitializeParallelWorkers(pvs->pcxt, pvs->shared->nworkers_for_table);
+	ReinitializeParallelWorkers(pvs->pcxt, nworkers);
 	LaunchParallelWorkers(pvs->pcxt);
 
 	if (pvs->pcxt->nworkers_launched > 0)
@@ -1097,7 +1101,7 @@ parallel_vacuum_table_scan_begin(ParallelVacuumState *pvs)
 			(errmsg(ngettext("launched %d parallel vacuum worker for table processing (planned: %d)",
 							 "launched %d parallel vacuum workers for table processing (planned: %d)",
 							 pvs->pcxt->nworkers_launched),
-					pvs->pcxt->nworkers_launched, pvs->shared->nworkers_for_table)));
+					pvs->pcxt->nworkers_launched, nworkers)));
 
 	return pvs->pcxt->nworkers_launched;
 }
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index b70e50133fa..ab6b6cde759 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -371,7 +371,7 @@ extern void parallel_vacuum_bulkdel_all_indexes(ParallelVacuumState *pvs,
 extern void parallel_vacuum_cleanup_all_indexes(ParallelVacuumState *pvs,
 												long num_table_tuples,
 												bool estimated_count);
-extern int	parallel_vacuum_table_scan_begin(ParallelVacuumState *pvs);
+extern int	parallel_vacuum_table_scan_begin(ParallelVacuumState *pvs, int nworkers_request);
 extern void parallel_vacuum_table_scan_end(ParallelVacuumState *pvs);
 extern int	parallel_vacuum_get_nworkers_table(ParallelVacuumState *pvs);
 extern int	parallel_vacuum_get_nworkers_index(ParallelVacuumState *pvs);
-- 
2.43.5

