From 0ebda44e3d0f59ebd7b6ee26a5d93c6d53bf9c3c Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Fri, 13 Jun 2025 10:58:18 -0700 Subject: [PATCH v19 5/5] Add more parallel vacuum tests. --- src/backend/access/heap/vacuumlazy.c | 22 ++++- src/backend/commands/vacuumparallel.c | 21 +++- .../injection_points/t/002_parallel_vacuum.pl | 97 +++++++++++++++++++ 3 files changed, 135 insertions(+), 5 deletions(-) create mode 100644 src/test/modules/injection_points/t/002_parallel_vacuum.pl diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index e6ca9c60e8a..6b7b22816b9 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/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; diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index 49e43b95132..7f0869ee4dc 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/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 */ diff --git a/src/test/modules/injection_points/t/002_parallel_vacuum.pl b/src/test/modules/injection_points/t/002_parallel_vacuum.pl new file mode 100644 index 00000000000..f0ef33ed86b --- /dev/null +++ b/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(); + -- 2.51.0