From 2dbd64dcb55a88b925ba26fe7c6d48c9a714c02d Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Wed, 15 Jul 2026 23:43:53 +0000 Subject: [PATCH v20 2/2] Support parallel heap vacuum for collecting dead items Extend vacuum to scan the heap, pruning and freezing tuples and collecting dead item TIDs, using parallel workers. This adds to the existing parallel index vacuuming, so scanning the heap, vacuuming indexes, and cleaning up indexes can all run in parallel. The leader splits the heap into fixed-size block chunks and hands them out to workers, itself included, each claiming the next chunk atomically. The leader always participates as a worker. The heap scan and the index phases may use different worker counts, so workers are launched at the start of each phase and exit at its end. Freezing cutoffs are computed once by the leader and shared, so freezing decisions are identical across workers. Each worker prunes using its own visibility state, so workers may remove slightly different sets of recently-dead tuples, which is safe. Each worker tracks the oldest XID/MXID it observes and its own scan counters; the leader then combines these, taking the globally oldest XID/MXID to advance the relation's frozen markers. Eager scanning of all-visible pages for early freezing is distributed across workers too. When the dead-item store fills mid-scan, the heap scan pauses, the indexes and heap are vacuumed, and the scan resumes. Per-worker scan state is kept in shared memory so a worker can resume where it left off, and the leader finishes any chunk left partly scanned when a later round runs with fewer workers. Vacuum is entered through the table AM, so the generic parallel vacuum code must not call back up into the table AM. Instead an access method supplies a ParallelVacuumCallbacks struct to parallel_vacuum_init() and the generic code calls it without knowing the access method, like logical decoding's output plugin callbacks. Worker callbacks cannot be passed as pointers across processes, so the leader stores the access method name in shared memory and each worker resolves the callbacks locally by that name, as the executor does for custom scan methods. The heap scan itself is what gets parallelized here. The later pass that reaps dead line pointers from the heap is not. Testing showed it is already the fastest phase, so parallelizing it gave little overall benefit. A TAP test exercises this using injection points, including worker ramp-down across multiple scan rounds and the leader completing the unfinished scans, plus the planned worker counts for different PARALLEL requests and table sizes. Performance numbers to follow. Author: Masahiko Sawada Author: Bharath Rupireddy Reviewed-by: Amit Kapila Reviewed-by: Hayato Kuroda Reviewed-by: Peter Smith Reviewed-by: Tomas Vondra Reviewed-by: Dilip Kumar Reviewed-by: Melanie Plageman Reviewed-by: Andres Freund Discussion: https://postgr.es/m/CAD21AoAEfCNv-GgaDheDJ+s-p_Lv1H24AiJeNoPGCmZNSwL1YA@mail.gmail.com --- doc/src/sgml/ref/vacuum.sgml | 40 +- src/backend/access/heap/vacuumlazy.c | 1242 +++++++++++++++-- src/backend/commands/vacuumparallel.c | 515 +++++-- src/include/access/heapam.h | 8 + src/include/commands/vacuum.h | 84 +- src/test/modules/injection_points/Makefile | 2 + src/test/modules/injection_points/meson.build | 5 + .../t/parallel_heap_vacuum.pl | 138 ++ src/test/regress/expected/vacuum_parallel.out | 7 + src/test/regress/sql/vacuum_parallel.sql | 8 + src/tools/pgindent/typedefs.list | 8 + 11 files changed, 1848 insertions(+), 209 deletions(-) create mode 100644 src/test/modules/injection_points/t/parallel_heap_vacuum.pl diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml index 38ee973ea05..b07958ee93c 100644 --- a/doc/src/sgml/ref/vacuum.sgml +++ b/doc/src/sgml/ref/vacuum.sgml @@ -266,9 +266,43 @@ VACUUM [ ( option [, ...] ) ] [ PARALLEL - Specifies the maximum number of parallel workers that can be used - for , which is further limited - by . This + Perform the scanning heap, + vacuuming indexes, and + cleaning up indexes phases of + VACUUM in parallel using + integer background workers + (see for details on each phase). + + + The number of workers used for the scanning heap + phase is determined based on the size of the table. A table can + participate in parallel scanning heap if and only if + its size is more than + . During this phase, + the 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. + + + The number of workers used for the vacuuming indexes + and cleaning up indexes phases is equal to the number + of indexes on the table that support parallel vacuum. An index can + participate in parallel vacuum if and only if its size is more than + . Only one worker can + be used per index. So parallel workers for these phases are launched only + when there are at least 2 indexes in the table. + + + Workers for vacuum are launched before the start of each phase and exit + at the end of the phase. The number of workers used in each phase is + capped by the number specified with the PARALLEL + option, if any, and in turn by + . It is not + guaranteed that the number of parallel workers specified by + integer will be used during + execution; a vacuum may run with fewer workers than specified, or with + none at all. These behaviors might change in a future release. This option can't be used with the FULL option. diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index affd7221c74..7226e3031a0 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -99,6 +99,44 @@ * After pruning and freezing, pages that are newly all-visible and all-frozen * are marked as such in the visibility map. * + * Parallel Vacuum: + * + * Vacuum on heap tables supports parallel processing for phase I, which + * scans heap pages, prunes and freezes tuples, and saves dead tuples' TIDs in + * the TID store, and phase II, which deletes the dead index entries referenced + * in the TID store. Before starting phase I, we initialize the parallel vacuum + * state, ParallelVacuumState, and allocate the TID store in a DSA area if we + * can use parallel mode for either phase. + * + * We may need a different number of parallel workers for each phase, depending + * on factors such as the table size and the number of indexes. Parallel + * workers are launched at the beginning of each phase and exit at the end of + * it. + * + * Vacuum cutoffs are shared between the leader and worker processes, but each + * process uses its own GlobalVisState, so some workers may remove fewer tuples + * than others. During the parallel heap scan, each worker tracks the oldest + * existing XID and MXID it observes; after the scan, the leader computes the + * globally oldest XID and MXID while gathering each worker's scan results + * (see LVScanData). + * + * The parallel heap scan (phase I) is driven by ParallelLVScanDesc together + * with the read stream. The table is split into chunks that are distributed + * among the parallel workers. Because the read stream's look-ahead can leave + * pinned buffers, we cannot stop phase I abruptly when the dead_items TID + * space exceeds its limit. Instead, once the limit is reached, we stop + * requesting new blocks and process the pages already pinned until the read + * stream is exhausted. This may exceed the memory limit slightly, but not by + * much, because processing a few tens to hundreds of buffers does not + * substantially grow the dead_items TID space. Each worker's scan state, + * ParallelLVScanWorkerData, is stored in DSM so that a worker can resume + * phase I from where it left off. + * + * If the leader resumes phase I with fewer workers than the previous round, + * some blocks within a worker's chunk may remain un-scanned. The leader + * finishes any such unfinished scans at the end of the parallel heap scan + * (see parallel_lazy_scan_heap_complete()). + * * Dead TID Storage: * * The major space usage for vacuuming is storage for the dead tuple IDs that @@ -144,6 +182,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 +253,23 @@ */ #define PREFETCH_SIZE ((BlockNumber) 32) +/* + * DSM keys for parallel heap vacuum. Like the keys in vacuumparallel.c, these + * don't need to worry about conflicting with plan_node_id. But since they + * share a DSM segment with those keys, we use a high range here to avoid + * conflicting with the small integers used there. + */ +#define PARALLEL_LV_KEY_SHARED 0xFFFF0001 +#define PARALLEL_LV_KEY_SCANDESC 0xFFFF0002 +#define PARALLEL_LV_KEY_SCANWORKER 0xFFFF0003 +#define PARALLEL_LV_KEY_SCANDATA 0xFFFF0004 + /* * 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 @@ -249,6 +300,12 @@ typedef enum */ #define EAGER_SCAN_REGION_SIZE 4096 +/* + * Number of blocks each worker (including the leader) retrieves at a time + * during the parallel heap scan. + */ +#define PARALLEL_LV_CHUNK_SIZE 1024 + /* * Data and counters updated during lazy heap scan. */ @@ -298,6 +355,131 @@ typedef struct LVScanData bool skippedallvis; } LVScanData; +/* + * In DSM. Written once by the leader before workers launch; read-only in + * workers. + */ +typedef struct ParallelLVShared +{ + bool aggressive; + bool skipwithvm; + + /* The current oldest extant XID/MXID shared by the leader process */ + TransactionId NewRelfrozenXid; + MultiXactId NewRelminMxid; + + /* Cutoffs for freezing and pruning, computed once by the leader */ + struct VacuumCutoffs cutoffs; + + /* + * The first chunk size varies depending on the first eager scan region + * size. If eager scan is disabled, we use the default chunk size + * PARALLEL_LV_CHUNK_SIZE for the first chunk. + */ + BlockNumber initial_chunk_size; + + /* + * Per-chunk failure cap for eager scanning, the parallel vacuum analog of + * LVRelState.eager_scan_max_fails_per_region (chunks, not regions, are + * the unit of work distribution here). + */ + BlockNumber eager_scan_max_fails_per_chunk; + + /* + * Each worker's share of the total eager scan success budget + * (LVRelState.eager_scan_remaining_successes), divided among the workers. + * A worker seeds its local counter from this; the shared value itself is + * not decremented. + */ + BlockNumber eager_scan_remaining_successes_per_worker; +} ParallelLVShared; + +/* + * In DSM. Shared work-allocation cursor for the parallel heap scan; workers + * advance nallocated atomically to claim block ranges. + */ +typedef struct ParallelLVScanDesc +{ + /* Number of blocks in the heap at start of scan */ + BlockNumber nblocks; + + /* Number of blocks allocated to workers so far */ + pg_atomic_uint64 nallocated; +} ParallelLVScanDesc; + +/* + * In DSM, one entry per worker. Each worker writes only its own entry; the + * leader reads all entries to resume scans and gather results. + */ +typedef struct ParallelLVScanWorkerData +{ + bool inited; + + /* Current number of blocks into the scan */ + BlockNumber nallocated; + + /* Number of blocks per chunk */ + BlockNumber chunk_size; + + /* Number of blocks left in this chunk */ + uint32 chunk_remaining; + + /* The last processed block number */ + pg_atomic_uint32 last_blkno; + + /* Eager scan state for resuming the scan */ + BlockNumber remaining_fails_save; + BlockNumber remaining_successes_save; + BlockNumber next_region_start_save; +} ParallelLVScanWorkerData; + +/* + * Process-local (one per leader and per worker). Holds pointers into the + * shared (DSM) objects above. + */ +typedef struct ParallelLVState +{ + /* Shared static information */ + ParallelLVShared *shared; + + /* Parallel scan description shared among parallel workers */ + ParallelLVScanDesc *scan_desc; + + /* This worker's entry in the shared scan_work_array */ + ParallelLVScanWorkerData *scan_work; +} ParallelLVState; + +/* + * Process-local, leader only. Bookkeeping for setting up the DSM objects and + * for gathering each worker's results. + */ +typedef struct ParallelLVLeader +{ + /* Shared memory size for each shared object */ + Size shared_len; + Size scan_desc_len; + Size scan_work_len; + Size scan_data_len; + + /* The number of workers launched for parallel heap scan */ + int nworkers_launched; + + /* + * Will the leader participate in the parallel heap scan? + * + * This is a parameter for testing and is always true unless disabled + * explicitly by the injection point. + */ + bool leaderparticipates; + + /* + * These fields point to the arrays of all per-worker scan states stored + * in DSM. + */ + ParallelLVScanWorkerData *scan_work_array; + LVScanData *scan_data_array; +} ParallelLVLeader; + typedef struct LVRelState { /* Target heap relation and its indexes */ @@ -370,12 +552,28 @@ typedef struct LVRelState */ PVWorkerUsage worker_usage; + /* Last processed block number */ + BlockNumber last_blkno; + + /* 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 */ 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 heap vacuum working state */ + ParallelLVState *plvstate; + + /* + * The leader state for parallel heap vacuum. NULL for parallel workers. + */ + ParallelLVLeader *leader; + /* State related to managing eager scanning of all-visible pages */ /* @@ -435,12 +633,14 @@ typedef struct LVSavedErrInfo /* non-export function prototypes */ static void lazy_scan_heap(LVRelState *vacrel); +static void do_lazy_scan_heap(LVRelState *vacrel, bool check_mem_usage); static void heap_vacuum_eager_scan_setup(LVRelState *vacrel, const 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 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); @@ -451,6 +651,35 @@ static int lazy_scan_prune(LVRelState *vacrel, Buffer buf, static bool lazy_scan_noprune(LVRelState *vacrel, Buffer buf, BlockNumber blkno, Page page, bool *has_lpdead_items); + +/* Parallel heap scan (phase I): internal helpers driving the workers */ +static void do_parallel_lazy_scan_heap(LVRelState *vacrel); +static BlockNumber parallel_lazy_scan_compute_min_scan_block(LVRelState *vacrel); +static void parallel_lazy_scan_heap_complete(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_results(LVRelState *vacrel); +static void parallel_lazy_scan_init_scan_worker(ParallelLVScanWorkerData *scan_work, + BlockNumber initial_chunk_size); +static BlockNumber parallel_lazy_scan_get_nextpage(LVRelState *vacrel, Relation rel, + ParallelLVScanDesc *scan_desc, + ParallelLVScanWorkerData *scan_work); + +/* Table AM parallel-vacuum callbacks implemented by heap (see vacuum.h) */ +static int heap_parallel_vacuum_compute_workers(Relation rel, + int nworkers_requested, + void *state); +static void heap_parallel_vacuum_estimate(Relation rel, ParallelContext *pcxt, + int nworkers, void *state); +static void heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt, + int nworkers, void *state); +static void heap_parallel_vacuum_initialize_worker(Relation rel, + ParallelVacuumState *pvs, + ParallelWorkerContext *pwcxt, + void **state_out); +static void heap_parallel_vacuum_collect_dead_items(Relation rel, + ParallelVacuumState *pvs, + void *state); static void lazy_vacuum(LVRelState *vacrel); static bool lazy_vacuum_all_indexes(LVRelState *vacrel); static void lazy_vacuum_heap_rel(LVRelState *vacrel); @@ -475,6 +704,7 @@ static BlockNumber count_nondeletable_pages(LVRelState *vacrel, static void dead_items_alloc(LVRelState *vacrel, int nworkers); static void dead_items_add(LVRelState *vacrel, BlockNumber blkno, OffsetNumber *offsets, int num_offsets); +static bool dead_items_check_memory_limit(LVRelState *vacrel); static void dead_items_reset(LVRelState *vacrel); static void dead_items_cleanup(LVRelState *vacrel); @@ -771,6 +1001,7 @@ heap_vacuum_rel(Relation rel, const VacuumParams *params, vacrel->num_index_scans = 0; vacrel->num_dead_items_resets = 0; vacrel->total_dead_items_bytes = 0; + vacrel->next_fsm_block_to_vacuum = 0; vacrel->worker_usage.vacuum.nlaunched = 0; vacrel->worker_usage.vacuum.nplanned = 0; @@ -1275,13 +1506,7 @@ heap_vacuum_rel(Relation rel, const VacuumParams *params, static void lazy_scan_heap(LVRelState *vacrel) { - ReadStream *stream; - BlockNumber rel_pages = vacrel->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->rel_pages; const int initprog_index[] = { PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_TOTAL_HEAP_BLKS, @@ -1301,6 +1526,80 @@ lazy_scan_heap(LVRelState *vacrel) vacrel->next_unskippable_eager_scanned = false; vacrel->next_unskippable_vmbuffer = InvalidBuffer; + /* Do the actual work */ + if (ParallelHeapVacuumIsActive(vacrel)) + do_parallel_lazy_scan_heap(vacrel); + else + do_lazy_scan_heap(vacrel, true); + + /* + * 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->heap_scan_data->scanned_pages, + vacrel->heap_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->heap_scan_data->recently_dead_tuples + + vacrel->heap_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(). + * + * If check_mem_usage is true, we check the memory usage during the heap scan. + * If the space of dead_items TIDs exceeds the limit, we stop the lazy heap scan + * and invoke a cycle of index vacuuming and heap vacuuming, and then resume the + * scan. If it's false, we continue doing lazy heap scan until the read stream + * is exhausted. + */ +static void +do_lazy_scan_heap(LVRelState *vacrel, bool check_mem_usage) +{ + ReadStream *stream; + BlockNumber blkno = InvalidBlockNumber; + BlockNumber orig_eager_scan_success_limit = + vacrel->eager_scan_remaining_successes; /* for logging */ + Buffer vmbuffer = InvalidBuffer; + + /* + * We should not set check_mem_usage to false unless during parallel heap + * vacuum. + */ + Assert(check_mem_usage || ParallelHeapVacuumIsActive(vacrel)); + /* * Set up the read stream for vacuum's first pass through the heap. * @@ -1336,8 +1635,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 is done only by the leader process. */ - if (vacrel->heap_scan_data->scanned_pages > 0 && + if (!IsParallelWorker() && + vacrel->heap_scan_data->scanned_pages > 0 && vacrel->heap_scan_data->scanned_pages % FAILSAFE_EVERY_PAGES == 0) lazy_check_wraparound_failsafe(vacrel); @@ -1345,12 +1647,9 @@ lazy_scan_heap(LVRelState *vacrel) * Consider if we definitely have enough space to process TIDs on page * already. If we are close to overrunning the available space for * dead_items TIDs, pause and do a cycle of vacuuming before we tackle - * this page. However, let's force at least one page-worth of tuples - * to be stored as to ensure we do at least some work when the memory - * configured is so low that we run out before storing anything. + * this page. */ - if (vacrel->dead_items_info->num_items > 0 && - TidStoreMemoryUsage(vacrel->dead_items) > vacrel->dead_items_info->max_bytes) + if (check_mem_usage && dead_items_check_memory_limit(vacrel)) { /* * Before beginning index vacuuming, we release any pin we may @@ -1373,15 +1672,16 @@ 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); } + /* Read the next block to process */ buf = read_stream_next_buffer(stream, &per_buffer_data); /* The relation is exhausted. */ @@ -1391,7 +1691,7 @@ lazy_scan_heap(LVRelState *vacrel) was_eager_scanned = *((bool *) per_buffer_data); CheckBufferIsPinnedOnce(buf); page = BufferGetPage(buf); - blkno = BufferGetBlockNumber(buf); + blkno = vacrel->last_blkno = BufferGetBlockNumber(buf); vacrel->heap_scan_data->scanned_pages++; if (was_eager_scanned) @@ -1554,13 +1854,34 @@ lazy_scan_heap(LVRelState *vacrel) * visible on upper FSM pages. This is done after vacuuming if the * table has indexes. There will only be newly-freed space if we * held the cleanup lock and lazy_scan_prune() was called. + * + * During parallel heap scanning, only the leader process vacuums + * the FSM. However, we cannot vacuum the FSM for blocks up to + * 'blk' because there may be un-scanned blocks or blocks being + * processed by workers before this point. Instead, parallel + * workers advertise the block numbers they have just processed, + * and the leader vacuums the FSM up to the smallest block number + * among them. This approach ensures we vacuum the FSM for + * consecutive processed blocks. */ if (got_cleanup_lock && vacrel->nindexes == 0 && ndeleted > 0 && - 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, + if (IsParallelWorker()) + pg_atomic_write_u32(&(vacrel->plvstate->scan_work->last_blkno), blkno); - next_fsm_block_to_vacuum = blkno; + else + { + BlockNumber fsmvac_upto = blkno; + + if (ParallelHeapVacuumIsActive(vacrel)) + fsmvac_upto = parallel_lazy_scan_compute_min_scan_block(vacrel); + + FreeSpaceMapVacuumRange(vacrel->rel, vacrel->next_fsm_block_to_vacuum, + fsmvac_upto); + } + + vacrel->next_fsm_block_to_vacuum = blkno; } } else @@ -1571,50 +1892,7 @@ lazy_scan_heap(LVRelState *vacrel) if (BufferIsValid(vmbuffer)) 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. - */ - 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->heap_scan_data->scanned_pages, - vacrel->heap_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->heap_scan_data->recently_dead_tuples + - vacrel->heap_scan_data->missed_dead_tuples; - read_stream_end(stream); - - /* - * 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 > next_fsm_block_to_vacuum) - FreeSpaceMapVacuumRange(vacrel->rel, 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); } /* @@ -1628,7 +1906,8 @@ lazy_scan_heap(LVRelState *vacrel) * 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. + * remaining blocks or the space of dead_items TIDs reaches the limit (only + * in parallel heap vacuum cases). * * The visibility status of the next block to process and whether or not it * was eager scanned is set in the per_buffer_data. @@ -1636,10 +1915,10 @@ lazy_scan_heap(LVRelState *vacrel) * 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->heap_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. + * relation are read from it. vacrel->heap_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 heap_vac_scan_next_block(ReadStream *stream, @@ -1649,8 +1928,42 @@ heap_vac_scan_next_block(ReadStream *stream, BlockNumber next_block; LVRelState *vacrel = callback_private_data; - /* relies on InvalidBlockNumber + 1 overflowing to 0 on first call */ - next_block = vacrel->current_block + 1; +retry: + next_block = InvalidBlockNumber; + + /* Get the next block to process */ + if (ParallelHeapVacuumIsActive(vacrel)) + { + /* + * Stop returning the next block to the read stream if we are close to + * overrunning the available space for dead_items TIDs so that the + * read stream returns pinned buffers in its buffers queue until the + * stream is exhausted. See the comments atop this file for details. + */ + if (dead_items_check_memory_limit(vacrel)) + { + if (BufferIsValid(vacrel->next_unskippable_vmbuffer)) + { + ReleaseBuffer(vacrel->next_unskippable_vmbuffer); + vacrel->next_unskippable_vmbuffer = InvalidBuffer; + } + + return InvalidBlockNumber; + + } + + next_block = parallel_lazy_scan_get_nextpage(vacrel, + vacrel->rel, + vacrel->plvstate->scan_desc, + vacrel->plvstate->scan_work); + } + else + { + /* relies on InvalidBlockNumber + 1 overflowing to 0 on first call */ + next_block = vacrel->current_block + 1; + } + + Assert(BlockNumberIsValid(next_block)); /* Have we reached the end of the relation? */ if (next_block >= vacrel->rel_pages) @@ -1675,8 +1988,41 @@ 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->scan_work->chunk_remaining + 1; + } + else + end_block = vacrel->rel_pages; - find_next_unskippable_block(vacrel, &skipsallvis); + found = find_next_unskippable_block(vacrel, &skipsallvis, next_block, end_block); + + /* + * We must have found the next unskippable block within the specified + * range in non-parallel cases as the end_block is always the last + * block + 1 and we must scan the last block. + */ + Assert(found || ParallelHeapVacuumIsActive(vacrel)); + + if (!found) + { + if (skipsallvis) + vacrel->heap_scan_data->skippedallvis = true; + + /* + * Skip all remaining blocks in the current chunk, and retry with + * the next chunk. + */ + vacrel->plvstate->scan_work->chunk_remaining = 0; + goto retry; + } + + Assert(vacrel->next_unskippable_block < end_block); /* * We now know the next block that we must process. It can be the @@ -1693,11 +2039,21 @@ 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->heap_scan_data->skippedallvis = true; + + /* Tell the parallel scans to skip blocks */ + if (ParallelHeapVacuumIsActive(vacrel)) + { + vacrel->plvstate->scan_work->chunk_remaining -= nblocks_skip; + vacrel->plvstate->scan_work->nallocated += nblocks_skip; + Assert(vacrel->plvstate->scan_work->chunk_remaining > 0); + } + + next_block = vacrel->next_unskippable_block; } } @@ -1728,9 +2084,86 @@ heap_vac_scan_next_block(ReadStream *stream, } } + +/* + * Initialize scan state of the given ParallelLVScanWorkerData. + */ +static void +parallel_lazy_scan_init_scan_worker(ParallelLVScanWorkerData *scan_work, + BlockNumber initial_chunk_size) +{ + Assert(BlockNumberIsValid(initial_chunk_size)); + + scan_work->inited = true; + scan_work->nallocated = 0; + scan_work->chunk_size = initial_chunk_size; + scan_work->chunk_remaining = 0; + pg_atomic_init_u32(&(scan_work->last_blkno), InvalidBlockNumber); +} + +/* + * Return the next page to process for parallel heap scan. + * + * If there is no block to scan for the worker, return the number of blocks in + * the relation. + */ +static BlockNumber +parallel_lazy_scan_get_nextpage(LVRelState *vacrel, Relation rel, + ParallelLVScanDesc *scan_desc, + ParallelLVScanWorkerData *scan_work) +{ + uint64 nallocated; + + if (scan_work->chunk_remaining > 0) + { + /* + * Give them the next block in the range and update the remaining + * number of blocks. + */ + nallocated = ++scan_work->nallocated; + scan_work->chunk_remaining--; + } + else + { + /* Get the new chunk */ + nallocated = scan_work->nallocated = + pg_atomic_fetch_add_u64(&scan_desc->nallocated, scan_work->chunk_size); + + /* + * Set the remaining number of blocks in this chunk so that subsequent + * calls from this worker continue on with this chunk until it's done. + */ + scan_work->chunk_remaining = scan_work->chunk_size - 1; + + /* We use the fixed size chunk for subsequent scans */ + scan_work->chunk_size = PARALLEL_LV_CHUNK_SIZE; + + /* + * Getting the new chunk also means to start the new eager scan + * region. + * + * Update next_eager_scan_region_start to the first block in the chunk + * so that we can reset the remaining_fails counter when checking the + * visibility of the first block in this chunk in + * find_next_unskippable_block(). + */ + vacrel->next_eager_scan_region_start = nallocated; + + } + + /* Clear the chunk_remaining if there is no more blocks to process */ + if (nallocated >= scan_desc->nblocks) + scan_work->chunk_remaining = 0; + + return Min(nallocated, scan_desc->nblocks); +} + /* - * Find the next unskippable block in a vacuum scan using the visibility map. - * The next unskippable block and its visibility information is updated in + * Find the next unskippable block in a vacuum scan using the visibility map, + * in a range of 'start' (inclusive) and 'end' (exclusive). + * + * If found, the next unskippable block and its visibility information is + * updated in vacrel. Otherwise, return false and reset the information in * vacrel. * * Note: our opinion of which blocks can be skipped can go stale immediately. @@ -1741,21 +2174,31 @@ 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, BlockNumber end) { BlockNumber rel_pages = vacrel->rel_pages; - BlockNumber next_unskippable_block = vacrel->next_unskippable_block + 1; + BlockNumber next_unskippable_block = start; Buffer next_unskippable_vmbuffer = vacrel->next_unskippable_vmbuffer; bool next_unskippable_eager_scanned = false; + bool found = true; *skipsallvis = false; for (;; next_unskippable_block++) { - uint8 mapbits = visibilitymap_get_status(vacrel->rel, - next_unskippable_block, - &next_unskippable_vmbuffer); + uint8 mapbits; + + /* Reach the end of range? */ + if (next_unskippable_block >= end) + { + found = false; + break; + } + + mapbits = visibilitymap_get_status(vacrel->rel, next_unskippable_block, + &next_unskippable_vmbuffer); /* @@ -1830,10 +2273,284 @@ find_next_unskippable_block(LVRelState *vacrel, bool *skipsallvis) *skipsallvis = true; } - /* write the local variables back to vacrel */ - vacrel->next_unskippable_block = next_unskippable_block; - vacrel->next_unskippable_eager_scanned = next_unskippable_eager_scanned; - vacrel->next_unskippable_vmbuffer = next_unskippable_vmbuffer; + if (found) + { + /* write the local variables back to vacrel */ + vacrel->next_unskippable_block = next_unskippable_block; + vacrel->next_unskippable_eager_scanned = next_unskippable_eager_scanned; + vacrel->next_unskippable_vmbuffer = next_unskippable_vmbuffer; + } + else + { + if (BufferIsValid(next_unskippable_vmbuffer)) + ReleaseBuffer(next_unskippable_vmbuffer); + + /* + * There is no unskippable block in the specified range. Reset the + * related fields in vacrel. + */ + vacrel->next_unskippable_block = InvalidBlockNumber; + vacrel->next_unskippable_eager_scanned = false; + vacrel->next_unskippable_vmbuffer = InvalidBuffer; + } + + 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) +{ + ParallelLVScanWorkerData scan_work; + + Assert(ParallelHeapVacuumIsActive(vacrel)); + Assert(!IsParallelWorker()); + + /* Setup the parallel scan description for the leader to join as a worker */ + parallel_lazy_scan_init_scan_worker(&scan_work, + vacrel->plvstate->shared->initial_chunk_size); + vacrel->plvstate->scan_work = &scan_work; + + /* Adjust the eager scan's success counter as a worker */ + vacrel->eager_scan_remaining_successes = + vacrel->plvstate->shared->eager_scan_remaining_successes_per_worker; + + for (;;) + { + BlockNumber fsmvac_upto; + + /* Launch parallel workers */ + parallel_lazy_scan_heap_begin(vacrel); + + /* + * Do lazy heap scan until the read stream is exhausted. We will stop + * retrieving new blocks for the read stream once the space of + * dead_items TIDs exceeds the limit. + */ + if (vacrel->leader->leaderparticipates) + do_lazy_scan_heap(vacrel, false); + + /* Wait for parallel workers to finish and gather scan results */ + parallel_lazy_scan_heap_end(vacrel); + + if (!dead_items_check_memory_limit(vacrel)) + break; + + /* Perform a round of index and heap vacuuming */ + vacrel->consider_bypass_optimization = false; + lazy_vacuum(vacrel); + + /* Compute the smallest processed block number */ + fsmvac_upto = 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 (fsmvac_upto > vacrel->next_fsm_block_to_vacuum) + { + FreeSpaceMapVacuumRange(vacrel->rel, vacrel->next_fsm_block_to_vacuum, + fsmvac_upto); + vacrel->next_fsm_block_to_vacuum = fsmvac_upto; + } + + /* Report that we are once again scanning the heap */ + pgstat_progress_update_param(PROGRESS_VACUUM_PHASE, + PROGRESS_VACUUM_PHASE_SCAN_HEAP); + } + + /* + * 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. + */ + parallel_lazy_scan_heap_complete(vacrel); +} + +/* + * Return the smallest block number that the leader and workers have scanned. + */ +static BlockNumber +parallel_lazy_scan_compute_min_scan_block(LVRelState *vacrel) +{ + BlockNumber min_blk; + + Assert(ParallelHeapVacuumIsActive(vacrel)); + + /* Initialized with the leader's value */ + min_blk = vacrel->last_blkno; + + for (int i = 0; i < vacrel->leader->nworkers_launched; i++) + { + ParallelLVScanWorkerData *scan_work = &(vacrel->leader->scan_work_array[i]); + BlockNumber blkno; + + /* Skip if no worker has been initialized the scan state */ + if (!scan_work->inited) + continue; + + blkno = pg_atomic_read_u32(&(scan_work->last_blkno)); + + if (!BlockNumberIsValid(min_blk) || min_blk > blkno) + min_blk = blkno; + } + + Assert(BlockNumberIsValid(min_blk)); + + return min_blk; +} + +/* + * Complete parallel heaps scans that have remaining blocks in their + * chunks. + */ +static void +parallel_lazy_scan_heap_complete(LVRelState *vacrel) +{ + int nworkers; + + Assert(!IsParallelWorker()); + + nworkers = parallel_vacuum_get_nworkers_table(vacrel->pvs); + + for (int i = 0; i < nworkers; i++) + { + ParallelLVScanWorkerData *scan_work = &(vacrel->leader->scan_work_array[i]); + + if (!scan_work->inited) + continue; + + if (scan_work->chunk_remaining == 0) + continue; + + /* Attach the worker's scan state */ + vacrel->plvstate->scan_work = scan_work; + + vacrel->next_fsm_block_to_vacuum = pg_atomic_read_u32(&(scan_work->last_blkno)); + vacrel->next_eager_scan_region_start = scan_work->next_region_start_save; + vacrel->eager_scan_remaining_fails = scan_work->remaining_fails_save; + + /* + * Complete the unfinished scan. Note that we might perform multiple + * cycles of index and heap vacuuming while completing the scan. + */ + do_lazy_scan_heap(vacrel, true); + } + + /* + * 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 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 heap scan. + */ +static void +parallel_lazy_scan_heap_end(LVRelState *vacrel) +{ + /* Wait for all parallel workers to finish */ + parallel_vacuum_collect_dead_items_end(vacrel->pvs); + + /* Gather the workers' scan results */ + parallel_lazy_scan_gather_results(vacrel); +} + +/* + * Accumulate each worker's scan results into the leader's. + */ +static void +parallel_lazy_scan_gather_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->leader->scan_data_array[i]); + ParallelLVScanWorkerData *scan_work = &(vacrel->leader->scan_work_array[i]); + + /* Accumulate the counters collected by workers */ +#define ACCUM_COUNT(item) vacrel->heap_scan_data->item += data->item + ACCUM_COUNT(scanned_pages); + ACCUM_COUNT(removed_pages); + ACCUM_COUNT(new_frozen_tuple_pages); + ACCUM_COUNT(new_all_visible_pages); + ACCUM_COUNT(new_all_visible_all_frozen_pages); + ACCUM_COUNT(new_all_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 + + /* + * Track the greatest non-empty page among values the workers + * collected as it's used to cut-off point of heap truncation. + */ + if (vacrel->heap_scan_data->nonempty_pages < data->nonempty_pages) + vacrel->heap_scan_data->nonempty_pages = data->nonempty_pages; + + /* + * All workers must have initialized both values with the values + * passed by the leader. + */ + Assert(TransactionIdIsValid(data->NewRelfrozenXid)); + Assert(MultiXactIdIsValid(data->NewRelminMxid)); + + /* + * During parallel heap scanning, since different workers process + * separate blocks, they may observe different existing XIDs and + * MXIDs. Therefore, we compute the oldest XID and MXID from the + * values observed by each worker (including the leader). These + * computations are crucial for correctly advancing both relfrozenxid + * and relmminmxid values. + */ + + if (TransactionIdPrecedes(data->NewRelfrozenXid, vacrel->heap_scan_data->NewRelfrozenXid)) + vacrel->heap_scan_data->NewRelfrozenXid = data->NewRelfrozenXid; + + if (MultiXactIdPrecedesOrEquals(data->NewRelminMxid, vacrel->heap_scan_data->NewRelminMxid)) + vacrel->heap_scan_data->NewRelminMxid = data->NewRelminMxid; + + /* Has any one of workers skipped all-visible page? */ + vacrel->heap_scan_data->skippedallvis |= data->skippedallvis; + + /* + * Gather the remaining success count so that we can distribute the + * success counter again in the next parallel heap scan. + */ + vacrel->eager_scan_remaining_successes += scan_work->remaining_successes_save; + } } /* @@ -2123,7 +2840,8 @@ lazy_scan_prune(LVRelState *vacrel, /* Can't truncate this page */ if (presult.hastup) - vacrel->heap_scan_data->nonempty_pages = blkno + 1; + vacrel->heap_scan_data->nonempty_pages = + Max(blkno + 1, vacrel->heap_scan_data->nonempty_pages); /* Did we find LP_DEAD items? */ *has_lpdead_items = (presult.lpdead_items > 0); @@ -2340,7 +3058,8 @@ lazy_scan_noprune(LVRelState *vacrel, /* Can't truncate this page */ if (hastup) - vacrel->heap_scan_data->nonempty_pages = blkno + 1; + vacrel->heap_scan_data->nonempty_pages = + Max(blkno + 1, vacrel->heap_scan_data->nonempty_pages); /* Did we find LP_DEAD items? */ *has_lpdead_items = (lpdead_items > 0); @@ -3417,12 +4136,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 @@ -3440,11 +4155,19 @@ dead_items_alloc(LVRelState *vacrel, int nworkers) vacrel->relname))); } else + { + /* + * We initialize the parallel vacuum state for the heap scan, + * 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); + vacrel->bstrategy, + &heap_parallel_vacuum_callbacks, + "heap", (void *) vacrel); + } /* * If parallel mode started, dead_items and dead_items_info spaces are @@ -3484,15 +4207,35 @@ 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); pgstat_progress_update_multi_param(2, prog_index, prog_val); } +/* + * Check the memory usage of the collected dead items and return true + * if we are close to overrunning the available space for dead_items TIDs. + * However, let's force at least one page-worth of tuples to be stored as + * to ensure we do at least some work when the memory configured is so low + * that we run out before storing anything. + */ +static bool +dead_items_check_memory_limit(LVRelState *vacrel) +{ + return vacrel->dead_items_info->num_items > 0 && + TidStoreMemoryUsage(vacrel->dead_items) > vacrel->dead_items_info->max_bytes; +} + /* * Forget all collected dead items. */ @@ -3781,6 +4524,317 @@ update_relstats_all_indexes(LVRelState *vacrel) } } +/* + * Callbacks that heap registers with vacuumparallel.c for parallel table + * vacuum. vacuumparallel.c resolves this struct by the AM name "heap"; see + * parallel_vacuum_main(). + */ +const ParallelVacuumCallbacks heap_parallel_vacuum_callbacks = { + .compute_workers = heap_parallel_vacuum_compute_workers, + .estimate = heap_parallel_vacuum_estimate, + .initialize = heap_parallel_vacuum_initialize, + .initialize_worker = heap_parallel_vacuum_initialize_worker, + .collect_dead_items = heap_parallel_vacuum_collect_dead_items, +}; + +/* + * Compute the number of workers for parallel heap vacuum. + */ +static int +heap_parallel_vacuum_compute_workers(Relation rel, int nworkers_requested, + void *state) +{ + BlockNumber relpages = RelationGetNumberOfBlocks(rel); + int parallel_workers = 0; + + /* + * Parallel heap vacuuming a small relation shouldn't take long. We use + * two times the chunk size as the size cutoff because the leader is + * assigned to one chunk. + */ + if (relpages < PARALLEL_LV_CHUNK_SIZE * 2 || relpages < min_parallel_table_scan_size) + return 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 = relpages - (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 = PARALLEL_LV_CHUNK_SIZE; + 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. + */ +static void +heap_parallel_vacuum_estimate(Relation rel, ParallelContext *pcxt, int nworkers, + void *state) +{ + LVRelState *vacrel = (LVRelState *) state; + Size size = 0; + bool leaderparticipates = true; + + vacrel->leader = palloc(sizeof(ParallelLVLeader)); + + /* Estimate space for ParallelLVShared */ + size = add_size(size, sizeof(ParallelLVShared)); + 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 ParallelLVScanDesc */ + vacrel->leader->scan_desc_len = sizeof(ParallelLVScanDesc); + shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->scan_desc_len); + shm_toc_estimate_keys(&pcxt->estimator, 1); + + /* Estimate space for an array of ParallelLVScanWorkerData */ + vacrel->leader->scan_work_len = mul_size(sizeof(ParallelLVScanWorkerData), + nworkers); + shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->scan_work_len); + shm_toc_estimate_keys(&pcxt->estimator, 1); + + /* Estimate space for an array of LVScanData */ + vacrel->leader->scan_data_len = mul_size(sizeof(LVScanData), nworkers); + shm_toc_estimate_chunk(&pcxt->estimator, vacrel->leader->scan_data_len); + shm_toc_estimate_keys(&pcxt->estimator, 1); + +#ifdef USE_INJECTION_POINTS + if (IS_INJECTION_POINT_ATTACHED("parallel-heap-vacuum-disable-leader-participation")) + leaderparticipates = false; +#endif + vacrel->leader->leaderparticipates = leaderparticipates; +} + +/* + * Set up shared memory for parallel heap vacuum. + */ +static void +heap_parallel_vacuum_initialize(Relation rel, ParallelContext *pcxt, int nworkers, + void *state) +{ + LVRelState *vacrel = (LVRelState *) state; + ParallelLVShared *shared; + ParallelLVScanDesc *scan_desc; + ParallelLVScanWorkerData *scan_work; + LVScanData *scan_data; + + vacrel->plvstate = palloc0(sizeof(ParallelLVState)); + + /* Initialize ParallelLVShared */ + + 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->heap_scan_data->NewRelfrozenXid; + shared->NewRelminMxid = vacrel->heap_scan_data->NewRelminMxid; + shared->initial_chunk_size = BlockNumberIsValid(vacrel->next_eager_scan_region_start) + ? vacrel->next_eager_scan_region_start + : PARALLEL_LV_CHUNK_SIZE; + + /* Calculate the per-chunk maximum failure count */ + shared->eager_scan_max_fails_per_chunk = + (BlockNumber) (vacrel->eager_scan_max_fails_per_region * + ((float) PARALLEL_LV_CHUNK_SIZE / EAGER_SCAN_REGION_SIZE)); + + /* including the leader too */ + shared->eager_scan_remaining_successes_per_worker = + vacrel->eager_scan_remaining_successes / + (vacrel->leader->leaderparticipates ? nworkers + 1 : nworkers); + + shm_toc_insert(pcxt->toc, PARALLEL_LV_KEY_SHARED, shared); + vacrel->plvstate->shared = shared; + + /* Initialize ParallelLVScanDesc */ + scan_desc = shm_toc_allocate(pcxt->toc, vacrel->leader->scan_desc_len); + scan_desc->nblocks = RelationGetNumberOfBlocks(rel); + pg_atomic_init_u64(&scan_desc->nallocated, 0); + shm_toc_insert(pcxt->toc, PARALLEL_LV_KEY_SCANDESC, scan_desc); + vacrel->plvstate->scan_desc = scan_desc; + + /* Initialize the array of ParallelLVScanWorkerData */ + scan_work = shm_toc_allocate(pcxt->toc, vacrel->leader->scan_work_len); + MemSet(scan_work, 0, vacrel->leader->scan_work_len); + shm_toc_insert(pcxt->toc, PARALLEL_LV_KEY_SCANWORKER, scan_work); + vacrel->leader->scan_work_array = scan_work; + + /* Initialize the array of LVScanData */ + scan_data = shm_toc_allocate(pcxt->toc, vacrel->leader->scan_data_len); + shm_toc_insert(pcxt->toc, PARALLEL_LV_KEY_SCANDATA, scan_data); + vacrel->leader->scan_data_array = scan_data; +} + +/* + * Initialize lazy vacuum state with the information retrieved from + * shared memory. + */ +static void +heap_parallel_vacuum_initialize_worker(Relation rel, ParallelVacuumState *pvs, + ParallelWorkerContext *pwcxt, + void **state_out) +{ + LVRelState *vacrel; + ParallelLVState *plvstate; + ParallelLVShared *shared; + ParallelLVScanDesc *scan_desc; + ParallelLVScanWorkerData *scan_work_array; + LVScanData *scan_data_array; + + /* Initialize ParallelLVState and prepare the related objects */ + + plvstate = palloc0(sizeof(ParallelLVState)); + + /* Prepare ParallelLVShared */ + shared = (ParallelLVShared *) shm_toc_lookup(pwcxt->toc, PARALLEL_LV_KEY_SHARED, false); + plvstate->shared = shared; + + /* Prepare ParallelLVScanDesc */ + scan_desc = shm_toc_lookup(pwcxt->toc, PARALLEL_LV_KEY_SCANDESC, false); + plvstate->scan_desc = scan_desc; + + /* Prepare ParallelLVScanWorkerData */ + scan_work_array = shm_toc_lookup(pwcxt->toc, PARALLEL_LV_KEY_SCANWORKER, false); + plvstate->scan_work = &(scan_work_array[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->bstrategy = parallel_vacuum_get_bstrategy(pvs); + vacrel->pvs = pvs; + vacrel->aggressive = shared->aggressive; + vacrel->skipwithvm = shared->skipwithvm; + vacrel->vistest = GlobalVisTestFor(rel); + vacrel->cutoffs = shared->cutoffs; + vacrel->dead_items = parallel_vacuum_get_dead_items(pvs, + &vacrel->dead_items_info); + vacrel->rel_pages = RelationGetNumberOfBlocks(rel); + + /* + * Set the per-region failure counter and per-worker success counter, + * which are not changed during parallel heap vacuum. + */ + vacrel->eager_scan_max_fails_per_region = + plvstate->shared->eager_scan_max_fails_per_chunk; + vacrel->eager_scan_remaining_successes = + plvstate->shared->eager_scan_remaining_successes_per_worker; + + /* Does this worker have un-scanned blocks in a chunk? */ + if (plvstate->scan_work->chunk_remaining > 0) + { + /* + * We restore the previous eager scan state of the already allocated + * chunk, if the worker's previous scan suspended due to the full of + * dead_items TIDs space. + */ + vacrel->next_eager_scan_region_start = plvstate->scan_work->next_region_start_save; + vacrel->eager_scan_remaining_fails = plvstate->scan_work->remaining_fails_save; + } + else + { + /* + * next_eager_scan_region_start will be set when the first chunk is + * assigned. + */ + vacrel->next_eager_scan_region_start = InvalidBlockNumber; + vacrel->eager_scan_remaining_fails = vacrel->eager_scan_max_fails_per_region; + } + + vacrel->plvstate = plvstate; + + /* Prepare LVScanData */ + scan_data_array = shm_toc_lookup(pwcxt->toc, PARALLEL_LV_KEY_SCANDATA, false); + vacrel->heap_scan_data = &(scan_data_array[ParallelWorkerNumber]); + MemSet(vacrel->heap_scan_data, 0, sizeof(LVScanData)); + vacrel->heap_scan_data->NewRelfrozenXid = shared->NewRelfrozenXid; + vacrel->heap_scan_data->NewRelminMxid = shared->NewRelminMxid; + vacrel->heap_scan_data->skippedallvis = false; + + /* + * 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->scan_work->inited) + parallel_lazy_scan_init_scan_worker(vacrel->plvstate->scan_work, + vacrel->plvstate->shared->initial_chunk_size); + + *state_out = (void *) vacrel; +} + +/* + * Parallel heap vacuum callback for collecting dead items (i.e., the heap + * scan). + */ +static void +heap_parallel_vacuum_collect_dead_items(Relation rel, ParallelVacuumState *pvs, + void *state) +{ + LVRelState *vacrel = (LVRelState *) state; + ErrorContextCallback errcallback; + + 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 */ + do_lazy_scan_heap(vacrel, false); + + /* Advertise the last processed block number */ + pg_atomic_write_u32(&(vacrel->plvstate->scan_work->last_blkno), vacrel->last_blkno); + + /* Save the eager scan state */ + vacrel->plvstate->scan_work->remaining_fails_save = vacrel->eager_scan_remaining_fails; + vacrel->plvstate->scan_work->remaining_successes_save = vacrel->eager_scan_remaining_successes; + vacrel->plvstate->scan_work->next_region_start_save = vacrel->next_eager_scan_region_start; + + /* Pop the error context stack */ + error_context_stack = errcallback.previous; +} + /* * Error context callback for errors occurring during vacuum. The error * context messages for index phases should match the messages set in parallel diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index 41cefcfde54..74935b6aec9 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/src/backend/commands/vacuumparallel.c @@ -5,18 +5,17 @@ * comments below, the word "vacuum" will refer to both vacuum and * autovacuum. * - * This file contains routines that are intended to support setting up, using, - * and tearing down a ParallelVacuumState. + * This file contains routines that are intended to support setting up, + * using, and tearing down a ParallelVacuumState, which contains shared + * information as well as the memory space for storing dead items + * allocated in the DSA area. * - * In a parallel vacuum, we perform both index bulk deletion and index cleanup - * with parallel worker processes. Individual indexes are processed by one - * vacuum process. ParallelVacuumState contains shared information as well as - * the memory space for storing dead items allocated in the DSA area. We - * launch parallel worker processes at the start of parallel index - * bulk-deletion and index cleanup and once all indexes are processed, the - * parallel worker processes exit. Each time we process indexes in parallel, - * the parallel context is re-initialized so that the same DSM can be used for - * multiple passes of index bulk-deletion and index cleanup. + * In parallel vacuum, we use parallel worker processes to collect dead items + * from the table, to vacuum and clean up indexes, or both, depending on how + * many workers each phase needs (which can differ between the table scan and + * index processing). Workers are launched at the start of a phase and exit + * when it completes. The parallel context is re-initialized before each phase + * so that the same DSM can serve multiple passes. * * For parallel autovacuum, we need to propagate cost-based vacuum delay * parameters from the leader to its workers, as the leader's parameters can @@ -35,8 +34,11 @@ */ #include "postgres.h" +#include "access/parallel.h" #include "access/amapi.h" +#include "access/heapam.h" #include "access/table.h" +#include "access/tableam.h" #include "access/xact.h" #include "commands/progress.h" #include "commands/vacuum.h" @@ -46,9 +48,27 @@ #include "storage/bufmgr.h" #include "storage/proc.h" #include "tcop/tcopprot.h" +#include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/rel.h" +/* + * Parallel table vacuum callbacks, resolved by access method name. The name, + * not a pointer (which isn't valid across processes), is what reaches the + * workers. Only the in-core heap access method is supported, so a fixed table + * is enough; out-of-core support would need a registration API. + */ +typedef struct VacuumCallbacksEntry +{ + const char *name; + const ParallelVacuumCallbacks *cbs; +} VacuumCallbacksEntry; + +static const VacuumCallbacksEntry InternalVacuumCallbacks[] = +{ + {"heap", &heap_parallel_vacuum_callbacks}, +}; + /* * DSM keys for parallel vacuum. Unlike other parallel execution code, since * we don't need to worry about DSM keys conflicting with plan_node_id we can @@ -60,6 +80,16 @@ #define PARALLEL_VACUUM_KEY_WAL_USAGE 4 #define PARALLEL_VACUUM_KEY_INDEX_STATS 5 +/* + * The kinds of parallel vacuum phases. + */ +typedef enum +{ + PV_WORK_PHASE_COLLECT_DEAD_ITEMS, /* scan the table for dead items */ + PV_WORK_PHASE_INDEX_VACUUM, /* index bulk-deletion */ + PV_WORK_PHASE_INDEX_CLEANUP, /* index cleanup */ +} PVWorkPhase; + /* * Struct for cost-based vacuum delay related parameters to share among an * autovacuum worker and its parallel vacuum workers. @@ -102,6 +132,19 @@ typedef struct PVShared int elevel; int64 queryid; + /* + * Name of the table access method. Workers use this to look up the + * parallel table vacuum callbacks locally (see parallel_vacuum_main()); + * we never pass function pointers through shared memory. + */ + char am_name[NAMEDATALEN]; + + /* + * Tell parallel workers what phase to perform, either processing indexes + * or collecting dead tuples from the table. + */ + PVWorkPhase work_phase; + /* * Fields for both index vacuum and cleanup. * @@ -213,9 +256,20 @@ struct ParallelVacuumState /* NULL for worker processes */ ParallelContext *pcxt; + /* Do we need to reinitialize parallel DSM? */ + bool need_reinitialize_dsm; + /* Parent Heap Relation */ Relation heaprel; + /* + * Parallel table vacuum callbacks. In the leader this is the struct + * passed to parallel_vacuum_init(); in a worker it is resolved locally by + * AM name (see parallel_vacuum_main()). Either way the pointer is valid + * only within the current process. + */ + const ParallelVacuumCallbacks *cbs; + /* Target indexes */ Relation *indrels; int nindexes; @@ -227,7 +281,7 @@ struct ParallelVacuumState * Shared index statistics among parallel vacuum workers. The array * element is allocated for every index, even those indexes where parallel * index vacuuming is unsafe or not worthwhile (e.g., - * will_parallel_vacuum[] is false). During parallel vacuum, + * idx_will_parallel_vacuum[] is false). During parallel vacuum, * IndexBulkDeleteResult of each index is kept in DSM and is copied into * local memory at the end of parallel vacuum. */ @@ -242,12 +296,18 @@ struct ParallelVacuumState /* Points to WAL usage area in DSM */ WalUsage *wal_usage; + /* + * The number of workers for parallel table vacuuming. If 0, the parallel + * table vacuum is disabled. + */ + int nworkers_for_table; + /* * False if the index is totally unsuitable target for all parallel * processing. For example, the index could be < * min_parallel_index_scan_size cutoff. */ - bool *will_parallel_vacuum; + bool *idx_will_parallel_vacuum; /* * The number of indexes that support parallel index bulk-deletion and @@ -281,8 +341,12 @@ static PVSharedCostParams *pv_shared_cost_params = NULL; */ static uint32 shared_params_generation_local = 0; -static int parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, - bool *will_parallel_vacuum); +static const ParallelVacuumCallbacks *parallel_vacuum_lookup_callbacks(const char *am_name); +static int parallel_vacuum_compute_workers(Relation rel, Relation *indrels, int nindexes, + int nrequested, int *nworkers_for_table, + bool *idx_will_parallel_vacuum, + const ParallelVacuumCallbacks *cbs, + void *state); static void parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scans, bool vacuum, PVWorkerStats *wstats); static void parallel_vacuum_process_safe_indexes(ParallelVacuumState *pvs); @@ -291,20 +355,29 @@ static void parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation PVIndStats *indstats); static bool parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans, bool vacuum); +static void parallel_vacuum_begin_work_phase(ParallelVacuumState *pvs, int nworkers, + PVWorkPhase work_phase); +static void parallel_vacuum_end_worker_phase(ParallelVacuumState *pvs); static void parallel_vacuum_error_callback(void *arg); static inline void parallel_vacuum_set_cost_parameters(PVSharedCostParams *params); static void parallel_vacuum_dsm_detach(dsm_segment *seg, Datum arg); /* - * Try to enter parallel mode and create a parallel context. Then initialize + * Try to enter parallel mode and create a parallel context. Then initialize * shared memory state. * - * On success, return parallel vacuum state. Otherwise return NULL. + * nrequested_workers is the requested parallel degree. 0 means that the + * parallel degrees for table and index vacuum are decided differently. See + * the comments of parallel_vacuum_compute_workers() for details. + * + * On success, return parallel vacuum state. Otherwise return NULL. */ ParallelVacuumState * parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, int nrequested_workers, int vac_work_mem, - int elevel, BufferAccessStrategy bstrategy) + int elevel, BufferAccessStrategy bstrategy, + const ParallelVacuumCallbacks *cbs, const char *am_name, + void *state) { ParallelVacuumState *pvs; ParallelContext *pcxt; @@ -313,46 +386,49 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, PVIndStats *indstats; BufferUsage *buffer_usage; WalUsage *wal_usage; - bool *will_parallel_vacuum; + bool *idx_will_parallel_vacuum; Size est_indstats_len; Size est_shared_len; int nindexes_mwm = 0; int parallel_workers = 0; + int nworkers_for_table; int querylen; - /* - * A parallel vacuum must be requested and there must be indexes on the - * relation - */ + /* A parallel vacuum must be requested */ Assert(nrequested_workers >= 0); - Assert(nindexes > 0); /* * Compute the number of parallel vacuum workers to launch */ - will_parallel_vacuum = palloc0_array(bool, nindexes); - parallel_workers = parallel_vacuum_compute_workers(indrels, nindexes, + idx_will_parallel_vacuum = palloc0_array(bool, nindexes); + parallel_workers = parallel_vacuum_compute_workers(rel, indrels, nindexes, nrequested_workers, - will_parallel_vacuum); + &nworkers_for_table, + idx_will_parallel_vacuum, + cbs, state); + if (parallel_workers <= 0) { /* Can't perform vacuum in parallel -- return NULL */ - pfree(will_parallel_vacuum); + pfree(idx_will_parallel_vacuum); return NULL; } pvs = palloc0_object(ParallelVacuumState); pvs->indrels = indrels; pvs->nindexes = nindexes; - pvs->will_parallel_vacuum = will_parallel_vacuum; + pvs->idx_will_parallel_vacuum = idx_will_parallel_vacuum; pvs->bstrategy = bstrategy; pvs->heaprel = rel; + pvs->cbs = cbs; EnterParallelMode(); pcxt = CreateParallelContext("postgres", "parallel_vacuum_main", parallel_workers); Assert(pcxt->nworkers > 0); pvs->pcxt = pcxt; + pvs->need_reinitialize_dsm = false; + pvs->nworkers_for_table = nworkers_for_table; /* Estimate size for index vacuum stats -- PARALLEL_VACUUM_KEY_INDEX_STATS */ est_indstats_len = mul_size(sizeof(PVIndStats), nindexes); @@ -389,6 +465,10 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, else querylen = 0; /* keep compiler quiet */ + /* Estimate AM-specific space for parallel table vacuum */ + if (pvs->nworkers_for_table > 0) + cbs->estimate(rel, pcxt, pvs->nworkers_for_table, state); + InitializeParallelDSM(pcxt); /* Prepare index vacuum stats */ @@ -407,7 +487,7 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0)); Assert(vacoptions <= VACUUM_OPTION_MAX_VALID_VALUE); - if (!will_parallel_vacuum[i]) + if (!idx_will_parallel_vacuum[i]) continue; if (indrel->rd_indam->amusemaintenanceworkmem) @@ -433,6 +513,7 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, shared->relid = RelationGetRelid(rel); shared->elevel = elevel; shared->queryid = pgstat_get_my_query_id(); + strlcpy(shared->am_name, am_name, sizeof(shared->am_name)); shared->maintenance_work_mem_worker = (nindexes_mwm > 0) ? vac_work_mem / Min(parallel_workers, nindexes_mwm) : @@ -498,6 +579,10 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, PARALLEL_VACUUM_KEY_QUERY_TEXT, sharedquery); } + /* Initialize AM-specific DSM space for parallel table vacuum */ + if (pvs->nworkers_for_table > 0) + cbs->initialize(rel, pcxt, pvs->nworkers_for_table, state); + /* Success -- return parallel vacuum state */ return pvs; } @@ -538,7 +623,7 @@ parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats) if (AmAutoVacuumWorkerProcess()) pv_shared_cost_params = NULL; - pfree(pvs->will_parallel_vacuum); + pfree(pvs->idx_will_parallel_vacuum); pfree(pvs); } @@ -555,6 +640,35 @@ parallel_vacuum_dsm_detach(dsm_segment *seg, Datum arg) pv_shared_cost_params = NULL; } +/* + * 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 with the table being vacuumed. + */ +Relation * +parallel_vacuum_get_table_indexes(ParallelVacuumState *pvs, int *nindexes) +{ + *nindexes = pvs->nindexes; + + return pvs->indrels; +} + +/* + * Return the buffer strategy for parallel vacuum. + */ +BufferAccessStrategy +parallel_vacuum_get_bstrategy(ParallelVacuumState *pvs) +{ + return pvs->bstrategy; +} + /* * Returns the dead items space and dead items information. */ @@ -726,32 +840,62 @@ parallel_vacuum_propagate_shared_delay_params(void) } /* - * Compute the number of parallel worker processes to request. Both index - * vacuum and index cleanup can be executed with parallel workers. - * The index is eligible for parallel vacuum iff its size is greater than - * min_parallel_index_scan_size as invoking workers for very small indexes - * can hurt performance. + * Resolve an access method's parallel table vacuum callbacks by name. Used by + * workers to get a process-valid pointer without reading one from shared + * memory. + */ +static const ParallelVacuumCallbacks * +parallel_vacuum_lookup_callbacks(const char *am_name) +{ + for (int i = 0; i < lengthof(InternalVacuumCallbacks); i++) + { + if (strcmp(InternalVacuumCallbacks[i].name, am_name) == 0) + return InternalVacuumCallbacks[i].cbs; + } + + elog(ERROR, "could not find parallel vacuum callbacks for table access method \"%s\"", + am_name); + return NULL; /* keep compiler quiet */ +} + +/* + * Compute the number of parallel worker processes to request for table + * vacuum and index vacuum/cleanup. Return the maximum number of parallel + * workers for table vacuuming and index vacuuming. + * + * nrequested is the number of parallel workers that user requested, which + * applies to both the number of workers for table vacuum and index vacuum. + * If nrequested is 0, we compute the parallel degree for them differently + * as described below. * - * nrequested is the number of parallel workers that user requested. If - * nrequested is 0, we compute the parallel degree based on nindexes, that is - * the number of indexes that support parallel vacuum. This function also - * sets will_parallel_vacuum to remember indexes that participate in parallel + * For parallel table vacuum, we ask AM-specific routine to compute the + * number of parallel worker processes. The result is set to nworkers_table_p. + * + * For parallel index vacuum, the index is eligible for parallel vacuum iff + * its size is greater than min_parallel_index_scan_size as invoking workers + * for very small indexes can hurt performance. This function sets + * idx_will_parallel_vacuum to remember indexes that participate in parallel * vacuum. */ static int -parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, - bool *will_parallel_vacuum) +parallel_vacuum_compute_workers(Relation rel, Relation *indrels, int nindexes, + int nrequested, int *nworkers_table_p, + bool *idx_will_parallel_vacuum, + const ParallelVacuumCallbacks *cbs, void *state) { int nindexes_parallel = 0; int nindexes_parallel_bulkdel = 0; int nindexes_parallel_cleanup = 0; - int parallel_workers; + int nworkers_table = 0; + int nworkers_index = 0; int max_workers; max_workers = AmAutoVacuumWorkerProcess() ? autovacuum_max_parallel_workers : max_parallel_maintenance_workers; + *nworkers_table_p = 0; + /* * We don't allow performing parallel operation in standalone backend or * when parallelism is disabled. @@ -759,6 +903,13 @@ parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, if (!IsUnderPostmaster || max_workers == 0) return 0; + /* Compute the number of workers for parallel table scan */ + if (cbs->compute_workers != NULL) + nworkers_table = cbs->compute_workers(rel, nrequested, state); + + /* Cap by max_parallel_maintenance_workers (or the autovacuum GUC) */ + nworkers_table = Min(nworkers_table, max_workers); + /* * Compute the number of indexes that can participate in parallel vacuum. */ @@ -772,7 +923,7 @@ parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, RelationGetNumberOfBlocks(indrel) < min_parallel_index_scan_size) continue; - will_parallel_vacuum[i] = true; + idx_will_parallel_vacuum[i] = true; if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0) nindexes_parallel_bulkdel++; @@ -787,18 +938,18 @@ parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, /* The leader process takes one index */ nindexes_parallel--; - /* No index supports parallel vacuum */ - if (nindexes_parallel <= 0) - return 0; - - /* Compute the parallel degree */ - parallel_workers = (nrequested > 0) ? - Min(nrequested, nindexes_parallel) : nindexes_parallel; + if (nindexes_parallel > 0) + { + /* Take into account the requested number of workers */ + nworkers_index = (nrequested > 0) ? + Min(nrequested, nindexes_parallel) : nindexes_parallel; - /* Cap by GUC variable */ - parallel_workers = Min(parallel_workers, max_workers); + /* Cap by max_parallel_maintenance_workers (or the autovacuum GUC) */ + nworkers_index = Min(nworkers_index, max_workers); + } - return parallel_workers; + *nworkers_table_p = nworkers_table; + return Max(nworkers_table, nworkers_index); } /* @@ -861,7 +1012,7 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan Assert(indstats->status == PARALLEL_INDVAC_STATUS_INITIAL); indstats->status = new_status; indstats->parallel_workers_can_process = - (pvs->will_parallel_vacuum[i] && + (pvs->idx_will_parallel_vacuum[i] && parallel_vacuum_index_is_parallel_safe(pvs->indrels[i], num_index_scans, vacuum)); @@ -873,44 +1024,14 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan /* Setup the shared cost-based vacuum delay and launch workers */ if (nworkers > 0) { - /* Reinitialize parallel context to relaunch parallel workers */ - if (num_index_scans > 0) - ReinitializeParallelDSM(pvs->pcxt); + /* Start parallel vacuum workers for processing indexes */ + parallel_vacuum_begin_work_phase(pvs, nworkers, + vacuum ? PV_WORK_PHASE_INDEX_VACUUM : + PV_WORK_PHASE_INDEX_CLEANUP); - /* - * Set up shared cost balance and the number of active workers for - * vacuum delay. We need to do this before launching workers as - * otherwise, they might not see the updated values for these - * parameters. - */ - pg_atomic_write_u32(&(pvs->shared->cost_balance), VacuumCostBalance); - pg_atomic_write_u32(&(pvs->shared->active_nworkers), 0); - - /* - * The number of workers can vary between bulkdelete and cleanup - * phase. - */ - ReinitializeParallelWorkers(pvs->pcxt, nworkers); - - LaunchParallelWorkers(pvs->pcxt); - - if (pvs->pcxt->nworkers_launched > 0) - { - /* - * Reset the local cost values for leader backend as we have - * already accumulated the remaining balance of heap. - */ - VacuumCostBalance = 0; - VacuumCostBalanceLocal = 0; - - /* Enable shared cost balance for leader backend */ - VacuumSharedCostBalance = &(pvs->shared->cost_balance); - VacuumActiveNWorkers = &(pvs->shared->active_nworkers); - - /* Update the statistics, if we asked to */ - if (wstats != NULL) - wstats->nlaunched += pvs->pcxt->nworkers_launched; - } + /* Update the statistics, if we asked to */ + if (wstats != NULL && pvs->pcxt->nworkers_launched > 0) + wstats->nlaunched += pvs->pcxt->nworkers_launched; if (vacuum) ereport(pvs->shared->elevel, @@ -940,13 +1061,7 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan * to finish, or we might get incomplete data.) */ if (nworkers > 0) - { - /* Wait for all vacuum workers to finish */ - WaitForParallelWorkersToFinish(pvs->pcxt); - - for (int i = 0; i < pvs->pcxt->nworkers_launched; i++) - InstrAccumParallelQuery(&pvs->buffer_usage[i], &pvs->wal_usage[i]); - } + parallel_vacuum_end_worker_phase(pvs); /* * Reset all index status back to initial (while checking that we have @@ -963,15 +1078,8 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan indstats->status = PARALLEL_INDVAC_STATUS_INITIAL; } - /* - * Carry the shared balance value to heap scan and disable shared costing - */ - if (VacuumSharedCostBalance) - { - VacuumCostBalance = pg_atomic_read_u32(VacuumSharedCostBalance); - VacuumSharedCostBalance = NULL; - VacuumActiveNWorkers = NULL; - } + /* Parallel DSM will need to be reinitialized for the next execution */ + pvs->need_reinitialize_dsm = true; } /* @@ -1187,6 +1295,91 @@ parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans, return true; } +/* + * Begin the parallel scan to collect dead items. Return the number of + * launched parallel workers. + * + * The caller must call parallel_vacuum_collect_dead_items_end() to finish + * the parallel scan. + */ +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 (nworkers == 0) + return 0; + + /* Start parallel vacuum workers for collecting dead items */ + 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 */ + if (pvs->pcxt->nworkers_launched > 0) + pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1); + + return pvs->pcxt->nworkers_launched; +} + +/* + * Wait for all workers for parallel vacuum workers launched by + * parallel_vacuum_collect_dead_items_begin(), and gather workers' statistics. + */ +void +parallel_vacuum_collect_dead_items_end(ParallelVacuumState *pvs) +{ + Assert(!IsParallelWorker()); + Assert(pvs->shared->work_phase == PV_WORK_PHASE_COLLECT_DEAD_ITEMS); + + if (pvs->nworkers_for_table == 0) + return; + + /* Wait for parallel workers to finish */ + parallel_vacuum_end_worker_phase(pvs); + + /* Decrement the worker count for the leader itself */ + if (VacuumActiveNWorkers) + pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1); +} + +/* + * The function is for parallel workers to execute the parallel scan to + * collect dead tuples. + */ +static void +parallel_vacuum_process_table(ParallelVacuumState *pvs, void *state) +{ + Assert(VacuumActiveNWorkers); + Assert(pvs->shared->work_phase == PV_WORK_PHASE_COLLECT_DEAD_ITEMS); + + /* Increment the active worker before starting the table vacuum */ + pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1); + + /* Do the parallel scan to collect dead tuples */ + pvs->cbs->collect_dead_items(pvs->heaprel, pvs, state); + + /* + * We have completed the table vacuum so decrement the active worker + * count. + */ + pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1); +} + /* * Perform work within a launched parallel process. * @@ -1206,6 +1399,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) WalUsage *wal_usage; int nindexes; char *sharedquery; + void *state; ErrorContextCallback errcallback; /* @@ -1238,7 +1432,6 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) * matched to the leader's one. */ vac_open_indexes(rel, RowExclusiveLock, &nindexes, &indrels); - Assert(nindexes > 0); /* * Apply the desired value of maintenance_work_mem within this process. @@ -1290,6 +1483,13 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) pvs.relname = pstrdup(RelationGetRelationName(rel)); pvs.heaprel = rel; + /* + * Resolve the parallel table vacuum callbacks locally from the AM name in + * shared memory. We never read function pointers from shared memory; the + * name is looked up in this process to obtain a process-valid pointer. + */ + pvs.cbs = parallel_vacuum_lookup_callbacks(shared->am_name); + /* These fields will be filled during index vacuum or cleanup */ pvs.indname = NULL; pvs.status = PARALLEL_INDVAC_STATUS_INITIAL; @@ -1298,6 +1498,16 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) pvs.bstrategy = GetAccessStrategyWithSize(BAS_VACUUM, shared->ring_nbuffers * (BLCKSZ / 1024)); + /* Initialize AM-specific vacuum state for parallel table vacuuming */ + if (shared->work_phase == PV_WORK_PHASE_COLLECT_DEAD_ITEMS) + { + ParallelWorkerContext pwcxt; + + pwcxt.toc = toc; + pwcxt.seg = seg; + pvs.cbs->initialize_worker(rel, &pvs, &pwcxt, &state); + } + /* Setup error traceback support for ereport() */ errcallback.callback = parallel_vacuum_error_callback; errcallback.arg = &pvs; @@ -1307,8 +1517,20 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) /* Prepare to track buffer usage during parallel execution */ InstrStartParallelQuery(); - /* Process indexes to perform vacuum/cleanup */ - parallel_vacuum_process_safe_indexes(&pvs); + switch (pvs.shared->work_phase) + { + case PV_WORK_PHASE_COLLECT_DEAD_ITEMS: + /* Scan the table to collect dead items */ + parallel_vacuum_process_table(&pvs, state); + break; + case PV_WORK_PHASE_INDEX_VACUUM: + case PV_WORK_PHASE_INDEX_CLEANUP: + /* Bulk-delete or clean up indexes (per-index status decides) */ + parallel_vacuum_process_safe_indexes(&pvs); + break; + default: + elog(ERROR, "unrecognized parallel vacuum phase %d", pvs.shared->work_phase); + } /* Report buffer/WAL usage during parallel execution */ buffer_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, false); @@ -1334,6 +1556,77 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) pv_shared_cost_params = NULL; } +/* + * Launch parallel vacuum workers for the given phase. If at least one + * worker launched, enable the shared vacuum delay costing. + */ +static void +parallel_vacuum_begin_work_phase(ParallelVacuumState *pvs, int nworkers, + PVWorkPhase work_phase) +{ + /* Set the work phase */ + pvs->shared->work_phase = work_phase; + + /* Reinitialize parallel context to relaunch parallel workers */ + if (pvs->need_reinitialize_dsm) + ReinitializeParallelDSM(pvs->pcxt); + + /* + * Set up shared cost balance and the number of active workers for vacuum + * delay. We need to do this before launching workers as otherwise, they + * might not see the updated values for these parameters. + */ + pg_atomic_write_u32(&(pvs->shared->cost_balance), VacuumCostBalance); + pg_atomic_write_u32(&(pvs->shared->active_nworkers), 0); + + /* + * The number of workers can vary between bulkdelete and cleanup phase. + */ + ReinitializeParallelWorkers(pvs->pcxt, nworkers); + + LaunchParallelWorkers(pvs->pcxt); + + /* Enable shared vacuum costing if we are able to launch any worker */ + if (pvs->pcxt->nworkers_launched > 0) + { + /* + * Reset the local cost values for leader backend as we have already + * accumulated the remaining balance of heap. + */ + VacuumCostBalance = 0; + VacuumCostBalanceLocal = 0; + + /* Enable shared cost balance for leader backend */ + VacuumSharedCostBalance = &(pvs->shared->cost_balance); + VacuumActiveNWorkers = &(pvs->shared->active_nworkers); + } +} + +/* + * Wait for parallel vacuum workers to finish, accumulate the statistics, + * and disable shared vacuum delay costing if enabled. + */ +static void +parallel_vacuum_end_worker_phase(ParallelVacuumState *pvs) +{ + /* Wait for all vacuum workers to finish */ + WaitForParallelWorkersToFinish(pvs->pcxt); + + for (int i = 0; i < pvs->pcxt->nworkers_launched; i++) + InstrAccumParallelQuery(&pvs->buffer_usage[i], &pvs->wal_usage[i]); + + /* Carry the shared balance value and disable shared costing */ + if (VacuumSharedCostBalance) + { + VacuumCostBalance = pg_atomic_read_u32(VacuumSharedCostBalance); + VacuumSharedCostBalance = NULL; + VacuumActiveNWorkers = NULL; + } + + /* Parallel DSM will need to be reinitialized for the next execution */ + pvs->need_reinitialize_dsm = true; +} + /* * Error context callback for errors occurring during parallel index vacuum. * The error context messages should match the messages set in the lazy vacuum diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index 5176478c295..9bf17e52f89 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -465,8 +465,16 @@ extern void log_heap_prune_and_freeze(Relation relation, Buffer buffer, OffsetNumber *unused, int nunused); /* in heap/vacuumlazy.c */ +struct ParallelVacuumState; +struct ParallelVacuumCallbacks; extern void heap_vacuum_rel(Relation rel, const VacuumParams *params, BufferAccessStrategy bstrategy); + +/* + * Callback struct that heap registers with vacuumparallel.c for parallel + * table vacuum. vacuumparallel.c resolves this by the AM name "heap". + */ +extern const struct ParallelVacuumCallbacks heap_parallel_vacuum_callbacks; #ifdef USE_ASSERT_CHECKING extern bool heap_page_is_all_visible(Relation rel, Buffer buf, GlobalVisState *vistest, diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h index 956d9cea36d..31d40692f83 100644 --- a/src/include/commands/vacuum.h +++ b/src/include/commands/vacuum.h @@ -67,6 +67,80 @@ /* Abstract type for parallel vacuum state */ typedef struct ParallelVacuumState ParallelVacuumState; +/* forward references for parallel table vacuum callbacks */ +struct ParallelContext; +struct ParallelWorkerContext; + +/* + * Callbacks for parallel table vacuum. + * + * These are the table-AM specific routines that vacuumparallel.c calls to set + * up and perform a parallel scan of a table to collect dead items. Rather than + * living in the table AM (TableAmRoutine), which would mean generic vacuum + * code calling back up through the table-AM boundary it was entered from, an + * access method registers a callback struct under a name and vacuumparallel.c + * looks it up. The leader passes the struct directly to parallel_vacuum_init(); + * workers only receive the name via shared memory and resolve it locally to a + * process-valid pointer (see parallel_vacuum_main()). + * + * compute_workers is optional; returning 0 (or leaving it NULL) disables + * parallel table vacuum, in which case the other callbacks are never called. + * If compute_workers returns a positive worker count, the remaining callbacks + * must all be provided. + */ +typedef struct ParallelVacuumCallbacks +{ + /* + * Compute the number of parallel workers for parallel table vacuum. The + * parallel degree for parallel vacuum is further limited by + * max_parallel_maintenance_workers. The function must return 0 to disable + * parallel table vacuum. + * + * 'nworkers_requested' is a >=0 number and the requested number of + * workers. This comes from the PARALLEL option. 0 means to choose the + * parallel degree based on the table AM specific factors such as table + * size. + */ + int (*compute_workers) (Relation rel, + int nworkers_requested, + void *state); + + /* + * Estimate the size of shared memory needed for a parallel table vacuum + * of this relation. + */ + void (*estimate) (Relation rel, + struct ParallelContext *pcxt, + int nworkers, + void *state); + + /* + * Initialize DSM space for parallel table vacuum. + */ + void (*initialize) (Relation rel, + struct ParallelContext *pcxt, + int nworkers, + void *state); + + /* + * Initialize AM-specific vacuum state for worker processes. + * + * The state_out is the output parameter so that arbitrary data can be + * passed to the subsequent callback, collect_dead_items. + */ + void (*initialize_worker) (Relation rel, + struct ParallelVacuumState *pvs, + struct ParallelWorkerContext *pwcxt, + void **state_out); + + /* + * Execute a parallel scan to collect dead items. + */ + void (*collect_dead_items) (Relation rel, + struct ParallelVacuumState *pvs, + void *state); +} ParallelVacuumCallbacks; + /*---------- * ANALYZE builds one of these structs for each attribute (column) that is * to be analyzed. The struct and subsidiary data are in anl_context, @@ -408,8 +482,14 @@ extern void VacuumUpdateCosts(void); extern ParallelVacuumState *parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, int nrequested_workers, int vac_work_mem, int elevel, - BufferAccessStrategy bstrategy); + BufferAccessStrategy bstrategy, + const ParallelVacuumCallbacks *cbs, + const char *am_name, + 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 BufferAccessStrategy parallel_vacuum_get_bstrategy(ParallelVacuumState *pvs); extern TidStore *parallel_vacuum_get_dead_items(ParallelVacuumState *pvs, VacDeadItemsInfo **dead_items_info_p); extern void parallel_vacuum_reset_dead_items(ParallelVacuumState *pvs); @@ -424,6 +504,8 @@ extern void parallel_vacuum_cleanup_all_indexes(ParallelVacuumState *pvs, PVWorkerStats *wstats); extern void parallel_vacuum_update_shared_delay_params(void); extern void parallel_vacuum_propagate_shared_delay_params(void); +extern int parallel_vacuum_collect_dead_items_begin(ParallelVacuumState *pvs); +extern void parallel_vacuum_collect_dead_items_end(ParallelVacuumState *pvs); extern void parallel_vacuum_main(dsm_segment *seg, shm_toc *toc); /* in commands/analyze.c */ diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index c01d2fb095c..3aca8c389de 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -24,6 +24,8 @@ ISOLATION = basic \ # some isolation tests require wal_level=replica ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/extra.conf +TAP_TESTS = 1 + # The injection points are cluster-wide, so disable installcheck NO_INSTALLCHECK = 1 diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 59dba1cb023..9a33cb4a4b0 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -60,4 +60,9 @@ tests += { '--temp-config', files('extra.conf'), ], }, + 'tap': { + 'tests': [ + 't/parallel_heap_vacuum.pl', + ], + }, } diff --git a/src/test/modules/injection_points/t/parallel_heap_vacuum.pl b/src/test/modules/injection_points/t/parallel_heap_vacuum.pl new file mode 100644 index 00000000000..54bdc4b74c7 --- /dev/null +++ b/src/test/modules/injection_points/t/parallel_heap_vacuum.pl @@ -0,0 +1,138 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Tests for parallel heap vacuum. It checks that the vacuum freezes correctly +# across multiple scan rounds (with injection points forcing worker ramp-down +# and leader resume), and that it plans the expected number of workers for +# different PARALLEL requests and table sizes. + +use strict; +use warnings FATAL => 'all'; +use locale; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# This test requires injection points; skip it when the build lacks them. +if ($ENV{enable_injection_points} ne 'yes') +{ + plan skip_all => 'Injection points not supported by this build'; +} + +my $node = PostgreSQL::Test::Cluster->new('main'); +$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 $psql = $node->background_psql('postgres', on_error_stop => 0); + +# Hold a transaction open to keep its xmin from advancing. +$psql->query_safe('begin; select pg_current_xact_id();'); + +# Advance the XID a few times, each in its own transaction. +$node->safe_psql('postgres', 'select pg_current_xact_id()') for 1 .. 5; + +# Insert most rows under an old XID. +$psql->query_safe(qq[insert into t select generate_series(1, $first);]); + +# Insert a few rows in a newer transaction, enough to fill at least one +# page so it is not frozen by the vacuum below. +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 the remaining rows and commit the open transaction. +$psql->query_safe(qq[insert into t select generate_series($first, $nrows);]); +$psql->query_safe(qq[commit;]); + +# Delete a range of rows so the vacuum has dead items to collect. +$node->safe_psql('postgres', qq[delete from t where i between 1 and 20000;]); + +# Run a parallel vacuum in multiple rounds. The low maintenance_work_mem +# fills the dead-item store repeatedly, and the injection points ramp the +# worker count down and disable leader participation. Everything except the +# newer transaction's rows should freeze, advancing relfrozenxid to its XID. +$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"); + +# Feature coverage for the number of parallel heap vacuum workers, that is, how +# many workers are planned for different PARALLEL requests, table sizes, and the +# max_parallel_maintenance_workers cap. We assert on the "planned" count in the +# VERBOSE output, which is deterministic, rather than the launched count, which +# depends on background worker availability. + +# Return the "planned" heap-vacuum worker count from a VACUUM (VERBOSE) run, or +# undef when no parallel-heap-vacuum message was emitted (i.e. serial). +sub planned_table_workers +{ + my ($sql) = @_; + my ($ret, $stdout, $stderr) = $node->psql('postgres', $sql); + is($ret, 0, "vacuum ran: $sql"); + if ($stderr =~ /for collecting dead tuples \(planned: (\d+)\)/) + { + return $1; + } + return undef; +} + +# PARALLEL 0 disables parallel heap vacuum, so there is no message at all. +is( planned_table_workers( + 'set min_parallel_table_scan_size to "128kB"; vacuum (parallel 0, verbose) t;' + ), + undef, + 'PARALLEL 0 launches no parallel heap vacuum workers'); + +# An explicit degree is honored (below the max_parallel_maintenance_workers cap). +is( planned_table_workers( + 'set min_parallel_table_scan_size to "128kB"; set max_parallel_maintenance_workers to 4; vacuum (parallel 2, verbose) t;' + ), + 2, + 'PARALLEL 2 plans 2 parallel heap vacuum workers'); + +# The request is capped by max_parallel_maintenance_workers. +is( planned_table_workers( + 'set min_parallel_table_scan_size to "128kB"; set max_parallel_maintenance_workers to 3; vacuum (parallel 8, verbose) t;' + ), + 3, + 'PARALLEL request is capped by max_parallel_maintenance_workers'); + +# A table below min_parallel_table_scan_size is vacuumed serially even when +# parallelism is requested. +$node->safe_psql('postgres', qq[ + create table small (i int) with (autovacuum_enabled = off); + insert into small select generate_series(1, 100); +]); +is( planned_table_workers( + 'set min_parallel_table_scan_size to "1GB"; vacuum (parallel 4, verbose) small;' + ), + undef, + 'small table is not vacuumed with parallel heap workers'); + +$node->stop; +done_testing(); diff --git a/src/test/regress/expected/vacuum_parallel.out b/src/test/regress/expected/vacuum_parallel.out index ddf0ee544b7..b793d8093c2 100644 --- a/src/test/regress/expected/vacuum_parallel.out +++ b/src/test/regress/expected/vacuum_parallel.out @@ -1,5 +1,6 @@ SET max_parallel_maintenance_workers TO 4; SET min_parallel_index_scan_size TO '128kB'; +SET min_parallel_table_scan_size TO '128kB'; -- Bug #17245: Make sure that we don't totally fail to VACUUM individual indexes that -- happen to be below min_parallel_index_scan_size during parallel VACUUM: CREATE TABLE parallel_vacuum_table (a int) WITH (autovacuum_enabled = off); @@ -43,7 +44,13 @@ VACUUM (PARALLEL 4, INDEX_CLEANUP ON) parallel_vacuum_table; -- Since vacuum_in_leader_small_index uses deduplication, we expect an -- assertion failure with bug #17245 (in the absence of bugfix): INSERT INTO parallel_vacuum_table SELECT i FROM generate_series(1, 10000) i; +-- Insert more tuples to use parallel heap vacuum. +INSERT INTO parallel_vacuum_table SELECT i FROM generate_series(1, 500_000) i; +VACUUM (PARALLEL 2) parallel_vacuum_table; +DELETE FROM parallel_vacuum_table WHERE a < 1000; +VACUUM (PARALLEL 1) parallel_vacuum_table; RESET max_parallel_maintenance_workers; RESET min_parallel_index_scan_size; +RESET min_parallel_table_scan_size; -- Deliberately don't drop table, to get further coverage from tools like -- pg_amcheck in some testing scenarios diff --git a/src/test/regress/sql/vacuum_parallel.sql b/src/test/regress/sql/vacuum_parallel.sql index 1d23f33e39c..5381023642f 100644 --- a/src/test/regress/sql/vacuum_parallel.sql +++ b/src/test/regress/sql/vacuum_parallel.sql @@ -1,5 +1,6 @@ SET max_parallel_maintenance_workers TO 4; SET min_parallel_index_scan_size TO '128kB'; +SET min_parallel_table_scan_size TO '128kB'; -- Bug #17245: Make sure that we don't totally fail to VACUUM individual indexes that -- happen to be below min_parallel_index_scan_size during parallel VACUUM: @@ -39,8 +40,15 @@ VACUUM (PARALLEL 4, INDEX_CLEANUP ON) parallel_vacuum_table; -- assertion failure with bug #17245 (in the absence of bugfix): INSERT INTO parallel_vacuum_table SELECT i FROM generate_series(1, 10000) i; +-- Insert more tuples to use parallel heap vacuum. +INSERT INTO parallel_vacuum_table SELECT i FROM generate_series(1, 500_000) i; +VACUUM (PARALLEL 2) parallel_vacuum_table; +DELETE FROM parallel_vacuum_table WHERE a < 1000; +VACUUM (PARALLEL 1) parallel_vacuum_table; + RESET max_parallel_maintenance_workers; RESET min_parallel_index_scan_size; +RESET min_parallel_table_scan_size; -- Deliberately don't drop table, to get further coverage from tools like -- pg_amcheck in some testing scenarios diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 223dfc80417..1ddc334e2e3 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2063,6 +2063,11 @@ PLpgSQL_type PLpgSQL_type_type PLpgSQL_var PLpgSQL_variable +ParallelLVLeader +ParallelLVScanDesc +ParallelLVScanWorkerData +ParallelLVShared +ParallelLVState PLwdatum PLword PLyArrayToOb @@ -2137,6 +2142,7 @@ PVIndVacStatus PVOID PVShared PVSharedCostParams +PVWorkPhase PVWorkerStats PVWorkerUsage PX_Alias @@ -2175,6 +2181,7 @@ ParallelState ParallelTableScanDesc ParallelTableScanDescData ParallelTransState +ParallelVacuumCallbacks ParallelVacuumState ParallelWorkerContext ParallelWorkerInfo @@ -3358,6 +3365,7 @@ VacAttrStatsP VacDeadItemsInfo VacErrPhase VacOptValue +VacuumCallbacksEntry VacuumCutoffs VacuumParams VacuumRelation -- 2.47.3