From 464d583bd13607382cf2e0c62da41a58674b9c01 Mon Sep 17 00:00:00 2001
From: "Sami Imseih (AWS)" <simseih@amazon.com>
Date: Wed, 5 Aug 2026 18:52:30 +0000
Subject: [PATCH v11 2/2] 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 <samimseih.pg@gmail.com>
Reviewed-by: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Discussion: https://postgr.es/m/CAA5RZ0uA-4qcD3+2hjcE_-zQUBhvWf5foPM2vzYneFKrJLsBDQ@mail.gmail.com
---
 doc/src/sgml/monitoring.sgml                  |  30 +-
 src/backend/utils/activity/pgstat.c           | 148 ++++--
 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           | 498 +++++++++++++++++-
 src/test/regress/sql/stats.sql                | 310 ++++++++++-
 src/tools/pgindent/typedefs.list              |   2 +
 22 files changed, 1486 insertions(+), 241 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index b403fb990a7..42f7523a8ac 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 <varname>PGSTAT_MIN_INTERVAL</varname>
    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
+   <function>pg_stat_force_next_flush</function>.  However,
    current-query information collected by <varname>track_activities</varname>
    is always up-to-date.
   </para>
@@ -4536,7 +4538,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        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
       </para></entry>
      </row>
 
@@ -4564,7 +4567,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        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
       </para></entry>
      </row>
 
@@ -5065,7 +5069,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        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
       </para></entry>
      </row>
 
@@ -5936,6 +5941,23 @@ description | Waiting for a newly initialized WAL file to reach durable storage
        </para></entry>
       </row>
 
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_stat_force_next_flush</primary>
+        </indexterm>
+        <function>pg_stat_force_next_flush</function> ()
+        <returnvalue>void</returnvalue>
+       </para>
+       <para>
+        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.
+       </para></entry>
+      </row>
+
       <row>
        <entry role="func_table_entry"><para role="func_signature">
         <indexterm>
diff --git a/src/backend/utils/activity/pgstat.c b/src/backend/utils/activity/pgstat.c
index 4615f610106..d06ca6eff81 100644
--- a/src/backend/utils/activity/pgstat.c
+++ b/src/backend/utils/activity/pgstat.c
@@ -250,6 +250,15 @@ static dlist_head pgStatPending = DLIST_STATIC_INIT(pgStatPending);
  */
 static bool pgStatForceNextFlush = false;
 
+/* Set while pgstat_report_stat() flushes pending entries. */
+static bool pgStatFlushInProgress = false;
+
+/* Current pending entry callback executing, or NULL when none is active. */
+static PgStat_EntryRef *pgStatCurrentFlushEntry = NULL;
+
+/* Generation number of the current pgstat_flush_pending_entries() pass. */
+static uint64 pgStatFlushGeneration = 0;
+
 /*
  * Force-clear existing snapshot before next use when stats_fetch_consistency
  * is changed.
@@ -342,7 +351,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 +741,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 +755,11 @@ pgstat_report_stat(bool force)
 	bool		nowait;
 
 	pgstat_assert_is_up();
-	Assert(!IsTransactionOrTransactionBlock());
+	Assert(force || !IsTransactionOrTransactionBlock());
+
+	/* a prior flush must have cleared this, even if it errored out */
+	Assert(!pgStatFlushInProgress);
+	Assert(pgStatCurrentFlushEntry == NULL);
 
 	/* "absorb" the forced flush even if there's nothing to flush */
 	if (pgStatForceNextFlush)
@@ -808,35 +822,45 @@ pgstat_report_stat(bool force)
 
 	partial_flush = false;
 
-	/* flush of variable-numbered stats tracked in pending entries list */
-	partial_flush |= pgstat_flush_pending_entries(nowait);
+	pgStatFlushInProgress = true;
 
-	/* flush of other stats kinds */
-	if (pgstat_report_fixed)
+	/* Clear flush-pass state before leaving this flush. */
+	PG_TRY();
 	{
-		for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++)
-		{
-			const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
+		/* flush of variable-numbered stats tracked in pending entries list */
+		partial_flush |= pgstat_flush_pending_entries(nowait);
 
-			if (!kind_info)
-				continue;
-			if (!kind_info->flush_static_cb)
-				continue;
-
-			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();
+	{
+		pgStatCurrentFlushEntry = NULL;
+		pgStatFlushInProgress = false;
+	}
+	PG_END_TRY();
 
 	last_flush = now;
 
-	/*
-	 * If some of the pending stats could not be flushed due to lock
-	 * contention, let the caller know when to retry.
-	 */
+	/* Let the caller know when to retry after a partial flush. */
 	if (partial_flush)
 	{
-		/* force should have prevented us from getting here */
-		Assert(!force);
+		/* with force, only active transaction state can cause a partial flush */
+		Assert(!force || IsTransactionOrTransactionBlock());
 
 		/* remember since when stats have been pending */
 		if (pending_since == 0)
@@ -858,6 +882,16 @@ pgstat_report_stat(bool force)
 void
 pgstat_force_next_flush(void)
 {
+	/*
+	 * A flush callback must not call this; doing so would recursively enter
+	 * pgstat_report_stat() while a flush is already in progress.
+	 */
+	Assert(!pgStatFlushInProgress);
+
+	/* During a transaction flush now unless a flush is already running. */
+	if (!pgStatFlushInProgress && IsTransactionOrTransactionBlock())
+		pgstat_report_stat(true);
+
 	pgStatForceNextFlush = true;
 }
 
@@ -1370,6 +1404,7 @@ pgstat_delete_pending_entry(PgStat_EntryRef *entry_ref)
 
 	pfree(pending_data);
 	entry_ref->pending = NULL;
+	entry_ref->flushed_generation = 0;
 
 	dlist_delete(&entry_ref->pending_node);
 }
@@ -1383,6 +1418,8 @@ pgstat_prep_pending_from_entry_ref(PgStat_EntryRef *entry_ref)
 	PgStat_Kind kind;
 
 	Assert(entry_ref != NULL);
+	/* The entry currently being flushed must not be made pending again. */
+	Assert(entry_ref != pgStatCurrentFlushEntry);
 
 	kind = entry_ref->shared_entry->key.kind;
 
@@ -1404,8 +1441,19 @@ pgstat_prep_pending_from_entry_ref(PgStat_EntryRef *entry_ref)
 		}
 
 		entry_ref->pending = MemoryContextAllocZero(pgStatPendingContext, entrysize);
+		entry_ref->flushed_generation = 0;
 		dlist_push_tail(&pgStatPending, &entry_ref->pending_node);
 	}
+	else if (pgStatFlushInProgress &&
+			 entry_ref->flushed_generation == pgStatFlushGeneration)
+	{
+		/*
+		 * The entry is already pending and was already visited in the current
+		 * flush pass.  Move it to the tail so the data just accumulated into
+		 * it is flushed again before the pass ends.
+		 */
+		dlist_move_tail(&pgStatPending, &entry_ref->pending_node);
+	}
 }
 
 /*
@@ -1422,12 +1470,19 @@ pgstat_flush_pending_entries(bool nowait)
 	 * Processing a pending entry may queue further pending entries to the end
 	 * of the list that we want to process, so a simple iteration won't do.
 	 * Further complicating matters is that we want to delete the current
-	 * entry in each iteration from the list if we flushed successfully.
+	 * entry in each iteration from the list if we flushed successfully,
+	 * though during a transaction a fully flushed entry is retained and only
+	 * deleted at a transaction boundary.
 	 *
-	 * So we just keep track of the next pointer in each loop iteration.
+	 * So we just keep track of the next pointer in each loop iteration, and
+	 * start a new flush-pass generation only when there are pending entries.
 	 */
+
 	if (!dlist_is_empty(&pgStatPending))
+	{
 		cur = dlist_head_node(&pgStatPending);
+		pgStatFlushGeneration++;
+	}
 
 	while (cur)
 	{
@@ -1436,25 +1491,56 @@ 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);
 
 		/* flush the stats, if possible */
-		did_flush = kind_info->flush_pending_cb(entry_ref, nowait);
+		pgStatCurrentFlushEntry = entry_ref;
+		result = kind_info->flush_pending_cb(entry_ref, nowait, xact_boundary);
+		pgStatCurrentFlushEntry = NULL;
 
-		Assert(did_flush || nowait);
+		/*
+		 * A lock conflict can only happen when we allowed the callback to
+		 * give up without waiting for a lock, and a partial flush can only
+		 * happen inside a transaction.
+		 */
+		Assert(result == PGSTAT_FLUSH_DONE ||
+			   (result == PGSTAT_FLUSH_LOCK_CONFLICT && nowait) ||
+			   (result == PGSTAT_FLUSH_PARTIAL && !xact_boundary));
+
+		/*
+		 * Mark the entry visited, unless we could not flush it at all. If a
+		 * later callback adds more data to this entry in the same pass,
+		 * pgstat_prep_pending_from_entry_ref() then re-queues it to the tail
+		 * so the new data is flushed again before the pass ends.
+		 */
+		if (result != PGSTAT_FLUSH_LOCK_CONFLICT)
+			entry_ref->flushed_generation = pgStatFlushGeneration;
 
-		/* determine 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 a1f9a4c6ac1..8e9584c03e5 100644
--- a/src/backend/utils/activity/pgstat_index.c
+++ b/src/backend/utils/activity/pgstat_index.c
@@ -23,16 +23,14 @@
 
 
 /*
- * Flush out pending stats for an index entry.
+ * Flush pending stats for an index entry.
  *
- * If nowait is true and the lock could not be immediately acquired, returns
- * false without flushing the entry.  Otherwise returns true.
- *
- * Some of the stats are copied to the corresponding pending database stats
- * entry when successfully flushing.
+ * If nowait is true and the lock could not be acquired, return
+ * PGSTAT_FLUSH_LOCK_CONFLICT.  Index stats are not transactional, so a flush
+ * always completes.
  */
-bool
-pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
+PgStat_FlushResult
+pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary)
 {
 	Oid			dboid;
 	PgStat_RelationStatus *lstats;	/* pending stats entry */
@@ -44,43 +42,46 @@ pgstat_index_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
 	lstats = (PgStat_RelationStatus *) entry_ref->pending;
 	shidxstats = (PgStatShared_Index *) entry_ref->shared_stats;
 
-	/*
-	 * Ignore entries that didn't accumulate any actual counts, such as
-	 * indexes that were opened by the planner but not used.
-	 */
-	if (pg_memory_is_all_zeros(&lstats->idx,
-							   sizeof(struct PgStat_IndexCounts)))
-		return true;
+	/* Skip entries with nothing new to flush. */
+	if (memcmp(&lstats->idx.counts, &lstats->idx.flushed,
+			   sizeof(struct PgStat_IndexCounts)) == 0)
+		return PGSTAT_FLUSH_DONE;
 
 	if (!pgstat_lock_entry(entry_ref, nowait))
-		return false;
+		return PGSTAT_FLUSH_LOCK_CONFLICT;
 
 	/* Add the values to the shared entry. */
 	idxentry = &shidxstats->stats;
 
-	idxentry->numscans += lstats->idx.numscans;
-	if (lstats->idx.numscans)
+	idxentry->numscans += lstats->idx.counts.numscans - lstats->idx.flushed.numscans;
+	if (lstats->idx.counts.numscans > lstats->idx.flushed.numscans)
 	{
-		TimestampTz t = GetCurrentTransactionStopTimestamp();
+		TimestampTz t = xact_boundary ?
+			GetCurrentTransactionStopTimestamp() :
+			GetCurrentStatementStartTimestamp();
 
 		if (t > idxentry->lastscan)
 			idxentry->lastscan = t;
 	}
-	idxentry->tuples_returned += lstats->idx.tuples_returned;
-	idxentry->tuples_fetched += lstats->idx.tuples_fetched;
-	idxentry->blocks_fetched += lstats->idx.blocks_fetched;
-	idxentry->blocks_hit += lstats->idx.blocks_hit;
+	idxentry->tuples_returned += lstats->idx.counts.tuples_returned - lstats->idx.flushed.tuples_returned;
+	idxentry->tuples_fetched += lstats->idx.counts.tuples_fetched - lstats->idx.flushed.tuples_fetched;
+	idxentry->blocks_fetched += lstats->idx.counts.blocks_fetched - lstats->idx.flushed.blocks_fetched;
+	idxentry->blocks_hit += lstats->idx.counts.blocks_hit - lstats->idx.flushed.blocks_hit;
 
 	pgstat_unlock_entry(entry_ref);
 
 	/* The entry was successfully flushed, add the same to database stats */
 	dbentry = pgstat_prep_database_pending(dboid);
-	dbentry->tuples_returned += lstats->idx.tuples_returned;
-	dbentry->tuples_fetched += lstats->idx.tuples_fetched;
-	dbentry->blocks_fetched += lstats->idx.blocks_fetched;
-	dbentry->blocks_hit += lstats->idx.blocks_hit;
+	dbentry->tuples_returned += lstats->idx.counts.tuples_returned - lstats->idx.flushed.tuples_returned;
+	dbentry->tuples_fetched += lstats->idx.counts.tuples_fetched - lstats->idx.flushed.tuples_fetched;
+	dbentry->blocks_fetched += lstats->idx.counts.blocks_fetched - lstats->idx.flushed.blocks_fetched;
+	dbentry->blocks_hit += lstats->idx.counts.blocks_hit - lstats->idx.flushed.blocks_hit;
+
+	/* Record the new baseline while counts keep accumulating. */
+	if (!xact_boundary)
+		lstats->idx.flushed = lstats->idx.counts;
 
-	return true;
+	return PGSTAT_FLUSH_DONE;
 }
 
 /*
diff --git a/src/backend/utils/activity/pgstat_io.c b/src/backend/utils/activity/pgstat_io.c
index 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 820e0c4dbfe..6cd3ca95777 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,104 +882,154 @@ 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;
+	}
 
 	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 = pgstat_prep_database_pending(dboid);
-	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 6979c7d1161..b92cec8cef5 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..e5f87194d89 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,14 @@ typedef struct PgStat_EntryRef
 	 */
 	void	   *pending;
 	dlist_node	pending_node;	/* membership in pgStatPending list */
+
+	/*
+	 * Flush-pass generation in which this entry was last flushed.  Used by
+	 * pgstat_prep_pending_from_entry_ref() to decide whether an entry being
+	 * accumulated into needs re-queuing during the current flush pass.  Zero
+	 * means the entry has not been flushed since it became pending.
+	 */
+	uint64		flushed_generation;
 } PgStat_EntryRef;
 
 
@@ -225,6 +236,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 +331,25 @@ typedef struct PgStat_KindInfo
 	 * For variable-numbered stats: flush pending stats. Required if pending
 	 * data is used. See flush_static_cb when dealing with stats data that
 	 * that cannot use PgStat_EntryRef->pending.
+	 *
+	 * If nowait is true and the lock cannot be acquired, give up without
+	 * flushing.  When xact_boundary is false, a callback 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 it must not accumulate into
+	 * the entry currently being flushed, and such dependencies must be
+	 * acyclic (see pgstat_flush_pending_entries()).
 	 */
-	bool		(*flush_pending_cb) (PgStat_EntryRef *sr, bool nowait);
+	PgStat_FlushResult (*flush_pending_cb) (PgStat_EntryRef *sr, bool nowait,
+											bool xact_boundary);
 
 	/*
 	 * For variable-numbered stats: delete pending stats. Optional.
@@ -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 50825652bc1..1312e7a37ad 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..38d44df9982 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,496 @@ 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,
+       n_tup_hot_upd + n_tup_newpage_upd AS upd_detail_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 - :upd_detail_before AS upd_detail_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 | upd_detail_delta 
+----------------+--------------------+-----------------+-----------------+------------------
+              3 |                 13 |               2 |               1 |                1
+(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..a890bd01a01 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,313 @@ 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,
+       n_tup_hot_upd + n_tup_newpage_upd AS upd_detail_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 - :upd_detail_before AS upd_detail_delta
+  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 6ad90774d09..5719decf636 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2342,8 +2342,10 @@ PgStat_Counter
 PgStat_EntryRef
 PgStat_EntryRefHashEntry
 PgStat_FetchConsistency
+PgStat_FlushResult
 PgStat_FunctionCallUsage
 PgStat_FunctionCounts
+PgStat_FunctionStatus
 PgStat_HashKey
 PgStat_IO
 PgStat_IndexCounts
-- 
2.50.1

