Hi,
This new version avoids updating the stats for non parallel queries.
I noticed that the tests are still not stable. I tried using tenk2
but fail to have stable plans. I'd love to have pointers on that front.
--
Benoit Lobréau
Consultant
http://dalibo.com
From 5e4401c865f77ed447b8b3f25aac0ffa9af0d700 Mon Sep 17 00:00:00 2001
From: benoit <benoit.lobr...@dalibo.com>
Date: Wed, 28 Aug 2024 02:27:13 +0200
Subject: [PATCH] Adds four parallel workers stat columns to pg_stat_database
* parallel_workers_planned
* parallel_workers_launched
* parallel_maint_workers_planned
* parallel_maint_workers_launched
---
doc/src/sgml/monitoring.sgml | 36 ++++++++++++++++++++
src/backend/access/brin/brin.c | 4 +++
src/backend/access/nbtree/nbtsort.c | 4 +++
src/backend/catalog/system_views.sql | 4 +++
src/backend/commands/vacuumparallel.c | 5 +++
src/backend/executor/execMain.c | 7 ++++
src/backend/executor/execUtils.c | 3 ++
src/backend/executor/nodeGather.c | 3 ++
src/backend/executor/nodeGatherMerge.c | 3 ++
src/backend/utils/activity/pgstat_database.c | 36 ++++++++++++++++++++
src/backend/utils/adt/pgstatfuncs.c | 12 +++++++
src/include/catalog/pg_proc.dat | 20 +++++++++++
src/include/nodes/execnodes.h | 3 ++
src/include/pgstat.h | 7 ++++
src/test/regress/expected/rules.out | 4 +++
src/test/regress/expected/stats.out | 17 +++++++++
src/test/regress/sql/stats.sql | 14 ++++++++
17 files changed, 182 insertions(+)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 55417a6fa9..8c4b11c11d 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -3508,6 +3508,42 @@ description | Waiting for a newly initialized WAL file
to reach durable storage
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>parallel_workers_planned</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of parallel workers planned by queries on this database
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>parallel_workers_launched</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of parallel workers obtained by queries on this database
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>parallel_maint_workers_planned</structfield>
<type>bigint</type>
+ </para>
+ <para>
+ Number of parallel workers planned by utilities on this database
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>parallel_maint_workers_launched</structfield>
<type>bigint</type>
+ </para>
+ <para>
+ Number of parallel workers obtained by utilities on this database
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>stats_reset</structfield> <type>timestamp with time
zone</type>
diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index 6467bed604..9eceb87b52 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -2540,6 +2540,10 @@ _brin_end_parallel(BrinLeader *brinleader,
BrinBuildState *state)
/* Shutdown worker processes */
WaitForParallelWorkersToFinish(brinleader->pcxt);
+ pgstat_update_parallel_maint_workers_stats(
+ (PgStat_Counter) brinleader->pcxt->nworkers_to_launch,
+ (PgStat_Counter) brinleader->pcxt->nworkers_launched);
+
/*
* Next, accumulate WAL usage. (This must wait for the workers to
finish,
* or we might get incomplete data.)
diff --git a/src/backend/access/nbtree/nbtsort.c
b/src/backend/access/nbtree/nbtsort.c
index f5d7b3b0c3..232e1a0942 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -1611,6 +1611,10 @@ _bt_end_parallel(BTLeader *btleader)
/* Shutdown worker processes */
WaitForParallelWorkersToFinish(btleader->pcxt);
+ pgstat_update_parallel_maint_workers_stats(
+ (PgStat_Counter) btleader->pcxt->nworkers_to_launch,
+ (PgStat_Counter) btleader->pcxt->nworkers_launched);
+
/*
* Next, accumulate WAL usage. (This must wait for the workers to
finish,
* or we might get incomplete data.)
diff --git a/src/backend/catalog/system_views.sql
b/src/backend/catalog/system_views.sql
index 19cabc9a47..48bf9e5535 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1073,6 +1073,10 @@ 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_planned(D.oid) as
parallel_workers_planned,
+ pg_stat_get_db_parallel_workers_launched(D.oid) as
parallel_workers_launched,
+ pg_stat_get_db_parallel_maint_workers_planned(D.oid) as
parallel_maint_workers_planned,
+ pg_stat_get_db_parallel_maint_workers_launched(D.oid) as
parallel_maint_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/commands/vacuumparallel.c
b/src/backend/commands/vacuumparallel.c
index 22c057fe61..f7603a0863 100644
--- a/src/backend/commands/vacuumparallel.c
+++ b/src/backend/commands/vacuumparallel.c
@@ -435,6 +435,11 @@ parallel_vacuum_end(ParallelVacuumState *pvs,
IndexBulkDeleteResult **istats)
{
Assert(!IsParallelWorker());
+ pgstat_update_parallel_maint_workers_stats(
+ (PgStat_Counter) pvs->pcxt->nworkers_to_launch,
+ (PgStat_Counter) pvs->pcxt->nworkers_launched
+ );
+
/* Copy the updated statistics */
for (int i = 0; i < pvs->nindexes; i++)
{
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 29e186fa73..9bf5f57b65 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -51,6 +51,7 @@
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "parser/parse_relation.h"
+#include "pgstat.h"
#include "rewrite/rewriteHandler.h"
#include "tcop/utility.h"
#include "utils/acl.h"
@@ -480,6 +481,12 @@ standard_ExecutorEnd(QueryDesc *queryDesc)
Assert(estate != NULL);
+ if (estate->es_workers_planned > 0) {
+ pgstat_update_parallel_workers_stats(
+ (PgStat_Counter) estate->es_workers_planned,
+ (PgStat_Counter) estate->es_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/executor/execUtils.c b/src/backend/executor/execUtils.c
index 5737f9f4eb..5919902075 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -162,6 +162,9 @@ CreateExecutorState(void)
estate->es_jit_flags = 0;
estate->es_jit = NULL;
+ estate->es_workers_launched = 0;
+ estate->es_workers_planned = 0;
+
/*
* Return the executor state structure
*/
diff --git a/src/backend/executor/nodeGather.c
b/src/backend/executor/nodeGather.c
index 5d4ffe989c..1271a0f7d1 100644
--- a/src/backend/executor/nodeGather.c
+++ b/src/backend/executor/nodeGather.c
@@ -182,6 +182,9 @@ ExecGather(PlanState *pstate)
/* We save # workers launched for the benefit of
EXPLAIN */
node->nworkers_launched = pcxt->nworkers_launched;
+ estate->es_workers_launched += pcxt->nworkers_launched;
+ estate->es_workers_planned += pcxt->nworkers_to_launch;
+
/* Set up tuple queue readers to read the results. */
if (pcxt->nworkers_launched > 0)
{
diff --git a/src/backend/executor/nodeGatherMerge.c
b/src/backend/executor/nodeGatherMerge.c
index 45f6017c29..677c450c3d 100644
--- a/src/backend/executor/nodeGatherMerge.c
+++ b/src/backend/executor/nodeGatherMerge.c
@@ -223,6 +223,9 @@ ExecGatherMerge(PlanState *pstate)
/* We save # workers launched for the benefit of
EXPLAIN */
node->nworkers_launched = pcxt->nworkers_launched;
+ estate->es_workers_launched += pcxt->nworkers_launched;
+ estate->es_workers_planned += pcxt->nworkers_to_launch;
+
/* Set up tuple queue readers to read the results. */
if (pcxt->nworkers_launched > 0)
{
diff --git a/src/backend/utils/activity/pgstat_database.c
b/src/backend/utils/activity/pgstat_database.c
index 29bc090974..9e72c286b2 100644
--- a/src/backend/utils/activity/pgstat_database.c
+++ b/src/backend/utils/activity/pgstat_database.c
@@ -262,6 +262,38 @@ AtEOXact_PgStat_Database(bool isCommit, bool parallel)
}
}
+/*
+ * reports parallel_workers_planned and parallel_workers_launched into
+ * PgStat_StatDBEntry
+ */
+void
+pgstat_update_parallel_workers_stats(PgStat_Counter parallel_workers_planned,
PgStat_Counter parallel_workers_launched)
+{
+ PgStat_StatDBEntry *dbentry;
+ if (!OidIsValid(MyDatabaseId))
+ return;
+
+ dbentry = pgstat_prep_database_pending(MyDatabaseId);
+ dbentry->parallel_workers_planned += parallel_workers_planned;
+ dbentry->parallel_workers_launched += parallel_workers_launched;
+}
+
+/*
+ * reports parallel_maint_workers_planned and parallel_maint_workers_launched
into
+ * PgStat_StatDBEntry
+ */
+void
+pgstat_update_parallel_maint_workers_stats(PgStat_Counter
parallel_maint_workers_planned, PgStat_Counter parallel_maint_workers_launched)
+{
+ PgStat_StatDBEntry *dbentry;
+ if (!OidIsValid(MyDatabaseId))
+ return;
+
+ dbentry = pgstat_prep_database_pending(MyDatabaseId);
+ dbentry->parallel_maint_workers_planned +=
parallel_maint_workers_planned;
+ dbentry->parallel_maint_workers_launched +=
parallel_maint_workers_launched;
+}
+
/*
* Subroutine for pgstat_report_stat(): Handle xact commit/rollback and I/O
* timings.
@@ -425,6 +457,10 @@ 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_planned);
+ PGSTAT_ACCUM_DBCOUNT(parallel_workers_launched);
+ PGSTAT_ACCUM_DBCOUNT(parallel_maint_workers_planned);
+ PGSTAT_ACCUM_DBCOUNT(parallel_maint_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 3221137123..377a0f6453 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1039,6 +1039,18 @@ 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_planned*/
+PG_STAT_GET_DBENTRY_INT64(parallel_workers_planned)
+
+/* pg_stat_get_db_parallel_workers_launched*/
+PG_STAT_GET_DBENTRY_INT64(parallel_workers_launched)
+
+/* pg_stat_get_db_parallel_maint_workers_planned*/
+PG_STAT_GET_DBENTRY_INT64(parallel_maint_workers_planned)
+
+/* pg_stat_get_db_parallel_maint_workers_launched*/
+PG_STAT_GET_DBENTRY_INT64(parallel_maint_workers_launched)
+
/* pg_stat_get_db_temp_bytes */
PG_STAT_GET_DBENTRY_INT64(temp_bytes)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 4abc6d9526..b1cd4fa1b0 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5751,6 +5751,26 @@
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 for queries',
+ proname => 'pg_stat_get_db_parallel_workers_planned', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_parallel_workers_planned' },
+{ oid => '8404',
+ descr => 'statistics: number of parallel workers effectively launched for
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 => '8405',
+ descr => 'statistics: number of parallel workers planned for utilities',
+ proname => 'pg_stat_get_db_parallel_maint_workers_planned', provolatile =>
's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_parallel_maint_workers_planned' },
+{ oid => '8406',
+ descr => 'statistics: number of parallel workers effectively launched for
utilities',
+ proname => 'pg_stat_get_db_parallel_maint_workers_launched', provolatile =>
's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_parallel_maint_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/nodes/execnodes.h b/src/include/nodes/execnodes.h
index af7d8fd1e7..1903ad60f8 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -724,6 +724,9 @@ typedef struct EState
*/
List *es_insert_pending_result_relations;
List *es_insert_pending_modifytables;
+
+ int es_workers_launched;
+ int es_workers_planned;
} EState;
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index f63159c55c..bad74a9f2d 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -383,6 +383,11 @@ typedef struct PgStat_StatDBEntry
PgStat_Counter sessions_fatal;
PgStat_Counter sessions_killed;
+ PgStat_Counter parallel_workers_planned;
+ PgStat_Counter parallel_workers_launched;
+ PgStat_Counter parallel_maint_workers_planned;
+ PgStat_Counter parallel_maint_workers_launched;
+
TimestampTz stat_reset_timestamp;
} PgStat_StatDBEntry;
@@ -578,6 +583,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
parallel_workers_planned, PgStat_Counter parallel_workers_launched);
+extern void pgstat_update_parallel_maint_workers_stats(PgStat_Counter
parallel_maint_workers_planned, PgStat_Counter parallel_maint_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 862433ee52..e8a4453cd5 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1861,6 +1861,10 @@ 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_planned(oid) AS parallel_workers_planned,
+ pg_stat_get_db_parallel_workers_launched(oid) AS parallel_workers_launched,
+ pg_stat_get_db_parallel_maint_workers_planned(oid) AS
parallel_maint_workers_planned,
+ pg_stat_get_db_parallel_maint_workers_launched(oid) AS
parallel_maint_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/stats.out
b/src/test/regress/expected/stats.out
index 6e08898b18..88d283a991 100644
--- a/src/test/regress/expected/stats.out
+++ b/src/test/regress/expected/stats.out
@@ -32,6 +32,11 @@ SELECT t.seq_scan, t.seq_tup_read, t.idx_scan,
t.idx_tup_fetch,
pg_catalog.pg_statio_user_tables AS b
WHERE t.relname='tenk2' AND b.relname='tenk2';
COMMIT;
+SELECT sum(parallel_workers_planned) AS parallel_workers_planned_before,
+ sum(parallel_workers_launched) AS parallel_workers_launched_before,
+ sum(parallel_maint_workers_planned) AS
parallel_maint_workers_planned_before,
+ sum(parallel_maint_workers_launched) AS
parallel_maint_workers_launched_before
+FROM pg_stat_database \gset
-- test effects of TRUNCATE on n_live_tup/n_dead_tup counters
CREATE TABLE trunc_stats_test(id serial);
CREATE TABLE trunc_stats_test1(id serial, stuff text);
@@ -862,6 +867,18 @@ WHERE pg_stat_get_backend_pid(beid) = pg_backend_pid();
t
(1 row)
+-- Test that parallel workers stats are updated in pg_stat_database
+SELECT
+ sum(parallel_workers_planned) > :'parallel_workers_planned_before' AS
wrk_planned,
+ sum(parallel_workers_launched) > :'parallel_workers_launched_before' AS
wrk_launched,
+ sum(parallel_maint_workers_planned) >
:'parallel_maint_workers_planned_before' AS maint_wrk_planned,
+ sum(parallel_maint_workers_launched) >
:'parallel_maint_workers_launched_before' AS maint_wrk_launched
+FROM pg_stat_database;
+ wrk_planned | wrk_launched | maint_wrk_planned | maint_wrk_launched
+-------------+--------------+-------------------+--------------------
+ t | t | t | t
+(1 row)
+
-----
-- Test that resetting stats works for reset timestamp
-----
diff --git a/src/test/regress/sql/stats.sql b/src/test/regress/sql/stats.sql
index d8ac0d06f4..3a59c75539 100644
--- a/src/test/regress/sql/stats.sql
+++ b/src/test/regress/sql/stats.sql
@@ -32,6 +32,12 @@ SELECT t.seq_scan, t.seq_tup_read, t.idx_scan,
t.idx_tup_fetch,
WHERE t.relname='tenk2' AND b.relname='tenk2';
COMMIT;
+SELECT sum(parallel_workers_planned) AS parallel_workers_planned_before,
+ sum(parallel_workers_launched) AS parallel_workers_launched_before,
+ sum(parallel_maint_workers_planned) AS
parallel_maint_workers_planned_before,
+ sum(parallel_maint_workers_launched) AS
parallel_maint_workers_launched_before
+FROM pg_stat_database \gset
+
-- test effects of TRUNCATE on n_live_tup/n_dead_tup counters
CREATE TABLE trunc_stats_test(id serial);
CREATE TABLE trunc_stats_test1(id serial, stuff text);
@@ -442,6 +448,14 @@ SELECT (current_schemas(true))[1] = ('pg_temp_' ||
beid::text) AS match
FROM pg_stat_get_backend_idset() beid
WHERE pg_stat_get_backend_pid(beid) = pg_backend_pid();
+-- Test that parallel workers stats are updated in pg_stat_database
+SELECT
+ sum(parallel_workers_planned) > :'parallel_workers_planned_before' AS
wrk_planned,
+ sum(parallel_workers_launched) > :'parallel_workers_launched_before' AS
wrk_launched,
+ sum(parallel_maint_workers_planned) >
:'parallel_maint_workers_planned_before' AS maint_wrk_planned,
+ sum(parallel_maint_workers_launched) >
:'parallel_maint_workers_launched_before' AS maint_wrk_launched
+FROM pg_stat_database;
+
-----
-- Test that resetting stats works for reset timestamp
-----
--
2.45.2
