From 55ded562fc39e175a45ae8a1fb20cac55fc1ab96 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 v5 1/1] pgstat: Allow pg_stat_force_next_flush() to work
 in-transaction

Previously, pg_stat_force_next_flush() deferred the actual flush until
after the transaction ended.  Extend it to also flush immediately when
called in-transaction.

Non-transactional counters (numscans, tuples_returned, tuples_fetched,
blocks_fetched, blocks_hit) are flushed right away since they reflect
completed work that does not depend on transaction outcome.
Transactional counters (tuples_inserted/updated/deleted and the derived
live/dead tuple counts) are deferred until transaction end, since their
final values depend on commit/abort.

pg_stat_force_next_flush() is not documented since it introduces new
behavior, and stats are no longer just flushed at transaction boundary.

Also remove a test query that checked last_seq_scan/last_idx_scan inside a
transaction; it only previously appeared to work because pg_stat_force_next_flush()
previously did not flush in-transaction, and would now be unstable.
---
 doc/src/sgml/monitoring.sgml                  |  26 +-
 src/backend/utils/activity/pgstat.c           |  96 +++-
 src/backend/utils/activity/pgstat_backend.c   |   2 +-
 src/backend/utils/activity/pgstat_database.c  |  13 +-
 src/backend/utils/activity/pgstat_function.c  |  50 +-
 src/backend/utils/activity/pgstat_io.c        |   5 +-
 src/backend/utils/activity/pgstat_lock.c      |   5 +-
 src/backend/utils/activity/pgstat_relation.c  | 168 +++++--
 src/backend/utils/activity/pgstat_slru.c      |   2 +-
 .../utils/activity/pgstat_subscription.c      |  16 +-
 src/backend/utils/activity/pgstat_wal.c       |   5 +-
 src/include/catalog/catversion.h              |   2 +-
 src/include/catalog/pg_proc.dat               |   4 +-
 src/include/pgstat.h                          |  30 +-
 src/include/utils/pgstat_internal.h           |  72 ++-
 .../test_custom_stats/test_custom_var_stats.c |  18 +-
 src/test/regress/expected/stats.out           | 475 +++++++++++++++++-
 src/test/regress/sql/stats.sql                | 311 +++++++++++-
 src/tools/pgindent/typedefs.list              |   2 +
 19 files changed, 1163 insertions(+), 139 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 099e9b6f4e9..969222ba8f3 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -4540,7 +4540,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 mid-transaction
       </para></entry>
      </row>
 
@@ -4568,7 +4569,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 mid-transaction
       </para></entry>
      </row>
 
@@ -5069,7 +5071,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 mid-transaction
       </para></entry>
      </row>
 
@@ -5922,6 +5925,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 50cd07822b4..ef19e25c9bd 100644
--- a/src/backend/utils/activity/pgstat.c
+++ b/src/backend/utils/activity/pgstat.c
@@ -326,7 +326,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,
@@ -716,8 +716,10 @@ 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 non-forced flush is only ever called outside of a transaction, so it is
+ * fair to use transaction stop time as an approximation of current time.  A
+ * forced flush may also happen within a transaction (e.g.
+ * pg_stat_force_next_flush()), and uses the current time instead.
  */
 long
 pgstat_report_stat(bool force)
@@ -729,7 +731,7 @@ pgstat_report_stat(bool force)
 	bool		nowait;
 
 	pgstat_assert_is_up();
-	Assert(!IsTransactionOrTransactionBlock());
+	Assert(force || !IsTransactionOrTransactionBlock());
 
 	/* "absorb" the forced flush even if there's nothing to flush */
 	if (pgStatForceNextFlush)
@@ -807,7 +809,8 @@ pgstat_report_stat(bool force)
 			if (!kind_info->flush_static_cb)
 				continue;
 
-			partial_flush |= kind_info->flush_static_cb(nowait);
+			partial_flush |= kind_info->flush_static_cb(nowait,
+														!IsTransactionOrTransactionBlock());
 		}
 	}
 
@@ -815,12 +818,13 @@ pgstat_report_stat(bool force)
 
 	/*
 	 * If some of the pending stats could not be flushed due to lock
-	 * contention, let the caller know when to retry.
+	 * contention, or only partially flushed due to in-transaction counters
+	 * being deferred, let the caller know when to retry.
 	 */
 	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)
@@ -842,6 +846,9 @@ pgstat_report_stat(bool force)
 void
 pgstat_force_next_flush(void)
 {
+	if (IsTransactionOrTransactionBlock())
+		pgstat_report_stat(true);
+
 	pgStatForceNextFlush = true;
 }
 
@@ -1354,6 +1361,7 @@ pgstat_delete_pending_entry(PgStat_EntryRef *entry_ref)
 
 	pfree(pending_data);
 	entry_ref->pending = NULL;
+	entry_ref->flushed_this_pass = false;
 
 	dlist_delete(&entry_ref->pending_node);
 }
@@ -1388,8 +1396,18 @@ pgstat_prep_pending_from_entry_ref(PgStat_EntryRef *entry_ref)
 		}
 
 		entry_ref->pending = MemoryContextAllocZero(pgStatPendingContext, entrysize);
+		entry_ref->flushed_this_pass = false;
 		dlist_push_tail(&pgStatPending, &entry_ref->pending_node);
 	}
+	else if (entry_ref->flushed_this_pass)
+	{
+		/*
+		 * The entry is already pending and was already visited in the current
+		 * flush pass.  Move it to the tail so the data just accumulated into
+		 * it is flushed again before the pass ends.
+		 */
+		dlist_move_tail(&pgStatPending, &entry_ref->pending_node);
+	}
 }
 
 /*
@@ -1403,12 +1421,22 @@ pgstat_flush_pending_entries(bool nowait)
 
 	/*
 	 * Need to be a bit careful iterating over the list of pending entries.
-	 * 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.
+	 * Processing a pending entry may add further pending entries to the end
+	 * of the list, or move an already-passed entry there (when a callback
+	 * accumulates into a dependent entry that was already flushed in this
+	 * pass), 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 from the list once it is fully flushed, though mid-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.
+	 *
+	 * NOTE: a callback must not create a cycle by accumulating into a
+	 * dependent entry whose own callback accumulates back into it, directly
+	 * or indirectly; that would re-queue forever.  CHECK_FOR_INTERRUPTS()
+	 * below keeps such a pathological cycle cancellable.
 	 */
 	if (!dlist_is_empty(&pgStatPending))
 		cur = dlist_head_node(&pgStatPending);
@@ -1420,25 +1448,59 @@ pgstat_flush_pending_entries(bool nowait)
 		PgStat_HashKey key = entry_ref->shared_entry->key;
 		PgStat_Kind kind = key.kind;
 		const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
-		bool		did_flush;
+		PgStat_FlushResult result;
+		bool		xact_boundary;
 		dlist_node *next;
 
+		CHECK_FOR_INTERRUPTS();
+
+		xact_boundary = !IsTransactionOrTransactionBlock();
+
 		Assert(!kind_info->fixed_amount);
 		Assert(kind_info->flush_pending_cb != NULL);
 
+		/*
+		 * Clear the per-pass flag before the callback so that a callback
+		 * accumulating into its own entry does not re-queue it.  Set again
+		 * below once the entry has been visited.
+		 */
+		entry_ref->flushed_this_pass = false;
+
 		/* flush the stats, if possible */
-		did_flush = kind_info->flush_pending_cb(entry_ref, nowait);
+		result = kind_info->flush_pending_cb(entry_ref, nowait, xact_boundary);
 
-		Assert(did_flush || nowait);
+		/*
+		 * A lock conflict can only happen when we allowed the callback to
+		 * give up without waiting, and a partial flush can only happen when
+		 * we are inside a transaction and the callback had to retain
+		 * transactional state.
+		 */
+		Assert(result == PGSTAT_FLUSH_DONE ||
+			   (result == PGSTAT_FLUSH_LOCK_CONFLICT && nowait) ||
+			   (result == PGSTAT_FLUSH_PARTIAL && !xact_boundary));
 
-		/* determine next entry, before deleting the pending entry */
+		/*
+		 * Mark the entry visited, unless we could not flush it at all.  A
+		 * later callback accumulating into it then re-queues it to the tail
+		 * so the new data is flushed again before the pass ends.
+		 */
+		if (result != PGSTAT_FLUSH_LOCK_CONFLICT)
+			entry_ref->flushed_this_pass = true;
+
+		/*
+		 * Determine 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)
+		/*
+		 * Never free pending entries mid-transaction; callers may hold
+		 * pointers.
+		 */
+		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..d4f7fcd54e8 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;
@@ -187,31 +187,49 @@ pgstat_end_function_usage(PgStat_FunctionCallUsage *fcu, bool finalize)
  * 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.
+ *
+ * Function stats are not transactional, so this always returns
+ * PGSTAT_FLUSH_DONE.  The entry may be flushed more than once per transaction;
+ * see PgStat_FunctionStatus for the counts/flushed delta scheme.
  */
-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;
+	instr_time	total_delta;
+	instr_time	self_delta;
 
-	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 */
 
 	if (!pgstat_lock_entry(entry_ref, nowait))
-		return false;
+		return PGSTAT_FLUSH_LOCK_CONFLICT;
 
-	shfuncent->stats.numcalls += localent->numcalls;
-	shfuncent->stats.total_time +=
-		INSTR_TIME_GET_MICROSEC(localent->total_time);
-	shfuncent->stats.self_time +=
-		INSTR_TIME_GET_MICROSEC(localent->self_time);
+	/*
+	 * Subtract the already-flushed instr_time baseline before converting to
+	 * microseconds, so that rounding does not drift across repeated flushes.
+	 */
+	total_delta = localent->counts.total_time;
+	INSTR_TIME_SUBTRACT(total_delta, localent->flushed.total_time);
+	self_delta = localent->counts.self_time;
+	INSTR_TIME_SUBTRACT(self_delta, localent->flushed.self_time);
+
+	shfuncent->stats.numcalls +=
+		localent->counts.numcalls - localent->flushed.numcalls;
+	shfuncent->stats.total_time += INSTR_TIME_GET_MICROSEC(total_delta);
+	shfuncent->stats.self_time += INSTR_TIME_GET_MICROSEC(self_delta);
 
 	pgstat_unlock_entry(entry_ref);
 
-	return true;
+	/* Record what has been flushed; counts stays cumulative. */
+	localent->flushed = localent->counts;
+
+	return PGSTAT_FLUSH_DONE;
 }
 
 void
@@ -233,7 +251,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_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 04f2eb21d0b..93f9477ce13 100644
--- a/src/backend/utils/activity/pgstat_relation.c
+++ b/src/backend/utils/activity/pgstat_relation.c
@@ -315,8 +315,13 @@ pgstat_report_analyze(Relation rel,
 			livetuples -= trans->tuples_inserted - trans->tuples_deleted;
 			deadtuples -= trans->tuples_updated + trans->tuples_deleted;
 		}
-		/* count stuff inserted by already-aborted subxacts, too */
-		deadtuples -= rel->pgstat_info->counts.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->counts.delta_dead_tuples -
+			rel->pgstat_info->flushed.delta_dead_tuples;
 		/* Since ANALYZE's counts are estimates, we could have underflowed */
 		livetuples = Max(livetuples, 0);
 		deadtuples = Max(deadtuples, 0);
@@ -808,98 +813,153 @@ pgstat_twophase_postabort(FullTransactionId fxid, uint16 info,
  * 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.
+ *
+ * The transactional counters here are tuples_inserted/updated/deleted and the
+ * derived live/dead tuple counts; per the flush_pending_cb contract they are
+ * retained (PGSTAT_FLUSH_PARTIAL) when flushing mid-transaction.
  *
  * Some of the stats are copied to the corresponding pending database stats
  * entry when successfully flushing.
  */
-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_TableStatus *lstats; /* pending stats entry  */
 	PgStatShared_Relation *shtabstats;
 	PgStat_StatTabEntry *tabentry;	/* table entry of shared stats */
 	PgStat_StatDBEntry *dbentry;	/* pending database entry */
+	bool		flush_txn;
 
 	dboid = entry_ref->shared_entry->key.dboid;
 	lstats = (PgStat_TableStatus *) entry_ref->pending;
 	shtabstats = (PgStatShared_Relation *) 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.
+	 * The transactional counters can be flushed once we reach a transaction
+	 * boundary, or when this relation has no active transaction state (i.e.
+	 * no pending DML whose outcome depends on commit/abort).
+	 */
+	flush_txn = (xact_boundary || lstats->trans == NULL);
+
+	/*
+	 * Ignore entries with nothing new since the last flush, such as indexes
+	 * opened by the planner but not used.  A mid-transaction entry with
+	 * active transaction state stays pending (PGSTAT_FLUSH_PARTIAL) as commit
+	 * will merge more counters into it.
 	 */
-	if (pg_memory_is_all_zeros(&lstats->counts,
-							   sizeof(struct PgStat_TableCounts)))
-		return true;
+	if (memcmp(&lstats->counts, &lstats->flushed,
+			   sizeof(struct PgStat_TableCounts)) == 0)
+		return flush_txn ? PGSTAT_FLUSH_DONE : PGSTAT_FLUSH_PARTIAL;
 
 	if (!pgstat_lock_entry(entry_ref, nowait))
-		return false;
+		return PGSTAT_FLUSH_LOCK_CONFLICT;
 
-	/* add the values to the shared entry. */
+	/* Flush non-transactional counters using deltas against the baseline. */
 	tabentry = &shtabstats->stats;
 
-	tabentry->numscans += lstats->counts.numscans;
-	if (lstats->counts.numscans)
+	tabentry->numscans += lstats->counts.numscans - lstats->flushed.numscans;
+	if (lstats->counts.numscans > lstats->flushed.numscans)
 	{
-		TimestampTz t = GetCurrentTransactionStopTimestamp();
+		TimestampTz t = xact_boundary ?
+			GetCurrentTransactionStopTimestamp() :
+			GetCurrentStatementStartTimestamp();
 
 		if (t > tabentry->lastscan)
 			tabentry->lastscan = t;
 	}
-	tabentry->tuples_returned += lstats->counts.tuples_returned;
-	tabentry->tuples_fetched += lstats->counts.tuples_fetched;
-	tabentry->tuples_inserted += lstats->counts.tuples_inserted;
-	tabentry->tuples_updated += lstats->counts.tuples_updated;
-	tabentry->tuples_deleted += lstats->counts.tuples_deleted;
-	tabentry->tuples_hot_updated += lstats->counts.tuples_hot_updated;
-	tabentry->tuples_newpage_updated += lstats->counts.tuples_newpage_updated;
+	tabentry->tuples_returned += lstats->counts.tuples_returned - lstats->flushed.tuples_returned;
+	tabentry->tuples_fetched += lstats->counts.tuples_fetched - lstats->flushed.tuples_fetched;
+	tabentry->blocks_fetched += lstats->counts.blocks_fetched - lstats->flushed.blocks_fetched;
+	tabentry->blocks_hit += lstats->counts.blocks_hit - lstats->flushed.blocks_hit;
 
 	/*
-	 * If table was truncated/dropped, first reset the live/dead counters.
+	 * Flush the transactional counters as a group, only at a transaction
+	 * boundary.  They are consistent only relative to each other (a reader
+	 * must never see tuples_hot_updated advance past tuples_updated), so a
+	 * partial flush of just some of them could expose an inconsistent state.
 	 */
-	if (lstats->counts.truncdropped)
+	if (flush_txn)
 	{
-		tabentry->live_tuples = 0;
-		tabentry->dead_tuples = 0;
-		tabentry->ins_since_vacuum = 0;
-	}
+		tabentry->tuples_inserted += lstats->counts.tuples_inserted - lstats->flushed.tuples_inserted;
+		tabentry->tuples_updated += lstats->counts.tuples_updated - lstats->flushed.tuples_updated;
+		tabentry->tuples_deleted += lstats->counts.tuples_deleted - lstats->flushed.tuples_deleted;
+		tabentry->tuples_hot_updated += lstats->counts.tuples_hot_updated - lstats->flushed.tuples_hot_updated;
+		tabentry->tuples_newpage_updated += lstats->counts.tuples_newpage_updated - lstats->flushed.tuples_newpage_updated;
 
-	tabentry->live_tuples += lstats->counts.delta_live_tuples;
-	tabentry->dead_tuples += lstats->counts.delta_dead_tuples;
-	tabentry->mod_since_analyze += lstats->counts.changed_tuples;
+		/*
+		 * If table was truncated/dropped, first reset the live/dead counters.
+		 * Commit zeroed counts.delta_live/dead_tuples, so zero their stale
+		 * flushed baselines too.  changed_tuples is not zeroed on truncate,
+		 * so its baseline is still valid.
+		 */
+		if (lstats->counts.truncdropped && !lstats->flushed.truncdropped)
+		{
+			tabentry->live_tuples = 0;
+			tabentry->dead_tuples = 0;
+			tabentry->ins_since_vacuum = 0;
+			lstats->flushed.delta_live_tuples = 0;
+			lstats->flushed.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->counts.tuples_inserted;
+		tabentry->live_tuples += lstats->counts.delta_live_tuples - lstats->flushed.delta_live_tuples;
+		tabentry->dead_tuples += lstats->counts.delta_dead_tuples - lstats->flushed.delta_dead_tuples;
+		tabentry->mod_since_analyze += lstats->counts.changed_tuples - lstats->flushed.changed_tuples;
 
-	tabentry->blocks_fetched += lstats->counts.blocks_fetched;
-	tabentry->blocks_hit += lstats->counts.blocks_hit;
+		/*
+		 * 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->counts.tuples_inserted - lstats->flushed.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->counts.tuples_returned;
-	dbentry->tuples_fetched += lstats->counts.tuples_fetched;
-	dbentry->tuples_inserted += lstats->counts.tuples_inserted;
-	dbentry->tuples_updated += lstats->counts.tuples_updated;
-	dbentry->tuples_deleted += lstats->counts.tuples_deleted;
-	dbentry->blocks_fetched += lstats->counts.blocks_fetched;
-	dbentry->blocks_hit += lstats->counts.blocks_hit;
-
-	return true;
+	dbentry->tuples_returned += lstats->counts.tuples_returned - lstats->flushed.tuples_returned;
+	dbentry->tuples_fetched += lstats->counts.tuples_fetched - lstats->flushed.tuples_fetched;
+	dbentry->blocks_fetched += lstats->counts.blocks_fetched - lstats->flushed.blocks_fetched;
+	dbentry->blocks_hit += lstats->counts.blocks_hit - lstats->flushed.blocks_hit;
+
+	if (flush_txn)
+	{
+		dbentry->tuples_inserted += lstats->counts.tuples_inserted - lstats->flushed.tuples_inserted;
+		dbentry->tuples_updated += lstats->counts.tuples_updated - lstats->flushed.tuples_updated;
+		dbentry->tuples_deleted += lstats->counts.tuples_deleted - lstats->flushed.tuples_deleted;
+
+		/*
+		 * Record everything as flushed.  Clear truncdropped on both so a
+		 * later truncate re-triggers the reset above.
+		 */
+		lstats->flushed = lstats->counts;
+		lstats->counts.truncdropped = false;
+		lstats->flushed.truncdropped = false;
+		return PGSTAT_FLUSH_DONE;
+	}
+
+	/*
+	 * Partial, in-transaction flush: record only the non-transactional
+	 * counters as flushed so the transactional ones flush at the boundary.
+	 */
+	lstats->flushed.numscans = lstats->counts.numscans;
+	lstats->flushed.tuples_returned = lstats->counts.tuples_returned;
+	lstats->flushed.tuples_fetched = lstats->counts.tuples_fetched;
+	lstats->flushed.blocks_fetched = lstats->counts.blocks_fetched;
+	lstats->flushed.blocks_hit = lstats->counts.blocks_hit;
+
+	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/include/catalog/catversion.h b/src/include/catalog/catversion.h
index f20ea73ed39..7598ed6e7e8 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -57,6 +57,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	202607273
+#define CATALOG_VERSION_NO	202608061
 
 #endif
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index f8a021987b5..b89145632de 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 58a44857f13..c5d38260b42 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -90,6 +90,22 @@ typedef struct PgStat_FunctionCounts
 	instr_time	self_time;
 } PgStat_FunctionCounts;
 
+/*
+ * 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 a mid-transaction flush intact and the shared
+ * totals are never double-counted.  The same counts/flushed scheme is used for
+ * relation stats; see PgStat_TableStatus.
+ */
+typedef struct PgStat_FunctionStatus
+{
+	PgStat_FunctionCounts counts;
+	PgStat_FunctionCounts flushed;
+} PgStat_FunctionStatus;
+
 /*
  * Working state needed to accumulate per-function-call timing statistics.
  */
@@ -121,9 +137,11 @@ typedef struct PgStat_BackendSubEntry
 /* ----------
  * PgStat_TableCounts			The actual per-table counts kept by a backend
  *
- * This struct 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.
+ * This struct should contain only actual event counters, because we byte
+ * compare it against the flushed baseline (see PgStat_TableStatus) to detect
+ * whether there are any unflushed stats updates to apply.  Both are zeroed on
+ * allocation and no field write ever touches the padding, so the byte compare
+ * is safe.
  *
  * It is a component of PgStat_TableStatus (within-backend state).
  *
@@ -183,6 +201,12 @@ typedef struct PgStat_TableStatus
 	bool		shared;			/* is it a shared catalog? */
 	struct PgStat_TableXactStatus *trans;	/* lowest subxact's counts */
 	PgStat_TableCounts counts;	/* event counts to be sent */
+
+	/*
+	 * Portion of counts already written to shared memory; a flush writes only
+	 * counts minus flushed.  See PgStat_FunctionStatus for the scheme.
+	 */
+	PgStat_TableCounts flushed;
 	Relation	relation;		/* rel that is using this entry */
 } PgStat_TableStatus;
 
diff --git a/src/include/utils/pgstat_internal.h b/src/include/utils/pgstat_internal.h
index b0a17691966..5c6277491d9 100644
--- a/src/include/utils/pgstat_internal.h
+++ b/src/include/utils/pgstat_internal.h
@@ -190,6 +190,15 @@ typedef struct PgStat_EntryRef
 	 */
 	void	   *pending;
 	dlist_node	pending_node;	/* membership in pgStatPending list */
+
+	/*
+	 * True once this entry has been flushed during the current
+	 * pgstat_flush_pending_entries() pass, which sets it.  Used by
+	 * pgstat_prep_pending_from_entry_ref() to decide whether an entry being
+	 * accumulated into needs re-queuing.  Cleared when the entry leaves the
+	 * list.
+	 */
+	bool		flushed_this_pass;
 } PgStat_EntryRef;
 
 
@@ -225,6 +234,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: 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 mid-transaction) and must be flushed again
+	 * at a transaction boundary.
+	 */
+	PGSTAT_FLUSH_PARTIAL,
+} PgStat_FlushResult;
+
 /*
  * Metadata for a specific kind of statistics.
  */
@@ -297,8 +329,18 @@ 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.
+	 *
+	 * A callback may accumulate into another kind's pending entry (e.g.
+	 * relation stats feed database stats); 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 +407,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.
@@ -711,7 +757,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);
 
@@ -743,7 +789,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);
 
 
@@ -751,7 +798,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);
 
 
@@ -761,7 +809,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);
@@ -770,7 +818,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);
@@ -784,7 +832,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);
 
@@ -832,7 +881,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);
@@ -843,7 +892,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);
@@ -853,7 +902,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/test_custom_var_stats.c b/src/test/modules/test_custom_stats/test_custom_var_stats.c
index a39ada0b67c..7ce44c7cbaf 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
@@ -90,8 +90,9 @@ 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);
+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 */
 static bool test_custom_stats_var_to_serialized_data(const PgStat_HashKey *key,
@@ -151,10 +152,13 @@ _PG_init(void)
  * Called by pgstat collector to flush accumulated local statistics
  * to shared memory where other backends can read them.
  *
- * Returns false only if nowait=true and lock acquisition fails.
+ * These stats are not transactional, so xact_boundary is unused; returns
+ * PGSTAT_FLUSH_LOCK_CONFLICT only if nowait=true and lock acquisition fails,
+ * otherwise PGSTAT_FLUSH_DONE.
  */
-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;
@@ -163,14 +167,14 @@ test_custom_stats_var_flush_pending_cb(PgStat_EntryRef *entry_ref, bool nowait)
 	shared_entry = (PgStatShared_CustomVarEntry *) entry_ref->shared_stats;
 
 	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;
 
 	pgstat_unlock_entry(entry_ref);
 
-	return true;
+	return PGSTAT_FLUSH_DONE;
 }
 
 /*
diff --git a/src/test/regress/expected/stats.out b/src/test/regress/expected/stats.out
index e230356de13..3d00f64226a 100644
--- a/src/test/regress/expected/stats.out
+++ b/src/test/regress/expected/stats.out
@@ -691,12 +691,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;
@@ -2068,4 +2062,473 @@ SELECT fastpath_exceeded > :backend_fastpath_exceeded_before
 (1 row)
 
 DROP TABLE part_test;
+--
+-- Test in-transaction flushes
+--
+CREATE TABLE partial_flush(id int) WITH (autovacuum_enabled = off);
+INSERT INTO partial_flush VALUES (1), (2), (3);
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+-- Record counters before the explicit transaction
+SELECT seq_scan AS seq_scan_before,
+       seq_tup_read AS seq_tup_read_before,
+       n_tup_ins AS n_tup_ins_before,
+       n_tup_upd AS n_tup_upd_before
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush' \gset
+BEGIN;
+SET LOCAL stats_fetch_consistency = none;
+-- Generate both transaction-safe and transaction-unsafe counters.
+SELECT count(*) FROM partial_flush;
+ count 
+-------
+     3
+(1 row)
+
+INSERT INTO partial_flush VALUES (4), (5);
+UPDATE partial_flush SET id = id WHERE id = 1;
+-- Flush in-transaction
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+-- Scans are visible in-transaction; ins and upd stay deferred.  upd_counts_ok
+-- checks the hot/newpage subsets were not flushed ahead of n_tup_upd.
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta,
+       n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta,
+       n_tup_hot_upd + n_tup_newpage_upd <= n_tup_upd AS upd_counts_ok
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush';
+ seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta | upd_counts_ok 
+----------------+--------------------+-----------------+-----------------+---------------
+              2 |                  8 |               0 |               0 | t
+(1 row)
+
+-- Generate more transaction-safe activity to verify no double counting.
+SELECT count(*) FROM partial_flush;
+ count 
+-------
+     5
+(1 row)
+
+-- Flush again in-transaction
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+-- Should show cumulative totals, not double-counted.
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta,
+       n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush';
+ seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta 
+----------------+--------------------+-----------------+-----------------
+              3 |                 13 |               0 |               0
+(1 row)
+
+COMMIT;
+-- After commit, all counters should be flushed.
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta,
+       n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta,
+       n_tup_hot_upd + n_tup_newpage_upd <= n_tup_upd AS upd_counts_ok
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush';
+ seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_tup_upd_delta | upd_counts_ok 
+----------------+--------------------+-----------------+-----------------+---------------
+              3 |                 13 |               2 |               1 | t
+(1 row)
+
+DROP TABLE partial_flush;
+--
+-- Roll back a savepoint after a mid-transaction flush.  The flush must not
+-- publish the subxact's inserts, and after commit the 10 inserts made outside
+-- the subxact count as live tuples while the 30 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 mid-transaction.  The scan is published; 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 10 inserts made outside the subxact count as live tuples,
+-- the 30 rolled back with the subxact count as dead tuples, and the
+-- mid-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 a mid-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 is published in this pass, not
+-- the next one.  To force that ordering, prime the database pending entry
+-- 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: a lower bound discriminates
+-- against that noise.  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 followed by rollback.  The rollback
+-- discards the deferred transactional counters, while the non-transactional
+-- counters already flushed to shared stats persist.
+--
+CREATE TABLE partial_flush_rollback(id int) WITH (autovacuum_enabled = off);
+INSERT INTO partial_flush_rollback SELECT generate_series(1, 50);
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+SELECT seq_scan AS seq_scan_before,
+       seq_tup_read AS seq_tup_read_before,
+       n_tup_ins AS n_tup_ins_before,
+       n_live_tup AS n_live_tup_before
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush_rollback' \gset
+BEGIN;
+SET LOCAL stats_fetch_consistency = none;
+-- Generate both non-transactional (scan) and transactional (insert) activity.
+SELECT count(*) FROM partial_flush_rollback;
+ count 
+-------
+    50
+(1 row)
+
+INSERT INTO partial_flush_rollback SELECT generate_series(51, 100);
+-- Flush mid-transaction.  The scans are published; the insert and its
+-- live-tuple delta 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,
+       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_live_tup - :n_live_tup_before AS n_live_tup_delta
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush_rollback';
+ seq_scan_delta | seq_tup_read_delta | n_tup_ins_delta | n_live_tup_delta 
+----------------+--------------------+-----------------+------------------
+              1 |                 50 |               0 |                0
+(1 row)
+
+ROLLBACK;
+-- After rollback the scans persist, since they were already in shared stats,
+-- while the inserts are discarded and n_live_tup is unchanged.
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_live_tup - :n_live_tup_before AS n_live_tup_delta
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush_rollback';
+ seq_scan_delta | seq_tup_read_delta | n_live_tup_delta 
+----------------+--------------------+------------------
+              1 |                 50 |                0
+(1 row)
+
+DROP TABLE partial_flush_rollback;
+--
+-- 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 mid-transaction.  The scan is published; everything else 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 mid-transaction.  The scan is published; 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 zeros live/dead, then only post-truncate DML counts.
+-- delta_live is inserted minus deleted, 10 - 3 = 7.
+-- delta_dead is updated plus deleted, 0 + 3 = 3.
+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.  A mid-transaction flush must not clear the
+-- pending counts that the transaction-local 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 two calls are visible in both
+-- the shared view and the transaction-local view after the flush.  The flush
+-- must not clear the transaction-local 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 4c265d1245c..ac1cdbe018f 100644
--- a/src/test/regress/sql/stats.sql
+++ b/src/test/regress/sql/stats.sql
@@ -316,7 +316,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
@@ -1024,4 +1023,314 @@ SELECT fastpath_exceeded > :backend_fastpath_exceeded_before
 
 DROP TABLE part_test;
 
+--
+-- Test in-transaction flushes
+--
+CREATE TABLE partial_flush(id int) WITH (autovacuum_enabled = off);
+INSERT INTO partial_flush VALUES (1), (2), (3);
+SELECT pg_stat_force_next_flush();
+
+-- Record counters before the explicit transaction
+SELECT seq_scan AS seq_scan_before,
+       seq_tup_read AS seq_tup_read_before,
+       n_tup_ins AS n_tup_ins_before,
+       n_tup_upd AS n_tup_upd_before
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush' \gset
+
+BEGIN;
+SET LOCAL stats_fetch_consistency = none;
+
+-- Generate both transaction-safe and transaction-unsafe counters.
+SELECT count(*) FROM partial_flush;
+INSERT INTO partial_flush VALUES (4), (5);
+UPDATE partial_flush SET id = id WHERE id = 1;
+
+-- Flush in-transaction
+SELECT pg_stat_force_next_flush();
+
+-- Scans are visible in-transaction; ins and upd stay deferred.  upd_counts_ok
+-- checks the hot/newpage subsets were not flushed ahead of n_tup_upd.
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta,
+       n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta,
+       n_tup_hot_upd + n_tup_newpage_upd <= n_tup_upd AS upd_counts_ok
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush';
+
+-- Generate more transaction-safe activity to verify no double counting.
+SELECT count(*) FROM partial_flush;
+
+-- Flush again in-transaction
+SELECT pg_stat_force_next_flush();
+
+-- Should show cumulative totals, not double-counted.
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta,
+       n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush';
+
+COMMIT;
+
+-- After commit, all counters should be flushed.
+
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_tup_ins - :n_tup_ins_before AS n_tup_ins_delta,
+       n_tup_upd - :n_tup_upd_before AS n_tup_upd_delta,
+       n_tup_hot_upd + n_tup_newpage_upd <= n_tup_upd AS upd_counts_ok
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush';
+
+DROP TABLE partial_flush;
+
+--
+-- Roll back a savepoint after a mid-transaction flush.  The flush must not
+-- publish the subxact's inserts, and after commit the 10 inserts made outside
+-- the subxact count as live tuples while the 30 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 mid-transaction.  The scan is published; 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 10 inserts made outside the subxact count as live tuples,
+-- the 30 rolled back with the subxact count as dead tuples, and the
+-- mid-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 a mid-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 is published in this pass, not
+-- the next one.  To force that ordering, prime the database pending entry
+-- 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: a lower bound discriminates
+-- against that noise.  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 followed by rollback.  The rollback
+-- discards the deferred transactional counters, while the non-transactional
+-- counters already flushed to shared stats persist.
+--
+CREATE TABLE partial_flush_rollback(id int) WITH (autovacuum_enabled = off);
+INSERT INTO partial_flush_rollback SELECT generate_series(1, 50);
+SELECT pg_stat_force_next_flush();
+
+SELECT seq_scan AS seq_scan_before,
+       seq_tup_read AS seq_tup_read_before,
+       n_tup_ins AS n_tup_ins_before,
+       n_live_tup AS n_live_tup_before
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush_rollback' \gset
+
+BEGIN;
+SET LOCAL stats_fetch_consistency = none;
+
+-- Generate both non-transactional (scan) and transactional (insert) activity.
+SELECT count(*) FROM partial_flush_rollback;
+INSERT INTO partial_flush_rollback SELECT generate_series(51, 100);
+
+-- Flush mid-transaction.  The scans are published; the insert and its
+-- live-tuple delta are deferred.
+SELECT pg_stat_force_next_flush();
+
+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_live_tup - :n_live_tup_before AS n_live_tup_delta
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush_rollback';
+
+ROLLBACK;
+
+-- After rollback the scans persist, since they were already in shared stats,
+-- while the inserts are discarded and n_live_tup is unchanged.
+SELECT pg_stat_force_next_flush();
+SELECT seq_scan - :seq_scan_before AS seq_scan_delta,
+       seq_tup_read - :seq_tup_read_before AS seq_tup_read_delta,
+       n_live_tup - :n_live_tup_before AS n_live_tup_delta
+  FROM pg_stat_user_tables WHERE relname = 'partial_flush_rollback';
+
+DROP TABLE partial_flush_rollback;
+
+--
+-- 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 mid-transaction.  The scan is published; everything else 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 mid-transaction.  The scan is published; 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 zeros live/dead, then only post-truncate DML counts.
+-- delta_live is inserted minus deleted, 10 - 3 = 7.
+-- delta_dead is updated plus deleted, 0 + 3 = 3.
+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.  A mid-transaction flush must not clear the
+-- pending counts that the transaction-local 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 two calls are visible in both
+-- the shared view and the transaction-local view after the flush.  The flush
+-- must not clear the transaction-local 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 85d989f395d..689e0164013 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2343,8 +2343,10 @@ PgStat_Counter
 PgStat_EntryRef
 PgStat_EntryRefHashEntry
 PgStat_FetchConsistency
+PgStat_FlushResult
 PgStat_FunctionCallUsage
 PgStat_FunctionCounts
+PgStat_FunctionStatus
 PgStat_HashKey
 PgStat_IO
 PgStat_KindInfo
-- 
2.50.1 (Apple Git-155)

