From f25c54069d9eeb7610c6be81cc3f7e0ca31364ad 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 v4 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.

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                  |   9 +-
 src/backend/utils/activity/pgstat.c           |  88 +++-
 src/backend/utils/activity/pgstat_backend.c   |   2 +-
 src/backend/utils/activity/pgstat_database.c  |  11 +-
 src/backend/utils/activity/pgstat_function.c  |  52 +-
 src/backend/utils/activity/pgstat_io.c        |   5 +-
 src/backend/utils/activity/pgstat_lock.c      |   5 +-
 src/backend/utils/activity/pgstat_relation.c  | 161 +++++--
 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/pg_proc.dat               |   2 +-
 src/include/pgstat.h                          |  13 +
 src/include/utils/pgstat_internal.h           |  72 ++-
 .../test_custom_stats/t/001_custom_stats.pl   |  34 ++
 .../test_custom_var_stats--1.0.sql            |  10 +
 .../test_custom_stats/test_custom_var_stats.c | 190 +++++++-
 src/test/regress/expected/stats.out           | 445 +++++++++++++++++-
 src/test/regress/sql/stats.sql                | 296 +++++++++++-
 src/tools/pgindent/typedefs.list              |   1 +
 20 files changed, 1287 insertions(+), 132 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 099e9b6f4e9..bd08c338a0c 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>
 
diff --git a/src/backend/utils/activity/pgstat.c b/src/backend/utils/activity/pgstat.c
index 50cd07822b4..72d350d4d2c 100644
--- a/src/backend/utils/activity/pgstat.c
+++ b/src/backend/utils/activity/pgstat.c
@@ -219,6 +219,11 @@ PgStat_LocalState pgStatLocal;
  */
 bool		pgstat_report_fixed = false;
 
+/*
+ * Request an additional pass over the pending list, used by flush_pending_cb.
+ */
+bool		pgStatPendingFlushExtra = false;
+
 /* ----------
  * Local data
  *
@@ -326,7 +331,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 +721,8 @@ 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, while a
+ * forced flush may also happen within one (e.g. pg_stat_force_next_flush()).
  */
 long
 pgstat_report_stat(bool force)
@@ -727,9 +732,10 @@ pgstat_report_stat(bool force)
 	bool		partial_flush;
 	TimestampTz now;
 	bool		nowait;
+	bool		xact_boundary;
 
 	pgstat_assert_is_up();
-	Assert(!IsTransactionOrTransactionBlock());
+	Assert(force || !IsTransactionOrTransactionBlock());
 
 	/* "absorb" the forced flush even if there's nothing to flush */
 	if (pgStatForceNextFlush)
@@ -789,6 +795,7 @@ pgstat_report_stat(bool force)
 
 	/* don't wait for lock acquisition when !force */
 	nowait = !force;
+	xact_boundary = !IsTransactionOrTransactionBlock();
 
 	partial_flush = false;
 
@@ -807,7 +814,7 @@ 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, xact_boundary);
 		}
 	}
 
@@ -815,12 +822,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 +850,9 @@ pgstat_report_stat(bool force)
 void
 pgstat_force_next_flush(void)
 {
+	if (IsTransactionOrTransactionBlock())
+		pgstat_report_stat(true);
+
 	pgStatForceNextFlush = true;
 }
 
@@ -1401,6 +1412,8 @@ pgstat_flush_pending_entries(bool nowait)
 	bool		have_pending = false;
 	dlist_node *cur = NULL;
 
+	pgStatPendingFlushExtra = false;
+
 	/*
 	 * 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
@@ -1409,6 +1422,12 @@ pgstat_flush_pending_entries(bool nowait)
 	 * entry in each iteration from the list if we flushed successfully.
 	 *
 	 * So we just keep track of the next pointer in each loop iteration.
+	 *
+	 * When flushing mid-transaction, a successfully flushed entry is retained
+	 * rather than deleted, since callers may hold pointers to it and later
+	 * flush more into it.  Because such entries stay on the list, a callback
+	 * can accumulate into a dependent entry that was already visited; those
+	 * are re-flushed by a second scan below (see pgStatPendingFlushExtra).
 	 */
 	if (!dlist_is_empty(&pgStatPending))
 		cur = dlist_head_node(&pgStatPending);
@@ -1420,16 +1439,25 @@ 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 = !IsTransactionOrTransactionBlock();
 		dlist_node *next;
 
 		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);
+		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 */
 		if (dlist_has_next(&pgStatPending, cur))
@@ -1437,8 +1465,11 @@ pgstat_flush_pending_entries(bool nowait)
 		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;
@@ -1446,6 +1477,37 @@ pgstat_flush_pending_entries(bool nowait)
 		cur = next;
 	}
 
+	/*
+	 * Second scan (see above) for dependent entries populated after they were
+	 * already visited.
+	 */
+	while (pgStatPendingFlushExtra && IsTransactionOrTransactionBlock())
+	{
+		bool		xact_boundary = false;
+
+		pgStatPendingFlushExtra = false;
+
+		cur = NULL;
+		if (!dlist_is_empty(&pgStatPending))
+			cur = dlist_head_node(&pgStatPending);
+
+		while (cur)
+		{
+			PgStat_EntryRef *entry_ref =
+				dlist_container(PgStat_EntryRef, pending_node, cur);
+			const PgStat_KindInfo *kind_info =
+				pgstat_get_kind_info(entry_ref->shared_entry->key.kind);
+			dlist_node *next;
+
+			next = dlist_has_next(&pgStatPending, cur) ?
+				dlist_next_node(&pgStatPending, cur) : NULL;
+
+			kind_info->flush_pending_cb(entry_ref, nowait, xact_boundary);
+
+			cur = next;
+		}
+	}
+
 	Assert(dlist_is_empty(&pgStatPending) == !have_pending);
 
 	return have_pending;
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..b5816dbd8ae 100644
--- a/src/backend/utils/activity/pgstat_database.c
+++ b/src/backend/utils/activity/pgstat_database.c
@@ -432,10 +432,11 @@ 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.
  */
-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 +445,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 +497,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..8883a73eb39 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)
@@ -119,12 +119,12 @@ pgstat_init_function_usage(FunctionCallInfo fcinfo,
 		}
 	}
 
-	pending = entry_ref->pending;
+	pending = (PgStat_FunctionStatus *) 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,45 @@ 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.
+ *
+ * This always returns PGSTAT_FLUSH_DONE.
+ *
+ * An entry flushed mid-transaction is retained and can be flushed more than
+ * once.  To avoid double counting, we add only the counts accumulated since
+ * the previous flush.
  */
-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	delta_total;
+	instr_time	delta_self;
 
-	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);
+	shfuncent->stats.numcalls +=
+		localent->counts.numcalls - localent->flushed.numcalls;
+
+	delta_total = localent->counts.total_time;
+	INSTR_TIME_SUBTRACT(delta_total, localent->flushed.total_time);
+	shfuncent->stats.total_time += INSTR_TIME_GET_MICROSEC(delta_total);
+
+	delta_self = localent->counts.self_time;
+	INSTR_TIME_SUBTRACT(delta_self, localent->flushed.self_time);
+	shfuncent->stats.self_time += INSTR_TIME_GET_MICROSEC(delta_self);
 
 	pgstat_unlock_entry(entry_ref);
 
-	return true;
+	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_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..8bb3e29530d 100644
--- a/src/backend/utils/activity/pgstat_relation.c
+++ b/src/backend/utils/activity/pgstat_relation.c
@@ -808,98 +808,155 @@ 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 (tuples_inserted/updated/deleted and the derived
+ * live/dead/mod counts) depend on the transaction outcome.  When flushing
+ * mid-transaction, they are held back and only the non-transactional counters
+ * are flushed, and the callback returns PGSTAT_FLUSH_PARTIAL.  Otherwise all
+ * counters are flushed and the callback returns PGSTAT_FLUSH_DONE.
+ *
+ * A retained entry can be flushed more than once, so each flush sends only the
+ * counts accumulated since the previous flush.
  *
  * 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).
 	 */
-	if (pg_memory_is_all_zeros(&lstats->counts,
-							   sizeof(struct PgStat_TableCounts)))
-		return true;
+	flush_txn = (xact_boundary || lstats->trans == NULL);
+
+	/*
+	 * Ignore entries with nothing new since the last flush, such as indexes
+	 * that were opened by the planner but not used.  Since counts and flushed
+	 * are both zero-initialized, a byte compare safely detects new counts.
+	 *
+	 * When flushing mid-transaction, keep such an entry pending if it still
+	 * has active transaction state, since commit will merge counters into it.
+	 */
+	if (memcmp(&lstats->counts, &lstats->flushed, sizeof(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 flushed 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 transactional counters only when it is safe to do so.
+	 *
+	 * These counters are grouped because they have cross-counter invariants:
+	 * hot_updated + newpage_updated <= updated, ins_since_vacuum tracks
+	 * tuples_inserted, and live/dead/changed tuples are derived from
+	 * inserted/updated/deleted.  Flushing them separately could expose
+	 * inconsistent intermediate states (e.g. hot_updated > 0 with updated ==
+	 * 0 after an abort).
 	 */
-	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
+		 * and the flushed baseline for delta counters.
+		 */
+		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;
+			lstats->flushed.changed_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;
+		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;
+	}
+
+	if (!pg_memory_is_all_zeros(dbentry, sizeof(*dbentry)))
+		pgStatPendingFlushExtra = true;
+
+	/*
+	 * Record what was flushed.  Transactional counters are retained until the
+	 * transaction boundary.
+	 */
+	if (flush_txn)
+	{
+		lstats->flushed = lstats->counts;
+		return PGSTAT_FLUSH_DONE;
+	}
+
+	/* Partial flush: record only the non-transactional counters as flushed */
+	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/pg_proc.dat b/src/include/catalog/pg_proc.dat
index f8a021987b5..5d971e42a90 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -6220,7 +6220,7 @@
 { oid => '2137',
   descr => 'statistics: force stats to be flushed after the next commit',
   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..c7606ef38de 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -90,6 +90,17 @@ typedef struct PgStat_FunctionCounts
 	instr_time	self_time;
 } PgStat_FunctionCounts;
 
+/*
+ * Pending function stats stored in PgStat_EntryRef->pending.
+ * Keeps cumulative counts (read by pg_stat_xact_user_functions) and a
+ * flushed baseline so that flush callbacks send only the delta.
+ */
+typedef struct PgStat_FunctionStatus
+{
+	PgStat_FunctionCounts counts;
+	PgStat_FunctionCounts flushed;
+} PgStat_FunctionStatus;
+
 /*
  * Working state needed to accumulate per-function-call timing statistics.
  */
@@ -183,6 +194,8 @@ 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 */
+	PgStat_TableCounts flushed; /* already-flushed baseline for delta
+								 * computation */
 	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..4c7d662ba36 100644
--- a/src/include/utils/pgstat_internal.h
+++ b/src/include/utils/pgstat_internal.h
@@ -225,6 +225,34 @@ typedef struct PgStat_SubXactStatus
 } PgStat_SubXactStatus;
 
 
+/*
+ * Result of a flush_pending_cb call, describing what happened to the pending
+ * entry.  The caller combines this with the flush context to decide whether the
+ * entry can be removed from the pending list.
+ */
+typedef enum PgStat_FlushResult
+{
+	/*
+	 * The lock could not be acquired without waiting (nowait was true).  The
+	 * entry must stay pending and be retried later.
+	 *
+	 * Must be first; in previous versions flush_pending_cb returned a bool
+	 * where false meant lock conflict.
+	 */
+	PGSTAT_FLUSH_LOCK_CONFLICT = 0,
+
+	/* Fully flushed; the entry can be removed from the pending list. */
+	PGSTAT_FLUSH_DONE,
+
+	/*
+	 * Only part of the entry was flushed; some state was intentionally
+	 * retained because it cannot be flushed in the current context (e.g.
+	 * transactional counters flushed mid-transaction).  The entry must stay
+	 * pending and be flushed again at a suitable boundary.
+	 */
+	PGSTAT_FLUSH_PARTIAL,
+}			PgStat_FlushResult;
+
 /*
  * Metadata for a specific kind of statistics.
  */
@@ -297,8 +325,19 @@ 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 entry's lock cannot be acquired without
+	 * waiting, the callback must return PGSTAT_FLUSH_LOCK_CONFLICT without
+	 * flushing.  xact_boundary tells the callback whether it is safe to flush
+	 * transactional state: it is true when called outside of a transaction
+	 * (e.g. at transaction end), and false when called mid-transaction.  When
+	 * xact_boundary is false, a callback holding state whose outcome depends
+	 * on the transaction must flush only its non-transactional counters,
+	 * retain the rest, and return PGSTAT_FLUSH_PARTIAL.  Otherwise it flushes
+	 * everything and returns PGSTAT_FLUSH_DONE.
 	 */
-	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.
@@ -368,7 +407,7 @@ typedef struct PgStat_KindInfo
 	 * "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 +750,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 +782,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 +791,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 +802,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 +811,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 +825,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 +874,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 +885,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 +895,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);
 
 
@@ -885,6 +928,13 @@ extern void pgstat_create_transactional(PgStat_Kind kind, Oid dboid, uint64 obji
  */
 extern PGDLLIMPORT bool pgstat_report_fixed;
 
+/*
+ * Set by flush_pending_cb to request an additional pass over the pending list
+ * while flushing mid-transaction, when a callback accumulates data into a
+ * dependent entry that was already visited in the current pass.
+ */
+extern bool pgStatPendingFlushExtra;
+
 /* Backend-local stats state */
 extern PGDLLIMPORT PgStat_LocalState pgStatLocal;
 
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..e60be88d1be 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
@@ -168,5 +168,39 @@ $result = $node->safe_psql('postgres',
 );
 is($result, "0", "report of fixed-sized after manual reset");
 
+# Test cascade flush (A -> B -> C).  A's flush accumulates into B, B's flush
+# accumulates into C, and pgStatPendingFlushExtra drives the extra passes.
+# Seed B and C on the pending list before A to exercise the ordering case
+# where dependent entries are visited before the entry that produces them.
+$node->safe_psql('postgres',
+	q(select test_custom_stats_var_create('cascade_test', 'cascade test')));
+
+my $b_before = $node->safe_psql('postgres',
+	q(SELECT coalesce(test_cascade_get_count('B'), 0)));
+my $c_before = $node->safe_psql('postgres',
+	q(SELECT coalesce(test_cascade_get_count('C'), 0)));
+
+$node->safe_psql('postgres', qq(
+	SET stats_fetch_consistency = none;
+	BEGIN;
+	SELECT test_cascade_seed('B');
+	SELECT test_cascade_seed('C');
+	SELECT test_custom_stats_var_update('cascade_test');
+	SELECT test_custom_stats_var_update('cascade_test');
+	SELECT test_custom_stats_var_update('cascade_test');
+	SELECT pg_stat_force_next_flush();
+	COMMIT;
+));
+
+my $b_after = $node->safe_psql('postgres',
+	q(SELECT test_cascade_get_count('B')));
+my $c_after = $node->safe_psql('postgres',
+	q(SELECT test_cascade_get_count('C')));
+
+is($b_after - $b_before, 3,
+	"cascade B receives 3 from A");
+is($c_after - $c_before, 3,
+	"cascade C receives 3 from B");
+
 # Test completed successfully
 done_testing();
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..4577c261b3c 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
@@ -24,3 +24,13 @@ CREATE FUNCTION test_custom_stats_var_report(INOUT name TEXT,
 RETURNS SETOF record
 AS 'MODULE_PATHNAME', 'test_custom_stats_var_report'
 LANGUAGE C STRICT PARALLEL UNSAFE;
+
+CREATE FUNCTION test_cascade_get_count(IN which TEXT)
+RETURNS BIGINT
+AS 'MODULE_PATHNAME', 'test_cascade_get_count'
+LANGUAGE C STRICT PARALLEL UNSAFE;
+
+CREATE FUNCTION test_cascade_seed(IN which TEXT)
+RETURNS void
+AS 'MODULE_PATHNAME', 'test_cascade_seed'
+LANGUAGE C STRICT PARALLEL UNSAFE;
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..80e05762a7f 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
@@ -43,6 +43,13 @@ PG_MODULE_MAGIC_EXT(
  */
 #define PGSTAT_KIND_TEST_CUSTOM_VAR_STATS 25
 
+/*
+ * Kind IDs for cascade flush test (A -> B -> C).
+ * Tests that pgStatPendingFlushExtra handles multi-level dependencies.
+ */
+#define PGSTAT_KIND_CASCADE_B 27
+#define PGSTAT_KIND_CASCADE_C 28
+
 /* File paths for auxiliary data serialization */
 #define TEST_CUSTOM_AUX_DATA_DESC "pg_stat/test_custom_var_stats_desc.stats"
 
@@ -70,6 +77,18 @@ typedef struct PgStatShared_CustomVarEntry
 	dsa_pointer description;	/* pointer to description string in DSA */
 } PgStatShared_CustomVarEntry;
 
+/* Pending and shared types for cascade kinds B and C */
+typedef struct PgStat_CascadeEntry
+{
+	PgStat_Counter count;
+}			PgStat_CascadeEntry;
+
+typedef struct PgStatShared_CascadeEntry
+{
+	PgStatShared_Common header;
+	PgStat_CascadeEntry stats;
+}			PgStatShared_CascadeEntry;
+
 /*--------------------------------------------------------------------------
  * Global Variables
  *--------------------------------------------------------------------------
@@ -90,8 +109,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,
@@ -127,6 +147,91 @@ static const PgStat_KindInfo custom_stats = {
 	.finish = test_custom_stats_var_finish,
 };
 
+/*
+ * cascade_b_flush_cb
+ *
+ * Flushes to shared memory and accumulates into cascade C.
+ */
+static PgStat_FlushResult
+cascade_b_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary)
+{
+	PgStat_CascadeEntry *pending;
+	PgStatShared_CascadeEntry *shared;
+	PgStat_EntryRef *c_ref;
+	PgStat_CascadeEntry *c_pending;
+
+	pending = (PgStat_CascadeEntry *) entry_ref->pending;
+	shared = (PgStatShared_CascadeEntry *) entry_ref->shared_stats;
+
+	if (pending->count == 0)
+		return PGSTAT_FLUSH_DONE;
+
+	if (!pgstat_lock_entry(entry_ref, nowait))
+		return PGSTAT_FLUSH_LOCK_CONFLICT;
+
+	shared->stats.count += pending->count;
+	pgstat_unlock_entry(entry_ref);
+
+	/* Accumulate into cascade C */
+	c_ref = pgstat_prep_pending_entry(PGSTAT_KIND_CASCADE_C, InvalidOid, 1, NULL);
+	c_pending = (PgStat_CascadeEntry *) c_ref->pending;
+	c_pending->count += pending->count;
+
+	pgStatPendingFlushExtra = true;
+
+	pending->count = 0;
+	return PGSTAT_FLUSH_DONE;
+}
+
+/*
+ * cascade_c_flush_cb
+ *
+ * Terminal node.  Flushes to shared memory only.
+ */
+static PgStat_FlushResult
+cascade_c_flush_cb(PgStat_EntryRef *entry_ref, bool nowait, bool xact_boundary)
+{
+	PgStat_CascadeEntry *pending;
+	PgStatShared_CascadeEntry *shared;
+
+	pending = (PgStat_CascadeEntry *) entry_ref->pending;
+	shared = (PgStatShared_CascadeEntry *) entry_ref->shared_stats;
+
+	if (pending->count == 0)
+		return PGSTAT_FLUSH_DONE;
+
+	if (!pgstat_lock_entry(entry_ref, nowait))
+		return PGSTAT_FLUSH_LOCK_CONFLICT;
+
+	shared->stats.count += pending->count;
+	pgstat_unlock_entry(entry_ref);
+
+	pending->count = 0;
+	return PGSTAT_FLUSH_DONE;
+}
+
+static const PgStat_KindInfo cascade_b_stats = {
+	.name = "cascade_b",
+	.fixed_amount = false,
+	.accessed_across_databases = true,
+	.shared_size = sizeof(PgStatShared_CascadeEntry),
+	.shared_data_off = offsetof(PgStatShared_CascadeEntry, stats),
+	.shared_data_len = sizeof(((PgStatShared_CascadeEntry *) 0)->stats),
+	.pending_size = sizeof(PgStat_CascadeEntry),
+	.flush_pending_cb = cascade_b_flush_cb,
+};
+
+static const PgStat_KindInfo cascade_c_stats = {
+	.name = "cascade_c",
+	.fixed_amount = false,
+	.accessed_across_databases = true,
+	.shared_size = sizeof(PgStatShared_CascadeEntry),
+	.shared_data_off = offsetof(PgStatShared_CascadeEntry, stats),
+	.shared_data_len = sizeof(((PgStatShared_CascadeEntry *) 0)->stats),
+	.pending_size = sizeof(PgStat_CascadeEntry),
+	.flush_pending_cb = cascade_c_flush_cb,
+};
+
 /*--------------------------------------------------------------------------
  * Module initialization
  *--------------------------------------------------------------------------
@@ -135,8 +240,9 @@ static const PgStat_KindInfo custom_stats = {
 void
 _PG_init(void)
 {
-	/* Register custom statistics kind */
 	pgstat_register_kind(PGSTAT_KIND_TEST_CUSTOM_VAR_STATS, &custom_stats);
+	pgstat_register_kind(PGSTAT_KIND_CASCADE_B, &cascade_b_stats);
+	pgstat_register_kind(PGSTAT_KIND_CASCADE_C, &cascade_c_stats);
 }
 
 /*--------------------------------------------------------------------------
@@ -151,26 +257,43 @@ _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;
+	PgStat_EntryRef *b_ref;
+	PgStat_CascadeEntry *b_pending;
 
 	pending_entry = (PgStat_StatCustomVarEntry *) entry_ref->pending;
 	shared_entry = (PgStatShared_CustomVarEntry *) entry_ref->shared_stats;
 
+	if (pending_entry->numcalls == 0)
+		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;
 
 	pgstat_unlock_entry(entry_ref);
 
-	return true;
+	/* Accumulate into cascade B */
+	b_ref = pgstat_prep_pending_entry(PGSTAT_KIND_CASCADE_B, InvalidOid, 1, NULL);
+	b_pending = (PgStat_CascadeEntry *) b_ref->pending;
+	b_pending->count += pending_entry->numcalls;
+
+	pgStatPendingFlushExtra = true;
+
+	memset(pending_entry, 0, sizeof(*pending_entry));
+
+	return PGSTAT_FLUSH_DONE;
 }
 
 /*
@@ -701,3 +824,54 @@ test_custom_stats_var_report(PG_FUNCTION_ARGS)
 
 	SRF_RETURN_DONE(funcctx);
 }
+
+/*
+ * test_cascade_get_count
+ *		Read the count from cascade B or C shared stats.
+ */
+PG_FUNCTION_INFO_V1(test_cascade_get_count);
+Datum
+test_cascade_get_count(PG_FUNCTION_ARGS)
+{
+	char	   *which = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	PgStat_Kind kind;
+	PgStat_CascadeEntry *entry;
+
+	if (strcmp(which, "B") == 0)
+		kind = PGSTAT_KIND_CASCADE_B;
+	else if (strcmp(which, "C") == 0)
+		kind = PGSTAT_KIND_CASCADE_C;
+	else
+		ereport(ERROR, (errmsg("argument must be 'B' or 'C'")));
+
+	entry = (PgStat_CascadeEntry *)
+		pgstat_fetch_entry(kind, InvalidOid, 1, NULL);
+
+	if (!entry)
+		PG_RETURN_INT64(0);
+
+	PG_RETURN_INT64(entry->count);
+}
+
+/*
+ * test_cascade_seed
+ *		Put cascade B or C on the pending list with zero data.
+ */
+PG_FUNCTION_INFO_V1(test_cascade_seed);
+Datum
+test_cascade_seed(PG_FUNCTION_ARGS)
+{
+	char	   *which = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	PgStat_Kind kind;
+
+	if (strcmp(which, "B") == 0)
+		kind = PGSTAT_KIND_CASCADE_B;
+	else if (strcmp(which, "C") == 0)
+		kind = PGSTAT_KIND_CASCADE_C;
+	else
+		ereport(ERROR, (errmsg("argument must be 'B' or 'C'")));
+
+	pgstat_prep_pending_entry(kind, InvalidOid, 1, NULL);
+
+	PG_RETURN_VOID();
+}
diff --git a/src/test/regress/expected/stats.out b/src/test/regress/expected/stats.out
index e230356de13..bdbaae0940c 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,443 @@ 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 into it, rather than a flush behind.
+--
+CREATE TABLE partial_flush_db(id int) WITH (autovacuum_enabled = off);
+INSERT INTO partial_flush_db SELECT generate_series(1, 100);
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+SELECT pg_stat_get_db_tuples_returned(oid) AS db_tup_returned_before
+  FROM pg_database WHERE datname = current_database() \gset
+BEGIN;
+SET LOCAL stats_fetch_consistency = none;
+SELECT count(*) FROM partial_flush_db;
+ count 
+-------
+   100
+(1 row)
+
+SELECT pg_stat_force_next_flush();
+ pg_stat_force_next_flush 
+--------------------------
+ 
+(1 row)
+
+-- The aggregate must have advanced by at least the relation's own scan in
+-- this same flush.  It also counts catalog scans, so this is not an equality.
+SELECT tup_returned - :db_tup_returned_before >= seq_tup_read AS db_kept_up
+  FROM pg_stat_database, pg_stat_user_tables
+  WHERE datname = current_database() AND relname = 'partial_flush_db';
+ db_kept_up 
+------------
+ t
+(1 row)
+
+COMMIT;
+DROP TABLE partial_flush_db;
+--
+-- 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 or double-count function call statistics.
+--
+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 already visible
+-- in-transaction after the flush.
+SELECT funcname, calls FROM pg_stat_user_functions WHERE funcid = :flush_func_test_oid;
+    funcname     | calls 
+-----------------+-------
+ flush_func_test |     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..92e6f566942 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,299 @@ 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 into it, rather than a flush behind.
+--
+CREATE TABLE partial_flush_db(id int) WITH (autovacuum_enabled = off);
+INSERT INTO partial_flush_db SELECT generate_series(1, 100);
+SELECT pg_stat_force_next_flush();
+
+SELECT pg_stat_get_db_tuples_returned(oid) AS db_tup_returned_before
+  FROM pg_database WHERE datname = current_database() \gset
+
+BEGIN;
+SET LOCAL stats_fetch_consistency = none;
+
+SELECT count(*) FROM partial_flush_db;
+
+SELECT pg_stat_force_next_flush();
+
+-- The aggregate must have advanced by at least the relation's own scan in
+-- this same flush.  It also counts catalog scans, so this is not an equality.
+SELECT tup_returned - :db_tup_returned_before >= seq_tup_read AS db_kept_up
+  FROM pg_stat_database, pg_stat_user_tables
+  WHERE datname = current_database() AND relname = 'partial_flush_db';
+
+COMMIT;
+
+DROP TABLE partial_flush_db;
+
+--
+-- 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 or double-count function call statistics.
+--
+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 already visible
+-- in-transaction after the flush.
+SELECT funcname, calls FROM pg_stat_user_functions WHERE funcid = :flush_func_test_oid;
+
+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..a880610cec2 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2345,6 +2345,7 @@ PgStat_EntryRefHashEntry
 PgStat_FetchConsistency
 PgStat_FunctionCallUsage
 PgStat_FunctionCounts
+PgStat_FunctionStatus
 PgStat_HashKey
 PgStat_IO
 PgStat_KindInfo
-- 
2.47.3

