From fe37433fd4304ddf19be1f8c2f637f79d7fc2ebb Mon Sep 17 00:00:00 2001 From: "Sami Imseih (AWS)" Date: Wed, 9 Mar 2022 01:54:08 +0000 Subject: [PATCH v3 1/3] Show progress for index vacuums Add 2 new columns to pg_stat_progress_vacuum. The columns are indexes_total as the total indexes to be vacuumed or cleaned and indexes_processed as the number of indexes vacuumed or cleaned up so far. Author: Sami Imseih, based on suggestions by Nathan Bossart, Peter Geoghegan and Masahiko Sawada Reviewed by: Nathan Bossart, Justin Pryzby Discussion: https://www.postgresql.org/message-id/flat/5478DFCD-2333-401A-B2F0-0D186AB09228@amazon.com --- doc/src/sgml/monitoring.sgml | 24 ++++ src/backend/access/heap/vacuumlazy.c | 156 ++++++++++++++++++++++- src/backend/catalog/system_views.sql | 3 +- src/backend/commands/vacuumparallel.c | 7 + src/backend/storage/ipc/ipci.c | 2 + src/backend/storage/lmgr/lwlocknames.txt | 1 + src/backend/utils/adt/pgstatfuncs.c | 16 ++- src/include/commands/progress.h | 5 + src/include/commands/vacuum.h | 7 + src/test/regress/expected/rules.out | 4 +- 10 files changed, 215 insertions(+), 10 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 9fb62fec8e..140fbd5fb9 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -6227,6 +6227,30 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid, Number of dead tuples collected since the last index vacuum cycle. + + + + indexes_total bigint + + + The number of indexes to be processed in the + vacuuming indexes + or cleaning up indexes phase. It is set to + 0 when vacuum is not in any of these phases. + + + + + + indexes_processed bigint + + + The number of indexes already processed in the + vacuuming indexes + or cleaning up indexes phase. It is set to + 0 when vacuum is not in any of these phases. + + diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 40101e0cb8..4c91e673d1 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -58,6 +58,7 @@ #include "postmaster/autovacuum.h" #include "storage/bufmgr.h" #include "storage/freespace.h" +#include "storage/ipc.h" #include "storage/lmgr.h" #include "tcop/tcopprot.h" #include "utils/lsyscache.h" @@ -244,6 +245,16 @@ typedef struct LVSavedErrInfo VacErrPhase phase; } LVSavedErrInfo; +/* + * Structs for tracking shared Progress information + * amongst worker ( and leader ) processes of a vacuum. + */ +typedef struct VacProgressEntry +{ + pid_t leader_pid; + int indexes_total; + int indexes_processed; +} VacProgressEntry; /* non-export function prototypes */ static void lazy_scan_heap(LVRelState *vacrel, int nworkers); @@ -290,6 +301,7 @@ static void update_vacuum_error_info(LVRelState *vacrel, static void restore_vacuum_error_info(LVRelState *vacrel, const LVSavedErrInfo *saved_vacrel); +static HTAB *VacuumWorkerProgressHash; /* * heap_vacuum_rel() -- perform VACUUM for one heap relation @@ -2305,9 +2317,13 @@ lazy_vacuum_all_indexes(LVRelState *vacrel) /* Report that we are now vacuuming indexes */ pgstat_progress_update_param(PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_VACUUM_INDEX); + /* Advertise the number of indexes we are vacuuming */ + pgstat_progress_update_param(PROGRESS_VACUUM_TOTAL_INDEXES, vacrel->nindexes); if (!ParallelVacuumIsActive(vacrel)) { + int indexes_completed = 1; + for (int idx = 0; idx < vacrel->nindexes; idx++) { Relation indrel = vacrel->indrels[idx]; @@ -2317,6 +2333,8 @@ lazy_vacuum_all_indexes(LVRelState *vacrel) lazy_vacuum_one_index(indrel, istat, vacrel->old_live_tuples, vacrel); + pgstat_progress_update_param(PROGRESS_VACUUM_INDEXES_COMPLETED, indexes_completed++); + if (lazy_check_wraparound_failsafe(vacrel)) { /* Wraparound emergency -- end current index scan */ @@ -2327,9 +2345,22 @@ lazy_vacuum_all_indexes(LVRelState *vacrel) } else { - /* Outsource everything to parallel variant */ - parallel_vacuum_bulkdel_all_indexes(vacrel->pvs, vacrel->old_live_tuples, - vacrel->num_index_scans); + /* + * Outsource everything to parallel variant. + * + * * parallel_vacuum_bulkdel_all_indexes will call vacuum_worker_update + * which updates shared memory for the index progress. To ensure shared + * memory cleanup, do the work with PG_ENSURE_ERROR_CLEANUP. + */ + PG_ENSURE_ERROR_CLEANUP(vacuum_worker_end_callback, Int32GetDatum(MyProcPid)); + { + + parallel_vacuum_bulkdel_all_indexes(vacrel->pvs, + vacrel->old_live_tuples, + vacrel->num_index_scans); + } + PG_END_ENSURE_ERROR_CLEANUP(vacuum_worker_end_callback, Int32GetDatum(MyProcPid)); + vacuum_worker_end(MyProcPid); /* * Do a postcheck to consider applying wraparound failsafe now. Note @@ -2339,6 +2370,10 @@ lazy_vacuum_all_indexes(LVRelState *vacrel) allindexes = false; } + /* reset index progress */ + pgstat_progress_update_param(PROGRESS_VACUUM_TOTAL_INDEXES, 0); + pgstat_progress_update_param(PROGRESS_VACUUM_INDEXES_COMPLETED, 0); + /* * We delete all LP_DEAD items from the first heap pass in all indexes on * each call here (except calls where we choose to do the failsafe). This @@ -2617,6 +2652,8 @@ lazy_check_wraparound_failsafe(LVRelState *vacrel) vacrel->do_index_vacuuming = false; vacrel->do_index_cleanup = false; vacrel->do_rel_truncate = false; + pgstat_progress_update_param(PROGRESS_VACUUM_TOTAL_INDEXES, 0); + pgstat_progress_update_param(PROGRESS_VACUUM_INDEXES_COMPLETED, 0); ereport(WARNING, (errmsg("bypassing nonessential maintenance of table \"%s.%s.%s\" as a failsafe after %d index scans", @@ -2649,12 +2686,15 @@ lazy_cleanup_all_indexes(LVRelState *vacrel) /* Report that we are now cleaning up indexes */ pgstat_progress_update_param(PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_INDEX_CLEANUP); + /* Advertise the number of indexes we are cleaning up */ + pgstat_progress_update_param(PROGRESS_VACUUM_TOTAL_INDEXES, vacrel->nindexes); if (!ParallelVacuumIsActive(vacrel)) { double reltuples = vacrel->new_rel_tuples; bool estimated_count = vacrel->scanned_pages < vacrel->rel_pages; + int indexes_completed = 1; for (int idx = 0; idx < vacrel->nindexes; idx++) { @@ -2664,15 +2704,32 @@ lazy_cleanup_all_indexes(LVRelState *vacrel) vacrel->indstats[idx] = lazy_cleanup_one_index(indrel, istat, reltuples, estimated_count, vacrel); + + pgstat_progress_update_param(PROGRESS_VACUUM_INDEXES_COMPLETED, indexes_completed++); + } } else { - /* Outsource everything to parallel variant */ - parallel_vacuum_cleanup_all_indexes(vacrel->pvs, vacrel->new_rel_tuples, + /* + * Outsource everything to parallel variant + * + * See the lazy_vacuum_all_indexes comments + */ + PG_ENSURE_ERROR_CLEANUP(vacuum_worker_end_callback, Int32GetDatum(MyProcPid)); + { + parallel_vacuum_cleanup_all_indexes(vacrel->pvs, vacrel->new_rel_tuples, vacrel->num_index_scans, (vacrel->scanned_pages < vacrel->rel_pages)); + } + PG_END_ENSURE_ERROR_CLEANUP(vacuum_worker_end_callback, Int32GetDatum(MyProcPid)); + vacuum_worker_end(MyProcPid); } + + /* reset index progress */ + pgstat_progress_update_param(PROGRESS_VACUUM_TOTAL_INDEXES, 0); + pgstat_progress_update_param(PROGRESS_VACUUM_INDEXES_COMPLETED, 0); + } /* @@ -3453,3 +3510,92 @@ restore_vacuum_error_info(LVRelState *vacrel, vacrel->offnum = saved_vacrel->offnum; vacrel->phase = saved_vacrel->phase; } + +/* + * vacuum_worker_update --- sets the number of indexes processed so far + * in a parallel vacuum. + */ +void +vacuum_worker_update(int leader_pid) +{ + VacProgressEntry *entry; + bool found; + + LWLockAcquire(VacuumWorkerProgressLock, LW_EXCLUSIVE); + + entry = (VacProgressEntry *) hash_search(VacuumWorkerProgressHash, &leader_pid, HASH_ENTER_NULL, &found); + + if (!entry) + elog(ERROR, "cannot allocate shared memory for vacuum worker progress"); + + if (!found) + entry->indexes_processed = 1; + else + entry->indexes_processed++; + + LWLockRelease(VacuumWorkerProgressLock); +} + +/* + * vacuum_worker_end --- remove the leader_pid of a completed parallel vacuum + */ +void +vacuum_worker_end(int leader_pid) +{ + LWLockAcquire(VacuumWorkerProgressLock, LW_EXCLUSIVE); + + hash_search(VacuumWorkerProgressHash, &leader_pid, HASH_REMOVE, NULL); + + LWLockRelease(VacuumWorkerProgressLock); +} + +/* + * vacuum_worker_end wrapped as an on_shmem_exit callback function + */ +void +vacuum_worker_end_callback(int code, Datum arg) +{ + vacuum_worker_end(DatumGetInt32(arg)); +} + +/* + * set_vaccum_worker_progress --- updates the number of indexes that have been + * vacuumed or cleaned up in a parallel vacuum. + */ +void +set_vaccum_worker_progress(Datum *values) +{ + VacProgressEntry *entry; + int leader_pid = values[0]; + + LWLockAcquire(VacuumWorkerProgressLock, LW_SHARED); + + entry = (VacProgressEntry *) hash_search(VacuumWorkerProgressHash, &leader_pid, HASH_FIND, NULL); + + if (entry != NULL) + values[PGSTAT_NUM_PROGRESS_COMMON + PROGRESS_VACUUM_INDEXES_COMPLETED] = entry->indexes_processed; + + LWLockRelease(VacuumWorkerProgressLock); +} + +/* + * vacuum_worker_init --- initialize this module's shared memory hash + * to track the progress of a vacuum worker + */ +void +vacuum_worker_init(void) +{ + HASHCTL info; + long max_table_size = GetMaxBackends(); + + VacuumWorkerProgressHash = NULL; + + info.keysize = sizeof(pid_t); + info.entrysize = sizeof(VacProgressEntry); + + VacuumWorkerProgressHash = ShmemInitHash("Vacuum Progress Hash", + max_table_size, + max_table_size, + &info, + HASH_ELEM | HASH_BLOBS); +} diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 40b7bca5a9..04ce4a45d1 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1126,7 +1126,8 @@ CREATE VIEW pg_stat_progress_vacuum AS END AS phase, S.param2 AS heap_blks_total, S.param3 AS heap_blks_scanned, S.param4 AS heap_blks_vacuumed, S.param5 AS index_vacuum_count, - S.param6 AS max_dead_tuples, S.param7 AS num_dead_tuples + S.param6 AS max_dead_tuples, S.param7 AS num_dead_tuples, + S.param8 AS indexes_total, S.param9 AS indexes_processed FROM pg_stat_get_progress_info('VACUUM') AS S LEFT JOIN pg_database D ON S.datid = D.oid; diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index 974a29e7a9..9b465e12cc 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/src/backend/commands/vacuumparallel.c @@ -29,6 +29,7 @@ #include "access/amapi.h" #include "access/table.h" #include "catalog/index.h" +#include "commands/progress.h" #include "commands/vacuum.h" #include "optimizer/paths.h" #include "pgstat.h" @@ -101,6 +102,9 @@ typedef struct PVShared /* Counter for vacuuming and cleanup */ pg_atomic_uint32 idx; + + /* Leader PID of the vacuum */ + int leader_pid; } PVShared; /* Status used during parallel index vacuum or cleanup */ @@ -357,6 +361,7 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, (nindexes_mwm > 0) ? maintenance_work_mem / Min(parallel_workers, nindexes_mwm) : maintenance_work_mem; + shared->leader_pid = MyProcPid; pg_atomic_init_u32(&(shared->cost_balance), 0); pg_atomic_init_u32(&(shared->active_nworkers), 0); @@ -844,9 +849,11 @@ parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation indrel, { case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE: istat_res = vac_bulkdel_one_index(&ivinfo, istat, pvs->dead_items); + vacuum_worker_update(pvs->shared->leader_pid); break; case PARALLEL_INDVAC_STATUS_NEED_CLEANUP: istat_res = vac_cleanup_one_index(&ivinfo, istat); + vacuum_worker_update(pvs->shared->leader_pid); break; default: elog(ERROR, "unexpected parallel vacuum index status %d for index \"%s\"", diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index cd4ebe2fc5..83ecad4e4a 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -24,6 +24,7 @@ #include "access/twophase.h" #include "access/xlogrecovery.h" #include "commands/async.h" +#include "commands/vacuum.h" #include "miscadmin.h" #include "pgstat.h" #include "postmaster/autovacuum.h" @@ -296,6 +297,7 @@ CreateSharedMemoryAndSemaphores(void) BTreeShmemInit(); SyncScanShmemInit(); AsyncShmemInit(); + vacuum_worker_init(); #ifdef EXEC_BACKEND diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt index 6c7cf6c295..9a5fa0a0e0 100644 --- a/src/backend/storage/lmgr/lwlocknames.txt +++ b/src/backend/storage/lmgr/lwlocknames.txt @@ -53,3 +53,4 @@ XactTruncationLock 44 # 45 was XactTruncationLock until removal of BackendRandomLock WrapLimitsVacuumLock 46 NotifyQueueTailLock 47 +VacuumWorkerProgressLock 48 diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index eff45b16f2..ad7327fbd9 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -18,6 +18,8 @@ #include "access/xlog.h" #include "catalog/pg_authid.h" #include "catalog/pg_type.h" +#include "commands/progress.h" +#include "commands/vacuum.h" #include "common/ip.h" #include "funcapi.h" #include "miscadmin.h" @@ -452,11 +454,15 @@ pg_stat_get_backend_idset(PG_FUNCTION_ARGS) /* * Returns command progress information for the named command. + * + * A command type can optionally define a callback function + * which will derive Datum values rather than use values + * directly from the backends progress array. */ Datum pg_stat_get_progress_info(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_PROGRESS_COLS PGSTAT_NUM_PROGRESS_PARAM + 3 +#define PG_STAT_GET_PROGRESS_COLS PGSTAT_NUM_PROGRESS_PARAM + PGSTAT_NUM_PROGRESS_COMMON int num_backends = pgstat_fetch_stat_numbackends(); int curr_backend; char *cmd = text_to_cstring(PG_GETARG_TEXT_PP(0)); @@ -518,15 +524,19 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS) { values[2] = ObjectIdGetDatum(beentry->st_progress_command_target); for (i = 0; i < PGSTAT_NUM_PROGRESS_PARAM; i++) - values[i + 3] = Int64GetDatum(beentry->st_progress_param[i]); + values[i + PGSTAT_NUM_PROGRESS_COMMON] = Int64GetDatum(beentry->st_progress_param[i]); } else { nulls[2] = true; for (i = 0; i < PGSTAT_NUM_PROGRESS_PARAM; i++) - nulls[i + 3] = true; + nulls[i + PGSTAT_NUM_PROGRESS_COMMON] = true; } + /* Call the command specific function to override datum values */ + if (pg_strcasecmp(cmd, "VACUUM") == 0) + set_vaccum_worker_progress(values); + tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); } diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h index a28938caf4..c0dd1d7ab5 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -25,6 +25,8 @@ #define PROGRESS_VACUUM_NUM_INDEX_VACUUMS 4 #define PROGRESS_VACUUM_MAX_DEAD_TUPLES 5 #define PROGRESS_VACUUM_NUM_DEAD_TUPLES 6 +#define PROGRESS_VACUUM_TOTAL_INDEXES 7 +#define PROGRESS_VACUUM_INDEXES_COMPLETED 8 /* Phases of vacuum (as advertised via PROGRESS_VACUUM_PHASE) */ #define PROGRESS_VACUUM_PHASE_SCAN_HEAP 1 @@ -151,4 +153,7 @@ #define PROGRESS_COPY_TYPE_PIPE 3 #define PROGRESS_COPY_TYPE_CALLBACK 4 +/* Number of common fields at the start of progress views */ +#define PGSTAT_NUM_PROGRESS_COMMON 3 + #endif diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h index d64f6268f2..8661516ae2 100644 --- a/src/include/commands/vacuum.h +++ b/src/include/commands/vacuum.h @@ -336,4 +336,11 @@ extern double anl_random_fract(void); extern double anl_init_selection_state(int n); extern double anl_get_next_S(double t, int n, double *stateptr); +/* in commands/vacuumparallel.c */ +extern void vacuum_worker_init(void); +extern void vacuum_worker_end(int leader_pid); +extern void vacuum_worker_update(int leader_pid); +extern void vacuum_worker_end_callback(int code, Datum arg); +extern void set_vaccum_worker_progress(Datum *values); + #endif /* VACUUM_H */ diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index ac468568a1..d70a176514 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2002,7 +2002,9 @@ pg_stat_progress_vacuum| SELECT s.pid, s.param4 AS heap_blks_vacuumed, s.param5 AS index_vacuum_count, s.param6 AS max_dead_tuples, - s.param7 AS num_dead_tuples + s.param7 AS num_dead_tuples, + s.param8 AS indexes_total, + s.param9 AS indexes_processed FROM (pg_stat_get_progress_info('VACUUM'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20) LEFT JOIN pg_database d ON ((s.datid = d.oid))); pg_stat_replication| SELECT s.pid, -- 2.32.0