v2-0009-Expose-progress-for-the-vacuuming-indexes-and-cleani.patch

From 996254bdbe792bbf1cabb914f3eab883978911a0 Mon Sep 17 00:00:00 2001
From: "Sami Imseih (AWS)" <simseih@amazon.com>
Date: Thu, 3 Mar 2022 04:37:48 +0000
Subject: [PATCH 1/1] Expose progress for the "vacuuming indexes" and "cleaning
 up indexes" phase of a VACUUM operation.

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.
       </para></entry>
      </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>indexes_total</structfield> <type>bigint</type>
+      </para>
+      <para>
+       The number of indexes to be processed in the 
+       <literal>vacuuming indexes</literal> 
+       or <literal>cleaning up indexes</literal> phase. It is set to
+       <literal>0</literal> when vacuum is not in any of these phases.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>indexes_processed</structfield> <type>bigint</type>
+      </para>
+      <para>
+       The number of indexes already processed in the
+       <literal>vacuuming indexes</literal>
+       or <literal>cleaning up indexes</literal> phase. It is set to
+       <literal>0</literal> when vacuum is not in any of these phases.
+      </para></entry>
+     </row>
     </tbody>
    </tgroup>
   </table>
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 40101e0cb8..e9c081ada6 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 = 0;
+
 		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 = 0;
 
 		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 fd993d0d5f..f90c39d9b9 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));
@@ -543,15 +549,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(tupstore, tupdesc, 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