From 733ab4fade46df1b45d44b50a2bcc8e7582b6452 Mon Sep 17 00:00:00 2001 From: "Sami Imseih (AWS)" Date: Wed, 5 Aug 2026 18:52:30 +0000 Subject: [PATCH v8 3/3] 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 stay pending until a transaction boundary because their final values still depend on commit or abort. To support repeated flushes in one transaction, keep flushed baselines for relation, index, and function stats. 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, and recursive calls from a function. XXX catversion bump needed. Author: Sami Imseih Reviewed-by: Bertrand Drouvot Discussion: https://postgr.es/m/CAA5RZ0uA-4qcD3+2hjcE_-zQUBhvWf5foPM2vzYneFKrJLsBDQ@mail.gmail.com --- doc/src/sgml/monitoring.sgml | 30 +- src/backend/utils/activity/pgstat.c | 128 +++-- 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 | 178 ++++--- src/backend/utils/activity/pgstat_slru.c | 2 +- .../utils/activity/pgstat_subscription.c | 16 +- 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 | 69 ++- src/include/utils/pgstat_internal.h | 85 ++- .../test_custom_stats/t/001_custom_stats.pl | 64 ++- .../test_custom_var_stats--1.0.sql | 8 + .../test_custom_stats/test_custom_var_stats.c | 171 ++++-- src/test/regress/expected/stats.out | 497 +++++++++++++++++- src/test/regress/sql/stats.sql | 309 ++++++++++- src/tools/pgindent/typedefs.list | 2 + 22 files changed, 1465 insertions(+), 240 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 32cb6fdbd76..afd6bb72932 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. @@ -4540,7 +4542,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 @@ -4568,7 +4571,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 @@ -5069,7 +5073,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 @@ -5940,6 +5945,23 @@ description | Waiting for a newly initialized WAL file to reach durable storage + + + + pg_stat_force_next_flush + + pg_stat_force_next_flush () + void + + + Flushes the statistics pending in the current session to shared + memory, making them visible to other sessions. When called within + a transaction, counters tracking data modifications, such as the + numbers of inserted, updated, and deleted tuples, are deferred + until the transaction ends. + + + diff --git a/src/backend/utils/activity/pgstat.c b/src/backend/utils/activity/pgstat.c index 4615f610106..226427b7e8a 100644 --- a/src/backend/utils/activity/pgstat.c +++ b/src/backend/utils/activity/pgstat.c @@ -250,6 +250,9 @@ static dlist_head pgStatPending = DLIST_STATIC_INIT(pgStatPending); */ static bool pgStatForceNextFlush = false; +/* Set while pgstat_report_stat() flushes pending entries. */ +static bool pgStatFlushInProgress = false; + /* * Force-clear existing snapshot before next use when stats_fetch_consistency * is changed. @@ -342,7 +345,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, @@ -732,8 +735,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. A forced flush can happen during a transaction and + * uses the current time instead. */ long pgstat_report_stat(bool force) @@ -745,7 +749,10 @@ 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); /* "absorb" the forced flush even if there's nothing to flush */ if (pgStatForceNextFlush) @@ -808,35 +815,44 @@ 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 pgStatFlushInProgress on every exit path. */ + PG_TRY(); { - 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; + /* flush of variable-numbered stats tracked in pending entries list */ + partial_flush |= pgstat_flush_pending_entries(nowait); - partial_flush |= kind_info->flush_static_cb(nowait); + /* flush of other stats kinds */ + if (pgstat_report_fixed) + { + 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; + + partial_flush |= kind_info->flush_static_cb(nowait, + !IsTransactionOrTransactionBlock()); + } } } + PG_FINALLY(); + { + 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) @@ -858,6 +874,10 @@ pgstat_report_stat(bool force) void pgstat_force_next_flush(void) { + /* During a transaction flush now unless a flush is already running. */ + if (!pgStatFlushInProgress && IsTransactionOrTransactionBlock()) + pgstat_report_stat(true); + pgStatForceNextFlush = true; } @@ -1370,6 +1390,7 @@ pgstat_delete_pending_entry(PgStat_EntryRef *entry_ref) pfree(pending_data); entry_ref->pending = NULL; + entry_ref->flushed_this_pass = false; dlist_delete(&entry_ref->pending_node); } @@ -1404,8 +1425,18 @@ pgstat_prep_pending_from_entry_ref(PgStat_EntryRef *entry_ref) } entry_ref->pending = MemoryContextAllocZero(pgStatPendingContext, entrysize); + entry_ref->flushed_this_pass = false; dlist_push_tail(&pgStatPending, &entry_ref->pending_node); } + else if (entry_ref->flushed_this_pass) + { + /* + * 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); + } } /* @@ -1422,7 +1453,9 @@ 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. */ @@ -1436,25 +1469,60 @@ 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; + CHECK_FOR_INTERRUPTS(); + + xact_boundary = !IsTransactionOrTransactionBlock(); + Assert(!kind_info->fixed_amount); Assert(kind_info->flush_pending_cb != NULL); + /* + * Clear the per-pass flag before the callback so that a callback + * accumulating into its own entry does not re-queue it. Set again + * below once the entry has been visited. + */ + entry_ref->flushed_this_pass = false; + /* flush the stats, if possible */ - did_flush = kind_info->flush_pending_cb(entry_ref, nowait); + result = kind_info->flush_pending_cb(entry_ref, nowait, xact_boundary); + + /* + * 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)); - Assert(did_flush || nowait); + /* + * Mark the entry visited, unless we could not flush it at all. A + * later callback accumulating into it 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_this_pass = true; - /* determine next entry, before deleting the pending entry */ + /* + * 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 10d2350b7e0..e41dd16958c 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,13 +42,10 @@ 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; /* * Do this before mutating shared stats, so an ERROR leaves no partial @@ -59,33 +54,39 @@ pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) dbentry = pgstat_prep_database_pending(dboid); 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->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 8ec1aad5078..07a66e5e153 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 0b66c925f4f..ed5bb583985 100644 --- a/src/backend/utils/activity/pgstat_relation.c +++ b/src/backend/utils/activity/pgstat_relation.c @@ -368,8 +368,14 @@ 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.txn.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.txn.delta_dead_tuples - + rel->pgstat_info->tab.flushed.txn.delta_dead_tuples; + /* Since ANALYZE's counts are estimates, we could have underflowed */ livetuples = Max(livetuples, 0); deadtuples = Max(deadtuples, 0); @@ -876,37 +882,61 @@ 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. + * Transactional counters stay pending during a transaction, so that case + * returns PGSTAT_FLUSH_PARTIAL after flushing only the nontransactional + * group. */ -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 */ PgStatShared_Relation *shtabstats; PgStat_StatTabEntry *tabentry; /* table entry of shared stats */ PgStat_StatDBEntry *dbentry; /* pending database entry */ + bool flush_txn; + bool nontxn_changed; dboid = entry_ref->shared_entry->key.dboid; lstats = (PgStat_RelationStatus *) entry_ref->pending; 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. + * Transactional counters can flush at a transaction boundary, or when + * this relation has no active transaction state. + */ + flush_txn = (xact_boundary || lstats->tab.trans == NULL); + + /* + * During a transaction only the nontransactional group can flush, so + * compare that group on its own. A change only in the transactional + * group must not force a lock. + * + * counts and flushed are zeroed on allocation and no field write touches + * padding. StaticAssertDecl in pgstat.h checks that. */ - if (!lstats->tab.truncdropped && - pg_memory_is_all_zeros(&lstats->tab.counts, - sizeof(struct PgStat_TableCounts))) - return true; + nontxn_changed = memcmp(&lstats->tab.counts.nontxn, &lstats->tab.flushed.nontxn, + sizeof(PgStat_TableCountsNonTxn)) != 0; + + if (!nontxn_changed) + { + /* + * No nontransactional counters can flush now. During a transaction + * keep the entry pending for the boundary flush. At the boundary + * drop the entry only if nothing changed at all. + */ + if (!flush_txn) + return PGSTAT_FLUSH_PARTIAL; + if (!lstats->tab.truncdropped && + memcmp(&lstats->tab.counts.txn, &lstats->tab.flushed.txn, + sizeof(PgStat_TableCountsTxn)) == 0) + return PGSTAT_FLUSH_DONE; + } /* * Do this before mutating shared stats, so an ERROR leaves no partial @@ -915,70 +945,96 @@ pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) dbentry = pgstat_prep_database_pending(dboid); 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.nontxn.numscans; - if (lstats->tab.counts.nontxn.numscans) + tabentry->numscans += lstats->tab.counts.nontxn.numscans - lstats->tab.flushed.nontxn.numscans; + if (lstats->tab.counts.nontxn.numscans > lstats->tab.flushed.nontxn.numscans) { - TimestampTz t = GetCurrentTransactionStopTimestamp(); + TimestampTz t = xact_boundary ? + GetCurrentTransactionStopTimestamp() : + GetCurrentStatementStartTimestamp(); if (t > tabentry->lastscan) tabentry->lastscan = t; } - tabentry->tuples_returned += lstats->tab.counts.nontxn.tuples_returned; - tabentry->tuples_fetched += lstats->tab.counts.nontxn.tuples_fetched; - tabentry->tuples_inserted += lstats->tab.counts.txn.tuples_inserted; - tabentry->tuples_updated += lstats->tab.counts.txn.tuples_updated; - tabentry->tuples_deleted += lstats->tab.counts.txn.tuples_deleted; - tabentry->tuples_hot_updated += lstats->tab.counts.txn.tuples_hot_updated; - tabentry->tuples_newpage_updated += lstats->tab.counts.txn.tuples_newpage_updated; + tabentry->tuples_returned += lstats->tab.counts.nontxn.tuples_returned - lstats->tab.flushed.nontxn.tuples_returned; + tabentry->tuples_fetched += lstats->tab.counts.nontxn.tuples_fetched - lstats->tab.flushed.nontxn.tuples_fetched; + tabentry->blocks_fetched += lstats->tab.counts.nontxn.blocks_fetched - lstats->tab.flushed.nontxn.blocks_fetched; + tabentry->blocks_hit += lstats->tab.counts.nontxn.blocks_hit - lstats->tab.flushed.nontxn.blocks_hit; /* - * If table was truncated/dropped, first reset the live/dead counters. + * Flush transactional counters as a group at a transaction boundary. + * Readers must not see only part of this group. */ - if (lstats->tab.truncdropped) + if (flush_txn) { - tabentry->live_tuples = 0; - tabentry->dead_tuples = 0; - tabentry->ins_since_vacuum = 0; - } + tabentry->tuples_inserted += lstats->tab.counts.txn.tuples_inserted - lstats->tab.flushed.txn.tuples_inserted; + tabentry->tuples_updated += lstats->tab.counts.txn.tuples_updated - lstats->tab.flushed.txn.tuples_updated; + tabentry->tuples_deleted += lstats->tab.counts.txn.tuples_deleted - lstats->tab.flushed.txn.tuples_deleted; + tabentry->tuples_hot_updated += lstats->tab.counts.txn.tuples_hot_updated - lstats->tab.flushed.txn.tuples_hot_updated; + tabentry->tuples_newpage_updated += lstats->tab.counts.txn.tuples_newpage_updated - lstats->tab.flushed.txn.tuples_newpage_updated; - tabentry->live_tuples += lstats->tab.counts.txn.delta_live_tuples; - tabentry->dead_tuples += lstats->tab.counts.txn.delta_dead_tuples; - tabentry->mod_since_analyze += lstats->tab.counts.txn.changed_tuples; + /* + * 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.txn.delta_live_tuples = 0; + lstats->tab.flushed.txn.delta_dead_tuples = 0; + } - /* - * 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.txn.tuples_inserted; + tabentry->live_tuples += lstats->tab.counts.txn.delta_live_tuples - lstats->tab.flushed.txn.delta_live_tuples; + tabentry->dead_tuples += lstats->tab.counts.txn.delta_dead_tuples - lstats->tab.flushed.txn.delta_dead_tuples; + tabentry->mod_since_analyze += lstats->tab.counts.txn.changed_tuples - lstats->tab.flushed.txn.changed_tuples; - tabentry->blocks_fetched += lstats->tab.counts.nontxn.blocks_fetched; - tabentry->blocks_hit += lstats->tab.counts.nontxn.blocks_hit; + /* This still counts aborted inserts in ins_since_vacuum. */ + tabentry->ins_since_vacuum += lstats->tab.counts.txn.tuples_inserted - lstats->tab.flushed.txn.tuples_inserted; - /* Clamp live_tuples in case of negative delta_live_tuples */ - tabentry->live_tuples = Max(tabentry->live_tuples, 0); - /* Likewise for dead_tuples */ - tabentry->dead_tuples = Max(tabentry->dead_tuples, 0); + /* Clamp live_tuples in case of negative delta_live_tuples */ + tabentry->live_tuples = Max(tabentry->live_tuples, 0); + /* Likewise for dead_tuples */ + tabentry->dead_tuples = Max(tabentry->dead_tuples, 0); + } pgstat_unlock_entry(entry_ref); /* The entry was successfully flushed, add the same to database stats */ - dbentry->tuples_returned += lstats->tab.counts.nontxn.tuples_returned; - dbentry->tuples_fetched += lstats->tab.counts.nontxn.tuples_fetched; - dbentry->tuples_inserted += lstats->tab.counts.txn.tuples_inserted; - dbentry->tuples_updated += lstats->tab.counts.txn.tuples_updated; - dbentry->tuples_deleted += lstats->tab.counts.txn.tuples_deleted; - dbentry->blocks_fetched += lstats->tab.counts.nontxn.blocks_fetched; - dbentry->blocks_hit += lstats->tab.counts.nontxn.blocks_hit; - - return true; + dbentry->tuples_returned += lstats->tab.counts.nontxn.tuples_returned - lstats->tab.flushed.nontxn.tuples_returned; + dbentry->tuples_fetched += lstats->tab.counts.nontxn.tuples_fetched - lstats->tab.flushed.nontxn.tuples_fetched; + dbentry->blocks_fetched += lstats->tab.counts.nontxn.blocks_fetched - lstats->tab.flushed.nontxn.blocks_fetched; + dbentry->blocks_hit += lstats->tab.counts.nontxn.blocks_hit - lstats->tab.flushed.nontxn.blocks_hit; + + if (flush_txn) + { + dbentry->tuples_inserted += lstats->tab.counts.txn.tuples_inserted - lstats->tab.flushed.txn.tuples_inserted; + dbentry->tuples_updated += lstats->tab.counts.txn.tuples_updated - lstats->tab.flushed.txn.tuples_updated; + dbentry->tuples_deleted += lstats->tab.counts.txn.tuples_deleted - lstats->tab.flushed.txn.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.truncdropped = false; + } + return PGSTAT_FLUSH_DONE; + } + + /* For a partial flush record only the nontransactional baseline. */ + lstats->tab.flushed.nontxn = lstats->tab.counts.nontxn; + + return PGSTAT_FLUSH_PARTIAL; } void 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..a79120d899a 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; @@ -128,7 +131,7 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) /* 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 +142,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 5c3b1d51091..c530d4fffae 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 1d2a9db3262..5ce755b89a7 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -6218,9 +6218,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 990d6371e5a..1e034b4ec5d 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. */ @@ -121,15 +148,14 @@ typedef struct PgStat_BackendSubEntry /* ---------- * PgStat_TableCounts The actual per-table counts kept by a backend * - * The counters are split into two structs. PgStat_TableCountsNonTxn holds - * counters that are recorded whether the transaction commits or aborts, while - * PgStat_TableCountsTxn holds counters whose effect depends on the transaction - * outcome. Both are combined in PgStat_TableCounts. + * Split per table counters into two groups. PgStat_TableCountsNonTxn holds + * counters recorded whether the transaction commits or aborts. + * PgStat_TableCountsTxn holds counters whose effect depends on transaction + * outcome. * - * These structs should contain only actual event counters, because we make use - * of pg_memory_is_all_zeros() to detect whether there are any stats updates - * to apply. StaticAssertDecl() checks below ensure that these substructs - * contain no padding. + * These structs contain only counters. They are byte compared against zero + * and against the flushed baseline in PgStat_RelationStatus. + * StaticAssertDecl checks below ensure that they contain no padding. * * It is a component of PgStat_RelationStatus (within-backend state, for * table data). @@ -168,7 +194,7 @@ typedef struct PgStat_TableCountsTxn PgStat_Counter changed_tuples; } PgStat_TableCountsTxn; -/* Keep these in sync with the fields in the corresponding counter structs. */ +/* Keep these counts in sync with the fields above. */ #define PGSTAT_TABLECOUNTS_NON_TXN_NUM_COUNTERS 5 #define PGSTAT_TABLECOUNTS_TXN_NUM_COUNTERS 8 @@ -206,6 +232,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 * @@ -238,10 +271,16 @@ typedef struct PgStat_RelationStatus bool truncdropped; /* pending truncate/drop reset */ struct PgStat_TableXactStatus *trans; /* lowest subxact's counts */ PgStat_TableCounts counts; /* event counts to be sent */ + PgStat_TableCounts flushed; /* Portion of counts already written + * to shared memory */ } tab; /* index counters */ - PgStat_IndexCounts idx; + struct + { + PgStat_IndexCounts counts; + PgStat_IndexCounts flushed; + } idx; }; } PgStat_RelationStatus; @@ -821,7 +860,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.nontxn.tuples_fetched++; \ } \ @@ -831,7 +870,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) \ @@ -839,7 +878,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) \ @@ -847,7 +886,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.nontxn.blocks_fetched++; \ } \ @@ -857,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_hit++; \ + (rel)->pgstat_info->idx.counts.blocks_hit++; \ else \ (rel)->pgstat_info->tab.counts.nontxn.blocks_hit++; \ } \ diff --git a/src/include/utils/pgstat_internal.h b/src/include/utils/pgstat_internal.h index 14369e59a1c..e71a262685d 100644 --- a/src/include/utils/pgstat_internal.h +++ b/src/include/utils/pgstat_internal.h @@ -42,7 +42,10 @@ * * 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. An entry only partially flushed during a transaction (its + * transactional counters deferred) is instead retained until the transaction + * boundary. * * Each stat kind in the shared hash table has a fixed member * PgStatShared_Common as the first element. @@ -190,6 +193,15 @@ typedef struct PgStat_EntryRef */ void *pending; dlist_node pending_node; /* membership in pgStatPending list */ + + /* + * True once this entry has been flushed during the current + * pgstat_flush_pending_entries() pass, which sets it. Used by + * pgstat_prep_pending_from_entry_ref() to decide whether an entry being + * accumulated into needs re-queuing. Cleared when the entry leaves the + * list. + */ + bool flushed_this_pass; } PgStat_EntryRef; @@ -225,6 +237,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. */ @@ -297,8 +332,24 @@ 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 with + * transaction-dependent state flushes only its non-transactional + * counters. 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 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. @@ -365,10 +416,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 + * transaction-dependent 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. @@ -717,7 +772,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); @@ -749,7 +804,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); @@ -757,7 +813,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); @@ -767,7 +824,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); @@ -776,7 +833,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); @@ -790,7 +847,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); @@ -799,7 +857,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); @@ -847,7 +905,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); @@ -858,7 +916,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); @@ -868,7 +926,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 69f2284229e..5bb5ffdd748 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 @@ -78,31 +78,83 @@ $node->safe_psql('postgres', q(select test_custom_stats_fixed_update())); $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", "report for variable-sized data of entry1"); $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", "report for variable-sized data of entry2"); $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", "report for variable-sized data of entry3"); $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", "report for variable-sized data of entry4"); $result = $node->safe_psql('postgres', q(select * from test_custom_stats_fixed_report())); is($result, "3|", "report for fixed-sized stats"); +# 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"); +$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"); + +$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"; +$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"); +$node->safe_psql('postgres', + q(select test_custom_stats_var_drop('txn_entry'))); + # Test drop of variable-sized stats. $node->safe_psql('postgres', q(select * from test_custom_stats_var_drop('entry3'))); @@ -122,13 +174,13 @@ $node->start(); $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 persist after clean restart"); $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 persist after clean restart"); $result = $node->safe_psql('postgres', 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 5ed8cfc2dcf..903ea549b5b 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' @@ -20,6 +25,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 bf4c8fa69c3..338b33e9b3c 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,21 +57,37 @@ PG_MODULE_MAGIC_EXT( */ /* - * Backend-local pending statistics before flush to shared memory. + * Backend local pending stats before flush. * - * numcalls is non-transactional and is flushed on any flush, including an - * in-transaction one. numcalls_txn is transactional and becomes visible in - * shared memory only at a transaction boundary; this demonstrates how a custom - * kind can defer transaction-dependent counters (see the flush callback). + * nontxn flushes on any flush. txn flushes only at a transaction + * boundary. */ -typedef struct PgStat_StatCustomVarEntry +typedef struct PgStat_StatCustomVarCountsNonTxn { PgStat_Counter numcalls; /* times statistic was incremented */ - PgStat_Counter numcalls_txn; /* likewise, but flushed only at a - * transaction boundary */ + 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 */ @@ -98,21 +114,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); /*-------------------------------------------------------------------------- @@ -154,32 +171,80 @@ _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. + * StaticAssertDecl above checks that the compared structs have no + * padding. + */ + 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; } /* @@ -583,8 +648,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 @@ -599,7 +664,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(); } @@ -628,8 +721,9 @@ test_custom_stats_var_drop(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 @@ -662,8 +756,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; @@ -696,12 +790,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 cbe0e793f22..d1279376ed0 100644 --- a/src/test/regress/expected/stats.out +++ b/src/test/regress/expected/stats.out @@ -715,12 +715,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; @@ -2092,4 +2086,495 @@ SELECT fastpath_exceeded > :backend_fastpath_exceeded_before (1 row) DROP TABLE part_test; +-- +-- Test in-transaction flushes +-- +CREATE TABLE partial_flush(id int) WITH (autovacuum_enabled = off); +INSERT INTO partial_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 + FROM pg_stat_user_tables WHERE relname = 'partial_flush' \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +-- Generate both transaction-safe and transaction-unsafe counters. +SELECT count(*) FROM partial_flush; + count +------- + 3 +(1 row) + +INSERT INTO partial_flush VALUES (4), (5); +UPDATE partial_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 = 'partial_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 = 'partial_flush'; + n_tup_ins | n_tup_upd +-----------+----------- + 2 | 1 +(1 row) + +-- Generate more transaction-safe activity to verify no double counting. +SELECT count(*) FROM partial_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 = 'partial_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 <= n_tup_upd AS upd_counts_ok + FROM pg_stat_user_tables WHERE relname = 'partial_flush'; + seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta | upd_counts_ok +----------------+--------------------+-----------------+-----------------+--------------- + 3 | 13 | 2 | 1 | t +(1 row) + +DROP TABLE partial_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 partial_flush_idx(id int primary key) WITH (autovacuum_enabled = off); +INSERT INTO partial_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 = 'partial_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 partial_flush_idx WHERE id = 1; + id +---- + 1 +(1 row) + +SELECT id FROM partial_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 = 'partial_flush_idx_pkey'::regclass; + idx_scan_delta +---------------- + 2 +(1 row) + +-- Another index scan, then flush again to verify no double counting. +SELECT id FROM partial_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 = 'partial_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 = 'partial_flush_idx_pkey'::regclass; + idx_scan_delta +---------------- + 3 +(1 row) + +DROP TABLE partial_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 partial flush with TRUNCATE. The truncate's reset of +-- live/dead counters is transactional and must not reach shared stats until +-- commit. +-- +CREATE TABLE partial_flush_truncate(id int) WITH (autovacuum_enabled = off); +INSERT INTO partial_flush_truncate SELECT generate_series(1, 100); +DELETE FROM partial_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 = 'partial_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 partial_flush_truncate; + count +------- + 80 +(1 row) + +INSERT INTO partial_flush_truncate SELECT generate_series(101, 110); +UPDATE partial_flush_truncate SET id = id WHERE id = 1; +TRUNCATE partial_flush_truncate; +-- DML after truncate. +INSERT INTO partial_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 = 'partial_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 = 'partial_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 = 'partial_flush_truncate' \gset +BEGIN; +SET LOCAL stats_fetch_consistency = none; +-- DML before truncate. +SELECT count(*) FROM partial_flush_truncate; + count +------- + 80 +(1 row) + +INSERT INTO partial_flush_truncate SELECT generate_series(101, 110); +UPDATE partial_flush_truncate SET id = id WHERE id = 1; +TRUNCATE partial_flush_truncate; +-- DML after truncate. +INSERT INTO partial_flush_truncate SELECT generate_series(1, 10); +DELETE FROM partial_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 = 'partial_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 = 'partial_flush_truncate'; + n_live_tup | n_dead_tup +------------+------------ + 7 | 3 +(1 row) + +DROP TABLE partial_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; -- End of Stats Test diff --git a/src/test/regress/sql/stats.sql b/src/test/regress/sql/stats.sql index 0e83218c437..e27f101fa09 100644 --- a/src/test/regress/sql/stats.sql +++ b/src/test/regress/sql/stats.sql @@ -330,7 +330,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 @@ -1038,4 +1037,312 @@ SELECT fastpath_exceeded > :backend_fastpath_exceeded_before DROP TABLE part_test; +-- +-- Test in-transaction flushes +-- +CREATE TABLE partial_flush(id int) WITH (autovacuum_enabled = off); +INSERT INTO partial_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 + FROM pg_stat_user_tables WHERE relname = 'partial_flush' \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +-- Generate both transaction-safe and transaction-unsafe counters. +SELECT count(*) FROM partial_flush; +INSERT INTO partial_flush VALUES (4), (5); +UPDATE partial_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 = 'partial_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 = 'partial_flush'; + +-- Generate more transaction-safe activity to verify no double counting. +SELECT count(*) FROM partial_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 = 'partial_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 <= n_tup_upd AS upd_counts_ok + FROM pg_stat_user_tables WHERE relname = 'partial_flush'; + +DROP TABLE partial_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 partial_flush_idx(id int primary key) WITH (autovacuum_enabled = off); +INSERT INTO partial_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 = 'partial_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 partial_flush_idx WHERE id = 1; +SELECT id FROM partial_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 = 'partial_flush_idx_pkey'::regclass; + +-- Another index scan, then flush again to verify no double counting. +SELECT id FROM partial_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 = 'partial_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 = 'partial_flush_idx_pkey'::regclass; + +DROP TABLE partial_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 partial flush with TRUNCATE. The truncate's reset of +-- live/dead counters is transactional and must not reach shared stats until +-- commit. +-- +CREATE TABLE partial_flush_truncate(id int) WITH (autovacuum_enabled = off); +INSERT INTO partial_flush_truncate SELECT generate_series(1, 100); +DELETE FROM partial_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 = 'partial_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 partial_flush_truncate; +INSERT INTO partial_flush_truncate SELECT generate_series(101, 110); +UPDATE partial_flush_truncate SET id = id WHERE id = 1; + +TRUNCATE partial_flush_truncate; + +-- DML after truncate. +INSERT INTO partial_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 = 'partial_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 = 'partial_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 = 'partial_flush_truncate' \gset + +BEGIN; +SET LOCAL stats_fetch_consistency = none; + +-- DML before truncate. +SELECT count(*) FROM partial_flush_truncate; +INSERT INTO partial_flush_truncate SELECT generate_series(101, 110); +UPDATE partial_flush_truncate SET id = id WHERE id = 1; + +TRUNCATE partial_flush_truncate; + +-- DML after truncate. +INSERT INTO partial_flush_truncate SELECT generate_series(1, 10); +DELETE FROM partial_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 = 'partial_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 = 'partial_flush_truncate'; + +DROP TABLE partial_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; + -- End of Stats Test diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 0ecb22a8f2d..5cfaedc4f29 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2343,8 +2343,10 @@ PgStat_Counter PgStat_EntryRef PgStat_EntryRefHashEntry PgStat_FetchConsistency +PgStat_FlushResult PgStat_FunctionCallUsage PgStat_FunctionCounts +PgStat_FunctionStatus PgStat_HashKey PgStat_IO PgStat_IndexCounts -- 2.47.3