From e8ed07d43e2dfb7564c8d4fe6c41aca466c24499 Mon Sep 17 00:00:00 2001 From: Sami Imseih Date: Fri, 6 Nov 2026 19:39:38 +0000 Subject: [PATCH v5 3/4] pg_stat_statements: Store statement entries as custom cumulative statistics pg_stat_statements kept its entries in a fixed-size shared hash table, sized by pg_stat_statements.max and protected by a single LWLock taken by every statement, exclusively to create entries, and saved them in its own dump file at shutdown. When statements churn, each new entry past the maximum scanned, sorted and evicted entries while holding that lock exclusively, stalling all other backends. Statement entries are now a custom kind of cumulative statistics, with their own hash table, and pg_stat_statements_info is a second, fixed kind. Both are saved and restored with the cumulative statistics file, with each entry's query text written after it. As the number of entries no longer sizes shared memory, pg_stat_statements.max can be changed with a reload. The kind counts its entries, so the entry_count column of pg_stat_kind_info shows the entries allocated, including evicted ones not freed yet. Executions no longer take the global lock. They count in pending statistics, added to the entry when flushed, and update the entry's usage under its lock if it is free, leaving the usage to the flush otherwise. The view first flushes the reading backend's own pending statistics, so version 1.14 of the extension marks pg_stat_statements() as PARALLEL RESTRICTED. Query texts stay in the external file, with the same appending, reading and garbage collection, and entries are created as before: the text is appended with the global lock held shared, then the lock is taken exclusively to create the entry. Texts of evicted entries become garbage in the file. As before, a statement whose text cannot be written is not tracked. Eviction keeps the existing usage-based policy: each pass decays the usage of the entries it scans, under their lock, skipping busy ones, which miss that pass's decay. A single backend evicts at a time, under a lock of its own rather than the global lock, so that executions and query texts don't wait for passes; below pg_stat_statements.max plus USAGE_DEALLOC_PERCENT, the excess a pass removes, new entries don't wait for one either. At that limit, they wait for the running pass, and run one if the excess remains. The kind sets release_refs_after_flush, so that backends release their references to evicted entries once they flush their pending statistics, and to the entries they create at the end of the creating transaction, which frees evicted entries. Should evicted entries still pile up, as many waiting to be freed as there can be live ones, new statements are not tracked until they are freed, and executions of tracked statements are counted as usual, which bounds the entries allocated to twice pg_stat_statements.max. Entries are found by a stats object ID, a 64-bit hash of their key: like statements whose queryids collide, statements whose keys hash to the same ID share an entry. The XXX comments list the cumulative statistics APIs that would remove the need to use its internals. If the module is removed from shared_preload_libraries, all cumulative statistics are discarded at the next start, as the statistics file then contains entries of an unknown kind. Add TAP tests for persistence across restarts, and for eviction and changes of pg_stat_statements.max. The two kinds use IDs 31 and 32, reserved at https://wiki.postgresql.org/wiki/CustomCumulativeStats. --- contrib/pg_stat_statements/Makefile | 1 + .../expected/oldextversions.out | 30 + contrib/pg_stat_statements/meson.build | 2 + .../pg_stat_statements--1.13--1.14.sql | 7 + .../pg_stat_statements/pg_stat_statements.c | 1499 +++++++++++------ .../pg_stat_statements.control | 2 +- .../pg_stat_statements/sql/oldextversions.sql | 16 + contrib/pg_stat_statements/t/010_restart.pl | 78 + contrib/pg_stat_statements/t/011_eviction.pl | 100 ++ doc/src/sgml/pgstatstatements.sgml | 56 +- src/tools/pgindent/typedefs.list | 3 + 11 files changed, 1299 insertions(+), 495 deletions(-) create mode 100644 contrib/pg_stat_statements/pg_stat_statements--1.13--1.14.sql create mode 100644 contrib/pg_stat_statements/t/011_eviction.pl diff --git a/contrib/pg_stat_statements/Makefile b/contrib/pg_stat_statements/Makefile index 2fe2a077ade..f48fedf8f94 100644 --- a/contrib/pg_stat_statements/Makefile +++ b/contrib/pg_stat_statements/Makefile @@ -7,6 +7,7 @@ OBJS = \ EXTENSION = pg_stat_statements DATA = pg_stat_statements--1.4.sql \ + pg_stat_statements--1.13--1.14.sql \ pg_stat_statements--1.12--1.13.sql \ pg_stat_statements--1.11--1.12.sql pg_stat_statements--1.10--1.11.sql \ pg_stat_statements--1.9--1.10.sql pg_stat_statements--1.8--1.9.sql \ diff --git a/contrib/pg_stat_statements/expected/oldextversions.out b/contrib/pg_stat_statements/expected/oldextversions.out index 726383a99d7..5c76ecd4986 100644 --- a/contrib/pg_stat_statements/expected/oldextversions.out +++ b/contrib/pg_stat_statements/expected/oldextversions.out @@ -474,4 +474,34 @@ SELECT count(*) > 0 AS has_data FROM pg_stat_statements; t (1 row) +-- Before 1.14, these functions are PARALLEL SAFE, and work in a worker +SET max_parallel_workers_per_gather = 2; +SET debug_parallel_query = on; +SELECT count(*) > 0 AS has_data FROM pg_stat_statements; + has_data +---------- + t +(1 row) + +SELECT pg_stat_statements_reset() IS NOT NULL AS t; + t +--- + t +(1 row) + +RESET debug_parallel_query; +RESET max_parallel_workers_per_gather; +-- pg_stat_statements() restricted to the leader in 1.14 +AlTER EXTENSION pg_stat_statements UPDATE TO '1.14'; +SELECT oid::regprocedure, proparallel +FROM pg_proc +WHERE oid IN ('pg_stat_statements(boolean)'::regprocedure, + 'pg_stat_statements_reset(oid, oid, bigint, boolean)'::regprocedure) +ORDER BY oid::regprocedure::text; + oid | proparallel +--------------------------------------------------+------------- + pg_stat_statements(boolean) | r + pg_stat_statements_reset(oid,oid,bigint,boolean) | s +(2 rows) + DROP EXTENSION pg_stat_statements; diff --git a/contrib/pg_stat_statements/meson.build b/contrib/pg_stat_statements/meson.build index a2f957836aa..8bf1e55b1c7 100644 --- a/contrib/pg_stat_statements/meson.build +++ b/contrib/pg_stat_statements/meson.build @@ -21,6 +21,7 @@ contrib_targets += pg_stat_statements install_data( 'pg_stat_statements.control', 'pg_stat_statements--1.4.sql', + 'pg_stat_statements--1.13--1.14.sql', 'pg_stat_statements--1.12--1.13.sql', 'pg_stat_statements--1.11--1.12.sql', 'pg_stat_statements--1.10--1.11.sql', @@ -71,6 +72,7 @@ tests += { 'tap': { 'tests': [ 't/010_restart.pl', + 't/011_eviction.pl', ], }, } diff --git a/contrib/pg_stat_statements/pg_stat_statements--1.13--1.14.sql b/contrib/pg_stat_statements/pg_stat_statements--1.13--1.14.sql new file mode 100644 index 00000000000..edcfd029b57 --- /dev/null +++ b/contrib/pg_stat_statements/pg_stat_statements--1.13--1.14.sql @@ -0,0 +1,7 @@ +/* contrib/pg_stat_statements/pg_stat_statements--1.13--1.14.sql */ + +-- complain if script is sourced in psql, rather than via ALTER EXTENSION +\echo Use "ALTER EXTENSION pg_stat_statements UPDATE TO '1.14'" to load this file. \quit + +-- Run in the leader, which can flush its pending statistics first +ALTER FUNCTION pg_stat_statements(boolean) PARALLEL RESTRICTED; diff --git a/contrib/pg_stat_statements/pg_stat_statements.c b/contrib/pg_stat_statements/pg_stat_statements.c index 8db4b3d6e24..ca2f07b8e6a 100644 --- a/contrib/pg_stat_statements/pg_stat_statements.c +++ b/contrib/pg_stat_statements/pg_stat_statements.c @@ -4,9 +4,10 @@ * Track statement planning and execution times as well as resource * usage across a whole database cluster. * - * Execution costs are totaled for each distinct source query, and kept in - * a shared hashtable. (We track only as many distinct queries as will fit - * in the designated amount of shared memory.) + * Execution costs are totaled for each distinct source query, and kept as + * entries of a custom cumulative statistics kind. (We track about as many + * distinct queries as pg_stat_statements.max; see entry_dealloc() for how + * far the count can exceed it.) * * Starting in Postgres 9.2, this module normalized query entries. As of * Postgres 14, the normalization is done by the core if compute_query_id is @@ -19,19 +20,23 @@ * these strings in a temporary external query-texts file. Offsets into this * file are kept in shared memory. * - * Note about locking issues: to create or delete an entry in the shared - * hashtable, one must hold pgss->lock exclusively. Modifying any field - * in an entry except the counters requires the same. To look up an entry, - * one must hold the lock shared. To read or update the counters within - * an entry, one must hold the lock shared or exclusive (so the entry doesn't - * disappear!) and also take the entry's mutex spinlock. + * Note about locking issues: the cumulative statistics system manages the + * hash table and the lifetime of entries. The kind sets own_hash, so its + * entries live in a dedicated hash table and DSA area, and its hash table + * partition locks are not shared with other kinds. An entry's fields are + * protected by its lock; executions count in this backend's pending entry, + * which a flush adds to the shared entry under the lock. An eviction pass + * holds pgss->dealloc_lock exclusively, so that passes run one at a time; + * see entry_dealloc(). * The shared state variable pgss->extent (the next free spot in the external * query-text file) should be accessed only while holding either the * pgss->mutex spinlock, or exclusive lock on pgss->lock. We use the mutex to * allow reserving file space while holding only shared lock on pgss->lock. * Rewriting the entire external query-text file, eg for garbage collection, * requires holding pgss->lock exclusively; this allows individual entries - * in the file to be read or written while holding only shared lock. + * in the file to be read or written while holding only shared lock. The + * query offset and length of an entry are set or changed only while holding + * pgss->lock exclusively. * * * Copyright (c) 2008-2026, PostgreSQL Global Development Group @@ -50,24 +55,28 @@ #include "access/htup_details.h" #include "access/parallel.h" #include "catalog/pg_authid.h" +#include "common/hashfn.h" #include "executor/instrument.h" #include "funcapi.h" #include "jit/jit.h" +#include "lib/dshash.h" #include "mb/pg_wchar.h" #include "miscadmin.h" #include "nodes/queryjumble.h" #include "optimizer/planner.h" #include "parser/analyze.h" #include "pgstat.h" +#include "port/atomics.h" #include "storage/fd.h" -#include "storage/ipc.h" #include "storage/lwlock.h" #include "storage/shmem.h" #include "storage/spin.h" #include "tcop/utility.h" #include "utils/acl.h" #include "utils/builtins.h" +#include "utils/dsa.h" #include "utils/memutils.h" +#include "utils/pgstat_internal.h" #include "utils/timestamp.h" #include "utils/tuplestore.h" @@ -76,20 +85,18 @@ PG_MODULE_MAGIC_EXT( .version = PG_VERSION ); -/* Location of permanent stats file (valid when database is shut down) */ -#define PGSS_DUMP_FILE PGSTAT_STAT_PERMANENT_DIRECTORY "/pg_stat_statements.stat" +/* + * Custom cumulative statistics kinds, for statement entries and for global + * statistics. See https://wiki.postgresql.org/wiki/CustomCumulativeStats. + */ +#define PGSS_STATS_KIND 31 +#define PGSS_GLOBAL_STATS_KIND 32 /* * Location of external query text file. */ #define PGSS_TEXT_FILE PG_STAT_TMP_DIR "/pgss_query_texts.stat" -/* Magic number identifying the stats file format */ -static const uint32 PGSS_FILE_HEADER = 0x20250731; - -/* PostgreSQL major version number, changes in which invalidate all entries */ -static const uint32 PGSS_PG_MAJOR_VERSION = PG_VERSION_NUM / 100; - /* XXX: Should USAGE_EXEC reflect execution time and/or buffer usage? */ #define USAGE_EXEC(duration) (1.0) #define USAGE_INIT (1.0) /* including initial planning */ @@ -136,9 +143,8 @@ typedef enum pgssStoreKind * Hashtable key that defines the identity of a hashtable entry. We separate * queries by user and by database even if they are otherwise identical. * - * If you add a new key to this struct, make sure to teach pgss_store() to - * zero the padding bytes. Otherwise, things will break, because pgss_hash is - * created using HASH_BLOBS, and thus tag_hash is used to hash this. + * If you add a new key to this struct, teach pgss_stats_objid() about it, + * as entries are identified by the object ID it computes from the key. */ typedef struct pgssHashKey { @@ -153,7 +159,7 @@ typedef struct pgssHashKey */ typedef struct Counters { - int64 calls[PGSS_NUMKIND]; /* # of times planned/executed */ + PgStat_Counter calls[PGSS_NUMKIND]; /* # of times planned/executed */ double total_time[PGSS_NUMKIND]; /* total planning/execution time, * in msec */ double min_time[PGSS_NUMKIND]; /* minimum planning/execution time in @@ -164,17 +170,17 @@ typedef struct Counters * msec */ double sum_var_time[PGSS_NUMKIND]; /* sum of variances in * planning/execution time in msec */ - int64 rows; /* total # of retrieved or affected rows */ - int64 shared_blks_hit; /* # of shared buffer hits */ - int64 shared_blks_read; /* # of shared disk blocks read */ - int64 shared_blks_dirtied; /* # of shared disk blocks dirtied */ - int64 shared_blks_written; /* # of shared disk blocks written */ - int64 local_blks_hit; /* # of local buffer hits */ - int64 local_blks_read; /* # of local disk blocks read */ - int64 local_blks_dirtied; /* # of local disk blocks dirtied */ - int64 local_blks_written; /* # of local disk blocks written */ - int64 temp_blks_read; /* # of temp blocks read */ - int64 temp_blks_written; /* # of temp blocks written */ + PgStat_Counter rows; /* total # of retrieved or affected rows */ + PgStat_Counter shared_blks_hit; /* # of shared buffer hits */ + PgStat_Counter shared_blks_read; /* # of shared disk blocks read */ + PgStat_Counter shared_blks_dirtied; /* # of shared disk blocks dirtied */ + PgStat_Counter shared_blks_written; /* # of shared disk blocks written */ + PgStat_Counter local_blks_hit; /* # of local buffer hits */ + PgStat_Counter local_blks_read; /* # of local disk blocks read */ + PgStat_Counter local_blks_dirtied; /* # of local disk blocks dirtied */ + PgStat_Counter local_blks_written; /* # of local disk blocks written */ + PgStat_Counter temp_blks_read; /* # of temp blocks read */ + PgStat_Counter temp_blks_written; /* # of temp blocks written */ double shared_blk_read_time; /* time spent reading shared blocks, * in msec */ double shared_blk_write_time; /* time spent writing shared blocks, @@ -186,32 +192,33 @@ typedef struct Counters double temp_blk_read_time; /* time spent reading temp blocks, in msec */ double temp_blk_write_time; /* time spent writing temp blocks, in * msec */ - double usage; /* usage factor */ - int64 wal_records; /* # of WAL records generated */ - int64 wal_fpi; /* # of WAL full page images generated */ + PgStat_Counter wal_records; /* # of WAL records generated */ + PgStat_Counter wal_fpi; /* # of WAL full page images generated */ uint64 wal_bytes; /* total amount of WAL generated in bytes */ - int64 wal_buffers_full; /* # of times the WAL buffers became full */ - int64 jit_functions; /* total number of JIT functions emitted */ + PgStat_Counter wal_buffers_full; /* # of times the WAL buffers became + * full */ + PgStat_Counter jit_functions; /* total number of JIT functions emitted */ double jit_generation_time; /* total time to generate jit code */ - int64 jit_inlining_count; /* number of times inlining time has been - * > 0 */ + PgStat_Counter jit_inlining_count; /* number of times inlining time has + * been > 0 */ double jit_deform_time; /* total time to deform tuples in jit code */ - int64 jit_deform_count; /* number of times deform time has been > - * 0 */ + PgStat_Counter jit_deform_count; /* number of times deform time has + * been > 0 */ double jit_inlining_time; /* total time to inline jit code */ - int64 jit_optimization_count; /* number of times optimization time - * has been > 0 */ + PgStat_Counter jit_optimization_count; /* number of times optimization + * time has been > 0 */ double jit_optimization_time; /* total time to optimize jit code */ - int64 jit_emission_count; /* number of times emission time has been - * > 0 */ + PgStat_Counter jit_emission_count; /* number of times emission time has + * been > 0 */ double jit_emission_time; /* total time to emit jit code */ - int64 parallel_workers_to_launch; /* # of parallel workers planned - * to be launched */ - int64 parallel_workers_launched; /* # of parallel workers actually - * launched */ - int64 generic_plan_calls; /* number of calls using a generic plan */ - int64 custom_plan_calls; /* number of calls using a custom plan */ + PgStat_Counter parallel_workers_to_launch; /* # of parallel workers + * planned to be launched */ + PgStat_Counter parallel_workers_launched; /* # of parallel workers + * actually launched */ + PgStat_Counter generic_plan_calls; /* number of calls using a generic + * plan */ + PgStat_Counter custom_plan_calls; /* number of calls using a custom plan */ } Counters; /* @@ -229,37 +236,80 @@ typedef struct pgssGlobalStats * Note: in event of a failure in garbage collection of the query text file, * we reset query_offset to zero and query_len to -1. This will be seen as * an invalid state by qtext_fetch(). + * + * stats_since is zero while the entry is not live, before entry_alloc() and + * after entry_remove(); scans skip such entries. + * + * Executions add to usage under the entry lock, or through the pending entry + * if the lock is busy. The first usage update sets executed; until then, + * eviction treats the entry as sticky. + * + * A min/max reset cannot reach the pending entries of backends, so it bumps + * minmax_generation, and pending min/max values of an older generation are + * discarded instead of merged. minmax_generation is atomic because + * executions read it without the entry lock. */ typedef struct pgssEntry { pgssHashKey key; /* hash key of entry - MUST BE FIRST */ Counters counters; /* the statistics for this query */ + double usage; /* usage factor */ + bool executed; /* executed since creation? */ Size query_offset; /* query text offset in external file */ int query_len; /* # of valid bytes in query string, or -1 */ int encoding; /* query text encoding */ TimestampTz stats_since; /* timestamp of entry allocation */ TimestampTz minmax_stats_since; /* timestamp of last min/max values reset */ - slock_t mutex; /* protects the counters only */ + pg_atomic_uint32 minmax_generation; /* bumped by min/max resets */ } pgssEntry; +/* + * Statement entry in the hash table of the cumulative statistics system + */ +typedef struct pgssSharedEntry +{ + PgStatShared_Common header; /* includes the entry lock */ + pgssEntry entry; +} pgssSharedEntry; + +/* + * Pending statistics of a statement, which a flush adds to the entry + */ +typedef struct pgssPendingEntry +{ + Counters counters; + double usage; /* of executions finding the entry locked */ + uint32 minmax_generation; /* of the entry, as of min/max values */ +} pgssPendingEntry; + +/* + * Element of the array of entries sorted by entry_dealloc() + */ +typedef struct pgssDeallocEntry +{ + PgStat_HashKey key; /* hash key of the entry */ + TimestampTz stats_since; /* identifies the entry's incarnation */ + double usage; /* usage of the entry */ +} pgssDeallocEntry; + /* * Global shared state */ typedef struct pgssSharedState { - LWLockPadded lock; /* protects hashtable search/modification */ + LWLockPadded lock; /* protects the query text file */ + LWLock dealloc_lock; /* held exclusively by entry_dealloc() */ + pg_atomic_uint64 num_entries; /* # of live entries; see entry_remove() */ + slock_t mutex; /* protects following fields only: */ double cur_median_usage; /* current median usage in hashtable */ Size mean_query_len; /* current mean entry text length */ - slock_t mutex; /* protects following fields only: */ Size extent; /* current extent of query file */ int n_writers; /* number of active writers to query file */ int gc_count; /* query file garbage collection cycle count */ - pgssGlobalStats stats; /* global statistics for pgss */ } pgssSharedState; /* Links to shared memory state */ static pgssSharedState *pgss; -static HTAB *pgss_hash; static void pgss_shmem_request(void *arg); static void pgss_shmem_init(void *arg); @@ -274,6 +324,11 @@ static const ShmemCallbacks pgss_shmem_callbacks = { /* Current nesting depth of planner/ExecutorRun/ProcessUtility calls */ static int nesting_level = 0; +/* Query text file loaded at shutdown to save the texts with the entries */ +static char *pgss_saved_qbuffer = NULL; +static Size pgss_saved_qbuffer_size = 0; +static bool pgss_saved_qbuffer_loaded = false; + /* * State saved in a QueryDesc at ExecutorStart, for use at ExecutorEnd. A * query may end at a different nesting level than the one it started at, @@ -348,7 +403,17 @@ PG_FUNCTION_INFO_V1(pg_stat_statements_1_13); PG_FUNCTION_INFO_V1(pg_stat_statements); PG_FUNCTION_INFO_V1(pg_stat_statements_info); -static void pgss_shmem_shutdown(int code, Datum arg); +static bool pgss_to_serialized_data_cb(const PgStat_HashKey *key, + const PgStatShared_Common *header, + FILE *statfile); +static bool pgss_from_serialized_data_cb(const PgStat_HashKey *key, + PgStatShared_Common *header, + FILE *statfile); +static void pgss_finish_cb(PgStat_StatsFileOp status); +static void pgss_global_stats_init_shmem_cb(void *stats); +static inline pgssGlobalStats *pgss_global_stats(void); +static void pgss_global_stats_reset_all_cb(TimestampTz ts); +static void pgss_global_stats_snapshot_cb(void); static void pgss_post_parse_analyze(ParseState *pstate, Query *query, const JumbleState *jstate); static PlannedStmt *pgss_planner(Query *parse, @@ -382,8 +447,19 @@ static void pgss_store(const char *query, int64 queryId, static void pg_stat_statements_internal(FunctionCallInfo fcinfo, pgssVersion api_version, bool showtext); -static pgssEntry *entry_alloc(pgssHashKey *key, Size query_offset, int query_len, - int encoding, bool sticky); +static PgStat_EntryRef *entry_alloc(const pgssHashKey *key, + Size query_offset, int query_len, + int encoding, bool sticky); +static uint64 pgss_stats_objid(const pgssHashKey *key); +static inline pgssSharedEntry *entry_from_hash(PgStatShared_HashEntry *hash_entry); +static void entry_seq_init(dshash_seq_status *hstat); +static pgssSharedEntry *entry_seq_next(dshash_seq_status *hstat, + PgStat_HashKey *key); +static void entry_drop(const PgStat_HashKey *key); +static PgStat_EntryRef *entry_find(const pgssHashKey *key); +static void entry_add_usage(pgssEntry *entry, double usage); +static PgStat_FlushResult pgss_flush_pending_cb(PgStat_EntryRef *entry_ref, + bool nowait, bool xact_boundary); static void entry_dealloc(void); static bool qtext_store(const char *query, int query_len, Size *query_offset, int *gc_count); @@ -392,11 +468,51 @@ static char *qtext_fetch(Size query_offset, int query_len, char *buffer, Size buffer_size); static bool need_gc_qtexts(void); static void gc_qtexts(void); +static void qtext_reset_file(void); +static bool entry_remove(pgssSharedEntry *shared_entry, + PgStat_EntryRef *entry_ref, + const pgssDeallocEntry *victim); +static void entry_reset_minmax(Counters *counters); +static void entry_reset_shared_minmax(pgssSharedEntry *shared_entry, + TimestampTz stats_reset); +static void entry_reset_single(const pgssHashKey *key, bool minmax_only, + TimestampTz stats_reset); +static bool entry_reset_matches(const pgssEntry *entry, const pgssHashKey *key); +static bool entry_reset_drop_matching(PgStatShared_HashEntry *hash_entry, + Datum match_data); static TimestampTz entry_reset(Oid userid, Oid dbid, int64 queryid, bool minmax_only); static char *generate_normalized_query(const JumbleState *jstate, const char *query, int query_loc, int *query_len_p); +static const PgStat_KindInfo pgss_stats_kind = { + .name = "pg_stat_statements", + .write_to_file = true, + .accessed_across_databases = true, + .own_hash = true, + .track_entry_count = true, + .release_refs_after_flush = true, + .shared_size = sizeof(pgssSharedEntry), + .shared_data_off = offsetof(pgssSharedEntry, entry), + .shared_data_len = sizeof(pgssEntry), + .pending_size = sizeof(pgssPendingEntry), + .flush_pending_cb = pgss_flush_pending_cb, + .to_serialized_data = pgss_to_serialized_data_cb, + .from_serialized_data = pgss_from_serialized_data_cb, + .finish = pgss_finish_cb, +}; + +static const PgStat_KindInfo pgss_global_stats_kind = { + .name = "pg_stat_statements_info", + .fixed_amount = true, + .write_to_file = true, + .shared_size = sizeof(pgssGlobalStats), + .shared_data_len = sizeof(pgssGlobalStats), + .init_shmem_cb = pgss_global_stats_init_shmem_cb, + .reset_all_cb = pgss_global_stats_reset_all_cb, + .snapshot_cb = pgss_global_stats_snapshot_cb, +}; + /* * Module load callback */ @@ -430,7 +546,7 @@ _PG_init(void) 5000, 100, INT_MAX / 2, - PGC_POSTMASTER, + PGC_SIGHUP, 0, NULL, NULL, @@ -483,6 +599,12 @@ _PG_init(void) MarkGUCPrefixReserved("pg_stat_statements"); + /* + * Register our custom kinds of cumulative statistics. + */ + pgstat_register_kind(PGSS_STATS_KIND, &pgss_stats_kind); + pgstat_register_kind(PGSS_GLOBAL_STATS_KIND, &pgss_global_stats_kind); + /* * Register our shared memory needs. */ @@ -524,13 +646,6 @@ _PG_init(void) static void pgss_shmem_request(void *arg) { - ShmemRequestHash(.name = "pg_stat_statements hash", - .nelems = (int64) pgss_max, - .hash_info.keysize = sizeof(pgssHashKey), - .hash_info.entrysize = sizeof(pgssEntry), - .hash_flags = HASH_ELEM | HASH_BLOBS, - .ptr = &pgss_hash, - ); ShmemRequestStruct(.name = "pg_stat_statements", .size = sizeof(pgssSharedState), .ptr = (void **) &pgss, @@ -540,22 +655,11 @@ pgss_shmem_request(void *arg) /* * shmem init callback: Initialize our shared memory data structures at * postmaster startup. - * - * Load any pre-existing statistics from file. Also create and load the - * query-texts file, which is expected to exist (even if empty) while the - * module is enabled. */ static void pgss_shmem_init(void *arg) { int tranche_id; - FILE *file = NULL; - FILE *qfile = NULL; - uint32 header; - int64 num; - uint32 pgver; - int buffer_size; - char *buffer = NULL; /* * We already checked that we're loaded from shared_preload_libraries in @@ -568,281 +672,292 @@ pgss_shmem_init(void *arg) */ tranche_id = LWLockNewTrancheId("pg_stat_statements"); LWLockInitialize(&pgss->lock.lock, tranche_id); + tranche_id = LWLockNewTrancheId("pg_stat_statements_dealloc"); + LWLockInitialize(&pgss->dealloc_lock, tranche_id); pgss->cur_median_usage = ASSUMED_MEDIAN_INIT; pgss->mean_query_len = ASSUMED_LENGTH_INIT; SpinLockInit(&pgss->mutex); pgss->extent = 0; pgss->n_writers = 0; pgss->gc_count = 0; - pgss->stats.dealloc = 0; - pgss->stats.stats_reset = GetCurrentTimestamp(); + pg_atomic_init_u64(&pgss->num_entries, 0); - /* The hash table must've also been initialized by now */ - Assert(pgss_hash != NULL); + /* Restored entries write their texts into an empty file */ + qtext_reset_file(); +} - /* - * Set up a shmem exit hook to dump the statistics to disk on postmaster - * (or standalone backend) exit. - */ - on_shmem_exit(pgss_shmem_shutdown, (Datum) 0); +/* + * Compute the stats object ID of an entry from its key. Different keys may + * map to the same ID; like statements whose queryids collide, they then share + * an entry. + */ +static uint64 +pgss_stats_objid(const pgssHashKey *key) +{ + uint64 hash; - /* - * Load any pre-existing statistics from file. - * - * Note: we don't bother with locks here, because there should be no other - * processes running when this code is reached. - */ + hash = murmurhash64(((uint64) key->userid << 32) | key->dbid); + hash = hash_combine64(hash, murmurhash64((uint64) key->queryid)); + hash = hash_combine64(hash, (uint64) key->toplevel); - /* Unlink query text file possibly left over from crash */ - unlink(PGSS_TEXT_FILE); + return hash; +} - /* Allocate new query text temp file */ - qfile = AllocateFile(PGSS_TEXT_FILE, PG_BINARY_W); - if (qfile == NULL) - goto write_error; +/* + * Return the statement entry of a hash table entry, or NULL if it has been + * dropped. Caller must hold the hash table entry's partition lock. + */ +static inline pgssSharedEntry * +entry_from_hash(PgStatShared_HashEntry *hash_entry) +{ + if (hash_entry->dropped) + return NULL; - /* - * If we were told not to load old statistics, we're done. (Note we do - * not try to unlink any old dump file in this case. This seems a bit - * questionable but it's the historical behavior.) - */ - if (!pgss_save) - { - FreeFile(qfile); - return; - } + return (pgssSharedEntry *) + dsa_get_address(pgStatLocal.kind_dsa[PGSS_STATS_KIND], + hash_entry->body); +} - /* - * Attempt to load old statistics from the dump file. - */ - file = AllocateFile(PGSS_DUMP_FILE, PG_BINARY_R); - if (file == NULL) - { - if (errno != ENOENT) - goto read_error; - /* No existing persisted stats file, so we're done */ - FreeFile(qfile); - return; - } +/* + * Scan the statement entries, skipping dropped ones. entry_seq_next() + * returns NULL at the end, and the hash key of the entry in *key if not NULL. + * The entry's partition lock is held until the next call. + * + * XXX: The cumulative statistics system has no API to iterate over the + * entries of a kind, so scans use its hash table and DSA area directly, and + * lock entries without a reference. + */ +static void +entry_seq_init(dshash_seq_status *hstat) +{ + dshash_seq_init(hstat, pgStatLocal.kind_hash[PGSS_STATS_KIND], false); +} - buffer_size = 2048; - buffer = (char *) palloc(buffer_size); +static pgssSharedEntry * +entry_seq_next(dshash_seq_status *hstat, PgStat_HashKey *key) +{ + PgStatShared_HashEntry *hash_entry; - if (fread(&header, sizeof(uint32), 1, file) != 1 || - fread(&pgver, sizeof(uint32), 1, file) != 1 || - fread(&num, sizeof(int64), 1, file) != 1) - goto read_error; + /* Skip dropped entries */ + do + hash_entry = dshash_seq_next(hstat); + while (hash_entry != NULL && hash_entry->dropped); - if (header != PGSS_FILE_HEADER || - pgver != PGSS_PG_MAJOR_VERSION) - goto data_error; + if (hash_entry == NULL) + return NULL; + if (key) + *key = hash_entry->key; + return entry_from_hash(hash_entry); +} - for (int64 i = 0; i < num; i++) - { - pgssEntry temp; - pgssEntry *entry; - Size query_offset; +/* + * Drop an entry, without requesting a garbage collection of references, see + * release_refs_after_flush. + */ +static void +entry_drop(const PgStat_HashKey *key) +{ + (void) pgstat_drop_entry(key->kind, key->dboid, key->objid, true); +} - if (fread(&temp, sizeof(pgssEntry), 1, file) != 1) - goto read_error; +/* + * Return a reference to the entry for a key, or NULL if there is none. The + * entry may be evicted after this returns; callers that must not use an + * evicted entry check under its lock. + */ +static PgStat_EntryRef * +entry_find(const pgssHashKey *key) +{ + return pgstat_get_entry_ref(PGSS_STATS_KIND, InvalidOid, + pgss_stats_objid(key), false, NULL); +} - /* Encoding is the only field we can easily sanity-check */ - if (!PG_VALID_BE_ENCODING(temp.encoding)) - goto data_error; +/* + * Write the query text of an entry after it in the statistics file, a length + * of -2 if it has no text, or of -1 if the entry is not to be restored. This + * runs at shutdown, when nothing else uses the entries, and loads the query + * text file once. + */ +static bool +pgss_to_serialized_data_cb(const PgStat_HashKey *key, + const PgStatShared_Common *header, FILE *statfile) +{ + const pgssEntry *entry = &((const pgssSharedEntry *) header)->entry; + const char *qstr = NULL; + int len = -1; - /* Resize buffer as needed */ - if (temp.query_len >= buffer_size) + /* Don't save texts if told not to, nor uninitialized or sticky entries */ + if (pgss_save && entry->stats_since != 0 && !IS_STICKY(entry->counters)) + { + if (!pgss_saved_qbuffer_loaded) { - buffer_size = Max(buffer_size * 2, temp.query_len + 1); - buffer = repalloc(buffer, buffer_size); + pgss_saved_qbuffer = qtext_load_file(&pgss_saved_qbuffer_size); + pgss_saved_qbuffer_loaded = true; } + qstr = qtext_fetch(entry->query_offset, entry->query_len, + pgss_saved_qbuffer, pgss_saved_qbuffer_size); + len = qstr ? entry->query_len : -2; + } - if (fread(buffer, 1, temp.query_len + 1, file) != temp.query_len + 1) - goto read_error; - - /* Should have a trailing null, but let's make sure */ - buffer[temp.query_len] = '\0'; - - /* Skip loading "sticky" entries */ - if (IS_STICKY(temp.counters)) - continue; + if (fwrite(&len, sizeof(len), 1, statfile) != 1) + return false; + if (len > 0 && fwrite(qstr, 1, len, statfile) != (size_t) len) + return false; - /* Store the query text */ - query_offset = pgss->extent; - if (fwrite(buffer, 1, temp.query_len + 1, qfile) != temp.query_len + 1) - goto write_error; - pgss->extent += temp.query_len + 1; - - /* make the hashtable entry (discards old entries if too many) */ - entry = entry_alloc(&temp.key, query_offset, temp.query_len, - temp.encoding, - false); - - /* copy in the actual stats */ - entry->counters = temp.counters; - entry->stats_since = temp.stats_since; - entry->minmax_stats_since = temp.minmax_stats_since; - } + return true; +} - /* Read global statistics for pg_stat_statements */ - if (fread(&pgss->stats, sizeof(pgssGlobalStats), 1, file) != 1) - goto read_error; +/* + * Read the query text of an entry from the statistics file into the query + * text file, dropping entries that cannot be restored. An entry is kept + * without its text if it was saved without one, or if the text cannot be + * written. This runs at startup, when nothing else uses the entries. + */ +static bool +pgss_from_serialized_data_cb(const PgStat_HashKey *key, + PgStatShared_Common *header, FILE *statfile) +{ + pgssEntry *entry = &((pgssSharedEntry *) header)->entry; + char *qstr; + int len; - pfree(buffer); - FreeFile(file); - FreeFile(qfile); + if (fread(&len, sizeof(len), 1, statfile) != 1) + return false; + if (len < -2 || (len >= 0 && (Size) len >= MaxAllocSize)) + return false; /* - * Remove the persisted stats file so it's not included in - * backups/replication standbys, etc. A new file will be written on next - * shutdown. - * - * Note: it's okay if the PGSS_TEXT_FILE is included in a basebackup, - * because we remove that file on startup; it acts inversely to - * PGSS_DUMP_FILE, in that it is only supposed to be around when the - * server is running, whereas PGSS_DUMP_FILE is only supposed to be around - * when the server is not running. Leaving the file creates no danger of - * a newly restored database having a spurious record of execution costs, - * which is what we're really concerned about here. + * Restore the entry only if it was saved, and looks sane: initialized, + * executed, and with a key matching its stats object ID. Encoding is the + * only other field we can easily sanity-check. A length of -2 is an + * entry saved without its text. */ - unlink(PGSS_DUMP_FILE); + if (len == -1 || !pgss_save || + entry->stats_since == 0 || IS_STICKY(entry->counters) || + entry->key.queryid == INT64CONST(0) || + !PG_VALID_BE_ENCODING(entry->encoding) || + key->dboid != InvalidOid || + key->objid != pgss_stats_objid(&entry->key)) + { + if (len > 0 && fseek(statfile, len, SEEK_CUR) != 0) + return false; - return; + /* Drop the entry, keeping the rest of the file */ + entry_drop(key); + return true; + } -read_error: - ereport(LOG, - (errcode_for_file_access(), - errmsg("could not read file \"%s\": %m", - PGSS_DUMP_FILE))); - goto fail; -data_error: - ereport(LOG, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("ignoring invalid data in file \"%s\"", - PGSS_DUMP_FILE))); - goto fail; -write_error: - ereport(LOG, - (errcode_for_file_access(), - errmsg("could not write file \"%s\": %m", - PGSS_TEXT_FILE))); -fail: - if (buffer) - pfree(buffer); - if (file) - FreeFile(file); - if (qfile) - FreeFile(qfile); - /* If possible, throw away the bogus file; ignore any error */ - unlink(PGSS_DUMP_FILE); + /* Store the query text, if any */ + entry->query_offset = 0; + entry->query_len = -1; + if (len >= 0) + { + qstr = palloc(len + 1); + if (len > 0 && fread(qstr, 1, len, statfile) != (size_t) len) + { + pfree(qstr); + return false; + } + qstr[len] = '\0'; + if (qtext_store(qstr, len, &entry->query_offset, NULL)) + entry->query_len = len; + pfree(qstr); + } - /* - * Don't unlink PGSS_TEXT_FILE here; it should always be around while the - * server is running with pg_stat_statements enabled - */ + pg_atomic_fetch_add_u64(&pgss->num_entries, 1); + + return true; } /* - * shmem_shutdown hook: Dump statistics into file. - * - * Note: we don't bother with acquiring lock, because there should be no - * other processes running when this is called. + * Clean up after the statistics file has been written, read or discarded. */ static void -pgss_shmem_shutdown(int code, Datum arg) +pgss_finish_cb(PgStat_StatsFileOp status) { - FILE *file; - char *qbuffer = NULL; - Size qbuffer_size = 0; - HASH_SEQ_STATUS hash_seq; - int64 num_entries; - pgssEntry *entry; - - /* Don't try to dump during a crash. */ - if (code) - return; - - /* Safety check ... shouldn't get here unless shmem is set up. */ - if (!pgss || !pgss_hash) - return; - - /* Don't dump if told not to. */ - if (!pgss_save) - return; - - file = AllocateFile(PGSS_DUMP_FILE ".tmp", PG_BINARY_W); - if (file == NULL) - goto error; - - if (fwrite(&PGSS_FILE_HEADER, sizeof(uint32), 1, file) != 1) - goto error; - if (fwrite(&PGSS_PG_MAJOR_VERSION, sizeof(uint32), 1, file) != 1) - goto error; - num_entries = hash_get_num_entries(pgss_hash); - if (fwrite(&num_entries, sizeof(int64), 1, file) != 1) - goto error; + switch (status) + { + case STATS_WRITE: + if (pgss_saved_qbuffer) + pfree(pgss_saved_qbuffer); + pgss_saved_qbuffer = NULL; + pgss_saved_qbuffer_size = 0; + pgss_saved_qbuffer_loaded = false; + break; - qbuffer = qtext_load_file(&qbuffer_size); - if (qbuffer == NULL) - goto error; + case STATS_READ: + /* Global statistics are restored only if entries are as well */ + if (!pgss_save) + pgss_global_stats_reset_all_cb(GetCurrentTimestamp()); - /* - * When serializing to disk, we store query texts immediately after their - * entry data. Any orphaned query texts are thereby excluded. - */ - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) - { - int len = entry->query_len; - char *qstr = qtext_fetch(entry->query_offset, len, - qbuffer, qbuffer_size); + /* Deallocate entries beyond pg_stat_statements.max */ + if (pg_atomic_read_u64(&pgss->num_entries) > (uint64) pgss_max) + entry_dealloc(); + break; - if (qstr == NULL) - continue; /* Ignore any entries with bogus texts */ + case STATS_DISCARD: - if (fwrite(entry, sizeof(pgssEntry), 1, file) != 1 || - fwrite(qstr, 1, len + 1, file) != len + 1) - { - /* note: we assume hash_seq_term won't change errno */ - hash_seq_term(&hash_seq); - goto error; - } + /* + * After a crash or a corrupted file, all entries have been + * dropped, so start again with an empty query text file. + */ + pg_atomic_write_u64(&pgss->num_entries, 0); + qtext_reset_file(); + pgss->extent = 0; + pgss->mean_query_len = ASSUMED_LENGTH_INIT; + break; } +} - /* Dump global statistics for pg_stat_statements */ - if (fwrite(&pgss->stats, sizeof(pgssGlobalStats), 1, file) != 1) - goto error; +/* + * Initialize the global statistics. + */ +static void +pgss_global_stats_init_shmem_cb(void *stats) +{ + pgssGlobalStats *gstats = (pgssGlobalStats *) stats; - pfree(qbuffer); - qbuffer = NULL; + gstats->dealloc = 0; + gstats->stats_reset = GetCurrentTimestamp(); +} - if (FreeFile(file)) - { - file = NULL; - goto error; - } +/* + * Return the global statistics in shared memory, protected by pgss->mutex. + */ +static inline pgssGlobalStats * +pgss_global_stats(void) +{ + return (pgssGlobalStats *) + pgstat_get_custom_shmem_data(PGSS_GLOBAL_STATS_KIND); +} - /* - * Rename file into place, so we atomically replace any old one. - */ - (void) durable_rename(PGSS_DUMP_FILE ".tmp", PGSS_DUMP_FILE, LOG); +/* + * Reset the global statistics, as of "ts". + */ +static void +pgss_global_stats_reset_all_cb(TimestampTz ts) +{ + pgssGlobalStats *gstats = pgss_global_stats(); - /* Unlink query-texts file; it's not needed while shutdown */ - unlink(PGSS_TEXT_FILE); + SpinLockAcquire(&pgss->mutex); + gstats->dealloc = 0; + gstats->stats_reset = ts; + SpinLockRelease(&pgss->mutex); +} - return; +/* + * Copy the global statistics into the cumulative statistics snapshot. + */ +static void +pgss_global_stats_snapshot_cb(void) +{ + pgssGlobalStats *snap = (pgssGlobalStats *) + pgstat_get_custom_snapshot_data(PGSS_GLOBAL_STATS_KIND); + pgssGlobalStats *gstats = pgss_global_stats(); -error: - ereport(LOG, - (errcode_for_file_access(), - errmsg("could not write file \"%s\": %m", - PGSS_DUMP_FILE ".tmp"))); - if (qbuffer) - pfree(qbuffer); - if (file) - FreeFile(file); - unlink(PGSS_DUMP_FILE ".tmp"); - unlink(PGSS_TEXT_FILE); + SpinLockAcquire(&pgss->mutex); + *snap = *gstats; + SpinLockRelease(&pgss->mutex); } /* @@ -855,7 +970,7 @@ pgss_post_parse_analyze(ParseState *pstate, Query *query, const JumbleState *jst prev_post_parse_analyze_hook(pstate, query, jstate); /* Safety check... */ - if (!pgss || !pgss_hash || !pgss_enabled(nesting_level)) + if (!pgss || !pgss_enabled(nesting_level)) return; /* @@ -1317,14 +1432,16 @@ pgss_store(const char *query, int64 queryId, PlannedStmtOrigin planOrigin) { pgssHashKey key; - pgssEntry *entry; + PgStat_EntryRef *entry_ref; + pgssSharedEntry *shared_entry; + pgssPendingEntry *entry; char *norm_query = NULL; int encoding = GetDatabaseEncoding(); Assert(query != NULL); /* Safety check... */ - if (!pgss || !pgss_hash) + if (!pgss) return; /* @@ -1351,19 +1468,29 @@ pgss_store(const char *query, int64 queryId, key.queryid = queryId; key.toplevel = toplevel; - /* Lookup the hash table entry with shared lock. */ - LWLockAcquire(&pgss->lock.lock, LW_SHARED); - - entry = (pgssEntry *) hash_search(pgss_hash, &key, HASH_FIND, NULL); + /* Lookup the hash table entry */ + entry_ref = entry_find(&key); /* Create new entry, if not present */ - if (!entry) + if (!entry_ref) { Size query_offset; int gc_count; bool stored; bool do_gc; + /* + * If we're at twice the pg_stat_statements.max limit, don't create a + * new entry. This is to prevent unbounded growth of the hash table + * and should be rare. + */ + if (pgstat_get_entry_count(PGSS_STATS_KIND) >= 2 * (uint64) pgss_max) + return; + + /* Make space if needed */ + if (pg_atomic_read_u64(&pgss->num_entries) >= (uint64) pgss_max) + entry_dealloc(); + /* * Create a new, normalized query string if caller asked. We don't * need to hold the lock while doing this work. (Note: in any case, @@ -1372,15 +1499,12 @@ pgss_store(const char *query, int64 queryId, * handled by entry_alloc.) */ if (jstate) - { - LWLockRelease(&pgss->lock.lock); norm_query = generate_normalized_query(jstate, query, query_location, &query_len); - LWLockAcquire(&pgss->lock.lock, LW_SHARED); - } /* Append new query text to file with only shared lock held */ + LWLockAcquire(&pgss->lock.lock, LW_SHARED); stored = qtext_store(norm_query ? norm_query : query, query_len, &query_offset, &gc_count); @@ -1406,33 +1530,62 @@ pgss_store(const char *query, int64 queryId, stored = qtext_store(norm_query ? norm_query : query, query_len, &query_offset, NULL); - /* If we failed to write to the text file, give up */ - if (!stored) - goto done; - /* OK to create a new hashtable entry */ - entry = entry_alloc(&key, query_offset, query_len, encoding, - jstate != NULL); + if (stored) + entry_ref = entry_alloc(&key, query_offset, query_len, encoding, + jstate != NULL); /* If needed, perform garbage collection while exclusive lock held */ if (do_gc) gc_qtexts(); + LWLockRelease(&pgss->lock.lock); + + if (norm_query) + pfree(norm_query); + + /* Give up if we failed to write the text, or the entry was evicted */ + if (!entry_ref) + return; } + shared_entry = (pgssSharedEntry *) entry_ref->shared_stats; + /* Increment the counts, except when jstate is not NULL */ if (!jstate) { + uint32 minmax_generation; + Assert(kind == PGSS_PLAN || kind == PGSS_EXEC); /* - * Grab the spinlock while updating the counters (see comment about - * locking rules at the head of the file) + * If the entry was evicted since entry_find(), our reference keeps + * its memory: the usage goes to the evicted entry, and the flush + * discards the counts. */ - SpinLockAcquire(&entry->mutex); + pgstat_prep_pending_from_entry_ref(entry_ref); + entry = (pgssPendingEntry *) entry_ref->pending; + + /* Discard min/max values from before a min/max reset, see pgssEntry */ + minmax_generation = pg_atomic_read_u32(&shared_entry->entry.minmax_generation); + if (entry->minmax_generation != minmax_generation) + { + entry_reset_minmax(&entry->counters); + entry->minmax_generation = minmax_generation; + } - /* "Unstick" entry if it was previously sticky */ - if (IS_STICKY(entry->counters)) - entry->counters.usage = USAGE_INIT; + /* + * Update the usage under the entry lock, so that eviction passes see + * it. If the lock is busy, the statement is running concurrently, so + * hot enough not to be evicted meanwhile: leave its usage to the + * flush rather than wait on the lock. + */ + if (pgstat_lock_entry(entry_ref, true)) + { + entry_add_usage(&shared_entry->entry, USAGE_EXEC(total_time)); + pgstat_unlock_entry(entry_ref); + } + else + entry->usage += USAGE_EXEC(total_time); entry->counters.calls[kind] += 1; entry->counters.total_time[kind] += total_time; @@ -1491,7 +1644,6 @@ pgss_store(const char *query, int64 queryId, entry->counters.local_blk_write_time += INSTR_TIME_GET_MILLISEC(bufusage->local_blk_write_time); entry->counters.temp_blk_read_time += INSTR_TIME_GET_MILLISEC(bufusage->temp_blk_read_time); entry->counters.temp_blk_write_time += INSTR_TIME_GET_MILLISEC(bufusage->temp_blk_write_time); - entry->counters.usage += USAGE_EXEC(total_time); entry->counters.wal_records += walusage->wal_records; entry->counters.wal_fpi += walusage->wal_fpi; entry->counters.wal_bytes += walusage->wal_bytes; @@ -1527,16 +1679,7 @@ pgss_store(const char *query, int64 queryId, entry->counters.generic_plan_calls++; else if (planOrigin == PLAN_STMT_CACHE_CUSTOM) entry->counters.custom_plan_calls++; - - SpinLockRelease(&entry->mutex); } - -done: - LWLockRelease(&pgss->lock.lock); - - /* We postpone this clean-up until we're out of the lock */ - if (norm_query) - pfree(norm_query); } /* @@ -1714,8 +1857,11 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, Size qbuffer_size = 0; Size extent = 0; int gc_count = 0; - HASH_SEQ_STATUS hash_seq; - pgssEntry *entry; + dshash_seq_status hstat; + pgssSharedEntry *shared_entry; + pgssEntry *entries; + int nentries = 0; + int nalloc; /* * Superusers or roles with the privileges of pg_read_all_stats members @@ -1724,7 +1870,7 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, is_allowed_role = has_privs_of_role(userid, ROLE_PG_READ_ALL_STATS); /* hash table must exist already */ - if (!pgss || !pgss_hash) + if (!pgss) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("pg_stat_statements must be loaded via \"shared_preload_libraries\""))); @@ -1784,6 +1930,14 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, elog(ERROR, "incorrect number of output arguments"); } + /* + * Make this backend's pending statistics visible to this read, as this + * flushes the pending statistics of all kinds. Versions before 1.14 may + * run this in a parallel worker, which cannot do that. + */ + if (!IsParallelWorker()) + pgstat_force_next_flush(); + /* * We'd like to load the query text file (if needed) while not holding any * lock on pgss->lock. In the worst case we'll have to do this again @@ -1812,14 +1966,9 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, /* * Get shared lock, load or reload the query text file if we must, and - * iterate over the hashtable entries. - * - * With a large hash table, we might be holding the lock rather longer - * than one could wish. However, this only blocks creation of new hash - * table entries, and the larger the hash table the less likely that is to - * be needed. So we can hope this is okay. Perhaps someday we'll decide - * we need to partition the hash table to limit the time spent holding any - * one lock. + * copy the initialized entries, so that rows are built without holding + * the lock nor the hash table partition locks, which block the creation + * of entries. */ LWLockAcquire(&pgss->lock.lock, LW_SHARED); @@ -1845,9 +1994,38 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, } } - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) + nalloc = 64; + entries = palloc_array(pgssEntry, nalloc); + entry_seq_init(&hstat); + while ((shared_entry = entry_seq_next(&hstat, NULL)) != NULL) { + pgssEntry *entry; + + if (nentries >= nalloc) + { + nalloc *= 2; + entries = repalloc_huge(entries, nalloc * sizeof(pgssEntry)); + } + + LWLockAcquire(&shared_entry->header.lock, LW_SHARED); + entry = &entries[nentries]; + *entry = shared_entry->entry; + LWLockRelease(&shared_entry->header.lock); + + /* + * Skip entry if not initialized yet, or unexecuted (ie, it's a + * pending "sticky" entry) + */ + if (entry->stats_since != 0 && !IS_STICKY(entry->counters)) + nentries++; + } + dshash_seq_term(&hstat); + + LWLockRelease(&pgss->lock.lock); + + for (int n = 0; n < nentries; n++) + { + pgssEntry *entry = &entries[n]; Datum values[PG_STAT_STATEMENTS_COLS]; bool nulls[PG_STAT_STATEMENTS_COLS]; int i = 0; @@ -1918,22 +2096,10 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, nulls[i++] = true; } - /* copy counters to a local variable to keep locking time short */ - SpinLockAcquire(&entry->mutex); tmp = entry->counters; - SpinLockRelease(&entry->mutex); - - /* - * The spinlock is not required when reading these two as they are - * always updated when holding pgss->lock exclusively. - */ stats_since = entry->stats_since; minmax_stats_since = entry->minmax_stats_since; - /* Skip entry if unexecuted (ie, it's a pending "sticky" entry) */ - if (IS_STICKY(tmp)) - continue; - /* Note that we rely on PGSS_PLAN being 0 and PGSS_EXEC being 1. */ for (int kind = 0; kind < PGSS_NUMKIND; kind++) { @@ -2058,8 +2224,7 @@ pg_stat_statements_internal(FunctionCallInfo fcinfo, tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); } - - LWLockRelease(&pgss->lock.lock); + pfree(entries); if (qbuffer) pfree(qbuffer); @@ -2079,7 +2244,7 @@ pg_stat_statements_info(PG_FUNCTION_ARGS) Datum values[PG_STAT_STATEMENTS_INFO_COLS] = {0}; bool nulls[PG_STAT_STATEMENTS_INFO_COLS] = {0}; - if (!pgss || !pgss_hash) + if (!pgss) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("pg_stat_statements must be loaded via \"shared_preload_libraries\""))); @@ -2090,7 +2255,7 @@ pg_stat_statements_info(PG_FUNCTION_ARGS) /* Read global statistics for pg_stat_statements */ SpinLockAcquire(&pgss->mutex); - stats = pgss->stats; + stats = *pgss_global_stats(); SpinLockRelease(&pgss->mutex); values[0] = Int64GetDatum(stats.dealloc); @@ -2100,10 +2265,10 @@ pg_stat_statements_info(PG_FUNCTION_ARGS) } /* - * Allocate a new hashtable entry. - * caller must hold an exclusive lock on pgss->lock - * - * "query" need not be null-terminated; we rely on query_len instead + * Allocate a new hashtable entry, with its key and the offset of its query + * text in the external file. If another backend created the entry since we + * looked it up, return theirs: the caller counts in it, with their text. + * Caller must hold pgss->lock exclusively. * * If "sticky" is true, make the new entry artificially sticky so that it will * probably still be there when the query finishes execution. We do this by @@ -2111,45 +2276,182 @@ pg_stat_statements_info(PG_FUNCTION_ARGS) * speaking, query strings are normalized on a best effort basis, though it * would be difficult to demonstrate this even under artificial conditions.) * - * Note: despite needing exclusive lock, it's not an error for the target - * entry to already exist. This is because pgss_store releases and - * reacquires lock after failing to find a match; so someone else could - * have made the entry while we waited to get exclusive lock. + * Returns NULL if the entry was evicted meanwhile. + * + * XXX: The cumulative statistics system makes a new entry visible to other + * backends, zeroed, before its creator can initialize it. A way to + * initialize an entry's data atomically with its insertion, for example a + * callback of PgStat_KindInfo called by pgstat_get_entry_ref() while the hash + * table partition lock is held, would make that window go away. */ -static pgssEntry * -entry_alloc(pgssHashKey *key, Size query_offset, int query_len, int encoding, - bool sticky) +static PgStat_EntryRef * +entry_alloc(const pgssHashKey *key, Size query_offset, int query_len, + int encoding, bool sticky) { + PgStat_EntryRef *entry_ref; pgssEntry *entry; - bool found; + bool created; + + entry_ref = pgstat_get_entry_ref(PGSS_STATS_KIND, InvalidOid, + pgss_stats_objid(key), true, &created); + if (!created) + return entry_ref; + + entry = &((pgssSharedEntry *) entry_ref->shared_stats)->entry; + pgstat_lock_entry(entry_ref, false); + + if (pgstat_entry_ref_is_stale(entry_ref)) + { + pgstat_unlock_entry(entry_ref); + return NULL; + } + entry->key = *key; + + /* Set the initial usage, unless concurrent executions set it already */ + if (!entry->executed && sticky) + { + SpinLockAcquire(&pgss->mutex); + entry->usage = pgss->cur_median_usage; + SpinLockRelease(&pgss->mutex); + } + else if (!entry->executed) + entry->usage = USAGE_INIT; + /* ... and don't forget the query text metadata */ + entry->query_offset = query_offset; + entry->query_len = query_len; + entry->encoding = encoding; + + /* + * Count the entry before marking it live, as entry_remove() only + * decrements the count for live entries. + */ + pg_atomic_fetch_add_u64(&pgss->num_entries, 1); + + entry->stats_since = GetCurrentTimestamp(); + entry->minmax_stats_since = entry->stats_since; - /* Make space if needed */ - while (hash_get_num_entries(pgss_hash) >= (int64) pgss_max) - entry_dealloc(); + pgstat_unlock_entry(entry_ref); - /* Find or create an entry with desired hash code */ - entry = (pgssEntry *) hash_search(pgss_hash, key, HASH_ENTER, &found); + return entry_ref; +} + +/* + * Add executions to the usage of an entry, "unsticking" it at its first + * execution. Caller must hold the entry lock. + */ +static void +entry_add_usage(pgssEntry *entry, double usage) +{ + entry->usage = (entry->executed ? entry->usage : USAGE_INIT) + usage; + entry->executed = true; +} - if (!found) +/* + * Add pending statistics to the entry. + */ +static PgStat_FlushResult +pgss_flush_pending_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary) +{ + pgssSharedEntry *shared_entry = (pgssSharedEntry *) entry_ref->shared_stats; + pgssPendingEntry *pending = (pgssPendingEntry *) entry_ref->pending; + Counters *dst = &shared_entry->entry.counters; + Counters *src = &pending->counters; + bool merge_minmax; + + /* + * Discard the statistics of an evicted entry; the reference is released + * after the flush. + */ + if (pgstat_entry_ref_is_stale(entry_ref)) { - /* New entry, initialize it */ - - /* reset the statistics */ - memset(&entry->counters, 0, sizeof(Counters)); - /* set the appropriate initial usage count */ - entry->counters.usage = sticky ? pgss->cur_median_usage : USAGE_INIT; - /* re-initialize the mutex each time ... we assume no one using it */ - SpinLockInit(&entry->mutex); - /* ... and don't forget the query text metadata */ - Assert(query_len >= 0); - entry->query_offset = query_offset; - entry->query_len = query_len; - entry->encoding = encoding; - entry->stats_since = GetCurrentTimestamp(); - entry->minmax_stats_since = entry->stats_since; + memset(pending, 0, sizeof(pgssPendingEntry)); + return PGSTAT_FLUSH_DONE; } - return entry; + if (!pgstat_lock_entry(entry_ref, nowait)) + return PGSTAT_FLUSH_LOCK_CONFLICT; + + merge_minmax = pending->minmax_generation == + pg_atomic_read_u32(&shared_entry->entry.minmax_generation); + + if (pending->usage > 0) + entry_add_usage(&shared_entry->entry, pending->usage); + + for (int kind = 0; kind < PGSS_NUMKIND; kind++) + { + PgStat_Counter n1 = dst->calls[kind]; + PgStat_Counter n2 = src->calls[kind]; + double delta; + + if (n2 == 0) + continue; + + /* Combine means and variances of both sets of calls */ + delta = src->mean_time[kind] - dst->mean_time[kind]; + dst->calls[kind] = n1 + n2; + dst->mean_time[kind] += delta * n2 / (n1 + n2); + dst->sum_var_time[kind] += src->sum_var_time[kind] + + delta * delta * n1 * n2 / (n1 + n2); + dst->total_time[kind] += src->total_time[kind]; + + /* min = 0 and max = 0 means that the values were reset */ + if (merge_minmax && (src->min_time[kind] != 0 || src->max_time[kind] != 0)) + { + if (dst->min_time[kind] == 0 && dst->max_time[kind] == 0) + { + dst->min_time[kind] = src->min_time[kind]; + dst->max_time[kind] = src->max_time[kind]; + } + else + { + dst->min_time[kind] = Min(dst->min_time[kind], src->min_time[kind]); + dst->max_time[kind] = Max(dst->max_time[kind], src->max_time[kind]); + } + } + } + + dst->rows += src->rows; + dst->shared_blks_hit += src->shared_blks_hit; + dst->shared_blks_read += src->shared_blks_read; + dst->shared_blks_dirtied += src->shared_blks_dirtied; + dst->shared_blks_written += src->shared_blks_written; + dst->local_blks_hit += src->local_blks_hit; + dst->local_blks_read += src->local_blks_read; + dst->local_blks_dirtied += src->local_blks_dirtied; + dst->local_blks_written += src->local_blks_written; + dst->temp_blks_read += src->temp_blks_read; + dst->temp_blks_written += src->temp_blks_written; + dst->shared_blk_read_time += src->shared_blk_read_time; + dst->shared_blk_write_time += src->shared_blk_write_time; + dst->local_blk_read_time += src->local_blk_read_time; + dst->local_blk_write_time += src->local_blk_write_time; + dst->temp_blk_read_time += src->temp_blk_read_time; + dst->temp_blk_write_time += src->temp_blk_write_time; + dst->wal_records += src->wal_records; + dst->wal_fpi += src->wal_fpi; + dst->wal_bytes += src->wal_bytes; + dst->wal_buffers_full += src->wal_buffers_full; + dst->jit_functions += src->jit_functions; + dst->jit_generation_time += src->jit_generation_time; + dst->jit_inlining_count += src->jit_inlining_count; + dst->jit_deform_time += src->jit_deform_time; + dst->jit_deform_count += src->jit_deform_count; + dst->jit_inlining_time += src->jit_inlining_time; + dst->jit_optimization_count += src->jit_optimization_count; + dst->jit_optimization_time += src->jit_optimization_time; + dst->jit_emission_count += src->jit_emission_count; + dst->jit_emission_time += src->jit_emission_time; + dst->parallel_workers_to_launch += src->parallel_workers_to_launch; + dst->parallel_workers_launched += src->parallel_workers_launched; + dst->generic_plan_calls += src->generic_plan_calls; + dst->custom_plan_calls += src->custom_plan_calls; + + pgstat_unlock_entry(entry_ref); + + /* The pending entry may be kept for reuse; don't add it twice */ + memset(pending, 0, sizeof(pgssPendingEntry)); + + return PGSTAT_FLUSH_DONE; } /* @@ -2158,8 +2460,8 @@ entry_alloc(pgssHashKey *key, Size query_offset, int query_len, int encoding, static int entry_cmp(const void *lhs, const void *rhs) { - double l_usage = (*(pgssEntry *const *) lhs)->counters.usage; - double r_usage = (*(pgssEntry *const *) rhs)->counters.usage; + double l_usage = ((const pgssDeallocEntry *) lhs)->usage; + double r_usage = ((const pgssDeallocEntry *) rhs)->usage; if (l_usage < r_usage) return -1; @@ -2170,20 +2472,45 @@ entry_cmp(const void *lhs, const void *rhs) } /* - * Deallocate least-used entries. + * Deallocate least-used entries, plus any excess over pg_stat_statements.max, + * once there are that many entries. * - * Caller must hold an exclusive lock on pgss->lock. + * Passes run one at a time, under pgss->dealloc_lock, which new entries + * don't need: they wait for a pass only at the limit. If another backend is + * running one, return at once, or once that pass is done if entries exceed + * the maximum by USAGE_DEALLOC_PERCENT percent, the excess a pass removes. */ static void entry_dealloc(void) { - HASH_SEQ_STATUS hash_seq; - pgssEntry **entries; - pgssEntry *entry; + dshash_seq_status hstat; + pgssSharedEntry *shared_entry; + PgStat_HashKey key; + pgssDeallocEntry *entries; + int nalloc; + uint64 num_entries; int nvictims; int i; Size tottextlen; int nvalidtexts; + pgssGlobalStats *gstats; + uint64 max = (uint64) pgss_max; + bool wait; + + /* At the limit, wait for the running pass instead of giving up */ + wait = pg_atomic_read_u64(&pgss->num_entries) >= + max + Max(max * USAGE_DEALLOC_PERCENT / 100, 1); + if (wait ? !LWLockAcquireOrWait(&pgss->dealloc_lock, LW_EXCLUSIVE) : + !LWLockConditionalAcquire(&pgss->dealloc_lock, LW_EXCLUSIVE)) + return; + + /* Someone else may have deallocated entries meanwhile */ + num_entries = pg_atomic_read_u64(&pgss->num_entries); + if (num_entries < max) + { + LWLockRelease(&pgss->dealloc_lock); + return; + } /* * Sort entries by usage and deallocate USAGE_DEALLOC_PERCENT of them. @@ -2195,58 +2522,115 @@ entry_dealloc(void) * Hopefully, that doesn't affect the mean too much; it doesn't seem worth * making two passes to get a more current result. Likewise, the new * cur_median_usage includes the entries we're about to zap. + * + * Entries whose lock is busy are in use, so skip them rather than wait; + * they miss this pass's decay. Entries may be initialized meanwhile, so + * grow the array as needed. */ - - entries = palloc_array(pgssEntry *, hash_get_num_entries(pgss_hash)); + nalloc = (int) Min(Max(num_entries, 64), + (uint64) MaxAllocHugeSize / sizeof(pgssDeallocEntry)); + entries = palloc_extended(nalloc * sizeof(pgssDeallocEntry), MCXT_ALLOC_HUGE); i = 0; tottextlen = 0; nvalidtexts = 0; - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) + entry_seq_init(&hstat); + while ((shared_entry = entry_seq_next(&hstat, &key)) != NULL) { - entries[i++] = entry; + pgssEntry *entry; + + if (!LWLockConditionalAcquire(&shared_entry->header.lock, LW_EXCLUSIVE)) + continue; + + entry = &shared_entry->entry; + if (entry->stats_since == 0) + { + LWLockRelease(&shared_entry->header.lock); + continue; + } + /* "Sticky" entries get a different usage decay rate. */ - if (IS_STICKY(entry->counters)) - entry->counters.usage *= STICKY_DECREASE_FACTOR; + if (!entry->executed) + entry->usage *= STICKY_DECREASE_FACTOR; else - entry->counters.usage *= USAGE_DECREASE_FACTOR; + entry->usage *= USAGE_DECREASE_FACTOR; + + if (i >= nalloc) + { + nalloc *= 2; + entries = repalloc_huge(entries, nalloc * sizeof(pgssDeallocEntry)); + } + entries[i].key = key; + entries[i].stats_since = entry->stats_since; + entries[i].usage = entry->usage; + i++; /* In the mean length computation, ignore dropped texts. */ if (entry->query_len >= 0) { tottextlen += entry->query_len + 1; nvalidtexts++; } + + LWLockRelease(&shared_entry->header.lock); } + dshash_seq_term(&hstat); /* Sort into increasing order by usage */ - qsort(entries, i, sizeof(pgssEntry *), entry_cmp); + qsort(entries, i, sizeof(pgssDeallocEntry), entry_cmp); + /* + * Under the mutex, as entries are created meanwhile and a garbage + * collection also sets the mean query length + */ + SpinLockAcquire(&pgss->mutex); /* Record the (approximate) median usage */ if (i > 0) - pgss->cur_median_usage = entries[i / 2]->counters.usage; + pgss->cur_median_usage = entries[i / 2].usage; /* Record the mean query length */ if (nvalidtexts > 0) pgss->mean_query_len = tottextlen / nvalidtexts; else pgss->mean_query_len = ASSUMED_LENGTH_INIT; + SpinLockRelease(&pgss->mutex); - /* Now zap an appropriate fraction of lowest-usage entries */ - nvictims = Max(10, i * USAGE_DEALLOC_PERCENT / 100); + /* + * Now zap an appropriate fraction of lowest-usage entries, plus the + * entries exceeding pg_stat_statements.max. + */ + nvictims = Max(10, (int) (Min((uint64) i, max) * + USAGE_DEALLOC_PERCENT / 100)); + num_entries = pg_atomic_read_u64(&pgss->num_entries); + if (num_entries > max) + nvictims += (int) Min(num_entries - max, (uint64) i); nvictims = Min(nvictims, i); + /* + * Remove the victims, unless used or reused since, removed by someone + * else, or busy; the next pass deallocates any excess left + */ for (i = 0; i < nvictims; i++) { - hash_search(pgss_hash, &entries[i]->key, HASH_REMOVE, NULL); + PgStat_EntryRef *entry_ref; + + entry_ref = pgstat_get_entry_ref(entries[i].key.kind, + entries[i].key.dboid, + entries[i].key.objid, false, NULL); + if (entry_ref != NULL && + entry_remove((pgssSharedEntry *) entry_ref->shared_stats, + entry_ref, &entries[i])) + entry_drop(&entries[i].key); } pfree(entries); /* Increment the number of times entries are deallocated */ + gstats = pgss_global_stats(); SpinLockAcquire(&pgss->mutex); - pgss->stats.dealloc += 1; + gstats->dealloc += 1; SpinLockRelease(&pgss->mutex); + + LWLockRelease(&pgss->dealloc_lock); } /* @@ -2465,10 +2849,12 @@ static bool need_gc_qtexts(void) { Size extent; + Size mean_query_len; - /* Read shared extent pointer */ + /* Read shared extent pointer and mean query length */ SpinLockAcquire(&pgss->mutex); extent = pgss->extent; + mean_query_len = pgss->mean_query_len; SpinLockRelease(&pgss->mutex); /* @@ -2488,12 +2874,31 @@ need_gc_qtexts(void) * query length in order to prevent garbage collection from thrashing * uselessly. */ - if ((uint64) extent < (uint64) pgss->mean_query_len * pgss_max * 2) + if ((uint64) extent < (uint64) mean_query_len * pgss_max * 2) return false; return true; } +/* + * Start again with an empty query text file. + */ +static void +qtext_reset_file(void) +{ + FILE *qfile; + + unlink(PGSS_TEXT_FILE); + qfile = AllocateFile(PGSS_TEXT_FILE, PG_BINARY_W); + if (qfile == NULL) + ereport(LOG, + (errcode_for_file_access(), + errmsg("could not create file \"%s\": %m", + PGSS_TEXT_FILE))); + else + FreeFile(qfile); +} + /* * Garbage-collect orphaned query texts in external file. * @@ -2516,8 +2921,8 @@ gc_qtexts(void) char *qbuffer; Size qbuffer_size; FILE *qfile = NULL; - HASH_SEQ_STATUS hash_seq; - pgssEntry *entry; + dshash_seq_status hstat; + pgssSharedEntry *shared_entry; Size extent; int nentries; @@ -2559,9 +2964,10 @@ gc_qtexts(void) extent = 0; nentries = 0; - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) + entry_seq_init(&hstat); + while ((shared_entry = entry_seq_next(&hstat, NULL)) != NULL) { + pgssEntry *entry = &shared_entry->entry; int query_len = entry->query_len; char *qry = qtext_fetch(entry->query_offset, query_len, @@ -2583,7 +2989,7 @@ gc_qtexts(void) (errcode_for_file_access(), errmsg("could not write file \"%s\": %m", PGSS_TEXT_FILE))); - hash_seq_term(&hash_seq); + dshash_seq_term(&hstat); goto gc_fail; } @@ -2591,6 +2997,7 @@ gc_qtexts(void) extent += query_len + 1; nentries++; } + dshash_seq_term(&hstat); /* * Truncate away any now-unused space. If this fails for some odd reason, @@ -2622,10 +3029,12 @@ gc_qtexts(void) * Also update the mean query length, to be sure that need_gc_qtexts() * won't still think we have a problem. */ + SpinLockAcquire(&pgss->mutex); if (nentries > 0) pgss->mean_query_len = extent / nentries; else pgss->mean_query_len = ASSUMED_LENGTH_INIT; + SpinLockRelease(&pgss->mutex); pfree(qbuffer); @@ -2651,12 +3060,15 @@ gc_fail: * Since the contents of the external file are now uncertain, mark all * hashtable entries as having invalid texts. */ - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) + entry_seq_init(&hstat); + while ((shared_entry = entry_seq_next(&hstat, NULL)) != NULL) { + pgssEntry *entry = &shared_entry->entry; + entry->query_offset = 0; entry->query_len = -1; } + dshash_seq_term(&hstat); /* * Destroy the query text file and create a new, empty one @@ -2675,7 +3087,9 @@ gc_fail: pgss->extent = 0; /* Reset mean_query_len to match the new state */ + SpinLockAcquire(&pgss->mutex); pgss->mean_query_len = ASSUMED_LENGTH_INIT; + SpinLockRelease(&pgss->mutex); /* * Bump the GC count even though we failed. @@ -2691,23 +3105,139 @@ gc_fail: record_gc_qtexts(); } -#define SINGLE_ENTRY_RESET(e) \ -if (e) { \ - if (minmax_only) { \ - /* When requested reset only min/max statistics of an entry */ \ - for (int kind = 0; kind < PGSS_NUMKIND; kind++) \ - { \ - e->counters.max_time[kind] = 0; \ - e->counters.min_time[kind] = 0; \ - } \ - e->minmax_stats_since = stats_reset; \ - } \ - else \ - { \ - /* Remove the key otherwise */ \ - hash_search(pgss_hash, &e->key, HASH_REMOVE, NULL); \ - num_remove++; \ - } \ +/* + * Remove a live entry: readers skip it, and pgss->num_entries no longer + * counts it. Its query text becomes garbage in the external file. Returns + * true if this call removed it, for the caller to drop it, or false if it was + * not live, or another removal got it first. + * + * With entry_ref, also return false if the entry was reused since the + * reference was obtained. With victim, also return false if the entry is + * not the one entry_dealloc() chose, was used since, or its lock is busy. + * + * XXX: pgss->num_entries is needed as the entry counts of the cumulative + * statistics system include dropped entries until their references are + * released. + */ +static bool +entry_remove(pgssSharedEntry *shared_entry, PgStat_EntryRef *entry_ref, + const pgssDeallocEntry *victim) +{ + pgssEntry *entry = &shared_entry->entry; + bool removed = false; + + if (victim == NULL) + LWLockAcquire(&shared_entry->header.lock, LW_EXCLUSIVE); + else if (!LWLockConditionalAcquire(&shared_entry->header.lock, LW_EXCLUSIVE)) + return false; + + if (entry->stats_since != 0 && + (entry_ref == NULL || !pgstat_entry_ref_is_stale(entry_ref)) && + (victim == NULL || + (entry->stats_since == victim->stats_since && + entry->usage == victim->usage))) + { + entry->stats_since = 0; + pg_atomic_sub_fetch_u64(&pgss->num_entries, 1); + removed = true; + } + LWLockRelease(&shared_entry->header.lock); + + return removed; +} + +/* + * Reset min/max statistics. For a shared entry's counters, caller must hold + * the entry lock exclusively. + */ +static void +entry_reset_minmax(Counters *counters) +{ + for (int kind = 0; kind < PGSS_NUMKIND; kind++) + { + counters->max_time[kind] = 0; + counters->min_time[kind] = 0; + } +} + +/* + * Reset the min/max statistics of a shared entry, and have backends discard + * their pending ones. Caller must hold the entry lock exclusively. + */ +static void +entry_reset_shared_minmax(pgssSharedEntry *shared_entry, + TimestampTz stats_reset) +{ + entry_reset_minmax(&shared_entry->entry.counters); + shared_entry->entry.minmax_stats_since = stats_reset; + pg_atomic_fetch_add_u32(&shared_entry->entry.minmax_generation, 1); +} + +/* + * Reset the entry of a key, if it exists. + */ +static void +entry_reset_single(const pgssHashKey *key, bool minmax_only, + TimestampTz stats_reset) +{ + PgStat_EntryRef *entry_ref; + pgssSharedEntry *shared_entry; + + entry_ref = entry_find(key); + if (!entry_ref) + return; + + shared_entry = (pgssSharedEntry *) entry_ref->shared_stats; + + if (minmax_only) + { + pgstat_lock_entry(entry_ref, false); + if (!pgstat_entry_ref_is_stale(entry_ref) && + shared_entry->entry.stats_since != 0) + entry_reset_shared_minmax(shared_entry, stats_reset); + pgstat_unlock_entry(entry_ref); + } + /* Remove the entry, unless it is not initialized or was evicted meanwhile */ + else if (entry_remove(shared_entry, entry_ref, NULL)) + entry_drop(&entry_ref->shared_entry->key); +} + +/* + * Does an entry match the key fields of entry_reset()? Zero fields match + * anything. + */ +static bool +entry_reset_matches(const pgssEntry *entry, const pgssHashKey *key) +{ + return (!key->userid || entry->key.userid == key->userid) && + (!key->dbid || entry->key.dbid == key->dbid) && + (!key->queryid || entry->key.queryid == key->queryid); +} + +/* + * pgstat_drop_matching_entries() callback for entry_reset(), removing the + * matching entries. Caller holds the entry's partition lock exclusively. + */ +static bool +entry_reset_drop_matching(PgStatShared_HashEntry *hash_entry, Datum match_data) +{ + pgssHashKey *key = (pgssHashKey *) DatumGetPointer(match_data); + pgssSharedEntry *shared_entry = entry_from_hash(hash_entry); + bool matches; + + /* pgstat_drop_matching_entries() skips dropped entries */ + Assert(shared_entry != NULL); + + /* + * Skip entries not initialized yet, which have no key to match, and + * entries being removed by someone else. + */ + LWLockAcquire(&shared_entry->header.lock, LW_SHARED); + matches = shared_entry->entry.stats_since != 0 && + entry_reset_matches(&shared_entry->entry, key); + LWLockRelease(&shared_entry->header.lock); + + return matches && entry_remove(shared_entry, NULL, NULL); } /* @@ -2716,83 +3246,78 @@ if (e) { \ static TimestampTz entry_reset(Oid userid, Oid dbid, int64 queryid, bool minmax_only) { - HASH_SEQ_STATUS hash_seq; - pgssEntry *entry; FILE *qfile; - int64 num_entries; - int64 num_remove = 0; pgssHashKey key; TimestampTz stats_reset; - if (!pgss || !pgss_hash) + if (!pgss) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("pg_stat_statements must be loaded via \"shared_preload_libraries\""))); - LWLockAcquire(&pgss->lock.lock, LW_EXCLUSIVE); - num_entries = hash_get_num_entries(pgss_hash); + memset(&key, 0, sizeof(pgssHashKey)); + key.userid = userid; + key.dbid = dbid; + key.queryid = queryid; stats_reset = GetCurrentTimestamp(); + /* Removals hold pgss->lock exclusively, to reset the query text file */ + if (!minmax_only) + LWLockAcquire(&pgss->lock.lock, LW_EXCLUSIVE); + if (userid != 0 && dbid != 0 && queryid != INT64CONST(0)) { /* If all the parameters are available, use the fast path. */ - memset(&key, 0, sizeof(pgssHashKey)); - key.userid = userid; - key.dbid = dbid; - key.queryid = queryid; /* * Reset the entry if it exists, starting with the non-top-level * entry. */ key.toplevel = false; - entry = (pgssEntry *) hash_search(pgss_hash, &key, HASH_FIND, NULL); - - SINGLE_ENTRY_RESET(entry); + entry_reset_single(&key, minmax_only, stats_reset); /* Also reset the top-level entry if it exists. */ key.toplevel = true; - entry = (pgssEntry *) hash_search(pgss_hash, &key, HASH_FIND, NULL); - - SINGLE_ENTRY_RESET(entry); + entry_reset_single(&key, minmax_only, stats_reset); } - else if (userid != 0 || dbid != 0 || queryid != INT64CONST(0)) + else if (minmax_only) { - /* Reset entries corresponding to valid parameters. */ - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) + dshash_seq_status hstat; + pgssSharedEntry *shared_entry; + + /* Reset min/max of entries corresponding to valid parameters, or all */ + entry_seq_init(&hstat); + while ((shared_entry = entry_seq_next(&hstat, NULL)) != NULL) { - if ((!userid || entry->key.userid == userid) && - (!dbid || entry->key.dbid == dbid) && - (!queryid || entry->key.queryid == queryid)) - { - SINGLE_ENTRY_RESET(entry); - } + LWLockAcquire(&shared_entry->header.lock, LW_EXCLUSIVE); + if (shared_entry->entry.stats_since != 0 && + entry_reset_matches(&shared_entry->entry, &key)) + entry_reset_shared_minmax(shared_entry, stats_reset); + LWLockRelease(&shared_entry->header.lock); } + dshash_seq_term(&hstat); } else { - /* Reset all entries. */ - hash_seq_init(&hash_seq, pgss_hash); - while ((entry = hash_seq_search(&hash_seq)) != NULL) - { - SINGLE_ENTRY_RESET(entry); - } + /* Reset entries corresponding to valid parameters, or all */ + pgstat_drop_matching_entries(PGSS_STATS_KIND, + entry_reset_drop_matching, + PointerGetDatum(&key)); } + if (minmax_only) + return stats_reset; + /* All entries are removed? */ - if (num_entries != num_remove) + if (pg_atomic_read_u64(&pgss->num_entries) != 0) goto release_lock; /* * Reset global statistics for pg_stat_statements since all entries are * removed. */ - SpinLockAcquire(&pgss->mutex); - pgss->stats.dealloc = 0; - pgss->stats.stats_reset = stats_reset; - SpinLockRelease(&pgss->mutex); + pgss_global_stats_reset_all_cb(stats_reset); /* * Write new empty query file, perhaps even creating a new one to recover diff --git a/contrib/pg_stat_statements/pg_stat_statements.control b/contrib/pg_stat_statements/pg_stat_statements.control index 2eee0ceffa8..61ae41efc14 100644 --- a/contrib/pg_stat_statements/pg_stat_statements.control +++ b/contrib/pg_stat_statements/pg_stat_statements.control @@ -1,5 +1,5 @@ # pg_stat_statements extension comment = 'track planning and execution statistics of all SQL statements executed' -default_version = '1.13' +default_version = '1.14' module_pathname = '$libdir/pg_stat_statements' relocatable = true diff --git a/contrib/pg_stat_statements/sql/oldextversions.sql b/contrib/pg_stat_statements/sql/oldextversions.sql index e416efe9ffb..a6da6f6482a 100644 --- a/contrib/pg_stat_statements/sql/oldextversions.sql +++ b/contrib/pg_stat_statements/sql/oldextversions.sql @@ -68,4 +68,20 @@ AlTER EXTENSION pg_stat_statements UPDATE TO '1.13'; \d pg_stat_statements SELECT count(*) > 0 AS has_data FROM pg_stat_statements; +-- Before 1.14, these functions are PARALLEL SAFE, and work in a worker +SET max_parallel_workers_per_gather = 2; +SET debug_parallel_query = on; +SELECT count(*) > 0 AS has_data FROM pg_stat_statements; +SELECT pg_stat_statements_reset() IS NOT NULL AS t; +RESET debug_parallel_query; +RESET max_parallel_workers_per_gather; + +-- pg_stat_statements() restricted to the leader in 1.14 +AlTER EXTENSION pg_stat_statements UPDATE TO '1.14'; +SELECT oid::regprocedure, proparallel +FROM pg_proc +WHERE oid IN ('pg_stat_statements(boolean)'::regprocedure, + 'pg_stat_statements_reset(oid, oid, bigint, boolean)'::regprocedure) +ORDER BY oid::regprocedure::text; + DROP EXTENSION pg_stat_statements; diff --git a/contrib/pg_stat_statements/t/010_restart.pl b/contrib/pg_stat_statements/t/010_restart.pl index eab26c01f8d..012b617352d 100644 --- a/contrib/pg_stat_statements/t/010_restart.pl +++ b/contrib/pg_stat_statements/t/010_restart.pl @@ -27,6 +27,13 @@ is( $node->safe_psql( "CREATE TABLE t1 (a int)\nSELECT a FROM t1", 'pg_stat_statements populated'); +my $count_query = + "SELECT count(*) FROM pg_stat_statements WHERE query NOT LIKE '%pg_stat_statements%'"; +my $stats_reset_query = + "SELECT stats_reset::text FROM pg_stat_statements_info"; + +my $stats_reset = $node->safe_psql('postgres', $stats_reset_query); + $node->restart; is( $node->safe_psql( @@ -36,6 +43,15 @@ is( $node->safe_psql( "CREATE TABLE t1 (a int)\nSELECT a FROM t1", 'pg_stat_statements data kept across restart'); +is( $node->safe_psql( + 'postgres', + "SELECT calls FROM pg_stat_statements WHERE query = 'SELECT a FROM t1'" + ), + '1', + 'pg_stat_statements counters kept across restart'); +is($node->safe_psql('postgres', $stats_reset_query), + $stats_reset, 'pg_stat_statements_info kept across restart'); + $node->append_conf('postgresql.conf', "pg_stat_statements.save = false"); $node->reload; @@ -48,6 +64,68 @@ is( $node->safe_psql( '0', 'pg_stat_statements data not kept across restart with .save=false'); +# Entries are not restored if pg_stat_statements.save is off at startup +# only, and neither are global statistics. +$node->append_conf('postgresql.conf', "pg_stat_statements.save = true"); +$node->reload; +$node->safe_psql('postgres', 'SELECT a FROM t1'); +$stats_reset = $node->safe_psql('postgres', $stats_reset_query); +$node->stop; +$node->append_conf('postgresql.conf', "pg_stat_statements.save = false"); +$node->start; + +is($node->safe_psql('postgres', $count_query), + '0', 'pg_stat_statements data not restored with .save=false at startup'); +isnt($node->safe_psql('postgres', $stats_reset_query), + $stats_reset, + 'pg_stat_statements_info not restored with .save=false at startup'); + +# Entries are saved without their texts if pg_stat_statements.save is off +# at shutdown, and are not restored then. +$node->safe_psql('postgres', 'SELECT a FROM t1'); +$node->stop; +$node->append_conf('postgresql.conf', "pg_stat_statements.save = true"); +$node->start; + +is($node->safe_psql('postgres', $count_query), + '0', + 'pg_stat_statements data not restored with .save=false at shutdown'); +is( $node->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_stat_statements WHERE query IS NULL"), + '0', + 'pg_stat_statements entries without texts not restored'); + +# Statistics are discarded after a crash. +$node->safe_psql('postgres', 'SELECT a FROM t1'); +$stats_reset = $node->safe_psql('postgres', $stats_reset_query); +$node->stop('immediate'); +$node->start; + +is($node->safe_psql('postgres', $count_query), + '0', 'pg_stat_statements data discarded after a crash'); +isnt($node->safe_psql('postgres', $stats_reset_query), + $stats_reset, 'pg_stat_statements_info reset after a crash'); + +# Statistics are discarded with a corrupted statistics file, and new entries +# still work. +$node->safe_psql('postgres', 'SELECT a FROM t1'); +$node->stop; +my $statsfile = $node->data_dir . '/pg_stat/pgstat.stat'; +truncate($statsfile, int((-s $statsfile) / 2)) + or die "could not truncate \"$statsfile\": $!"; +$node->start; + +is($node->safe_psql('postgres', $count_query), + '0', 'pg_stat_statements data discarded with a corrupted stats file'); +$node->safe_psql('postgres', 'SELECT a FROM t1'); +is( $node->safe_psql( + 'postgres', + "SELECT query FROM pg_stat_statements WHERE query NOT LIKE '%pg_stat_statements%' ORDER BY query" + ), + 'SELECT a FROM t1', + 'pg_stat_statements works after a corrupted stats file'); + $node->stop; done_testing(); diff --git a/contrib/pg_stat_statements/t/011_eviction.pl b/contrib/pg_stat_statements/t/011_eviction.pl new file mode 100644 index 00000000000..4b7f4bfec79 --- /dev/null +++ b/contrib/pg_stat_statements/t/011_eviction.pl @@ -0,0 +1,100 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Tests for the eviction of pg_stat_statements entries, and for changes of +# pg_stat_statements.max with a reload and across restarts. + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $node = PostgreSQL::Test::Cluster->new('main'); +$node->init; +$node->append_conf( + 'postgresql.conf', qq{ +shared_preload_libraries = 'pg_stat_statements' +pg_stat_statements.max = 200 +pg_stat_statements.track_utility = off +}); +$node->start; + +$node->safe_psql('postgres', 'CREATE EXTENSION pg_stat_statements'); + +# Run "count" distinct statements, which differ by their number of columns, +# starting from "first". +sub run_distinct +{ + my ($first, $count) = @_; + my $sql = ''; + + for my $n ($first .. $first + $count - 1) + { + $sql .= 'SELECT ' . join(', ', (1) x $n) . ";\n"; + } + $node->safe_psql('postgres', $sql); +} + +my $count_query = "SELECT count(*) FROM pg_stat_statements"; + +# A backend running a statement again in the same transaction, after its +# entry was removed by another backend, finds its reference to the removed +# entry stale, discards its pending statistics and counts in a new entry. +my $session = $node->background_psql('postgres'); +$session->query_safe("BEGIN"); +$session->query_safe("SELECT count(*) AS stale FROM pg_class"); +$node->safe_psql('postgres', "SELECT pg_stat_statements_reset()"); +$session->query_safe("SELECT count(*) AS stale FROM pg_class"); +$session->query_safe("COMMIT"); +$session->query_safe("SELECT pg_stat_force_next_flush()"); +$session->quit; +is( $node->safe_psql( + 'postgres', + "SELECT calls, query <> '' FROM pg_stat_statements WHERE query LIKE '%AS stale FROM pg_class'" + ), + '1|t', + 'statement counted in a new entry after its entry was removed'); + +# A statement executed often, which eviction keeps. +$node->safe_psql('postgres', + "SELECT 'hot' AS hot;\n" x 50 . "SELECT pg_stat_force_next_flush();"); + +run_distinct(1, 400); + +my $count = $node->safe_psql('postgres', $count_query); +cmp_ok($count, '<=', 200, 'number of entries bounded by .max'); +cmp_ok($count, '>=', 150, 'eviction keeps most entries'); +cmp_ok($node->safe_psql('postgres', + "SELECT dealloc FROM pg_stat_statements_info"), + '>', 0, 'deallocations counted'); +is( $node->safe_psql( + 'postgres', + "SELECT calls FROM pg_stat_statements WHERE query LIKE '%AS hot'"), + '50', + 'frequently executed statement kept'); + +# Lowering .max with a reload applies at the next new entry, and trims to +# the new maximum, not further. +$node->append_conf('postgresql.conf', 'pg_stat_statements.max = 100'); +$node->reload; +run_distinct(401, 1); + +$count = $node->safe_psql('postgres', $count_query); +cmp_ok($count, '<=', 100, 'number of entries bounded by lowered .max'); +cmp_ok($count, '>=', 80, 'lowering .max keeps most entries'); + +# Lowering .max across a restart trims the restored entries. +$node->append_conf('postgresql.conf', 'pg_stat_statements.max = 200'); +$node->reload; +run_distinct(500, 200); +$node->stop; +$node->append_conf('postgresql.conf', 'pg_stat_statements.max = 100'); +$node->start; + +$count = $node->safe_psql('postgres', $count_query); +cmp_ok($count, '<=', 100, 'restored entries trimmed to lowered .max'); +cmp_ok($count, '>=', 80, 'trimming restored entries keeps most of them'); + +$node->stop; + +done_testing(); diff --git a/doc/src/sgml/pgstatstatements.sgml b/doc/src/sgml/pgstatstatements.sgml index d753de5836e..1e5ea18c301 100644 --- a/doc/src/sgml/pgstatstatements.sgml +++ b/doc/src/sgml/pgstatstatements.sgml @@ -18,6 +18,10 @@ in postgresql.conf, because it requires additional shared memory. This means that a server restart is needed to add or remove the module. + Statement statistics are saved in the cumulative statistics file + (see ). If the module is removed from + shared_preload_libraries, all cumulative statistics, + not only those of the module, are discarded at the next server start. In addition, query identifier calculation must be enabled in order for the module to be active, which is done automatically if is set to auto or on, or any third-party @@ -748,6 +752,28 @@ calls | 2 recurrences. + + The statistics of an execution are kept in the local memory of its + session, and added to its statement's entry when the session next reports + cumulative statistics, normally within a second once it is idle outside a + transaction block (see ). Until + then, a statement whose executions are all still in local memory does + not appear in the view. A session reading the + pg_stat_statements view first adds its own + such statistics. + + + + The memory of a discarded statement is released once the sessions that + executed it have reported their statistics. When statements are + discarded so fast that as many wait for that as there can be tracked + statements, new statements are not tracked until the memory is released, + normally within a second; statements already tracked are still counted. + This keeps memory use bounded by twice + pg_stat_statements.max under any rate of new + statements, at the cost of not tracking some of them. + + plans and calls aren't always expected to match because planning and execution statistics are @@ -914,8 +940,21 @@ calls | 2 statements is discarded. The number of times such information was discarded can be seen in the pg_stat_statements_info view. + The number of statements can exceed this value by up to 5% while such + information is being discarded; past that, a session tracking a new + statement waits for the discarding, with the wait event + pg_stat_statements_dealloc of type + LWLock. The + entry_count column of the + pg_stat_kind_info + view shows the number of statements allocated in shared memory, + including discarded statements whose memory has not been released yet, + which is at most twice this value. The default value is 5000. - This parameter can only be set at server start. + This parameter can only be set in the postgresql.conf + file or on the server command line. A lowered value takes effect when + the next new statement is tracked, and at server start for the + statements saved at shutdown. @@ -996,8 +1035,13 @@ calls | 2 pg_stat_statements.save specifies whether to save statement statistics across server shutdowns. - If it is off then statistics are not saved at - shutdown nor reloaded at server start. + If it is off at shutdown, statements are saved + without their query texts, and are not reloaded at the next server + start. If it is off at server start, saved + statements are not reloaded, and the statistics of + pg_stat_statements_info are reset. + Statistics are not saved after a crash or an immediate shutdown, + like other cumulative statistics. The default value is on. This parameter can only be set in the postgresql.conf file or on the server command line. @@ -1007,10 +1051,8 @@ calls | 2 - The module requires additional shared memory proportional to - pg_stat_statements.max. Note that this - memory is consumed whenever the module is loaded, even if - pg_stat_statements.track is set to none. + The module stores statement statistics in dynamic shared memory allocated + as statements are tracked, proportional to their number. diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 8ca94ef2384..bb00d2a90a4 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -4126,10 +4126,13 @@ pgsa_stash_name_table_hash pgsa_writer_context pgsocket pgsql_thing_t +pgssDeallocEntry pgssEntry pgssGlobalStats pgssHashKey +pgssPendingEntry pgssQueryState +pgssSharedEntry pgssSharedState pgssStoreKind pgssVersion -- 2.50.1