Add two attributes to pg_stat_database for parallel workers activity

Two attributes are added to pg_stat_database:
* parallel_workers_to_launch, counting the total number of parallel
workers that were planned to be launched.
* parallel_workers_launched, counting the total number of parallel
workers actually launched.

The ratio of both fields can provide hints that there are not enough
slots available when launching parallel workers, also useful when
pg_stat_statements is not deployed on an instance (i.e. cf54a2c00254).

This commit relies on de3a2ea3b264, that has added two fields to EState,
that get incremented when executing Gather or GatherMerge nodes.

A test is added in select_parallel, where parallel workers are spawned.

Bump catalog version.

Author: Benoit Lobréau
Discussion: https://postgr.es/m/783bc7f7-659a-42fa-99dd-ee0565644e25@dalibo.com

10 files changed, 90 insertions, 1 deletions

diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 3456b821bc5..da9a8fe99f2 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1073,6 +1073,8 @@ CREATE VIEW pg_stat_database AS
             pg_stat_get_db_sessions_abandoned(D.oid) AS sessions_abandoned,
             pg_stat_get_db_sessions_fatal(D.oid) AS sessions_fatal,
             pg_stat_get_db_sessions_killed(D.oid) AS sessions_killed,
+            pg_stat_get_db_parallel_workers_to_launch(D.oid) as parallel_workers_to_launch,
+            pg_stat_get_db_parallel_workers_launched(D.oid) as parallel_workers_launched,
             pg_stat_get_db_stat_reset_time(D.oid) AS stats_reset
     FROM (
         SELECT 0 AS oid, NULL::name AS datname
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index cc9a594cba5..5ca856fd279 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -52,6 +52,7 @@
 #include "miscadmin.h"
 #include "nodes/queryjumble.h"
 #include "parser/parse_relation.h"
+#include "pgstat.h"
 #include "rewrite/rewriteHandler.h"
 #include "tcop/utility.h"
 #include "utils/acl.h"
@@ -483,6 +484,10 @@ standard_ExecutorEnd(QueryDesc *queryDesc)
 
 	Assert(estate != NULL);
 
+	if (estate->es_parallel_workers_to_launch > 0)
+		pgstat_update_parallel_workers_stats((PgStat_Counter) estate->es_parallel_workers_to_launch,
+											 (PgStat_Counter) estate->es_parallel_workers_launched);
+
 	/*
 	 * Check that ExecutorFinish was called, unless in EXPLAIN-only mode. This
 	 * Assert is needed because ExecutorFinish is new as of 9.1, and callers
diff --git a/src/backend/utils/activity/pgstat_database.c b/src/backend/utils/activity/pgstat_database.c
index 29bc0909748..7757d2ace74 100644
--- a/src/backend/utils/activity/pgstat_database.c
+++ b/src/backend/utils/activity/pgstat_database.c
@@ -263,6 +263,23 @@ AtEOXact_PgStat_Database(bool isCommit, bool parallel)
 }
 
 /*
+ * Notify the stats system about parallel worker information.
+ */
+void
+pgstat_update_parallel_workers_stats(PgStat_Counter workers_to_launch,
+									 PgStat_Counter workers_launched)
+{
+	PgStat_StatDBEntry *dbentry;
+
+	if (!OidIsValid(MyDatabaseId))
+		return;
+
+	dbentry = pgstat_prep_database_pending(MyDatabaseId);
+	dbentry->parallel_workers_to_launch += workers_to_launch;
+	dbentry->parallel_workers_launched += workers_launched;
+}
+
+/*
  * Subroutine for pgstat_report_stat(): Handle xact commit/rollback and I/O
  * timings.
  */
@@ -425,6 +442,8 @@ pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
 	PGSTAT_ACCUM_DBCOUNT(sessions_abandoned);
 	PGSTAT_ACCUM_DBCOUNT(sessions_fatal);
 	PGSTAT_ACCUM_DBCOUNT(sessions_killed);
+	PGSTAT_ACCUM_DBCOUNT(parallel_workers_to_launch);
+	PGSTAT_ACCUM_DBCOUNT(parallel_workers_launched);
 #undef PGSTAT_ACCUM_DBCOUNT
 
 	pgstat_unlock_entry(entry_ref);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index f7b50e0b5af..60a397dc561 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1039,6 +1039,12 @@ PG_STAT_GET_DBENTRY_INT64(sessions_fatal)
 /* pg_stat_get_db_sessions_killed */
 PG_STAT_GET_DBENTRY_INT64(sessions_killed)
 
+/* pg_stat_get_db_parallel_workers_to_launch */
+PG_STAT_GET_DBENTRY_INT64(parallel_workers_to_launch)
+
+/* pg_stat_get_db_parallel_workers_launched */
+PG_STAT_GET_DBENTRY_INT64(parallel_workers_launched)
+
 /* pg_stat_get_db_temp_bytes */
 PG_STAT_GET_DBENTRY_INT64(temp_bytes)
 
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 86436e03566..5dd91e190ae 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -57,6 +57,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	202411081
+#define CATALOG_VERSION_NO	202411111
 
 #endif
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index f23321a41f1..cbbe8acd382 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5813,6 +5813,16 @@
   proname => 'pg_stat_get_db_sessions_killed', provolatile => 's',
   proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
   prosrc => 'pg_stat_get_db_sessions_killed' },
+{ oid => '8403',
+  descr => 'statistics: number of parallel workers planned to be launched by queries',
+  proname => 'pg_stat_get_db_parallel_workers_to_launch', provolatile => 's',
+  proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+  prosrc => 'pg_stat_get_db_parallel_workers_to_launch' },
+{ oid => '8404',
+  descr => 'statistics: number of parallel workers effectively launched by queries',
+  proname => 'pg_stat_get_db_parallel_workers_launched', provolatile => 's',
+  proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+  prosrc => 'pg_stat_get_db_parallel_workers_launched' },
 { oid => '3195', descr => 'statistics: information about WAL archiver',
   proname => 'pg_stat_get_archiver', proisstrict => 'f', provolatile => 's',
   proparallel => 'r', prorettype => 'record', proargtypes => '',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index df53fa2d4f9..59c28b4aca8 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -386,6 +386,8 @@ typedef struct PgStat_StatDBEntry
 	PgStat_Counter sessions_abandoned;
 	PgStat_Counter sessions_fatal;
 	PgStat_Counter sessions_killed;
+	PgStat_Counter parallel_workers_to_launch;
+	PgStat_Counter parallel_workers_launched;
 
 	TimestampTz stat_reset_timestamp;
 } PgStat_StatDBEntry;
@@ -583,6 +585,8 @@ extern void pgstat_report_deadlock(void);
 extern void pgstat_report_checksum_failures_in_db(Oid dboid, int failurecount);
 extern void pgstat_report_checksum_failure(void);
 extern void pgstat_report_connect(Oid dboid);
+extern void pgstat_update_parallel_workers_stats(PgStat_Counter workers_to_launch,
+												 PgStat_Counter workers_launched);
 
 #define pgstat_count_buffer_read_time(n)							\
 	(pgStatBlockReadTime += (n))
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2b47013f113..3014d047fef 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1863,6 +1863,8 @@ pg_stat_database| SELECT oid AS datid,
     pg_stat_get_db_sessions_abandoned(oid) AS sessions_abandoned,
     pg_stat_get_db_sessions_fatal(oid) AS sessions_fatal,
     pg_stat_get_db_sessions_killed(oid) AS sessions_killed,
+    pg_stat_get_db_parallel_workers_to_launch(oid) AS parallel_workers_to_launch,
+    pg_stat_get_db_parallel_workers_launched(oid) AS parallel_workers_launched,
     pg_stat_get_db_stat_reset_time(oid) AS stats_reset
    FROM ( SELECT 0 AS oid,
             NULL::name AS datname
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index d17ade278b2..8c31f6460d3 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -1,6 +1,17 @@
 --
 -- PARALLEL
 --
+-- Save parallel worker stats, used for comparison at the end
+select pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+select parallel_workers_to_launch as parallel_workers_to_launch_before,
+       parallel_workers_launched as parallel_workers_launched_before
+  from pg_stat_database
+  where datname = current_database() \gset
 create function sp_parallel_restricted(int) returns int as
   $$begin return $1; end$$ language plpgsql parallel restricted;
 begin;
@@ -1407,3 +1418,19 @@ CREATE UNIQUE INDEX parallel_hang_idx
 SET debug_parallel_query = on;
 DELETE FROM parallel_hang WHERE 380 <= i AND i <= 420;
 ROLLBACK;
+-- Check parallel worker stats
+select pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+select parallel_workers_to_launch > :'parallel_workers_to_launch_before'  AS wrk_to_launch,
+       parallel_workers_launched > :'parallel_workers_launched_before' AS wrk_launched
+  from pg_stat_database
+  where datname = current_database();
+ wrk_to_launch | wrk_launched 
+---------------+--------------
+ t             | t
+(1 row)
+
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 9ba1328fd2e..5b4a6e1088f 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -2,6 +2,13 @@
 -- PARALLEL
 --
 
+-- Save parallel worker stats, used for comparison at the end
+select pg_stat_force_next_flush();
+select parallel_workers_to_launch as parallel_workers_to_launch_before,
+       parallel_workers_launched as parallel_workers_launched_before
+  from pg_stat_database
+  where datname = current_database() \gset
+
 create function sp_parallel_restricted(int) returns int as
   $$begin return $1; end$$ language plpgsql parallel restricted;
 
@@ -574,3 +581,10 @@ SET debug_parallel_query = on;
 DELETE FROM parallel_hang WHERE 380 <= i AND i <= 420;
 
 ROLLBACK;
+
+-- Check parallel worker stats
+select pg_stat_force_next_flush();
+select parallel_workers_to_launch > :'parallel_workers_to_launch_before'  AS wrk_to_launch,
+       parallel_workers_launched > :'parallel_workers_launched_before' AS wrk_launched
+  from pg_stat_database
+  where datname = current_database();