From 9797f3b6b4cc0e184c17182ddfc1ff5cba4e5c42 Mon Sep 17 00:00:00 2001 From: Sami Imseih Date: Thu, 12 Nov 2026 20:19:26 +0000 Subject: [PATCH v19 1/1] pgstat: Allow pg_stat_force_next_flush() to work in transaction Previously pg_stat_force_next_flush() only guaranteed that the next flush would happen after the transaction ended. Make it flush immediately when called during a transaction. Nontransactional counters flush right away because they describe completed work. Transactional counters flush once the (sub)transaction that made them has ended, so the running transaction's own counts are never published early. HOT and new-page update counts are kept in the per-(sub)transaction state with the update count, so a flush never shows them ahead of n_tup_upd. To support repeated flushes in one transaction, keep flushed baselines for relation, index, and function stats. Flush callbacks now receive an xact_boundary flag, and flush_pending_cb returns a PgStat_FlushResult instead of a bool. Update the custom stats test module to show one way to split transactional and nontransactional counters in a custom kind. Document pg_stat_force_next_flush() and add regression tests for repeated flushes, subtransactions, truncation, procedures that commit, and recursive calls from a function. While working in stats.sql, fixed a few cases where a CREATE table was missing a corresponding DROP. XXX catversion bump needed. Author: Sami Imseih Suggested-by: Kyotaro Horiguchi Reviewed-by: Bertrand Drouvot Reviewed-by: Michael Paquier Discussion: https://postgr.es/m/CAA5RZ0uA-4qcD3+2hjcE_-zQUBhvWf5foPM2vzYneFKrJLsBDQ@mail.gmail.com --- doc/src/sgml/monitoring.sgml | 24 +- src/backend/utils/activity/pgstat.c | 193 +++++-- src/backend/utils/activity/pgstat_backend.c | 2 +- src/backend/utils/activity/pgstat_database.c | 13 +- src/backend/utils/activity/pgstat_function.c | 48 +- src/backend/utils/activity/pgstat_index.c | 57 +- src/backend/utils/activity/pgstat_io.c | 5 +- src/backend/utils/activity/pgstat_lock.c | 5 +- src/backend/utils/activity/pgstat_relation.c | 179 ++++-- src/backend/utils/activity/pgstat_shmem.c | 4 +- src/backend/utils/activity/pgstat_slru.c | 2 +- .../utils/activity/pgstat_subscription.c | 18 +- src/backend/utils/activity/pgstat_wal.c | 5 +- src/backend/utils/adt/pgstatfuncs.c | 7 +- src/include/catalog/pg_proc.dat | 4 +- src/include/pgstat.h | 63 +- src/include/utils/pgstat_internal.h | 86 ++- .../test_custom_stats/t/001_custom_stats.pl | 68 ++- .../test_custom_var_stats--1.0.sql | 8 + .../test_custom_stats/test_custom_var_stats.c | 165 +++++- src/test/regress/expected/stats.out | 541 +++++++++++++++++- src/test/regress/sql/stats.sql | 353 +++++++++++- src/tools/pgindent/typedefs.list | 2 + 23 files changed, 1603 insertions(+), 249 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 445fbc9232c..5b5a0c0774f 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -239,7 +239,9 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser more frequently than once per PGSTAT_MIN_INTERVAL milliseconds (1 second unless altered while building the server); so a query or transaction still in progress does not affect the displayed totals - and the displayed information lags behind actual activity. However, + and the displayed information lags behind actual activity, unless the process + is asked to flush its pending statistics by calling + pg_stat_force_next_flush. However, current-query information collected by track_activities is always up-to-date. @@ -4555,7 +4557,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage The time of the last sequential scan on this table, based on the - most recent transaction stop time + most recent transaction stop time, or the statement start time + when flushed during a transaction @@ -4583,7 +4586,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage The time of the last index scan on this table, based on the - most recent transaction stop time + most recent transaction stop time, or the statement start time + when flushed during a transaction @@ -5084,7 +5088,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage The time of the last scan on this index, based on the - most recent transaction stop time + most recent transaction stop time, or the statement start time + when flushed during a transaction @@ -5965,11 +5970,12 @@ description | Waiting for a newly initialized WAL file to reach durable storage Forces the statistics accumulated by the current session to be - flushed to shared memory when the current transaction ends, - without waiting for the - PGSTAT_MIN_INTERVAL delay between flushes. This - is mainly useful in tests that need this flush to happen without - that delay. + flushed without waiting for the PGSTAT_MIN_INTERVAL + delay between flushes. When called within a transaction, counters + that should only be updated at the end of the transaction, such as + tuple inserts, updates, and deletes, are deferred until the + transaction ends. This is useful in tests, and to make the + statistics of a long-running transaction visible before it ends. diff --git a/src/backend/utils/activity/pgstat.c b/src/backend/utils/activity/pgstat.c index d31c0b48a1d..05a38bb87d2 100644 --- a/src/backend/utils/activity/pgstat.c +++ b/src/backend/utils/activity/pgstat.c @@ -249,6 +249,15 @@ static dlist_head pgStatPending = DLIST_STATIC_INIT(pgStatPending); */ static bool pgStatForceNextFlush = false; +/* Set while pgstat_report_stat() flushes pending entries. */ +static bool pgStatFlushInProgress = false; + +/* Current pending entry callback executing, or NULL when none is active. */ +static PgStat_EntryRef *pgStatCurrentFlushEntry = NULL; + +/* Generation number of the current pgstat_flush_pending_entries() pass. */ +static uint64 pgStatFlushGeneration = 0; + /* * Force-clear existing snapshot before next use when stats_fetch_consistency * is changed. @@ -341,7 +350,7 @@ static const PgStat_KindInfo pgstat_kind_builtin_infos[PGSTAT_KIND_BUILTIN_SIZE] .shared_size = sizeof(PgStatShared_Function), .shared_data_off = offsetof(PgStatShared_Function, stats), .shared_data_len = sizeof(((PgStatShared_Function *) 0)->stats), - .pending_size = sizeof(PgStat_FunctionCounts), + .pending_size = sizeof(PgStat_FunctionStatus), .flush_pending_cb = pgstat_function_flush_cb, .reset_timestamp_cb = pgstat_function_reset_timestamp_cb, @@ -639,36 +648,53 @@ pgstat_before_server_shutdown(int code, Datum arg) /* * Shut down a single backend's statistics reporting at process exit. * - * Flush out any remaining statistics counts. Without this, operations - * triggered during backend exit (such as temp table deletions) won't be - * counted. + * Flush out any remaining statistics counts when possible. Without this, + * operations triggered during backend exit (such as temp table deletions) + * won't be counted. */ static void pgstat_shutdown_hook(int code, Datum arg) { + bool discard_pending = false; + Assert(!pgstat_is_shutdown); Assert(IsUnderPostmaster || !IsPostmasterEnvironment); /* - * If we got as far as discovering our own database ID, we can flush out - * what we did so far. Otherwise, we'd be reporting an invalid database - * ID, so forget it. (This means that accesses to pg_database during - * failed backend starts might never get counted.) + * Don't recursively flush stats if process exit is reached while a flush + * is already in progress. Discarding pending stats during detach is + * preferable to reentering pgstat_report_stat(). */ - if (OidIsValid(MyDatabaseId)) - pgstat_report_disconnect(MyDatabaseId); + if (pgStatFlushInProgress) + { + pgStatCurrentFlushEntry = NULL; + pgStatFlushInProgress = false; + discard_pending = true; + } - pgstat_report_stat(true); + if (!discard_pending) + { + /* + * If we got as far as discovering our own database ID, we can flush + * out what we did so far. Otherwise, we'd be reporting an invalid + * database ID, so forget it. (This means that accesses to + * pg_database during failed backend starts might never get counted.) + */ + if (OidIsValid(MyDatabaseId)) + pgstat_report_disconnect(MyDatabaseId); + + pgstat_report_stat(true); - /* there shouldn't be any pending changes left */ - Assert(dlist_is_empty(&pgStatPending)); - dlist_init(&pgStatPending); + /* there shouldn't be any pending changes left */ + Assert(dlist_is_empty(&pgStatPending)); + dlist_init(&pgStatPending); + } /* drop the backend stats entry */ if (!pgstat_drop_entry(PGSTAT_KIND_BACKEND, InvalidOid, MyProcNumber, false)) pgstat_request_entry_refs_gc(); - pgstat_detach_shmem(); + pgstat_detach_shmem(discard_pending); #ifdef USE_ASSERT_CHECKING pgstat_is_shutdown = true; @@ -731,8 +757,9 @@ pgstat_initialize(void) * a timeout after which to call pgstat_report_stat(true), but are not * required to do so. * - * Note that this is called only when not within a transaction, so it is fair - * to use transaction stop time as an approximation of current time. + * A nonforced flush only happens outside a transaction, so transaction stop + * time is close enough for the flush interval checks. A forced flush can + * happen during a transaction and uses the current time instead. */ long pgstat_report_stat(bool force) @@ -744,7 +771,11 @@ pgstat_report_stat(bool force) bool nowait; pgstat_assert_is_up(); - Assert(!IsTransactionOrTransactionBlock()); + Assert(force || !IsTransactionOrTransactionBlock()); + + /* a prior flush must have cleared this, even if it errored out */ + Assert(!pgStatFlushInProgress); + Assert(pgStatCurrentFlushEntry == NULL); /* "absorb" the forced flush even if there's nothing to flush */ if (pgStatForceNextFlush) @@ -807,35 +838,45 @@ pgstat_report_stat(bool force) partial_flush = false; - /* flush of variable-numbered stats tracked in pending entries list */ - partial_flush |= pgstat_flush_pending_entries(nowait); + pgStatFlushInProgress = true; - /* flush of other stats kinds */ - if (pgstat_report_fixed) + /* Clear flush-pass state before leaving this flush. */ + PG_TRY(); { - for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++) + /* flush of variable-numbered stats tracked in pending entries list */ + partial_flush |= pgstat_flush_pending_entries(nowait); + + /* flush of other stats kinds */ + if (pgstat_report_fixed) { - const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind); + for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++) + { + const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind); - if (!kind_info) - continue; - if (!kind_info->flush_static_cb) - continue; + if (!kind_info) + continue; + if (!kind_info->flush_static_cb) + continue; - partial_flush |= kind_info->flush_static_cb(nowait); + partial_flush |= kind_info->flush_static_cb(nowait, + !IsTransactionOrTransactionBlock()); + } } } + PG_FINALLY(); + { + pgStatCurrentFlushEntry = NULL; + pgStatFlushInProgress = false; + } + PG_END_TRY(); last_flush = now; - /* - * If some of the pending stats could not be flushed due to lock - * contention, let the caller know when to retry. - */ + /* Let the caller know when to retry after a partial flush. */ if (partial_flush) { - /* force should have prevented us from getting here */ - Assert(!force); + /* with force, only active transaction state can cause a partial flush */ + Assert(!force || IsTransactionOrTransactionBlock()); /* remember since when stats have been pending */ if (pending_since == 0) @@ -851,12 +892,24 @@ pgstat_report_stat(bool force) } /* - * Force locally pending stats to be flushed during the next - * pgstat_report_stat() call. This is useful for writing tests. + * Force locally pending stats to be flushed. During a transaction this + * flushes immediately, and the next pgstat_report_stat() call is forced too. + * This is useful in tests, and for callers such as extensions that need + * statistics of a running transaction visible before it ends. */ void pgstat_force_next_flush(void) { + /* + * A flush callback must not call this; doing so would recursively enter + * pgstat_report_stat() while a flush is already in progress. + */ + Assert(!pgStatFlushInProgress); + + /* During a transaction flush now unless a flush is already running. */ + if (!pgStatFlushInProgress && IsTransactionOrTransactionBlock()) + pgstat_report_stat(true); + pgStatForceNextFlush = true; } @@ -1374,6 +1427,7 @@ pgstat_delete_pending_entry(PgStat_EntryRef *entry_ref) pfree(pending_data); entry_ref->pending = NULL; + entry_ref->flushed_generation = 0; dlist_delete(&entry_ref->pending_node); } @@ -1387,6 +1441,8 @@ pgstat_prep_pending_from_entry_ref(PgStat_EntryRef *entry_ref) PgStat_Kind kind; Assert(entry_ref != NULL); + /* The entry currently being flushed must not be made pending again. */ + Assert(entry_ref != pgStatCurrentFlushEntry); kind = entry_ref->shared_entry->key.kind; @@ -1408,8 +1464,19 @@ pgstat_prep_pending_from_entry_ref(PgStat_EntryRef *entry_ref) } entry_ref->pending = MemoryContextAllocZero(pgStatPendingContext, entrysize); + entry_ref->flushed_generation = 0; dlist_push_tail(&pgStatPending, &entry_ref->pending_node); } + else if (pgStatFlushInProgress && + entry_ref->flushed_generation == pgStatFlushGeneration) + { + /* + * The entry is already pending and was already visited in the current + * flush pass. Move it to the tail so the data just accumulated into + * it is flushed again before the pass ends. + */ + dlist_move_tail(&pgStatPending, &entry_ref->pending_node); + } } /* @@ -1426,12 +1493,19 @@ pgstat_flush_pending_entries(bool nowait) * Processing a pending entry may queue further pending entries to the end * of the list that we want to process, so a simple iteration won't do. * Further complicating matters is that we want to delete the current - * entry in each iteration from the list if we flushed successfully. + * entry in each iteration from the list if we flushed successfully, + * though during a transaction a fully flushed entry is retained and only + * deleted at a transaction boundary. * - * So we just keep track of the next pointer in each loop iteration. + * So we just keep track of the next pointer in each loop iteration, and + * start a new flush-pass generation only when there are pending entries. */ + if (!dlist_is_empty(&pgStatPending)) + { cur = dlist_head_node(&pgStatPending); + pgStatFlushGeneration++; + } while (cur) { @@ -1440,25 +1514,54 @@ pgstat_flush_pending_entries(bool nowait) PgStat_HashKey key = entry_ref->shared_entry->key; PgStat_Kind kind = key.kind; const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind); - bool did_flush; + PgStat_FlushResult result; + bool xact_boundary; dlist_node *next; + xact_boundary = !IsTransactionOrTransactionBlock(); + Assert(!kind_info->fixed_amount); Assert(kind_info->flush_pending_cb != NULL); /* flush the stats, if possible */ - did_flush = kind_info->flush_pending_cb(entry_ref, nowait); + pgStatCurrentFlushEntry = entry_ref; + result = kind_info->flush_pending_cb(entry_ref, nowait, xact_boundary); + pgStatCurrentFlushEntry = NULL; - Assert(did_flush || nowait); + /* + * A lock conflict can only happen when we allowed the callback to + * give up without waiting for a lock, and a partial flush can only + * happen inside a transaction. + */ + Assert(result == PGSTAT_FLUSH_DONE || + (result == PGSTAT_FLUSH_LOCK_CONFLICT && nowait) || + (result == PGSTAT_FLUSH_PARTIAL && !xact_boundary)); - /* determine next entry, before deleting the pending entry */ + /* + * Mark the entry visited, unless we could not flush it at all. If a + * later callback adds more data to this entry in the same pass, + * pgstat_prep_pending_from_entry_ref() then re-queues it to the tail + * so the new data is flushed again before the pass ends. + */ + if (result != PGSTAT_FLUSH_LOCK_CONFLICT) + entry_ref->flushed_generation = pgStatFlushGeneration; + + /* + * Determine the next entry after the callback ran (it may have + * re-queued an entry) and before possibly deleting the current one. + */ if (dlist_has_next(&pgStatPending, cur)) next = dlist_next_node(&pgStatPending, cur); else next = NULL; - /* if successfully flushed, remove entry */ - if (did_flush) + /* + * Only delete a fully flushed entry at a transaction boundary. + * Earlier in the transaction it may still be reused as more stats + * accumulate, and code holding a pointer into it relies on the entry + * staying put. + */ + if (result == PGSTAT_FLUSH_DONE && xact_boundary) pgstat_delete_pending_entry(entry_ref); else have_pending = true; diff --git a/src/backend/utils/activity/pgstat_backend.c b/src/backend/utils/activity/pgstat_backend.c index b736b2ccc6f..c071dddf296 100644 --- a/src/backend/utils/activity/pgstat_backend.c +++ b/src/backend/utils/activity/pgstat_backend.c @@ -383,7 +383,7 @@ pgstat_flush_backend(bool nowait, uint32 flags) * If some stats could not be flushed due to lock contention, return true. */ bool -pgstat_backend_flush_cb(bool nowait) +pgstat_backend_flush_cb(bool nowait, bool xact_boundary) { return pgstat_flush_backend(nowait, PGSTAT_BACKEND_FLUSH_ALL); } diff --git a/src/backend/utils/activity/pgstat_database.c b/src/backend/utils/activity/pgstat_database.c index 7f3bc016593..54b17f148d7 100644 --- a/src/backend/utils/activity/pgstat_database.c +++ b/src/backend/utils/activity/pgstat_database.c @@ -432,10 +432,13 @@ pgstat_reset_database_timestamp(Oid dboid, TimestampTz ts) * Flush out pending stats for the entry * * If nowait is true and the lock could not be immediately acquired, returns - * false without flushing the entry. Otherwise returns true. + * PGSTAT_FLUSH_LOCK_CONFLICT without flushing the entry. Database stats are + * not transactional, so xact_boundary is unused and this always returns + * PGSTAT_FLUSH_DONE once flushed. */ -bool -pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) +PgStat_FlushResult +pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, + bool xact_boundary) { PgStatShared_Database *sharedent; PgStat_StatDBEntry *pendingent; @@ -444,7 +447,7 @@ pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) sharedent = (PgStatShared_Database *) entry_ref->shared_stats; if (!pgstat_lock_entry(entry_ref, nowait)) - return false; + return PGSTAT_FLUSH_LOCK_CONFLICT; #define PGSTAT_ACCUM_DBCOUNT(item) \ (sharedent)->stats.item += (pendingent)->item @@ -496,7 +499,7 @@ pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) memset(pendingent, 0, sizeof(*pendingent)); - return true; + return PGSTAT_FLUSH_DONE; } void diff --git a/src/backend/utils/activity/pgstat_function.c b/src/backend/utils/activity/pgstat_function.c index f0366e13990..143a92f5f2a 100644 --- a/src/backend/utils/activity/pgstat_function.c +++ b/src/backend/utils/activity/pgstat_function.c @@ -73,7 +73,7 @@ pgstat_init_function_usage(FunctionCallInfo fcinfo, PgStat_FunctionCallUsage *fcu) { PgStat_EntryRef *entry_ref; - PgStat_FunctionCounts *pending; + PgStat_FunctionStatus *pending; bool created_entry; if (pgstat_track_functions <= fcinfo->flinfo->fn_stats) @@ -121,10 +121,10 @@ pgstat_init_function_usage(FunctionCallInfo fcinfo, pending = entry_ref->pending; - fcu->fs = pending; + fcu->fs = &pending->counts; /* save stats for this function, later used to compensate for recursion */ - fcu->save_f_total_time = pending->total_time; + fcu->save_f_total_time = pending->counts.total_time; /* save current backend-wide total time */ fcu->save_total = total_func_time; @@ -184,34 +184,48 @@ pgstat_end_function_usage(PgStat_FunctionCallUsage *fcu, bool finalize) } /* - * Flush out pending stats for the entry + * Flush pending stats for the entry. * - * If nowait is true and the lock could not be immediately acquired, returns - * false without flushing the entry. Otherwise returns true. + * If nowait is true and the lock could not be acquired, return + * PGSTAT_FLUSH_LOCK_CONFLICT. + * + * Function stats are not transactional, so this always returns + * PGSTAT_FLUSH_DONE. The entry can flush more than once per transaction. */ -bool -pgstat_function_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) +PgStat_FlushResult +pgstat_function_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, + bool xact_boundary) { - PgStat_FunctionCounts *localent; + PgStat_FunctionStatus *localent; PgStatShared_Function *shfuncent; - localent = (PgStat_FunctionCounts *) entry_ref->pending; + localent = (PgStat_FunctionStatus *) entry_ref->pending; shfuncent = (PgStatShared_Function *) entry_ref->shared_stats; - /* localent always has non-zero content */ + /* Nothing changed since the last flush. */ + if (memcmp(&localent->counts, &localent->flushed, + sizeof(PgStat_FunctionCounts)) == 0) + return PGSTAT_FLUSH_DONE; if (!pgstat_lock_entry(entry_ref, nowait)) - return false; + return PGSTAT_FLUSH_LOCK_CONFLICT; - shfuncent->stats.numcalls += localent->numcalls; + shfuncent->stats.numcalls += + localent->counts.numcalls - localent->flushed.numcalls; shfuncent->stats.total_time += - INSTR_TIME_GET_MICROSEC(localent->total_time); + INSTR_TIME_GET_MICROSEC(localent->counts.total_time) - + INSTR_TIME_GET_MICROSEC(localent->flushed.total_time); shfuncent->stats.self_time += - INSTR_TIME_GET_MICROSEC(localent->self_time); + INSTR_TIME_GET_MICROSEC(localent->counts.self_time) - + INSTR_TIME_GET_MICROSEC(localent->flushed.self_time); pgstat_unlock_entry(entry_ref); - return true; + /* Record the new baseline while counts keep accumulating. */ + if (!xact_boundary) + localent->flushed = localent->counts; + + return PGSTAT_FLUSH_DONE; } void @@ -233,7 +247,7 @@ find_funcstat_entry(Oid func_id) entry_ref = pgstat_fetch_pending_entry(PGSTAT_KIND_FUNCTION, MyDatabaseId, func_id); if (entry_ref) - return entry_ref->pending; + return &((PgStat_FunctionStatus *) entry_ref->pending)->counts; return NULL; } diff --git a/src/backend/utils/activity/pgstat_index.c b/src/backend/utils/activity/pgstat_index.c index a1f9a4c6ac1..8e9584c03e5 100644 --- a/src/backend/utils/activity/pgstat_index.c +++ b/src/backend/utils/activity/pgstat_index.c @@ -23,16 +23,14 @@ /* - * Flush out pending stats for an index entry. + * Flush pending stats for an index entry. * - * If nowait is true and the lock could not be immediately acquired, returns - * false without flushing the entry. Otherwise returns true. - * - * Some of the stats are copied to the corresponding pending database stats - * entry when successfully flushing. + * If nowait is true and the lock could not be acquired, return + * PGSTAT_FLUSH_LOCK_CONFLICT. Index stats are not transactional, so a flush + * always completes. */ -bool -pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) +PgStat_FlushResult +pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary) { Oid dboid; PgStat_RelationStatus *lstats; /* pending stats entry */ @@ -44,43 +42,46 @@ pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) lstats = (PgStat_RelationStatus *) entry_ref->pending; shidxstats = (PgStatShared_Index *) entry_ref->shared_stats; - /* - * Ignore entries that didn't accumulate any actual counts, such as - * indexes that were opened by the planner but not used. - */ - if (pg_memory_is_all_zeros(&lstats->idx, - sizeof(struct PgStat_IndexCounts))) - return true; + /* Skip entries with nothing new to flush. */ + if (memcmp(&lstats->idx.counts, &lstats->idx.flushed, + sizeof(struct PgStat_IndexCounts)) == 0) + return PGSTAT_FLUSH_DONE; if (!pgstat_lock_entry(entry_ref, nowait)) - return false; + return PGSTAT_FLUSH_LOCK_CONFLICT; /* Add the values to the shared entry. */ idxentry = &shidxstats->stats; - idxentry->numscans += lstats->idx.numscans; - if (lstats->idx.numscans) + idxentry->numscans += lstats->idx.counts.numscans - lstats->idx.flushed.numscans; + if (lstats->idx.counts.numscans > lstats->idx.flushed.numscans) { - TimestampTz t = GetCurrentTransactionStopTimestamp(); + TimestampTz t = xact_boundary ? + GetCurrentTransactionStopTimestamp() : + GetCurrentStatementStartTimestamp(); if (t > idxentry->lastscan) idxentry->lastscan = t; } - idxentry->tuples_returned += lstats->idx.tuples_returned; - idxentry->tuples_fetched += lstats->idx.tuples_fetched; - idxentry->blocks_fetched += lstats->idx.blocks_fetched; - idxentry->blocks_hit += lstats->idx.blocks_hit; + idxentry->tuples_returned += lstats->idx.counts.tuples_returned - lstats->idx.flushed.tuples_returned; + idxentry->tuples_fetched += lstats->idx.counts.tuples_fetched - lstats->idx.flushed.tuples_fetched; + idxentry->blocks_fetched += lstats->idx.counts.blocks_fetched - lstats->idx.flushed.blocks_fetched; + idxentry->blocks_hit += lstats->idx.counts.blocks_hit - lstats->idx.flushed.blocks_hit; pgstat_unlock_entry(entry_ref); /* The entry was successfully flushed, add the same to database stats */ dbentry = pgstat_prep_database_pending(dboid); - dbentry->tuples_returned += lstats->idx.tuples_returned; - dbentry->tuples_fetched += lstats->idx.tuples_fetched; - dbentry->blocks_fetched += lstats->idx.blocks_fetched; - dbentry->blocks_hit += lstats->idx.blocks_hit; + dbentry->tuples_returned += lstats->idx.counts.tuples_returned - lstats->idx.flushed.tuples_returned; + dbentry->tuples_fetched += lstats->idx.counts.tuples_fetched - lstats->idx.flushed.tuples_fetched; + dbentry->blocks_fetched += lstats->idx.counts.blocks_fetched - lstats->idx.flushed.blocks_fetched; + dbentry->blocks_hit += lstats->idx.counts.blocks_hit - lstats->idx.flushed.blocks_hit; + + /* Record the new baseline while counts keep accumulating. */ + if (!xact_boundary) + lstats->idx.flushed = lstats->idx.counts; - return true; + return PGSTAT_FLUSH_DONE; } /* diff --git a/src/backend/utils/activity/pgstat_io.c b/src/backend/utils/activity/pgstat_io.c index 3c8fcb20aba..2561c82633a 100644 --- a/src/backend/utils/activity/pgstat_io.c +++ b/src/backend/utils/activity/pgstat_io.c @@ -16,6 +16,7 @@ #include "postgres.h" +#include "access/xact.h" #include "executor/instrument.h" #include "storage/bufmgr.h" #include "utils/pgstat_internal.h" @@ -166,7 +167,7 @@ pgstat_fetch_stat_io(void) void pgstat_flush_io(bool nowait) { - (void) pgstat_io_flush_cb(nowait); + (void) pgstat_io_flush_cb(nowait, !IsTransactionOrTransactionBlock()); } /* @@ -178,7 +179,7 @@ pgstat_flush_io(bool nowait) * acquired. Otherwise, return false. */ bool -pgstat_io_flush_cb(bool nowait) +pgstat_io_flush_cb(bool nowait, bool xact_boundary) { LWLock *bktype_lock; PgStat_BktypeIO *bktype_shstats; diff --git a/src/backend/utils/activity/pgstat_lock.c b/src/backend/utils/activity/pgstat_lock.c index c20c7599683..02f5520ed1b 100644 --- a/src/backend/utils/activity/pgstat_lock.c +++ b/src/backend/utils/activity/pgstat_lock.c @@ -17,6 +17,7 @@ #include "postgres.h" +#include "access/xact.h" #include "utils/pgstat_internal.h" static PgStat_PendingLock PendingLockStats; @@ -36,7 +37,7 @@ pgstat_fetch_stat_lock(void) void pgstat_lock_flush(bool nowait) { - (void) pgstat_lock_flush_cb(nowait); + (void) pgstat_lock_flush_cb(nowait, !IsTransactionOrTransactionBlock()); } /* @@ -48,7 +49,7 @@ pgstat_lock_flush(bool nowait) * acquired. Otherwise, return false. */ bool -pgstat_lock_flush_cb(bool nowait) +pgstat_lock_flush_cb(bool nowait, bool xact_boundary) { LWLock *lckstat_lock; PgStatShared_Lock *shstats; diff --git a/src/backend/utils/activity/pgstat_relation.c b/src/backend/utils/activity/pgstat_relation.c index 82861e88e0e..2f7a32bd026 100644 --- a/src/backend/utils/activity/pgstat_relation.c +++ b/src/backend/utils/activity/pgstat_relation.c @@ -372,8 +372,13 @@ pgstat_report_analyze(Relation rel, livetuples -= trans->tuples_inserted - trans->tuples_deleted; deadtuples -= trans->tuples_updated + trans->tuples_deleted; } - /* count stuff inserted by already-aborted subxacts, too */ - deadtuples -= rel->pgstat_info->tab.counts_xact.delta_dead_tuples; + + /* + * Count stuff inserted by already-aborted subxacts, too, but only the + * part not yet flushed to shared stats. + */ + deadtuples -= rel->pgstat_info->tab.counts_xact.delta_dead_tuples - + rel->pgstat_info->tab.flushed_xact.delta_dead_tuples; /* Since ANALYZE's counts are estimates, we could have underflowed */ livetuples = Max(livetuples, 0); deadtuples = Max(deadtuples, 0); @@ -459,13 +464,13 @@ pgstat_count_heap_update(Relation rel, bool hot, bool newpage) pgstat_info->tab.trans->tuples_updated++; /* - * tuples_hot_updated and tuples_newpage_updated counters are - * nontransactional, so just advance them + * Keep these with tuples_updated until the (sub)transaction ends, so + * a flush does not publish them ahead of it. */ if (hot) - pgstat_info->tab.counts_xact.tuples_hot_updated++; + pgstat_info->tab.trans->tuples_hot_updated++; else if (newpage) - pgstat_info->tab.counts_xact.tuples_newpage_updated++; + pgstat_info->tab.trans->tuples_newpage_updated++; } } @@ -610,6 +615,10 @@ find_relstat_entry_kind(PgStat_Kind kind, Oid rel_id) relstatus->tab.counts_xact.tuples_inserted += trans->tuples_inserted; relstatus->tab.counts_xact.tuples_updated += trans->tuples_updated; relstatus->tab.counts_xact.tuples_deleted += trans->tuples_deleted; + relstatus->tab.counts_xact.tuples_hot_updated += + trans->tuples_hot_updated; + relstatus->tab.counts_xact.tuples_newpage_updated += + trans->tuples_newpage_updated; } return relstatus; @@ -643,6 +652,10 @@ AtEOXact_PgStat_Relations(PgStat_SubXactStatus *xact_state, bool isCommit) relstat->tab.counts_xact.tuples_inserted += trans->tuples_inserted; relstat->tab.counts_xact.tuples_updated += trans->tuples_updated; relstat->tab.counts_xact.tuples_deleted += trans->tuples_deleted; + relstat->tab.counts_xact.tuples_hot_updated += + trans->tuples_hot_updated; + relstat->tab.counts_xact.tuples_newpage_updated += + trans->tuples_newpage_updated; if (isCommit) { relstat->tab.truncdropped = trans->truncdropped; @@ -715,6 +728,10 @@ AtEOSubXact_PgStat_Relations(PgStat_SubXactStatus *xact_state, bool isCommit, in trans->upper->tuples_updated += trans->tuples_updated; trans->upper->tuples_deleted += trans->tuples_deleted; } + /* truncate/drop doesn't reset these */ + trans->upper->tuples_hot_updated += trans->tuples_hot_updated; + trans->upper->tuples_newpage_updated += + trans->tuples_newpage_updated; relstat->tab.trans = trans->upper; pfree(trans); } @@ -749,6 +766,10 @@ AtEOSubXact_PgStat_Relations(PgStat_SubXactStatus *xact_state, bool isCommit, in relstat->tab.counts_xact.tuples_inserted += trans->tuples_inserted; relstat->tab.counts_xact.tuples_updated += trans->tuples_updated; relstat->tab.counts_xact.tuples_deleted += trans->tuples_deleted; + relstat->tab.counts_xact.tuples_hot_updated += + trans->tuples_hot_updated; + relstat->tab.counts_xact.tuples_newpage_updated += + trans->tuples_newpage_updated; /* inserted tuples are dead, deleted tuples are unaffected */ relstat->tab.counts_xact.delta_dead_tuples += trans->tuples_inserted + trans->tuples_updated; @@ -799,6 +820,9 @@ AtPrepare_PgStat_Relations(PgStat_SubXactStatus *xact_state) * dead tuple counts are preserved in the 2PC state file. * * Note: AtEOXact_PgStat_Relations is not called during PREPARE. + * + * HOT and new-page update counts are not part of the 2PC record, so they + * are reported by this backend like the other nontransactional counts. */ void PostPrepare_PgStat_Relations(PgStat_SubXactStatus *xact_state) @@ -810,6 +834,10 @@ PostPrepare_PgStat_Relations(PgStat_SubXactStatus *xact_state) PgStat_RelationStatus *relstat; relstat = trans->parent; + relstat->tab.counts_xact.tuples_hot_updated += + trans->tuples_hot_updated; + relstat->tab.counts_xact.tuples_newpage_updated += + trans->tuples_newpage_updated; relstat->tab.trans = NULL; } } @@ -880,16 +908,16 @@ pgstat_twophase_postabort(FullTransactionId fxid, uint16 info, } /* - * Flush out pending stats for the entry + * Flush pending stats for the entry. * - * If nowait is true and the lock could not be immediately acquired, returns - * false without flushing the entry. Otherwise returns true. + * If nowait is true and the lock could not be acquired, return + * PGSTAT_FLUSH_LOCK_CONFLICT. * - * Some of the stats are copied to the corresponding pending database stats - * entry when successfully flushing. + * Nontransactional and transactional counters hold only final values (see + * PgStat_TableCountsXact), so everything pending is flushed. */ -bool -pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) +PgStat_FlushResult +pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary) { Oid dboid; PgStat_RelationStatus *lstats; /* pending stats entry */ @@ -902,65 +930,81 @@ pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) shtabstats = (PgStatShared_Relation *) entry_ref->shared_stats; /* - * Ignore entries that didn't accumulate any actual counts. If the table - * was truncated, we still need to flush the entry to reset the live/dead - * counters and ins_since_vacuum even when no other counts were - * accumulated. + * Skip entries with nothing new since the last flush, unless a truncate + * must reset the live/dead counters. */ if (!lstats->tab.truncdropped && - pg_memory_is_all_zeros(&lstats->tab.counts, - sizeof(struct PgStat_TableCounts)) && - pg_memory_is_all_zeros(&lstats->tab.counts_xact, - sizeof(struct PgStat_TableCountsXact))) - return true; + memcmp(&lstats->tab.counts, &lstats->tab.flushed, + sizeof(PgStat_TableCounts)) == 0 && + memcmp(&lstats->tab.counts_xact, &lstats->tab.flushed_xact, + sizeof(PgStat_TableCountsXact)) == 0) + return PGSTAT_FLUSH_DONE; if (!pgstat_lock_entry(entry_ref, nowait)) - return false; + return PGSTAT_FLUSH_LOCK_CONFLICT; - /* add the values to the shared entry. */ + /* Flush nontransactional counters using deltas from the baseline. */ tabentry = &shtabstats->stats; - tabentry->numscans += lstats->tab.counts.numscans; - if (lstats->tab.counts.numscans) + tabentry->numscans += lstats->tab.counts.numscans - + lstats->tab.flushed.numscans; + if (lstats->tab.counts.numscans > lstats->tab.flushed.numscans) { - TimestampTz t = GetCurrentTransactionStopTimestamp(); + TimestampTz t = xact_boundary ? + GetCurrentTransactionStopTimestamp() : + GetCurrentStatementStartTimestamp(); if (t > tabentry->lastscan) tabentry->lastscan = t; } - tabentry->tuples_returned += lstats->tab.counts.tuples_returned; - tabentry->tuples_fetched += lstats->tab.counts.tuples_fetched; - tabentry->tuples_inserted += lstats->tab.counts_xact.tuples_inserted; - tabentry->tuples_updated += lstats->tab.counts_xact.tuples_updated; - tabentry->tuples_deleted += lstats->tab.counts_xact.tuples_deleted; - tabentry->tuples_hot_updated += lstats->tab.counts_xact.tuples_hot_updated; - tabentry->tuples_newpage_updated += lstats->tab.counts_xact.tuples_newpage_updated; + tabentry->tuples_returned += lstats->tab.counts.tuples_returned - + lstats->tab.flushed.tuples_returned; + tabentry->tuples_fetched += lstats->tab.counts.tuples_fetched - + lstats->tab.flushed.tuples_fetched; + tabentry->blocks_fetched += lstats->tab.counts.blocks_fetched - + lstats->tab.flushed.blocks_fetched; + tabentry->blocks_hit += lstats->tab.counts.blocks_hit - + lstats->tab.flushed.blocks_hit; + + /* Flush transactional counters using deltas from the baseline. */ + tabentry->tuples_inserted += lstats->tab.counts_xact.tuples_inserted - + lstats->tab.flushed_xact.tuples_inserted; + tabentry->tuples_updated += lstats->tab.counts_xact.tuples_updated - + lstats->tab.flushed_xact.tuples_updated; + tabentry->tuples_deleted += lstats->tab.counts_xact.tuples_deleted - + lstats->tab.flushed_xact.tuples_deleted; + tabentry->tuples_hot_updated += + lstats->tab.counts_xact.tuples_hot_updated - + lstats->tab.flushed_xact.tuples_hot_updated; + tabentry->tuples_newpage_updated += + lstats->tab.counts_xact.tuples_newpage_updated - + lstats->tab.flushed_xact.tuples_newpage_updated; /* - * If table was truncated/dropped, first reset the live/dead counters. + * If the table was truncated, first reset live and dead counters and + * ins_since_vacuum. Commit zeroed delta_live_tuples and + * delta_dead_tuples, so reset their flushed baselines too. */ if (lstats->tab.truncdropped) { tabentry->live_tuples = 0; tabentry->dead_tuples = 0; tabentry->ins_since_vacuum = 0; + lstats->tab.flushed_xact.delta_live_tuples = 0; + lstats->tab.flushed_xact.delta_dead_tuples = 0; } - tabentry->live_tuples += lstats->tab.counts_xact.delta_live_tuples; - tabentry->dead_tuples += lstats->tab.counts_xact.delta_dead_tuples; - tabentry->mod_since_analyze += lstats->tab.counts_xact.changed_tuples; - - /* - * Using tuples_inserted to update ins_since_vacuum does mean that we'll - * track aborted inserts too. This isn't ideal, but otherwise probably - * not worth adding an extra field for. It may just amount to autovacuums - * triggering for inserts more often than they maybe should, which is - * probably not going to be common enough to be too concerned about here. - */ - tabentry->ins_since_vacuum += lstats->tab.counts_xact.tuples_inserted; + tabentry->live_tuples += lstats->tab.counts_xact.delta_live_tuples - + lstats->tab.flushed_xact.delta_live_tuples; + tabentry->dead_tuples += lstats->tab.counts_xact.delta_dead_tuples - + lstats->tab.flushed_xact.delta_dead_tuples; + tabentry->mod_since_analyze += lstats->tab.counts_xact.changed_tuples - + lstats->tab.flushed_xact.changed_tuples; - tabentry->blocks_fetched += lstats->tab.counts.blocks_fetched; - tabentry->blocks_hit += lstats->tab.counts.blocks_hit; + /* This still counts aborted inserts in ins_since_vacuum. */ + tabentry->ins_since_vacuum += + lstats->tab.counts_xact.tuples_inserted - + lstats->tab.flushed_xact.tuples_inserted; /* Clamp live_tuples in case of negative delta_live_tuples */ tabentry->live_tuples = Max(tabentry->live_tuples, 0); @@ -971,15 +1015,34 @@ pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) /* The entry was successfully flushed, add the same to database stats */ dbentry = pgstat_prep_database_pending(dboid); - dbentry->tuples_returned += lstats->tab.counts.tuples_returned; - dbentry->tuples_fetched += lstats->tab.counts.tuples_fetched; - dbentry->tuples_inserted += lstats->tab.counts_xact.tuples_inserted; - dbentry->tuples_updated += lstats->tab.counts_xact.tuples_updated; - dbentry->tuples_deleted += lstats->tab.counts_xact.tuples_deleted; - dbentry->blocks_fetched += lstats->tab.counts.blocks_fetched; - dbentry->blocks_hit += lstats->tab.counts.blocks_hit; - - return true; + dbentry->tuples_returned += lstats->tab.counts.tuples_returned - + lstats->tab.flushed.tuples_returned; + dbentry->tuples_fetched += lstats->tab.counts.tuples_fetched - + lstats->tab.flushed.tuples_fetched; + dbentry->blocks_fetched += lstats->tab.counts.blocks_fetched - + lstats->tab.flushed.blocks_fetched; + dbentry->blocks_hit += lstats->tab.counts.blocks_hit - + lstats->tab.flushed.blocks_hit; + dbentry->tuples_inserted += lstats->tab.counts_xact.tuples_inserted - + lstats->tab.flushed_xact.tuples_inserted; + dbentry->tuples_updated += lstats->tab.counts_xact.tuples_updated - + lstats->tab.flushed_xact.tuples_updated; + dbentry->tuples_deleted += lstats->tab.counts_xact.tuples_deleted - + lstats->tab.flushed_xact.tuples_deleted; + + /* + * Record everything as flushed while the entry stays around to gather + * more counts. Clear truncdropped so the next truncate is seen as new. + * At a transaction boundary the entry is deleted. + */ + if (!xact_boundary) + { + lstats->tab.flushed = lstats->tab.counts; + lstats->tab.flushed_xact = lstats->tab.counts_xact; + lstats->tab.truncdropped = false; + } + + return PGSTAT_FLUSH_DONE; } void diff --git a/src/backend/utils/activity/pgstat_shmem.c b/src/backend/utils/activity/pgstat_shmem.c index 8db33ac2fdf..883b2f0b65f 100644 --- a/src/backend/utils/activity/pgstat_shmem.c +++ b/src/backend/utils/activity/pgstat_shmem.c @@ -361,12 +361,12 @@ pgstat_attach_shmem(void) } void -pgstat_detach_shmem(void) +pgstat_detach_shmem(bool discard_pending) { Assert(pgStatLocal.num_var_hashes > 0); /* we shouldn't leave references to shared stats */ - pgstat_release_all_entry_refs(false); + pgstat_release_all_entry_refs(discard_pending); for (int h = 0; h < pgStatLocal.num_var_hashes; h++) { diff --git a/src/backend/utils/activity/pgstat_slru.c b/src/backend/utils/activity/pgstat_slru.c index 1863169a0ec..c261d41a752 100644 --- a/src/backend/utils/activity/pgstat_slru.c +++ b/src/backend/utils/activity/pgstat_slru.c @@ -137,7 +137,7 @@ pgstat_get_slru_index(const char *name) * acquired. Otherwise return false. */ bool -pgstat_slru_flush_cb(bool nowait) +pgstat_slru_flush_cb(bool nowait, bool xact_boundary) { PgStatShared_SLRU *stats_shmem = &pgStatLocal.shmem->slru; int i; diff --git a/src/backend/utils/activity/pgstat_subscription.c b/src/backend/utils/activity/pgstat_subscription.c index 3eaf3e0390f..3c61293eb5f 100644 --- a/src/backend/utils/activity/pgstat_subscription.c +++ b/src/backend/utils/activity/pgstat_subscription.c @@ -114,10 +114,13 @@ pgstat_fetch_stat_subscription(Oid subid) * Flush out pending stats for the entry * * If nowait is true and the lock could not be immediately acquired, returns - * false without flushing the entry. Otherwise returns true. + * PGSTAT_FLUSH_LOCK_CONFLICT without flushing the entry. Subscription stats + * are not transactional, so this always flushes everything and returns + * PGSTAT_FLUSH_DONE. */ -bool -pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) +PgStat_FlushResult +pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, + bool xact_boundary) { PgStat_BackendSubEntry *localent; PgStatShared_Subscription *shsubent; @@ -125,10 +128,8 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) localent = (PgStat_BackendSubEntry *) entry_ref->pending; shsubent = (PgStatShared_Subscription *) entry_ref->shared_stats; - /* localent always has non-zero content */ - if (!pgstat_lock_entry(entry_ref, nowait)) - return false; + return PGSTAT_FLUSH_LOCK_CONFLICT; #define SUB_ACC(fld) shsubent->stats.fld += localent->fld SUB_ACC(apply_error_count); @@ -139,7 +140,10 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) #undef SUB_ACC pgstat_unlock_entry(entry_ref); - return true; + + memset(localent, 0, sizeof(*localent)); + + return PGSTAT_FLUSH_DONE; } void diff --git a/src/backend/utils/activity/pgstat_wal.c b/src/backend/utils/activity/pgstat_wal.c index 183e0a7a97b..8745937bfe9 100644 --- a/src/backend/utils/activity/pgstat_wal.c +++ b/src/backend/utils/activity/pgstat_wal.c @@ -17,6 +17,7 @@ #include "postgres.h" +#include "access/xact.h" #include "executor/instrument.h" #include "utils/pgstat_internal.h" @@ -51,7 +52,7 @@ pgstat_report_wal(bool force) nowait = !force; /* flush wal stats */ - (void) pgstat_wal_flush_cb(nowait); + (void) pgstat_wal_flush_cb(nowait, !IsTransactionOrTransactionBlock()); pgstat_flush_backend(nowait, PGSTAT_BACKEND_FLUSH_WAL); /* flush IO stats */ @@ -88,7 +89,7 @@ pgstat_wal_have_pending(void) * acquired. Otherwise return false. */ bool -pgstat_wal_flush_cb(bool nowait) +pgstat_wal_flush_cb(bool nowait, bool xact_boundary) { PgStatShared_Wal *stats_shmem = &pgStatLocal.shmem->wal; WalUsage wal_usage_diff = {0}; diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index 64b6f60516c..fb5ad101ac4 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -1984,7 +1984,7 @@ CppConcat(pg_stat_get_xact_idx_,stat)(PG_FUNCTION_ARGS) \ if (!tabentry) \ result = 0; \ else \ - result = (int64) (tabentry->idx.stat); \ + result = (int64) (tabentry->idx.counts.stat); \ \ PG_RETURN_INT64(result); \ } @@ -2058,7 +2058,10 @@ pg_stat_clear_snapshot(PG_FUNCTION_ARGS) } -/* Force statistics to be reported at the next occasion */ +/* + * Force statistics to be reported. When called in a transaction this flushes + * immediately; otherwise the flush happens at the next occasion. + */ Datum pg_stat_force_next_flush(PG_FUNCTION_ARGS) { diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 50fb182cb79..52b43c5ef9f 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -6216,9 +6216,9 @@ proparallel => 'r', prorettype => 'void', proargtypes => '', prosrc => 'pg_stat_clear_snapshot' }, { oid => '2137', - descr => 'statistics: force stats to be flushed after the next commit', + descr => 'statistics: force stats to be flushed, immediately if within a transaction', proname => 'pg_stat_force_next_flush', proisstrict => 'f', provolatile => 'v', - proparallel => 'r', prorettype => 'void', proargtypes => '', + proparallel => 'u', prorettype => 'void', proargtypes => '', prosrc => 'pg_stat_force_next_flush' }, { oid => '2274', descr => 'statistics: reset collected statistics for current database', diff --git a/src/include/pgstat.h b/src/include/pgstat.h index 9b99062f853..e173a76e24c 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -90,6 +90,33 @@ typedef struct PgStat_FunctionCounts instr_time self_time; } PgStat_FunctionCounts; +/* Keep these counts in sync with the fields above. */ +#define PGSTAT_FUNCTIONCOUNTS_NUM_COUNTERS 1 +#define PGSTAT_FUNCTIONCOUNTS_NUM_INSTR_TIMES 2 + +StaticAssertDecl(sizeof(instr_time) == sizeof(int64), + "instr_time has padding"); +StaticAssertDecl(sizeof(PgStat_FunctionCounts) == + PGSTAT_FUNCTIONCOUNTS_NUM_COUNTERS * sizeof(PgStat_Counter) + + PGSTAT_FUNCTIONCOUNTS_NUM_INSTR_TIMES * sizeof(instr_time), + "PgStat_FunctionCounts has padding"); + +/* + * Pending function stats stored in PgStat_EntryRef->pending. + * + * counts accumulates for the whole transaction (it is what + * pg_stat_xact_user_functions reports); flushed is the portion already written + * to shared memory. A flush writes counts minus flushed and then sets flushed + * to counts, so counts survives an in-transaction flush intact and the shared + * totals are never double-counted. The same counts/flushed scheme is used for + * relation stats; see PgStat_RelationStatus. + */ +typedef struct PgStat_FunctionStatus +{ + PgStat_FunctionCounts counts; + PgStat_FunctionCounts flushed; +} PgStat_FunctionStatus; + /* * Working state needed to accumulate per-function-call timing statistics. */ @@ -164,6 +191,12 @@ StaticAssertDecl(sizeof(PgStat_TableCounts) == 5 * sizeof(PgStat_Counter), * actions, regardless of whether the transaction committed. delta_live_tuples, * delta_dead_tuples, and changed_tuples are set depending on commit or abort. * Note that delta_live_tuples and delta_dead_tuples can be negative! + * + * Counts are added here only when the (sub)transaction that made them ends. + * Until then they stay in PgStat_TableXactStatus. The exception is the + * nontransactional recovery of dead tuples, which updates delta_dead_tuples + * directly (see pgstat_update_heap_dead_tuples()). So this group can be + * flushed at any time. * ---------- */ typedef struct PgStat_TableCountsXact @@ -203,6 +236,13 @@ typedef struct PgStat_IndexCounts PgStat_Counter blocks_hit; } PgStat_IndexCounts; +/* Keep this count in sync with the fields above. */ +#define PGSTAT_INDEXCOUNTS_NUM_COUNTERS 5 + +StaticAssertDecl(sizeof(PgStat_IndexCounts) == + PGSTAT_INDEXCOUNTS_NUM_COUNTERS * sizeof(PgStat_Counter), + "PgStat_IndexCounts has padding"); + /* ---------- * PgStat_RelationStatus Per-relation pending status within a backend * @@ -236,10 +276,18 @@ typedef struct PgStat_RelationStatus struct PgStat_TableXactStatus *trans; /* lowest subxact's counts */ PgStat_TableCounts counts; /* event counts to be sent */ PgStat_TableCountsXact counts_xact; /* transactional counts */ + PgStat_TableCounts flushed; /* portion of counts already written + * to shared memory */ + PgStat_TableCountsXact flushed_xact; /* portion of counts_xact + * already written */ } tab; /* index counters */ - PgStat_IndexCounts idx; + struct + { + PgStat_IndexCounts counts; + PgStat_IndexCounts flushed; + } idx; }; } PgStat_RelationStatus; @@ -252,6 +300,9 @@ typedef struct PgStat_TableXactStatus PgStat_Counter tuples_inserted; /* tuples inserted in (sub)xact */ PgStat_Counter tuples_updated; /* tuples updated in (sub)xact */ PgStat_Counter tuples_deleted; /* tuples deleted in (sub)xact */ + PgStat_Counter tuples_hot_updated; /* tuples HOT updated in (sub)xact */ + PgStat_Counter tuples_newpage_updated; /* tuples updated to a new page in + * (sub)xact */ bool truncdropped; /* relation truncated/dropped in this * (sub)xact */ /* tuples i/u/d prior to truncate/drop */ @@ -819,7 +870,7 @@ extern void pgstat_report_analyze(Relation rel, if (pgstat_should_count_relation(rel)) \ { \ if ((rel)->pgstat_info->kind == PGSTAT_KIND_INDEX) \ - (rel)->pgstat_info->idx.tuples_fetched++; \ + (rel)->pgstat_info->idx.counts.tuples_fetched++; \ else \ (rel)->pgstat_info->tab.counts.tuples_fetched++; \ } \ @@ -829,7 +880,7 @@ extern void pgstat_report_analyze(Relation rel, if (pgstat_should_count_relation(rel)) \ { \ Assert((rel)->pgstat_info->kind == PGSTAT_KIND_INDEX); \ - (rel)->pgstat_info->idx.numscans++; \ + (rel)->pgstat_info->idx.counts.numscans++; \ } \ } while (0) #define pgstat_count_index_tuples(rel, n) \ @@ -837,7 +888,7 @@ extern void pgstat_report_analyze(Relation rel, if (pgstat_should_count_relation(rel)) \ { \ Assert((rel)->pgstat_info->kind == PGSTAT_KIND_INDEX); \ - (rel)->pgstat_info->idx.tuples_returned += (n); \ + (rel)->pgstat_info->idx.counts.tuples_returned += (n); \ } \ } while (0) #define pgstat_count_buffer_read(rel) \ @@ -845,7 +896,7 @@ extern void pgstat_report_analyze(Relation rel, if (pgstat_should_count_relation(rel)) \ { \ if ((rel)->pgstat_info->kind == PGSTAT_KIND_INDEX) \ - (rel)->pgstat_info->idx.blocks_fetched++; \ + (rel)->pgstat_info->idx.counts.blocks_fetched++; \ else \ (rel)->pgstat_info->tab.counts.blocks_fetched++; \ } \ @@ -855,7 +906,7 @@ extern void pgstat_report_analyze(Relation rel, if (pgstat_should_count_relation(rel)) \ { \ if ((rel)->pgstat_info->kind == PGSTAT_KIND_INDEX) \ - (rel)->pgstat_info->idx.blocks_hit++; \ + (rel)->pgstat_info->idx.counts.blocks_hit++; \ else \ (rel)->pgstat_info->tab.counts.blocks_hit++; \ } \ diff --git a/src/include/utils/pgstat_internal.h b/src/include/utils/pgstat_internal.h index 33fc58feb99..311901e8ab2 100644 --- a/src/include/utils/pgstat_internal.h +++ b/src/include/utils/pgstat_internal.h @@ -42,7 +42,9 @@ * * Once there are pending stats updates for a table PgStat_EntryRef->pending * is allocated to contain a working space for as-of-yet-unapplied stats - * updates. Once the stats are flushed, PgStat_EntryRef->pending is freed. + * updates. Once the stats are fully flushed, PgStat_EntryRef->pending is + * freed. During a transaction a flushed entry is instead retained until the + * transaction boundary. * * Each variable-numbered shared stats struct has PgStatShared_Common as the * first element. @@ -190,6 +192,14 @@ typedef struct PgStat_EntryRef */ void *pending; dlist_node pending_node; /* membership in pgStatPending list */ + + /* + * Flush-pass generation in which this entry was last flushed. Used by + * pgstat_prep_pending_from_entry_ref() to decide whether an entry being + * accumulated into needs re-queuing during the current flush pass. Zero + * means the entry has not been flushed since it became pending. + */ + uint64 flushed_generation; } PgStat_EntryRef; @@ -225,6 +235,29 @@ typedef struct PgStat_SubXactStatus } PgStat_SubXactStatus; +/* + * Result of a flush_pending_cb call, used to decide whether the pending entry + * can be removed from the pending list. + */ +typedef enum PgStat_FlushResult +{ + /* + * Lock not acquired (nowait was true); retry later. Must be 0 because + * earlier versions returned a bool where false meant lock conflict. + */ + PGSTAT_FLUSH_LOCK_CONFLICT = 0, + + /* Fully flushed; the entry can be removed. */ + PGSTAT_FLUSH_DONE, + + /* + * Only the non-transactional counters were flushed; transactional state + * was retained (e.g. flushing during a transaction) and must be flushed + * again at a transaction boundary. + */ + PGSTAT_FLUSH_PARTIAL, +} PgStat_FlushResult; + /* * Metadata for a specific kind of statistics. */ @@ -307,8 +340,25 @@ typedef struct PgStat_KindInfo * For variable-numbered stats: flush pending stats. Required if pending * data is used. See flush_static_cb when dealing with stats data that * that cannot use PgStat_EntryRef->pending. + * + * If nowait is true and the lock cannot be acquired, give up without + * flushing. When xact_boundary is false, a callback must not flush the + * transactional counters of the running transaction. Returns a + * PgStat_FlushResult reporting what happened. + * + * When xact_boundary is false the pending entry is retained, so the + * callback may run again for it before the transaction ends and must not + * merge the same pending data twice. Clear the flushed fields, or track + * a flushed baseline and merge only the delta. A callback must not call + * pgstat_force_next_flush(). + * + * A callback may accumulate into another kind's pending entry (e.g. + * relation stats feed database stats), but it must not accumulate into + * the entry currently being flushed, and such dependencies must be + * acyclic (see pgstat_flush_pending_entries()). */ - bool (*flush_pending_cb) (PgStat_EntryRef *sr, bool nowait); + PgStat_FlushResult (*flush_pending_cb) (PgStat_EntryRef *sr, bool nowait, + bool xact_boundary); /* * For variable-numbered stats: delete pending stats. Optional. @@ -375,10 +425,14 @@ typedef struct PgStat_KindInfo * Returns true if some of the stats could not be flushed, due to lock * contention for example. Optional. * + * xact_boundary is as for flush_pending_cb. The built-in static kinds + * are non-transactional and ignore it; a custom one may use it to defer + * transactional state. + * * "pgstat_report_fixed" needs to be set to trigger the flush of pending * stats. */ - bool (*flush_static_cb) (bool nowait); + bool (*flush_static_cb) (bool nowait, bool xact_boundary); /* * For fixed-numbered statistics: Reset All. @@ -732,7 +786,7 @@ extern void pgstat_archiver_snapshot_cb(void); #define PGSTAT_BACKEND_FLUSH_ALL (PGSTAT_BACKEND_FLUSH_IO | PGSTAT_BACKEND_FLUSH_WAL | PGSTAT_BACKEND_FLUSH_LOCK) extern bool pgstat_flush_backend(bool nowait, uint32 flags); -extern bool pgstat_backend_flush_cb(bool nowait); +extern bool pgstat_backend_flush_cb(bool nowait, bool xact_boundary); extern void pgstat_backend_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts); @@ -764,7 +818,8 @@ extern void AtEOXact_PgStat_Database(bool isCommit, bool parallel); extern PgStat_StatDBEntry *pgstat_prep_database_pending(Oid dboid); extern void pgstat_reset_database_timestamp(Oid dboid, TimestampTz ts); -extern bool pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, bool nowait); +extern PgStat_FlushResult pgstat_database_flush_cb(PgStat_EntryRef *entry_ref, + bool nowait, bool xact_boundary); extern void pgstat_database_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts); @@ -772,7 +827,8 @@ extern void pgstat_database_reset_timestamp_cb(PgStatShared_Common *header, Time * Functions in pgstat_function.c */ -extern bool pgstat_function_flush_cb(PgStat_EntryRef *entry_ref, bool nowait); +extern PgStat_FlushResult pgstat_function_flush_cb(PgStat_EntryRef *entry_ref, + bool nowait, bool xact_boundary); extern void pgstat_function_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts); @@ -782,7 +838,7 @@ extern void pgstat_function_reset_timestamp_cb(PgStatShared_Common *header, Time extern void pgstat_flush_io(bool nowait); -extern bool pgstat_io_flush_cb(bool nowait); +extern bool pgstat_io_flush_cb(bool nowait, bool xact_boundary); extern void pgstat_io_init_shmem_cb(void *stats); extern void pgstat_io_reset_all_cb(TimestampTz ts); extern void pgstat_io_snapshot_cb(void); @@ -791,7 +847,7 @@ extern void pgstat_io_snapshot_cb(void); * Functions in pgstat_lock.c */ -extern bool pgstat_lock_flush_cb(bool nowait); +extern bool pgstat_lock_flush_cb(bool nowait, bool xact_boundary); extern void pgstat_lock_init_shmem_cb(void *stats); extern void pgstat_lock_reset_all_cb(TimestampTz ts); extern void pgstat_lock_snapshot_cb(void); @@ -805,7 +861,8 @@ extern void AtEOSubXact_PgStat_Relations(PgStat_SubXactStatus *xact_state, bool extern void AtPrepare_PgStat_Relations(PgStat_SubXactStatus *xact_state); extern void PostPrepare_PgStat_Relations(PgStat_SubXactStatus *xact_state); -extern bool pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait); +extern PgStat_FlushResult pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, + bool nowait, bool xact_boundary); extern void pgstat_relation_delete_pending_cb(PgStat_EntryRef *entry_ref); extern void pgstat_relation_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts); @@ -814,7 +871,7 @@ extern void pgstat_relation_reset_timestamp_cb(PgStatShared_Common *header, Time * Functions in pgstat_index.c */ -extern bool pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait); +extern PgStat_FlushResult pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary); extern void pgstat_index_delete_pending_cb(PgStat_EntryRef *entry_ref); extern void pgstat_index_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts); @@ -833,7 +890,7 @@ extern bool pgstat_replslot_from_serialized_name_cb(const NameData *name, PgStat */ extern void pgstat_attach_shmem(void); -extern void pgstat_detach_shmem(void); +extern void pgstat_detach_shmem(bool discard_pending); extern PgStat_EntryRef *pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create, bool *created_entry); @@ -865,7 +922,7 @@ extern PgStatShared_Common *pgstat_init_entry(PgStat_Kind kind, * Functions in pgstat_slru.c */ -extern bool pgstat_slru_flush_cb(bool nowait); +extern bool pgstat_slru_flush_cb(bool nowait, bool xact_boundary); extern void pgstat_slru_init_shmem_cb(void *stats); extern void pgstat_slru_reset_all_cb(TimestampTz ts); extern void pgstat_slru_snapshot_cb(void); @@ -876,7 +933,7 @@ extern void pgstat_slru_snapshot_cb(void); */ extern void pgstat_wal_init_backend_cb(void); -extern bool pgstat_wal_flush_cb(bool nowait); +extern bool pgstat_wal_flush_cb(bool nowait, bool xact_boundary); extern void pgstat_wal_init_shmem_cb(void *stats); extern void pgstat_wal_reset_all_cb(TimestampTz ts); extern void pgstat_wal_snapshot_cb(void); @@ -886,7 +943,8 @@ extern void pgstat_wal_snapshot_cb(void); * Functions in pgstat_subscription.c */ -extern bool pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait); +extern PgStat_FlushResult pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, + bool nowait, bool xact_boundary); extern void pgstat_subscription_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts); diff --git a/src/test/modules/test_custom_stats/t/001_custom_stats.pl b/src/test/modules/test_custom_stats/t/001_custom_stats.pl index 484a2d73ff7..39413c6275d 100644 --- a/src/test/modules/test_custom_stats/t/001_custom_stats.pl +++ b/src/test/modules/test_custom_stats/t/001_custom_stats.pl @@ -93,27 +93,79 @@ foreach my $stats_kind (@var_stats_kinds) $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry1'))); is( $result, - "entry1|2|Test entry 1", + "entry1|2|2|0|0|Test entry 1", "variable-sized stats of entry1 for $kind_name"); $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry2'))); is( $result, - "entry2|3|Test entry 2", + "entry2|3|3|0|0|Test entry 2", "variable-sized stats of entry2 for $kind_name"); $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry3'))); is( $result, - "entry3|2|Test entry 3", + "entry3|2|2|0|0|Test entry 3", "variable-sized stats of entry3 for $kind_name"); $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry4'))); is( $result, - "entry4|3|Test entry 4", + "entry4|3|3|0|0|Test entry 4", "variable-sized stats of entry4 for $kind_name"); + # Repeated flushes in one transaction must not double count. + $node->safe_psql('postgres', + q(select test_custom_stats_var_create('xact_entry', 'Test xact entry'))); + $node->safe_psql( + 'postgres', q( + BEGIN; + select test_custom_stats_var_update('xact_entry'); + select pg_stat_force_next_flush(); + select test_custom_stats_var_update('xact_entry'); + select pg_stat_force_next_flush(); + select test_custom_stats_var_update('xact_entry'); + COMMIT;)); + $result = $node->safe_psql('postgres', + q(select * from test_custom_stats_var_report('xact_entry'))); + is( $result, + "xact_entry|3|3|0|0|Test xact entry", + "in-transaction flush does not double-count pending stats for $kind_name"); + $node->safe_psql('postgres', + q(select * from test_custom_stats_var_drop('xact_entry'))); + + # Transactional counters stay deferred until the transaction boundary. + $node->safe_psql('postgres', + q(select test_custom_stats_var_create('txn_entry', 'Test txn entry'))); + + my $bg = $node->background_psql('postgres'); + $bg->query_safe('BEGIN'); + $bg->query_safe(q(select test_custom_stats_var_update('txn_entry'))); + $bg->query_safe(q(select test_custom_stats_var_update_txn('txn_entry'))); + $bg->query_safe(q(select pg_stat_force_next_flush())); + + $result = $node->safe_psql('postgres', + q(select calls, calls2, calls_txn, calls_txn2 + from test_custom_stats_var_report('txn_entry'))); + is($result, "1|1|0|0", + "transactional counter deferred at an in-transaction flush for $kind_name"); + + $bg->query_safe('COMMIT'); + $bg->quit; + + # After commit the deferred counter is flushed too. + $node->poll_query_until('postgres', + q(select calls_txn = 1 and calls_txn2 = 1 + from test_custom_stats_var_report('txn_entry'))) + or die "timed out waiting for transactional counter to be flushed for $kind_name"; + $result = $node->safe_psql('postgres', + q(select calls, calls2, calls_txn, calls_txn2 + from test_custom_stats_var_report('txn_entry'))); + is($result, "1|1|1|1", + "transactional counter flushed at the transaction boundary for $kind_name"); + $node->safe_psql('postgres', + q(select test_custom_stats_var_drop('txn_entry'))); + # Test drop of variable-sized stats via pgstat_drop_entry(). $node->safe_psql('postgres', q(select * from test_custom_stats_var_drop('entry3'))); @@ -163,14 +215,14 @@ foreach my $stats_kind (@var_stats_kinds) $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry1'))); is( $result, - "entry1|2|Test entry 1", + "entry1|2|2|0|0|Test entry 1", "variable-sized stats of entry1 persist after clean restart for $kind_name" ); $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry2'))); is( $result, - "entry2|3|Test entry 2", + "entry2|3|3|0|0|Test entry 2", "variable-sized stats of entry2 persist after clean restart for $kind_name" ); @@ -179,13 +231,13 @@ foreach my $stats_kind (@var_stats_kinds) $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry1'))); is( $result, - "entry1|0|Test entry 1", + "entry1|0|0|0|0|Test entry 1", "variable-sized stats of entry1 reset for $kind_name"); $result = $node->safe_psql('postgres', q(select * from test_custom_stats_var_report('entry2'))); is( $result, - "entry2|0|Test entry 2", + "entry2|0|0|0|0|Test entry 2", "variable-sized stats of entry2 reset for $kind_name"); # Test loss after crash recovery. diff --git a/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql b/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql index d58a310aa14..f3249bc9168 100644 --- a/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql +++ b/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql @@ -13,6 +13,11 @@ RETURNS void AS 'MODULE_PATHNAME', 'test_custom_stats_var_update' LANGUAGE C STRICT PARALLEL UNSAFE; +CREATE FUNCTION test_custom_stats_var_update_txn(IN name TEXT) +RETURNS void +AS 'MODULE_PATHNAME', 'test_custom_stats_var_update_txn' +LANGUAGE C STRICT PARALLEL UNSAFE; + CREATE FUNCTION test_custom_stats_var_drop(IN name TEXT) RETURNS void AS 'MODULE_PATHNAME', 'test_custom_stats_var_drop' @@ -35,6 +40,9 @@ LANGUAGE C STRICT PARALLEL UNSAFE; CREATE FUNCTION test_custom_stats_var_report(INOUT name TEXT, OUT calls BIGINT, + OUT calls2 BIGINT, + OUT calls_txn BIGINT, + OUT calls_txn2 BIGINT, OUT description TEXT) RETURNS SETOF record AS 'MODULE_PATHNAME', 'test_custom_stats_var_report' diff --git a/src/test/modules/test_custom_stats/test_custom_var_stats.c b/src/test/modules/test_custom_stats/test_custom_var_stats.c index d572639e4ed..bc31626f739 100644 --- a/src/test/modules/test_custom_stats/test_custom_var_stats.c +++ b/src/test/modules/test_custom_stats/test_custom_var_stats.c @@ -57,13 +57,39 @@ PG_MODULE_MAGIC_EXT( *-------------------------------------------------------------------------- */ -/* Backend-local pending statistics before flush to shared memory */ -typedef struct PgStat_StatCustomVarEntry +/* + * Backend local pending stats before flush. + * + * nontxn flushes on any flush. txn flushes only at a transaction + * boundary. This example only defers txn; it does not discard the counts + * of an aborted transaction. + */ +typedef struct PgStat_StatCustomVarCountsNonTxn { PgStat_Counter numcalls; /* times statistic was incremented */ + PgStat_Counter numcalls2; /* second non-transactional counter */ +} PgStat_StatCustomVarCountsNonTxn; + +typedef struct PgStat_StatCustomVarCountsTxn +{ + PgStat_Counter numcalls_txn; /* flushed only at a transaction boundary */ + PgStat_Counter numcalls_txn2; /* second transactional counter */ +} PgStat_StatCustomVarCountsTxn; + +StaticAssertDecl(sizeof(PgStat_StatCustomVarCountsNonTxn) == + 2 * sizeof(PgStat_Counter), + "PgStat_StatCustomVarCountsNonTxn has padding"); +StaticAssertDecl(sizeof(PgStat_StatCustomVarCountsTxn) == + 2 * sizeof(PgStat_Counter), + "PgStat_StatCustomVarCountsTxn has padding"); + +typedef struct PgStat_StatCustomVarEntry +{ + PgStat_StatCustomVarCountsNonTxn nontxn; + PgStat_StatCustomVarCountsTxn txn; } PgStat_StatCustomVarEntry; -/* Shared memory statistics entry visible to all backends */ +/* Shared memory stats entry visible to all backends */ typedef struct PgStatShared_CustomVarEntry { PgStatShared_Common header; /* standard pgstat entry header */ @@ -90,21 +116,22 @@ static dsa_area *custom_stats_description_dsa = NULL; *-------------------------------------------------------------------------- */ -/* Flush callback: merge pending stats into shared memory */ -static bool test_custom_stats_var_flush_pending_cb(PgStat_EntryRef *entry_ref, - bool nowait); +/* Flush callback that merges pending stats into shared memory */ +static PgStat_FlushResult test_custom_stats_var_flush_pending_cb(PgStat_EntryRef *entry_ref, + bool nowait, + bool xact_boundary); -/* Serialization callback: write auxiliary entry data */ +/* Serialization callback that writes auxiliary entry data */ static bool test_custom_stats_var_to_serialized_data(const PgStat_HashKey *key, const PgStatShared_Common *header, FILE *statfile); -/* Deserialization callback: read auxiliary entry data */ +/* Deserialization callback that reads auxiliary entry data */ static bool test_custom_stats_var_from_serialized_data(const PgStat_HashKey *key, PgStatShared_Common *header, FILE *statfile); -/* Finish callback: end of statistics file operations */ +/* Finish callback for statistics file operations */ static void test_custom_stats_var_finish(PgStat_StatsFileOp status); /*-------------------------------------------------------------------------- @@ -185,32 +212,76 @@ _PG_init(void) */ /* - * test_custom_stats_var_flush_pending_cb - * Merge pending backend statistics into shared memory - * - * Called by pgstat collector to flush accumulated local statistics - * to shared memory where other backends can read them. + * Merge pending backend stats into shared memory. * - * Returns false only if nowait=true and lock acquisition fails. + * nontxn always flushes. txn flushes only at a transaction boundary. + * This callback uses the clear after flush pattern instead of a separate + * flushed baseline. */ -static bool -test_custom_stats_var_flush_pending_cb(PgStat_EntryRef *entry_ref, bool nowait) +static PgStat_FlushResult +test_custom_stats_var_flush_pending_cb(PgStat_EntryRef *entry_ref, bool nowait, + bool xact_boundary) { PgStat_StatCustomVarEntry *pending_entry; PgStatShared_CustomVarEntry *shared_entry; + static const PgStat_StatCustomVarCountsNonTxn zero_nontxn = {0}; + static const PgStat_StatCustomVarCountsTxn zero_txn = {0}; + bool nontxn_pending; + bool txn_pending; pending_entry = (PgStat_StatCustomVarEntry *) entry_ref->pending; shared_entry = (PgStatShared_CustomVarEntry *) entry_ref->shared_stats; + /* If nothing nontransactional can flush now, avoid taking the lock. */ + nontxn_pending = + memcmp(&pending_entry->nontxn, &zero_nontxn, + sizeof(PgStat_StatCustomVarCountsNonTxn)) != 0; + txn_pending = + memcmp(&pending_entry->txn, &zero_txn, + sizeof(PgStat_StatCustomVarCountsTxn)) != 0; + + if (!nontxn_pending) + { + if (!xact_boundary && txn_pending) + return PGSTAT_FLUSH_PARTIAL; + if (!txn_pending) + return PGSTAT_FLUSH_DONE; + } + if (!pgstat_lock_entry(entry_ref, nowait)) - return false; + return PGSTAT_FLUSH_LOCK_CONFLICT; - /* Add pending counts to shared totals */ - shared_entry->stats.numcalls += pending_entry->numcalls; + /* Always flush the nontransactional counters. */ + shared_entry->stats.nontxn.numcalls += pending_entry->nontxn.numcalls; + shared_entry->stats.nontxn.numcalls2 += pending_entry->nontxn.numcalls2; + + /* Flush transactional counters only at a transaction boundary. */ + if (xact_boundary) + { + shared_entry->stats.txn.numcalls_txn += pending_entry->txn.numcalls_txn; + shared_entry->stats.txn.numcalls_txn2 += pending_entry->txn.numcalls_txn2; + } pgstat_unlock_entry(entry_ref); - return true; + /* + * The pending entry is retained across in-transaction flushes (see + * pgstat_flush_pending_entries()), so clear what was just flushed to + * avoid counting it again on a later flush. Nothing reads these stats + * within the current transaction, so zeroing is correct. + */ + memset(&pending_entry->nontxn, 0, sizeof(PgStat_StatCustomVarCountsNonTxn)); + + if (!xact_boundary) + { + /* Transactional counters still pending; retain the entry. */ + if (txn_pending) + return PGSTAT_FLUSH_PARTIAL; + } + else + memset(&pending_entry->txn, 0, sizeof(PgStat_StatCustomVarCountsTxn)); + + return PGSTAT_FLUSH_DONE; } /* @@ -614,8 +685,8 @@ test_custom_stats_var_create(PG_FUNCTION_ARGS) * test_custom_stats_var_update * Increment custom statistic counter * - * Increments call count in backend-local memory. Changes are flushed - * to shared memory by the statistics collector. + * Increments the non-transactional counters in backend-local memory. + * Changes are flushed to shared memory by the statistics collector. */ PG_FUNCTION_INFO_V1(test_custom_stats_var_update); Datum @@ -630,7 +701,35 @@ test_custom_stats_var_update(PG_FUNCTION_ARGS) PGSTAT_CUSTOM_VAR_STATS_IDX(stat_name), NULL); pending_entry = (PgStat_StatCustomVarEntry *) entry_ref->pending; - pending_entry->numcalls++; + pending_entry->nontxn.numcalls++; + pending_entry->nontxn.numcalls2++; + + PG_RETURN_VOID(); +} + +/* + * test_custom_stats_var_update_txn + * Increment the transactional custom statistic counter + * + * Like test_custom_stats_var_update(), but increments the transactional + * counter, whose effect is made visible in shared memory only at a transaction + * boundary. + */ +PG_FUNCTION_INFO_V1(test_custom_stats_var_update_txn); +Datum +test_custom_stats_var_update_txn(PG_FUNCTION_ARGS) +{ + char *stat_name = text_to_cstring(PG_GETARG_TEXT_PP(0)); + PgStat_EntryRef *entry_ref; + PgStat_StatCustomVarEntry *pending_entry; + + /* Get pending entry in local memory */ + entry_ref = pgstat_prep_pending_entry(PGSTAT_KIND_TEST_CUSTOM_VAR_STATS, InvalidOid, + PGSTAT_CUSTOM_VAR_STATS_IDX(stat_name), NULL); + + pending_entry = (PgStat_StatCustomVarEntry *) entry_ref->pending; + pending_entry->txn.numcalls_txn++; + pending_entry->txn.numcalls_txn2++; PG_RETURN_VOID(); } @@ -738,8 +837,9 @@ test_custom_stats_var_reset(PG_FUNCTION_ARGS) * test_custom_stats_var_report * Retrieve custom statistic values * - * Returns single row with statistic name, call count, and description if the - * statistic exists, otherwise returns no rows. + * Returns single row with statistic name, the two non-transactional counters, + * the two transactional counters, and description if the statistic exists, + * otherwise returns no rows. */ PG_FUNCTION_INFO_V1(test_custom_stats_var_report); Datum @@ -772,8 +872,8 @@ test_custom_stats_var_report(PG_FUNCTION_ARGS) if (funcctx->call_cntr < funcctx->max_calls) { - Datum values[3]; - bool nulls[3] = {false, false, false}; + Datum values[6]; + bool nulls[6] = {false, false, false, false, false, false}; HeapTuple tuple; PgStat_EntryRef *entry_ref; PgStatShared_CustomVarEntry *shared_entry; @@ -806,12 +906,15 @@ test_custom_stats_var_report(PG_FUNCTION_ARGS) } values[0] = PointerGetDatum(cstring_to_text(stat_name)); - values[1] = Int64GetDatum(stat_entry->numcalls); + values[1] = Int64GetDatum(stat_entry->nontxn.numcalls); + values[2] = Int64GetDatum(stat_entry->nontxn.numcalls2); + values[3] = Int64GetDatum(stat_entry->txn.numcalls_txn); + values[4] = Int64GetDatum(stat_entry->txn.numcalls_txn2); if (description) - values[2] = PointerGetDatum(cstring_to_text(description)); + values[5] = PointerGetDatum(cstring_to_text(description)); else - nulls[2] = true; + nulls[5] = true; tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls); SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple)); diff --git a/src/test/regress/expected/stats.out b/src/test/regress/expected/stats.out index 8b15471248b..1936c9b1183 100644 --- a/src/test/regress/expected/stats.out +++ b/src/test/regress/expected/stats.out @@ -777,12 +777,6 @@ SELECT pg_stat_force_next_flush(); (1 row) -SELECT last_seq_scan, last_idx_scan FROM pg_stat_all_tables WHERE relid = 'test_last_scan'::regclass; - last_seq_scan | last_idx_scan ----------------+--------------- - | -(1 row) - COMMIT; SELECT stats_reset IS NOT NULL AS has_stats_reset FROM pg_stat_all_tables WHERE relid = 'test_last_scan'::regclass; @@ -999,6 +993,7 @@ SELECT idx_scan, stats_reset IS NOT NULL AS has_stats_reset 0 | t (1 row) +DROP TABLE test_last_scan; ----- -- Test reset of some stats for shared table ----- @@ -1561,6 +1556,7 @@ SELECT pg_stat_have_stats('index', :dboid, :stats_test_idx1_oid); t (1 row) +DROP TABLE stats_test_tab1; -- put enable_seqscan back to on SET enable_seqscan TO on; -- ensure that stats accessors handle NULL input correctly @@ -1836,6 +1832,7 @@ SELECT :io_sum_local_new_tblspc_writes > :io_sum_local_after_writes; (1 row) RESET temp_buffers; +DROP TABLE test_io_local; -- Test that reuse of strategy buffers and reads of blocks into these reused -- buffers while VACUUMing are tracked in pg_stat_io. If there is sufficient -- demand for shared buffers from concurrent queries, some buffers may be @@ -1949,6 +1946,8 @@ SELECT :my_io_stats_pre_reset > :my_io_stats_post_backend_reset; t (1 row) +DROP TABLE test_io_vac_strategy; +DROP TABLE test_io_bulkwrite_strategy; -- Check invalid input for pg_stat_get_backend_io() SELECT pg_stat_get_backend_io(NULL); pg_stat_get_backend_io @@ -2154,4 +2153,534 @@ SELECT fastpath_exceeded > :backend_fastpath_exceeded_before (1 row) DROP TABLE part_test; +-- +-- Test in-transaction flushes +-- +CREATE TABLE xact_flush(id int) WITH (autovacuum_enabled = off); +INSERT INTO xact_flush VALUES (1), (2), (3); +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +-- Record counters before the explicit transaction +SELECT seq_scan AS seq_scan_before, + seq_tup_read AS seq_tup_read_before, + n_tup_ins AS n_tup_ins_before, + n_tup_upd AS n_tup_upd_before, + n_tup_hot_upd + n_tup_newpage_upd AS upd_detail_before + FROM pg_stat_user_tables WHERE relname = 'xact_flush' \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +-- Generate both transaction-safe and transaction-unsafe counters. +SELECT count(*) FROM xact_flush; + count +------- + 3 +(1 row) + +INSERT INTO xact_flush VALUES (4), (5); +UPDATE xact_flush SET id = id WHERE id = 1; +-- Flush in-transaction +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +-- Scans are visible in-transaction; ins and upd stay deferred. +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta, + n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta, + n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush'; + seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta +----------------+--------------------+-----------------+----------------- + 2 | 8 | 0 | 0 +(1 row) + +-- The pg_stat_xact_user_tables view still reports this transaction's own +-- inserts and updates, which the flush leaves in the pending counts. +SELECT n_tup_ins, n_tup_upd + FROM pg_stat_xact_user_tables WHERE relname = 'xact_flush'; + n_tup_ins | n_tup_upd +-----------+----------- + 2 | 1 +(1 row) + +-- Generate more transaction-safe activity to verify no double counting. +SELECT count(*) FROM xact_flush; + count +------- + 5 +(1 row) + +-- Flush again in-transaction +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +-- Should show cumulative totals, not double-counted. +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta, + n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta, + n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush'; + seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta +----------------+--------------------+-----------------+----------------- + 3 | 13 | 0 | 0 +(1 row) + +COMMIT; +-- After commit, all counters should be flushed. +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta, + n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta, + n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta, + n_tup_hot_upd + n_tup_newpage_upd - :upd_detail_before AS upd_detail_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush'; + seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta | upd_detail_delta +----------------+--------------------+-----------------+-----------------+------------------ + 3 | 13 | 2 | 1 | 1 +(1 row) + +DROP TABLE xact_flush; +-- +-- Test in-transaction flushes of index statistics. An index scan reaches +-- shared stats during a transaction and must not be double-counted by a later flush. +-- +CREATE TABLE xact_flush_idx(id int primary key) WITH (autovacuum_enabled = off); +INSERT INTO xact_flush_idx SELECT generate_series(1, 100); +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT idx_scan AS idx_scan_before + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +SET LOCAL enable_seqscan = off; +SET LOCAL enable_bitmapscan = off; +-- Two index scans. +SELECT id FROM xact_flush_idx WHERE id = 1; + id +---- + 1 +(1 row) + +SELECT id FROM xact_flush_idx WHERE id = 2; + id +---- + 2 +(1 row) + +-- Flush in-transaction; index stats are non-transactional and reach shared +-- stats now. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT idx_scan - :idx_scan_before AS idx_scan_delta + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass; + idx_scan_delta +---------------- + 2 +(1 row) + +-- Another index scan, then flush again to verify no double counting. +SELECT id FROM xact_flush_idx WHERE id = 3; + id +---- + 3 +(1 row) + +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT idx_scan - :idx_scan_before AS idx_scan_delta + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass; + idx_scan_delta +---------------- + 3 +(1 row) + +COMMIT; +-- After commit the count is unchanged from the last in-transaction flush. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT idx_scan - :idx_scan_before AS idx_scan_delta + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass; + idx_scan_delta +---------------- + 3 +(1 row) + +DROP TABLE xact_flush_idx; +-- +-- Roll back a savepoint after an in-transaction flush. The flush must not +-- write the subxact's inserts to shared stats. After commit the inserts made +-- outside the subxact count as live tuples while those rolled back count as +-- dead. +-- +CREATE TABLE subxact_flush(id int) WITH (autovacuum_enabled = off); +INSERT INTO subxact_flush SELECT generate_series(1, 50); +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT n_live_tup AS n_live_tup_before, + n_dead_tup AS n_dead_tup_before, + seq_scan AS seq_scan_before + FROM pg_stat_user_tables WHERE relname = 'subxact_flush' \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +SELECT count(*) FROM subxact_flush; + count +------- + 50 +(1 row) + +SAVEPOINT sp; +INSERT INTO subxact_flush SELECT generate_series(51, 80); +-- Flush during a transaction. The scan reaches shared stats; the insert is deferred. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + n_live_tup = :n_live_tup_before AS live_tup_unchanged + FROM pg_stat_user_tables WHERE relname = 'subxact_flush'; + seq_scan_delta | live_tup_unchanged +----------------+-------------------- + 1 | t +(1 row) + +ROLLBACK TO SAVEPOINT sp; +INSERT INTO subxact_flush SELECT generate_series(81, 90); +COMMIT; +-- After commit the inserts made outside the subxact count as live tuples, +-- those rolled back count as dead, and the in-transaction scan is still there. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT n_live_tup - :n_live_tup_before AS n_live_tup_delta, + n_dead_tup - :n_dead_tup_before AS n_dead_tup_delta, + seq_scan - :seq_scan_before AS seq_scan_delta + FROM pg_stat_user_tables WHERE relname = 'subxact_flush'; + n_live_tup_delta | n_dead_tup_delta | seq_scan_delta +------------------+------------------+---------------- + 10 | 30 | 1 +(1 row) + +DROP TABLE subxact_flush; +-- +-- Test that an in-transaction flush keeps the database aggregate in step with +-- the relation counters that feed it, rather than a flush behind. +-- +-- The relation flush accumulates into the database pending entry. If that +-- database entry was already visited earlier in the same flush pass, it must +-- be re-queued so the relation's contribution reaches shared stats in this +-- pass, not the next one. To force that ordering, prime the database pending +-- ahead of the table. Scan a catalog and flush (which leaves the database +-- entry pending), then scan the table and flush again. +-- +CREATE TABLE flush_db_lag(id int) WITH (autovacuum_enabled = off); +INSERT INTO flush_db_lag SELECT generate_series(1, 100); +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +BEGIN; +SET LOCAL stats_fetch_consistency = none; +-- Prime the database pending entry (and flush it) before touching the table. +SELECT 1 FROM pg_class LIMIT 1; + ?column? +---------- + 1 +(1 row) + +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT pg_stat_get_db_tuples_returned(:dboid) AS db_before \gset +-- Now scan the table; its 100 tuples feed the already-visited database entry. +SELECT count(*) FROM flush_db_lag; + count +------- + 100 +(1 row) + +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +-- The database aggregate must have advanced by at least the relation's own +-- 100 tuples in this same flush. It also counts catalog scans, so the +-- increase can only be larger, never smaller. Without re-queue the relation's +-- contribution would be deferred and the increase would fall short of 100. +SELECT pg_stat_get_db_tuples_returned(:dboid) - :db_before >= 100 AS db_kept_up; + db_kept_up +------------ + t +(1 row) + +COMMIT; +DROP TABLE flush_db_lag; +-- +-- Test an in-transaction flush with TRUNCATE. The truncate's reset of +-- live/dead counters is transactional and must not reach shared stats until +-- commit. +-- +CREATE TABLE xact_flush_truncate(id int) WITH (autovacuum_enabled = off); +INSERT INTO xact_flush_truncate SELECT generate_series(1, 100); +DELETE FROM xact_flush_truncate WHERE id <= 20; +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT n_live_tup AS n_live_tup_before, + n_dead_tup AS n_dead_tup_before, + seq_scan AS seq_scan_before + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate' \gset +-- Case 1 runs DML, TRUNCATE, more DML, then ROLLBACK. The truncate's zeroing +-- and all transactional counters must not leak to shared stats. +BEGIN; +SET LOCAL stats_fetch_consistency = none; +-- DML before truncate. +SELECT count(*) FROM xact_flush_truncate; + count +------- + 80 +(1 row) + +INSERT INTO xact_flush_truncate SELECT generate_series(101, 110); +UPDATE xact_flush_truncate SET id = id WHERE id = 1; +TRUNCATE xact_flush_truncate; +-- DML after truncate. +INSERT INTO xact_flush_truncate SELECT generate_series(1, 10); +-- Flush during a transaction. The scan reaches shared stats; the rest is deferred. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + n_live_tup = :n_live_tup_before AS live_tup_unchanged, + n_dead_tup = :n_dead_tup_before AS dead_tup_unchanged + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + seq_scan_delta | live_tup_unchanged | dead_tup_unchanged +----------------+--------------------+-------------------- + 2 | t | t +(1 row) + +ROLLBACK; +-- After rollback live_tup is unchanged, but dead_tup increases since the +-- aborted inserts leave dead tuples behind. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT n_live_tup = :n_live_tup_before AS live_tup_unchanged, + n_dead_tup - :n_dead_tup_before AS dead_tup_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + live_tup_unchanged | dead_tup_delta +--------------------+---------------- + t | 10 +(1 row) + +-- Case 2 runs DML, TRUNCATE, INSERT, DELETE, then COMMIT. Update the baseline +-- to account for changes from case 1. +SELECT seq_scan AS seq_scan_before, + n_live_tup AS n_live_tup_before, + n_dead_tup AS n_dead_tup_before + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate' \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +-- DML before truncate. +SELECT count(*) FROM xact_flush_truncate; + count +------- + 80 +(1 row) + +INSERT INTO xact_flush_truncate SELECT generate_series(101, 110); +UPDATE xact_flush_truncate SET id = id WHERE id = 1; +TRUNCATE xact_flush_truncate; +-- DML after truncate. +INSERT INTO xact_flush_truncate SELECT generate_series(1, 10); +DELETE FROM xact_flush_truncate WHERE id <= 3; +-- Flush during a transaction. The scan reaches shared stats; transactional counters +-- are deferred. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + n_live_tup = :n_live_tup_before AS live_tup_unchanged, + n_dead_tup = :n_dead_tup_before AS dead_tup_unchanged + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + seq_scan_delta | live_tup_unchanged | dead_tup_unchanged +----------------+--------------------+-------------------- + 3 | t | t +(1 row) + +COMMIT; +-- After commit, TRUNCATE has zeroed live/dead, so the counts reflect only the +-- post-truncate inserts and deletes. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT n_live_tup, n_dead_tup + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + n_live_tup | n_dead_tup +------------+------------ + 7 | 3 +(1 row) + +DROP TABLE xact_flush_truncate; +-- +-- Test that pg_stat_force_next_flush() called inside a function does not +-- lose function call statistics. An in-transaction flush must not clear the +-- pending counts that the pg_stat_xact_user_functions view reports, and must +-- not lose or double-count the cumulative shared counts. +-- +SET track_functions TO 'all'; +CREATE FUNCTION flush_func_test() RETURNS VOID LANGUAGE plpgsql AS $$ +BEGIN + PERFORM pg_stat_force_next_flush(); +END; +$$; +SELECT 'flush_func_test()'::regprocedure::oid AS flush_func_test_oid \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +SELECT flush_func_test(); + flush_func_test +----------------- + +(1 row) + +SELECT flush_func_test(); + flush_func_test +----------------- + +(1 row) + +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +-- Function calls are not transactional, so the calls are visible in both +-- pg_stat_user_functions and pg_stat_xact_user_functions after the flush. The +-- flush must not clear the pending counts. +SELECT funcname, calls FROM pg_stat_user_functions WHERE funcid = :flush_func_test_oid; + funcname | calls +-----------------+------- + flush_func_test | 2 +(1 row) + +SELECT pg_stat_get_xact_function_calls(:flush_func_test_oid) AS xact_calls; + xact_calls +------------ + 2 +(1 row) + +SELECT flush_func_test(); + flush_func_test +----------------- + +(1 row) + +COMMIT; +-- After commit the third call is visible too, with no call double-counted. +SELECT pg_stat_force_next_flush(); + pg_stat_force_next_flush +-------------------------- + +(1 row) + +SELECT funcname, calls FROM pg_stat_user_functions WHERE funcid = :flush_func_test_oid; + funcname | calls +-----------------+------- + flush_func_test | 3 +(1 row) + +DROP FUNCTION flush_func_test; +-- +-- Test that a forced flush in a procedure publishes counters of a +-- transaction that already ended, but not those of the running one. +-- This includes HOT and new-page update counts. +-- +CREATE TABLE proc_flush(id int, v text); +ALTER TABLE proc_flush ALTER COLUMN v SET STORAGE PLAIN; +CREATE PROCEDURE proc_flush_probe(INOUT ins bigint DEFAULT NULL, + INOUT upd bigint DEFAULT NULL, + INOUT hot_upd bigint DEFAULT NULL, + INOUT newpage_upd bigint DEFAULT NULL) +LANGUAGE plpgsql AS $$ +DECLARE + big int := current_setting('block_size')::int * 6 / 10; +BEGIN + -- Two versions of this row do not fit on one page. + INSERT INTO proc_flush VALUES (1, repeat('x', big)); + UPDATE proc_flush SET v = repeat('y', big) WHERE id = 1; -- new page + UPDATE proc_flush SET v = 'z' WHERE id = 1; -- HOT + COMMIT; + INSERT INTO proc_flush VALUES (2, 'a'); + UPDATE proc_flush SET v = 'b' WHERE id = 2; -- HOT + PERFORM pg_stat_force_next_flush(); + SELECT n_tup_ins, n_tup_upd, n_tup_hot_upd, n_tup_newpage_upd + INTO ins, upd, hot_upd, newpage_upd + FROM pg_stat_user_tables WHERE relname = 'proc_flush'; +END; +$$; +SET stats_fetch_consistency = none; +CALL proc_flush_probe(); + ins | upd | hot_upd | newpage_upd +-----+-----+---------+------------- + 1 | 2 | 1 | 1 +(1 row) + +RESET stats_fetch_consistency; +DROP PROCEDURE proc_flush_probe; +DROP TABLE proc_flush; -- End of Stats Test diff --git a/src/test/regress/sql/stats.sql b/src/test/regress/sql/stats.sql index 674637e172b..b98a7c289db 100644 --- a/src/test/regress/sql/stats.sql +++ b/src/test/regress/sql/stats.sql @@ -362,7 +362,6 @@ BEGIN; CREATE TEMPORARY TABLE test_last_scan(idx_col int primary key, noidx_col int); INSERT INTO test_last_scan(idx_col, noidx_col) VALUES(1, 1); SELECT pg_stat_force_next_flush(); -SELECT last_seq_scan, last_idx_scan FROM pg_stat_all_tables WHERE relid = 'test_last_scan'::regclass; COMMIT; SELECT stats_reset IS NOT NULL AS has_stats_reset @@ -449,6 +448,8 @@ SELECT pg_stat_reset_single_index_counters('test_last_scan_pkey'::regclass); SELECT idx_scan, stats_reset IS NOT NULL AS has_stats_reset FROM pg_stat_all_indexes WHERE indexrelid = 'test_last_scan_pkey'::regclass; +DROP TABLE test_last_scan; + ----- -- Test reset of some stats for shared table ----- @@ -700,6 +701,8 @@ DROP index stats_test_idx1; ROLLBACK; SELECT pg_stat_have_stats('index', :dboid, :stats_test_idx1_oid); +DROP TABLE stats_test_tab1; + -- put enable_seqscan back to on SET enable_seqscan TO on; @@ -860,6 +863,8 @@ SELECT sum(writes) AS io_sum_local_new_tblspc_writes SELECT :io_sum_local_new_tblspc_writes > :io_sum_local_after_writes; RESET temp_buffers; +DROP TABLE test_io_local; + -- Test that reuse of strategy buffers and reads of blocks into these reused -- buffers while VACUUMing are tracked in pg_stat_io. If there is sufficient -- demand for shared buffers from concurrent queries, some buffers may be @@ -922,6 +927,9 @@ SELECT sum(evictions) + sum(reuses) + sum(extends) + sum(fsyncs) + sum(reads) + FROM pg_stat_get_backend_io(pg_backend_pid()) \gset SELECT :my_io_stats_pre_reset > :my_io_stats_post_backend_reset; +DROP TABLE test_io_vac_strategy; +DROP TABLE test_io_bulkwrite_strategy; + -- Check invalid input for pg_stat_get_backend_io() SELECT pg_stat_get_backend_io(NULL); SELECT pg_stat_get_backend_io(0); @@ -1070,4 +1078,347 @@ SELECT fastpath_exceeded > :backend_fastpath_exceeded_before DROP TABLE part_test; +-- +-- Test in-transaction flushes +-- +CREATE TABLE xact_flush(id int) WITH (autovacuum_enabled = off); +INSERT INTO xact_flush VALUES (1), (2), (3); +SELECT pg_stat_force_next_flush(); + +-- Record counters before the explicit transaction +SELECT seq_scan AS seq_scan_before, + seq_tup_read AS seq_tup_read_before, + n_tup_ins AS n_tup_ins_before, + n_tup_upd AS n_tup_upd_before, + n_tup_hot_upd + n_tup_newpage_upd AS upd_detail_before + FROM pg_stat_user_tables WHERE relname = 'xact_flush' \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +-- Generate both transaction-safe and transaction-unsafe counters. +SELECT count(*) FROM xact_flush; +INSERT INTO xact_flush VALUES (4), (5); +UPDATE xact_flush SET id = id WHERE id = 1; + +-- Flush in-transaction +SELECT pg_stat_force_next_flush(); + +-- Scans are visible in-transaction; ins and upd stay deferred. +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta, + n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta, + n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush'; + +-- The pg_stat_xact_user_tables view still reports this transaction's own +-- inserts and updates, which the flush leaves in the pending counts. +SELECT n_tup_ins, n_tup_upd + FROM pg_stat_xact_user_tables WHERE relname = 'xact_flush'; + +-- Generate more transaction-safe activity to verify no double counting. +SELECT count(*) FROM xact_flush; + +-- Flush again in-transaction +SELECT pg_stat_force_next_flush(); + +-- Should show cumulative totals, not double-counted. +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta, + n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta, + n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush'; + +COMMIT; + +-- After commit, all counters should be flushed. + +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta, + n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta, + n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta, + n_tup_hot_upd + n_tup_newpage_upd - :upd_detail_before AS upd_detail_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush'; + +DROP TABLE xact_flush; + +-- +-- Test in-transaction flushes of index statistics. An index scan reaches +-- shared stats during a transaction and must not be double-counted by a later flush. +-- +CREATE TABLE xact_flush_idx(id int primary key) WITH (autovacuum_enabled = off); +INSERT INTO xact_flush_idx SELECT generate_series(1, 100); +SELECT pg_stat_force_next_flush(); + +SELECT idx_scan AS idx_scan_before + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; +SET LOCAL enable_seqscan = off; +SET LOCAL enable_bitmapscan = off; + +-- Two index scans. +SELECT id FROM xact_flush_idx WHERE id = 1; +SELECT id FROM xact_flush_idx WHERE id = 2; + +-- Flush in-transaction; index stats are non-transactional and reach shared +-- stats now. +SELECT pg_stat_force_next_flush(); +SELECT idx_scan - :idx_scan_before AS idx_scan_delta + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass; + +-- Another index scan, then flush again to verify no double counting. +SELECT id FROM xact_flush_idx WHERE id = 3; +SELECT pg_stat_force_next_flush(); +SELECT idx_scan - :idx_scan_before AS idx_scan_delta + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass; + +COMMIT; + +-- After commit the count is unchanged from the last in-transaction flush. +SELECT pg_stat_force_next_flush(); +SELECT idx_scan - :idx_scan_before AS idx_scan_delta + FROM pg_stat_user_indexes WHERE indexrelid = 'xact_flush_idx_pkey'::regclass; + +DROP TABLE xact_flush_idx; + +-- +-- Roll back a savepoint after an in-transaction flush. The flush must not +-- write the subxact's inserts to shared stats. After commit the inserts made +-- outside the subxact count as live tuples while those rolled back count as +-- dead. +-- +CREATE TABLE subxact_flush(id int) WITH (autovacuum_enabled = off); +INSERT INTO subxact_flush SELECT generate_series(1, 50); +SELECT pg_stat_force_next_flush(); + +SELECT n_live_tup AS n_live_tup_before, + n_dead_tup AS n_dead_tup_before, + seq_scan AS seq_scan_before + FROM pg_stat_user_tables WHERE relname = 'subxact_flush' \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +SELECT count(*) FROM subxact_flush; + +SAVEPOINT sp; +INSERT INTO subxact_flush SELECT generate_series(51, 80); + +-- Flush during a transaction. The scan reaches shared stats; the insert is deferred. +SELECT pg_stat_force_next_flush(); +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + n_live_tup = :n_live_tup_before AS live_tup_unchanged + FROM pg_stat_user_tables WHERE relname = 'subxact_flush'; + +ROLLBACK TO SAVEPOINT sp; +INSERT INTO subxact_flush SELECT generate_series(81, 90); + +COMMIT; + +-- After commit the inserts made outside the subxact count as live tuples, +-- those rolled back count as dead, and the in-transaction scan is still there. +SELECT pg_stat_force_next_flush(); +SELECT n_live_tup - :n_live_tup_before AS n_live_tup_delta, + n_dead_tup - :n_dead_tup_before AS n_dead_tup_delta, + seq_scan - :seq_scan_before AS seq_scan_delta + FROM pg_stat_user_tables WHERE relname = 'subxact_flush'; + +DROP TABLE subxact_flush; + +-- +-- Test that an in-transaction flush keeps the database aggregate in step with +-- the relation counters that feed it, rather than a flush behind. +-- +-- The relation flush accumulates into the database pending entry. If that +-- database entry was already visited earlier in the same flush pass, it must +-- be re-queued so the relation's contribution reaches shared stats in this +-- pass, not the next one. To force that ordering, prime the database pending +-- ahead of the table. Scan a catalog and flush (which leaves the database +-- entry pending), then scan the table and flush again. +-- +CREATE TABLE flush_db_lag(id int) WITH (autovacuum_enabled = off); +INSERT INTO flush_db_lag SELECT generate_series(1, 100); +SELECT pg_stat_force_next_flush(); + +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +-- Prime the database pending entry (and flush it) before touching the table. +SELECT 1 FROM pg_class LIMIT 1; +SELECT pg_stat_force_next_flush(); + +SELECT pg_stat_get_db_tuples_returned(:dboid) AS db_before \gset + +-- Now scan the table; its 100 tuples feed the already-visited database entry. +SELECT count(*) FROM flush_db_lag; +SELECT pg_stat_force_next_flush(); + +-- The database aggregate must have advanced by at least the relation's own +-- 100 tuples in this same flush. It also counts catalog scans, so the +-- increase can only be larger, never smaller. Without re-queue the relation's +-- contribution would be deferred and the increase would fall short of 100. +SELECT pg_stat_get_db_tuples_returned(:dboid) - :db_before >= 100 AS db_kept_up; + +COMMIT; + +DROP TABLE flush_db_lag; + +-- +-- Test an in-transaction flush with TRUNCATE. The truncate's reset of +-- live/dead counters is transactional and must not reach shared stats until +-- commit. +-- +CREATE TABLE xact_flush_truncate(id int) WITH (autovacuum_enabled = off); +INSERT INTO xact_flush_truncate SELECT generate_series(1, 100); +DELETE FROM xact_flush_truncate WHERE id <= 20; +SELECT pg_stat_force_next_flush(); + +SELECT n_live_tup AS n_live_tup_before, + n_dead_tup AS n_dead_tup_before, + seq_scan AS seq_scan_before + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate' \gset + +-- Case 1 runs DML, TRUNCATE, more DML, then ROLLBACK. The truncate's zeroing +-- and all transactional counters must not leak to shared stats. +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +-- DML before truncate. +SELECT count(*) FROM xact_flush_truncate; +INSERT INTO xact_flush_truncate SELECT generate_series(101, 110); +UPDATE xact_flush_truncate SET id = id WHERE id = 1; + +TRUNCATE xact_flush_truncate; + +-- DML after truncate. +INSERT INTO xact_flush_truncate SELECT generate_series(1, 10); + +-- Flush during a transaction. The scan reaches shared stats; the rest is deferred. +SELECT pg_stat_force_next_flush(); + +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + n_live_tup = :n_live_tup_before AS live_tup_unchanged, + n_dead_tup = :n_dead_tup_before AS dead_tup_unchanged + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + +ROLLBACK; + +-- After rollback live_tup is unchanged, but dead_tup increases since the +-- aborted inserts leave dead tuples behind. +SELECT pg_stat_force_next_flush(); +SELECT n_live_tup = :n_live_tup_before AS live_tup_unchanged, + n_dead_tup - :n_dead_tup_before AS dead_tup_delta + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + +-- Case 2 runs DML, TRUNCATE, INSERT, DELETE, then COMMIT. Update the baseline +-- to account for changes from case 1. +SELECT seq_scan AS seq_scan_before, + n_live_tup AS n_live_tup_before, + n_dead_tup AS n_dead_tup_before + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate' \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +-- DML before truncate. +SELECT count(*) FROM xact_flush_truncate; +INSERT INTO xact_flush_truncate SELECT generate_series(101, 110); +UPDATE xact_flush_truncate SET id = id WHERE id = 1; + +TRUNCATE xact_flush_truncate; + +-- DML after truncate. +INSERT INTO xact_flush_truncate SELECT generate_series(1, 10); +DELETE FROM xact_flush_truncate WHERE id <= 3; + +-- Flush during a transaction. The scan reaches shared stats; transactional counters +-- are deferred. +SELECT pg_stat_force_next_flush(); + +SELECT seq_scan - :seq_scan_before AS seq_scan_delta, + n_live_tup = :n_live_tup_before AS live_tup_unchanged, + n_dead_tup = :n_dead_tup_before AS dead_tup_unchanged + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + +COMMIT; + +-- After commit, TRUNCATE has zeroed live/dead, so the counts reflect only the +-- post-truncate inserts and deletes. +SELECT pg_stat_force_next_flush(); +SELECT n_live_tup, n_dead_tup + FROM pg_stat_user_tables WHERE relname = 'xact_flush_truncate'; + +DROP TABLE xact_flush_truncate; + +-- +-- Test that pg_stat_force_next_flush() called inside a function does not +-- lose function call statistics. An in-transaction flush must not clear the +-- pending counts that the pg_stat_xact_user_functions view reports, and must +-- not lose or double-count the cumulative shared counts. +-- +SET track_functions TO 'all'; +CREATE FUNCTION flush_func_test() RETURNS VOID LANGUAGE plpgsql AS $$ +BEGIN + PERFORM pg_stat_force_next_flush(); +END; +$$; +SELECT 'flush_func_test()'::regprocedure::oid AS flush_func_test_oid \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; +SELECT flush_func_test(); +SELECT flush_func_test(); +SELECT pg_stat_force_next_flush(); + +-- Function calls are not transactional, so the calls are visible in both +-- pg_stat_user_functions and pg_stat_xact_user_functions after the flush. The +-- flush must not clear the pending counts. +SELECT funcname, calls FROM pg_stat_user_functions WHERE funcid = :flush_func_test_oid; +SELECT pg_stat_get_xact_function_calls(:flush_func_test_oid) AS xact_calls; + +SELECT flush_func_test(); +COMMIT; + +-- After commit the third call is visible too, with no call double-counted. +SELECT pg_stat_force_next_flush(); +SELECT funcname, calls FROM pg_stat_user_functions WHERE funcid = :flush_func_test_oid; + +DROP FUNCTION flush_func_test; + +-- +-- Test that a forced flush in a procedure publishes counters of a +-- transaction that already ended, but not those of the running one. +-- This includes HOT and new-page update counts. +-- +CREATE TABLE proc_flush(id int, v text); +ALTER TABLE proc_flush ALTER COLUMN v SET STORAGE PLAIN; +CREATE PROCEDURE proc_flush_probe(INOUT ins bigint DEFAULT NULL, + INOUT upd bigint DEFAULT NULL, + INOUT hot_upd bigint DEFAULT NULL, + INOUT newpage_upd bigint DEFAULT NULL) +LANGUAGE plpgsql AS $$ +DECLARE + big int := current_setting('block_size')::int * 6 / 10; +BEGIN + -- Two versions of this row do not fit on one page. + INSERT INTO proc_flush VALUES (1, repeat('x', big)); + UPDATE proc_flush SET v = repeat('y', big) WHERE id = 1; -- new page + UPDATE proc_flush SET v = 'z' WHERE id = 1; -- HOT + COMMIT; + INSERT INTO proc_flush VALUES (2, 'a'); + UPDATE proc_flush SET v = 'b' WHERE id = 2; -- HOT + PERFORM pg_stat_force_next_flush(); + SELECT n_tup_ins, n_tup_upd, n_tup_hot_upd, n_tup_newpage_upd + INTO ins, upd, hot_upd, newpage_upd + FROM pg_stat_user_tables WHERE relname = 'proc_flush'; +END; +$$; +SET stats_fetch_consistency = none; +CALL proc_flush_probe(); +RESET stats_fetch_consistency; +DROP PROCEDURE proc_flush_probe; +DROP TABLE proc_flush; + -- End of Stats Test diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index f10eb469266..9cb1676fffb 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2324,8 +2324,10 @@ PgStat_Counter PgStat_EntryRef PgStat_EntryRefHashEntry PgStat_FetchConsistency +PgStat_FlushResult PgStat_FunctionCallUsage PgStat_FunctionCounts +PgStat_FunctionStatus PgStat_HashKey PgStat_IO PgStat_IndexCounts -- 2.50.1