From 92cd53dff4e9a3da1278e7b666c15c03132c434d 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 v4 4/4] 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:
---
 src/backend/access/heap/vacuumlazy.c | 175 +++++++++++++++++++--------
 1 file changed, 122 insertions(+), 53 deletions(-)

diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 10991666e0b..1ab34732833 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -158,6 +158,7 @@ typedef struct LVRelScanStats
 	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 */
@@ -186,11 +187,15 @@ typedef struct PHVShared
 	MultiXactId NewRelminMxid;
 
 	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 */
 	LVRelScanStats worker_scan_stats[FLEXIBLE_ARRAY_MEMBER];
 }			PHVShared;
@@ -352,6 +357,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);
@@ -530,6 +536,7 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 	scan_stats->lpdead_item_pages = 0;
 	scan_stats->missed_dead_pages = 0;
 	scan_stats->nonempty_pages = 0;
+	scan_stats->vacuumed_pages = 0;
 
 	/* Initialize remaining counters (be tidy) */
 	scan_stats->tuples_deleted = 0;
@@ -2362,46 +2369,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;
@@ -2439,26 +2414,100 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 
 		UnlockReleaseBuffer(buf);
 		RecordPageWithFreeSpace(vacrel->rel, blkno, freespace);
-		vacuumed_pages++;
+		vacrel->scan_stats->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;
+
+	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_stats->vacuumed_pages = 0;
+
+	if (ParallelHeapVacuumIsActive(vacrel))
+	{
+		PHVState   *phvstate = vacrel->phvstate;
+
+		iter = TidStoreBeginIterateShared(vacrel->dead_items);
+
+		phvstate->shared->do_heap_vacuum = true;
+		phvstate->shared->shared_iter_handle = TidStoreGetSharedIterHandle(iter);
+
+		/* launch workers */
+		vacrel->phvstate->nworkers_launched = parallel_vacuum_table_scan_begin(vacrel->pvs);
+	}
+	else
+		iter = TidStoreBeginIterate(vacrel->dead_items);
+
+	/* do the real work */
+	do_lazy_vacuum_heap_rel(vacrel, iter);
+
+	if (ParallelHeapVacuumIsActive(vacrel))
+	{
+		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++)
+		{
+			LVRelScanStats *ss = &(phvstate->shared->worker_scan_stats[i]);
+
+			vacrel->scan_stats->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_stats->lpdead_items &&
-			vacuumed_pages == vacrel->scan_stats->lpdead_item_pages));
+			vacrel->scan_stats->vacuumed_pages == vacrel->scan_stats->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_stats->vacuumed_pages)));
 
 	/* Revert to the previous phase information for error traceback */
 	restore_vacuum_error_info(vacrel, &saved_err_info);
@@ -3514,6 +3563,7 @@ heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt,
 	shared->NewRelfrozenXid = vacrel->scan_stats->NewRelfrozenXid;
 	shared->NewRelminMxid = vacrel->scan_stats->NewRelminMxid;
 	shared->skippedallvis = vacrel->scan_stats->skippedallvis;
+	shared->do_index_vacuuming = vacrel->do_index_vacuuming;
 
 	/*
 	 * XXX: we copy the contents of vistest to the shared area, but in order
@@ -3566,7 +3616,6 @@ heap_parallel_vacuum_scan_worker(Relation rel, ParallelVacuumState *pvs,
 	PHVScanWorkerState *scanstate;
 	LVRelScanStats *scan_stats;
 	ErrorContextCallback errcallback;
-	bool		scan_done;
 
 	phvstate = palloc(sizeof(PHVState));
 
@@ -3603,10 +3652,11 @@ heap_parallel_vacuum_scan_worker(Relation rel, ParallelVacuumState *pvs,
 	/* initialize per-worker relation statistics */
 	MemSet(scan_stats, 0, sizeof(LVRelScanStats));
 
-	/* Set fields necessary for heap scan */
+	/* Set fields necessary for heap scan and vacuum */
 	vacrel.scan_stats->NewRelfrozenXid = shared->NewRelfrozenXid;
 	vacrel.scan_stats->NewRelminMxid = shared->NewRelminMxid;
 	vacrel.scan_stats->skippedallvis = shared->skippedallvis;
+	vacrel.do_index_vacuuming = shared->do_index_vacuuming;
 
 	/* Initialize the per-worker scan state if not yet */
 	if (!phvstate->myscanstate->initialized)
@@ -3628,25 +3678,44 @@ heap_parallel_vacuum_scan_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;
 }
 
 /*
@@ -3736,7 +3805,6 @@ parallel_heap_vacuum_gather_scan_stats(LVRelState *vacrel)
 		vacrel->scan_stats->frozen_pages += ss->frozen_pages;
 		vacrel->scan_stats->lpdead_item_pages += ss->lpdead_item_pages;
 		vacrel->scan_stats->missed_dead_pages += ss->missed_dead_pages;
-		vacrel->scan_stats->vacuumed_pages += ss->vacuumed_pages;
 		vacrel->scan_stats->tuples_deleted += ss->tuples_deleted;
 		vacrel->scan_stats->tuples_frozen += ss->tuples_frozen;
 		vacrel->scan_stats->lpdead_items += ss->lpdead_items;
@@ -3774,6 +3842,7 @@ do_parallel_lazy_scan_heap(LVRelState *vacrel)
 	Assert(!IsParallelWorker());
 
 	/* launcher workers */
+	vacrel->phvstate->shared->do_heap_vacuum = false;
 	vacrel->phvstate->nworkers_launched = parallel_vacuum_table_scan_begin(vacrel->pvs);
 
 	/* initialize parallel scan description to join as a worker */
-- 
2.43.5

