Add more parallel vacuum tests.

Author: MasahikoSawada

src/backend/access/heap/vacuumlazy.c

@@ -192,6 +192,7 @@
 #include "storage/freespace.h"
 #include "storage/lmgr.h"
 #include "storage/read_stream.h"
+#include "utils/injection_point.h"
 #include "utils/lsyscache.h"
 #include "utils/pg_rusage.h"
 #include "utils/timestamp.h"
@@ -466,6 +467,14 @@ typedef struct ParallelLVLeader
 	/* The number of workers launched for parallel lazy heap scan */
 	int			nworkers_launched;
 
+	/*
+	 * Will the leader participate to parallel lazy heap scan?
+	 *
+	 * This is a parameter for testing and always true unless it is disabled
+	 * explicitly by the injection point.
+	 */
+	bool		leaderparticipate;
+
 	/*
 	 * These fields point to the arrays of all per-worker scan states stored
 	 * in DSM.
@@ -2251,7 +2260,8 @@ do_parallel_lazy_scan_heap(LVRelState *vacrel)
 		 * retrieving new blocks for the read stream once the space of
 		 * dead_items TIDs exceeds the limit.
 		 */
-		do_lazy_scan_heap(vacrel, false);
+		if (vacrel->leader->leaderparticipate)
+			do_lazy_scan_heap(vacrel, false);
 
 		/* Wait for parallel workers to finish and gather scan results */
 		parallel_lazy_scan_heap_end(vacrel);
@@ -4543,6 +4553,7 @@ heap_parallel_vacuum_estimate(Relation rel, ParallelContext *pcxt, int nworkers,
 {
 	LVRelState *vacrel = (LVRelState *) state;
 	Size		size = 0;
+	bool		leaderparticipate = true;
 
 	vacrel->leader = palloc(sizeof(ParallelLVLeader));
 
@@ -4567,6 +4578,12 @@ heap_parallel_vacuum_estimate(Relation rel, ParallelContext *pcxt, int nworkers,
 	vacrel->leader->scandata_len = mul_size(sizeof(LVScanData), nworkers);
 	shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->scandata_len);
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+#ifdef USE_INJECTION_POINTS
+	if (IS_INJECTION_POINT_ATTACHED("parallel-heap-vacuum-disable-leader-participation"))
+		leaderparticipate = false;
+#endif
+	vacrel->leader->leaderparticipate = leaderparticipate;
 }
 
 /*
@@ -4604,7 +4621,8 @@ heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt, int nworker
 
 	/* including the leader too */
 	shared->eager_scan_remaining_successes_per_worker =
-		vacrel->eager_scan_remaining_successes / (nworkers + 1);
+		vacrel->eager_scan_remaining_successes /
+		(vacrel->leader->leaderparticipate ? nworkers + 1 : nworkers);
 
 	shm_toc_insert(pcxt->toc, PARALLEL_LV_KEY_SHARED, shared);
 	vacrel->plvstate->shared = shared;

src/backend/commands/vacuumparallel.c

@@ -39,6 +39,7 @@
 #include "pgstat.h"
 #include "storage/bufmgr.h"
 #include "tcop/tcopprot.h"
+#include "utils/injection_point.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
 
@@ -1035,14 +1036,28 @@ parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans,
 int
 parallel_vacuum_collect_dead_items_begin(ParallelVacuumState *pvs)
 {
+	int			nworkers = pvs->nworkers_for_table;
+#ifdef USE_INJECTION_POINTS
+	static int	ntimes = 0;
+#endif
+
 	Assert(!IsParallelWorker());
 
-	if (pvs->nworkers_for_table == 0)
+	if (nworkers == 0)
 		return 0;
 
 	/* Start parallel vacuum workers for collecting dead items */
-	Assert(pvs->nworkers_for_table <= pvs->pcxt->nworkers);
-	parallel_vacuum_begin_work_phase(pvs, pvs->nworkers_for_table,
+	Assert(nworkers <= pvs->pcxt->nworkers);
+
+#ifdef USE_INJECTION_POINTS
+	if (IS_INJECTION_POINT_ATTACHED("parallel-vacuum-ramp-down-workers"))
+	{
+		nworkers = pvs->nworkers_for_table - Min(ntimes, pvs->nworkers_for_table);
+		ntimes++;
+	}
+#endif
+
+	parallel_vacuum_begin_work_phase(pvs, nworkers,
 									 PV_WORK_PHASE_COLLECT_DEAD_ITEMS);
 
 	/* Include the worker count for the leader itself */

src/test/modules/injection_points/t/002_parallel_vacuum.pl

@@ -0,0 +1,97 @@
+
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# Tests for parallel heap vacuum.
+
+use strict;
+use warnings FATAL => 'all';
+use locale;
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Test persistency of statistics generated for injection points.
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+my $node = PostgreSQL::Test::Cluster->new('master');
+$node->init;
+$node->start;
+$node->safe_psql('postgres', qq[create extension injection_points;]);
+
+$node->safe_psql('postgres', qq[
+create table t (i int) with (autovacuum_enabled = off);
+create index on t (i);
+		 ]);
+my $nrows = 1_000_000;
+my $first = int($nrows * rand());
+my $second = $nrows - $first;
+
+my $psql = $node->background_psql('postgres', on_error_stop => 0);
+
+# Begin the transaciton that holds xmin.
+$psql->query_safe('begin; select pg_current_xact_id();');
+
+# consume some xids
+$node->safe_psql('postgres', qq[
+select pg_current_xact_id();
+select pg_current_xact_id();
+select pg_current_xact_id();
+select pg_current_xact_id();
+select pg_current_xact_id();
+		 ]);
+
+# While inserting $nrows tuples into the table with an older XID,
+# we inject some tuples with a newer XID filling one page somewhere
+# in the table.
+
+# Insert the first part of rows.
+$psql->query_safe(qq[insert into t select generate_series(1, $first);]);
+
+# Insert some rows with a newer XID, which needs to fill at least
+# one page to prevent the page from begin frozen in the following
+# vacuum.
+my $xid = $node->safe_psql('postgres', qq[
+begin;
+insert into t select 0 from generate_series(1, 300);
+select pg_current_xact_id()::xid;
+commit;
+]);
+
+# Insert remaining rows and commit.
+$psql->query_safe(qq[insert into t select generate_series($first, $nrows);]);
+$psql->query_safe(qq[commit;]);
+
+# Delete some rows.
+$node->safe_psql('postgres', qq[delete from t where i between 1 and 20000;]);
+
+# Execute parallel vacuum that freezes all rows except for the
+# tuple inserted by $psql. We should update the relfrozenxid up to
+# that XID. Setting a lower value to maintenance_work_mem invokes
+# multiple rounds of heap scanning and the number of parallel workers
+# will ramp-down thanks to the injection points.
+$node->safe_psql('postgres', qq[
+set vacuum_freeze_min_age to 5;
+set max_parallel_maintenance_workers TO 5;
+set maintenance_work_mem TO 256;
+select injection_points_set_local();
+select injection_points_attach('parallel-vacuum-ramp-down-workers', 'notice');
+select injection_points_attach('parallel-heap-vacuum-disable-leader-participation', 'notice');
+vacuum (parallel 5, verbose) t;
+		 ]);
+
+is( $node->safe_psql('postgres', qq[select relfrozenxid from pg_class where relname = 't';]),
+    "$xid", "relfrozenxid is updated as expected");
+
+# Check if we have successfully frozen the table in the previous
+# vacuum by scanning all tuples.
+$node->safe_psql('postgres', qq[vacuum (freeze, parallel 0, verbose, disable_page_skipping) t;]);
+is( $node->safe_psql('postgres', qq[select $xid < relfrozenxid::text::int from pg_class where relname = 't';]),
+    "t", "all rows are frozen");
+
+$node->stop;
+done_testing();
+