v26-0001-Machinery-for-grabbing-an-extended-vacuum-statistics.patch

From f96d5079774fe129fff32761bba4ab9089e491bd Mon Sep 17 00:00:00 2001
From: Alena Rybakina <a.rybakina@postgrespro.ru>
Date: Tue, 9 Dec 2025 09:56:34 +0300
Subject: [PATCH 1/5] Machinery for grabbing an extended vacuum statistics on
 table relations.

Value of total_blks_hit, total_blks_read, total_blks_dirtied are number of
hitted, missed and dirtied pages in shared buffers during a vacuum operation
respectively.

total_blks_dirtied means 'dirtied only by this action'. So, if this page was
dirty before the vacuum operation, it doesn't count this page as 'dirtied'.

The tuples_deleted parameter is the number of tuples cleaned up by the vacuum
operation.

The delay_time value means total vacuum sleep time in vacuum delay point.
The pages_removed value is the number of pages by which the physical data
storage of the relation was reduced.
The value of pages_deleted parameter is the number of freed pages in the table
(file size may not have changed).

Tracking of IO during an (auto)vacuum operation.
Introduced variables blk_read_time and blk_write_time tracks only access to
buffer pages and flushing them to disk. Reading operation is trivial, but
writing measurement technique is not obvious.
So, during a vacuum writing time can be zero incremented because no any flushing
operations were performed.

System time and user time are parameters that describes how much time a vacuum
operation has spent in executing of code in user space and kernel space
accordingly. Also, accumulate total time of a vacuum that is a diff between
timestamps in start and finish points in the vacuum code.
Remember about idle time, when vacuum waited for IO and locks, so total time
isn't equal a sum of user and system time, but no less.

pages_frozen is a number of pages that are marked as frozen in vm during vacuum.
This parameter is incremented if page is marked as all-frozen.
pages_all_visible is a number of pages that are marked as all-visible in vm during
vacuum.

wraparound_failsafe_count is a number of times when the vacuum starts urgent cleanup
to prevent wraparound problem which is critical for the database.

Authors: Alena Rybakina <lena.ribackina@yandex.ru>,
	 Andrei Lepikhov <a.lepikhov@postgrespro.ru>,
	 Andrei Zubkov <a.zubkov@postgrespro.ru>
Reviewed-by: Dilip Kumar <dilipbalaut@gmail.com>, Masahiko Sawada <sawada.mshk@gmail.com>,
	     Ilia Evdokimov <ilya.evdokimov@tantorlabs.com>, jian he <jian.universality@gmail.com>,
	     Kirill Reshke <reshkekirill@gmail.com>, Alexander Korotkov <aekorotkov@gmail.com>,
	     Jim Nasby <jnasby@upgrade.com>, Sami Imseih <samimseih@gmail.com>,
	     Karina Litskevich <litskevichkarina@gmail.com>
---
 src/backend/access/heap/vacuumlazy.c          | 145 ++++-
 src/backend/access/heap/visibilitymap.c       |  10 +
 src/backend/catalog/system_views.sql          |  52 +-
 src/backend/commands/vacuum.c                 |   4 +
 src/backend/commands/vacuumparallel.c         |   1 +
 src/backend/utils/activity/pgstat_relation.c  |  46 +-
 src/backend/utils/adt/pgstatfuncs.c           |  86 +++
 src/backend/utils/misc/postgresql.conf.sample |   1 +
 src/include/catalog/pg_proc.dat               |  18 +
 src/include/commands/vacuum.h                 |   1 +
 src/include/pgstat.h                          |  92 ++-
 .../vacuum-extending-in-repetable-read.out    |  53 ++
 src/test/isolation/isolation_schedule         |   1 +
 .../vacuum-extending-in-repetable-read.spec   |  53 ++
 .../t/050_vacuum_extending_basic_test.pl      | 571 ++++++++++++++++++
 .../t/051_vacuum_extending_freeze_test.pl     | 395 ++++++++++++
 src/test/regress/expected/rules.out           |  44 +-
 src/test/regress/parallel_schedule            |   2 +-
 18 files changed, 1565 insertions(+), 10 deletions(-)
 create mode 100644 src/test/isolation/expected/vacuum-extending-in-repetable-read.out
 create mode 100644 src/test/isolation/specs/vacuum-extending-in-repetable-read.spec
 create mode 100644 src/test/recovery/t/050_vacuum_extending_basic_test.pl
 create mode 100644 src/test/recovery/t/051_vacuum_extending_freeze_test.pl

diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 30778a15639..66e09d0a0cf 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -289,6 +289,7 @@ typedef struct LVRelState
 	/* Error reporting state */
 	char	   *dbname;
 	char	   *relnamespace;
+	Oid			reloid;
 	char	   *relname;
 	char	   *indname;		/* Current index name */
 	BlockNumber blkno;			/* used only for heap operations */
@@ -407,6 +408,10 @@ typedef struct LVRelState
 	 * been permanently disabled.
 	 */
 	BlockNumber eager_scan_remaining_fails;
+
+	int32		wraparound_failsafe_count;	/* number of emergency vacuums to
+											 * prevent anti-wraparound
+											 * shutdown */
 } LVRelState;
 
 
@@ -418,6 +423,18 @@ typedef struct LVSavedErrInfo
 	VacErrPhase phase;
 } LVSavedErrInfo;
 
+/*
+ * Counters and usage data for extended stats tracking.
+ */
+typedef struct LVExtStatCounters
+{
+	TimestampTz starttime;
+	WalUsage	walusage;
+	BufferUsage bufusage;
+	double		VacuumDelayTime;
+	PgStat_Counter blocks_fetched;
+	PgStat_Counter blocks_hit;
+}			LVExtStatCounters;
 
 /* non-export function prototypes */
 static void lazy_scan_heap(LVRelState *vacrel);
@@ -487,6 +504,102 @@ static void update_vacuum_error_info(LVRelState *vacrel,
 static void restore_vacuum_error_info(LVRelState *vacrel,
 									  const LVSavedErrInfo *saved_vacrel);
 
+/* ----------
+ * extvac_stats_start() -
+ *
+ * Save cut-off values of extended vacuum counters before start of a relation
+ * processing.
+ * ----------
+ */
+static void
+extvac_stats_start(Relation rel, LVExtStatCounters * counters)
+{
+	TimestampTz starttime;
+
+	memset(counters, 0, sizeof(LVExtStatCounters));
+
+	starttime = GetCurrentTimestamp();
+
+	counters->starttime = starttime;
+	counters->walusage = pgWalUsage;
+	counters->bufusage = pgBufferUsage;
+	counters->VacuumDelayTime = VacuumDelayTime;
+	counters->blocks_fetched = 0;
+	counters->blocks_hit = 0;
+
+	if (!rel->pgstat_info || !pgstat_track_counts)
+
+		/*
+		 * if something goes wrong or user doesn't want to track a database
+		 * activity - just suppress it.
+		 */
+		return;
+
+	counters->blocks_fetched = rel->pgstat_info->counts.blocks_fetched;
+	counters->blocks_hit = rel->pgstat_info->counts.blocks_hit;
+}
+
+/* ----------
+ * extvac_stats_end() -
+ *
+ *	Called to finish an extended vacuum statistic gathering and form a report.
+ * ----------
+ */
+static void
+extvac_stats_end(Relation rel, LVExtStatCounters * counters,
+				 ExtVacReport * report)
+{
+	WalUsage	walusage;
+	BufferUsage bufusage;
+	TimestampTz endtime;
+	long		secs;
+	int			usecs;
+
+	/* Calculate diffs of global stat parameters on WAL and buffer usage. */
+	memset(&walusage, 0, sizeof(WalUsage));
+	WalUsageAccumDiff(&walusage, &pgWalUsage, &counters->walusage);
+
+	memset(&bufusage, 0, sizeof(BufferUsage));
+	BufferUsageAccumDiff(&bufusage, &pgBufferUsage, &counters->bufusage);
+
+	endtime = GetCurrentTimestamp();
+	TimestampDifference(counters->starttime, endtime, &secs, &usecs);
+
+	memset(report, 0, sizeof(ExtVacReport));
+
+	/*
+	 * Fill additional statistics on a vacuum processing operation.
+	 */
+	report->total_blks_read = bufusage.local_blks_read + bufusage.shared_blks_read;
+	report->total_blks_hit = bufusage.local_blks_hit + bufusage.shared_blks_hit;
+	report->total_blks_dirtied = bufusage.local_blks_dirtied + bufusage.shared_blks_dirtied;
+	report->total_blks_written = bufusage.shared_blks_written;
+
+	report->wal_records = walusage.wal_records;
+	report->wal_fpi = walusage.wal_fpi;
+	report->wal_bytes = walusage.wal_bytes;
+
+	report->blk_read_time = INSTR_TIME_GET_MILLISEC(bufusage.local_blk_read_time);
+	report->blk_read_time += INSTR_TIME_GET_MILLISEC(bufusage.shared_blk_read_time);
+	report->blk_write_time = INSTR_TIME_GET_MILLISEC(bufusage.local_blk_write_time);
+	report->blk_write_time = INSTR_TIME_GET_MILLISEC(bufusage.shared_blk_write_time);
+	report->delay_time = VacuumDelayTime - counters->VacuumDelayTime;
+
+	report->total_time = secs * 1000. + usecs / 1000.;
+
+	if (!rel->pgstat_info || !pgstat_track_counts)
+
+		/*
+		 * if something goes wrong or an user doesn't want to track a database
+		 * activity - just suppress it.
+		 */
+		return;
+
+	report->blks_fetched =
+		rel->pgstat_info->counts.blocks_fetched - counters->blocks_fetched;
+	report->blks_hit =
+		rel->pgstat_info->counts.blocks_hit - counters->blocks_hit;
+}
 
 
 /*
@@ -645,6 +758,13 @@ heap_vacuum_rel(Relation rel, const VacuumParams params,
 	BufferUsage startbufferusage = pgBufferUsage;
 	ErrorContextCallback errcallback;
 	char	  **indnames = NULL;
+	LVExtStatCounters extVacCounters;
+	ExtVacReport extVacReport;
+	ExtVacReport allzero;
+
+	/* Initialize vacuum statistics */
+	memset(&allzero, 0, sizeof(ExtVacReport));
+	extVacReport = allzero;
 
 	verbose = (params.options & VACOPT_VERBOSE) != 0;
 	instrument = (verbose || (AmAutoVacuumWorkerProcess() &&
@@ -673,6 +793,8 @@ heap_vacuum_rel(Relation rel, const VacuumParams params,
 		pgstat_progress_update_param(PROGRESS_VACUUM_STARTED_BY,
 									 PROGRESS_VACUUM_STARTED_BY_MANUAL);
 
+	extvac_stats_start(rel, &extVacCounters);
+
 	/*
 	 * Setup error traceback support for ereport() first.  The idea is to set
 	 * up an error context callback to display additional information on any
@@ -689,6 +811,7 @@ heap_vacuum_rel(Relation rel, const VacuumParams params,
 	vacrel->dbname = get_database_name(MyDatabaseId);
 	vacrel->relnamespace = get_namespace_name(RelationGetNamespace(rel));
 	vacrel->relname = pstrdup(RelationGetRelationName(rel));
+	vacrel->reloid = RelationGetRelid(rel);
 	vacrel->indname = NULL;
 	vacrel->phase = VACUUM_ERRCB_PHASE_UNKNOWN;
 	vacrel->verbose = verbose;
@@ -797,6 +920,7 @@ heap_vacuum_rel(Relation rel, const VacuumParams params,
 	vacrel->aggressive = vacuum_get_cutoffs(rel, params, &vacrel->cutoffs);
 	vacrel->rel_pages = orig_rel_pages = RelationGetNumberOfBlocks(rel);
 	vacrel->vistest = GlobalVisTestFor(rel);
+	vacrel->wraparound_failsafe_count = 0;
 
 	/* Initialize state used to track oldest extant XID/MXID */
 	vacrel->NewRelfrozenXid = vacrel->cutoffs.OldestXmin;
@@ -951,6 +1075,23 @@ heap_vacuum_rel(Relation rel, const VacuumParams params,
 						vacrel->NewRelfrozenXid, vacrel->NewRelminMxid,
 						&frozenxid_updated, &minmulti_updated, false);
 
+	/* Make generic extended vacuum stats report */
+	extvac_stats_end(rel, &extVacCounters, &extVacReport);
+
+	/* Fill heap-specific extended stats fields */
+	extVacReport.pages_scanned = vacrel->scanned_pages;
+	extVacReport.pages_removed = vacrel->removed_pages;
+	extVacReport.vm_new_frozen_pages = vacrel->vm_new_frozen_pages;
+	extVacReport.vm_new_visible_pages = vacrel->vm_new_visible_pages;
+	extVacReport.vm_new_visible_frozen_pages = vacrel->vm_new_visible_frozen_pages;
+	extVacReport.tuples_deleted = vacrel->tuples_deleted;
+	extVacReport.tuples_frozen = vacrel->tuples_frozen;
+	extVacReport.recently_dead_tuples = vacrel->recently_dead_tuples;
+	extVacReport.missed_dead_tuples = vacrel->missed_dead_tuples;
+	extVacReport.missed_dead_pages = vacrel->missed_dead_pages;
+	extVacReport.index_vacuum_count = vacrel->num_index_scans;
+	extVacReport.wraparound_failsafe_count = vacrel->wraparound_failsafe_count;
+
 	/*
 	 * Report results to the cumulative stats system, too.
 	 *
@@ -965,7 +1106,8 @@ heap_vacuum_rel(Relation rel, const VacuumParams params,
 						 Max(vacrel->new_live_tuples, 0),
 						 vacrel->recently_dead_tuples +
 						 vacrel->missed_dead_tuples,
-						 starttime);
+						 starttime,
+						 &extVacReport);
 	pgstat_progress_end_command();
 
 	if (instrument)
@@ -3019,6 +3161,7 @@ lazy_check_wraparound_failsafe(LVRelState *vacrel)
 		int64		progress_val[3] = {0, 0, PROGRESS_VACUUM_MODE_FAILSAFE};
 
 		VacuumFailsafeActive = true;
+		vacrel->wraparound_failsafe_count++;
 
 		/*
 		 * Abandon use of a buffer access strategy to allow use of all of
diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c
index d14588e92ae..3030242d98e 100644
--- a/src/backend/access/heap/visibilitymap.c
+++ b/src/backend/access/heap/visibilitymap.c
@@ -92,6 +92,7 @@
 #include "access/xloginsert.h"
 #include "access/xlogutils.h"
 #include "miscadmin.h"
+#include "pgstat.h"
 #include "port/pg_bitutils.h"
 #include "storage/bufmgr.h"
 #include "storage/smgr.h"
@@ -161,6 +162,15 @@ visibilitymap_clear(Relation rel, BlockNumber heapBlk, Buffer vmbuf, uint8 flags
 
 	if (map[mapByte] & mask)
 	{
+		/*
+		 * As part of vacuum stats, track how often all-visible or all-frozen
+		 * bits are cleared.
+		 */
+		if (map[mapByte] >> mapOffset & flags & VISIBILITYMAP_ALL_VISIBLE)
+			pgstat_count_vm_rev_all_visible(rel);
+		if (map[mapByte] >> mapOffset & flags & VISIBILITYMAP_ALL_FROZEN)
+			pgstat_count_vm_rev_all_frozen(rel);
+
 		map[mapByte] &= ~mask;
 
 		MarkBufferDirty(vmbuf);
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 0a0f95f6bb9..ffb407d414f 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -727,7 +727,9 @@ CREATE VIEW pg_stat_all_tables AS
             pg_stat_get_total_autovacuum_time(C.oid) AS total_autovacuum_time,
             pg_stat_get_total_analyze_time(C.oid) AS total_analyze_time,
             pg_stat_get_total_autoanalyze_time(C.oid) AS total_autoanalyze_time,
-            pg_stat_get_stat_reset_time(C.oid) AS stats_reset
+            pg_stat_get_stat_reset_time(C.oid) AS stats_reset,
+            pg_stat_get_rev_all_frozen_pages(C.oid) AS rev_all_frozen_pages,
+            pg_stat_get_rev_all_visible_pages(C.oid) AS rev_all_visible_pages
     FROM pg_class C LEFT JOIN
          pg_index I ON C.oid = I.indrelid
          LEFT JOIN pg_namespace N ON (N.oid = C.relnamespace)
@@ -1452,3 +1454,51 @@ REVOKE ALL ON pg_aios FROM PUBLIC;
 GRANT SELECT ON pg_aios TO pg_read_all_stats;
 REVOKE EXECUTE ON FUNCTION pg_get_aios() FROM PUBLIC;
 GRANT EXECUTE ON FUNCTION pg_get_aios() TO pg_read_all_stats;
+--
+-- Show extended cumulative statistics on a vacuum operation over all tables and
+-- databases of the instance.
+-- Use Invalid Oid "0" as an input relation id to get stat on each table in a
+-- database.
+--
+
+CREATE VIEW pg_stat_vacuum_tables AS
+SELECT
+  ns.nspname AS schemaname,
+  rel.relname AS relname,
+  stats.relid as relid,
+
+  stats.total_blks_read AS total_blks_read,
+  stats.total_blks_hit AS total_blks_hit,
+  stats.total_blks_dirtied AS total_blks_dirtied,
+  stats.total_blks_written AS total_blks_written,
+
+  stats.rel_blks_read AS rel_blks_read,
+  stats.rel_blks_hit AS rel_blks_hit,
+
+  stats.pages_scanned AS pages_scanned,
+  stats.pages_removed AS pages_removed,
+  stats.vm_new_frozen_pages AS vm_new_frozen_pages,
+  stats.vm_new_visible_pages AS vm_new_visible_pages,
+  stats.vm_new_visible_frozen_pages AS vm_new_visible_frozen_pages,
+  stats.missed_dead_pages AS missed_dead_pages,
+  stats.tuples_deleted AS tuples_deleted,
+  stats.tuples_frozen AS tuples_frozen,
+  stats.recently_dead_tuples AS recently_dead_tuples,
+  stats.missed_dead_tuples AS missed_dead_tuples,
+
+  stats.wraparound_failsafe AS wraparound_failsafe,
+  stats.index_vacuum_count AS index_vacuum_count,
+  stats.wal_records AS wal_records,
+  stats.wal_fpi AS wal_fpi,
+  stats.wal_bytes AS wal_bytes,
+
+  stats.blk_read_time AS blk_read_time,
+  stats.blk_write_time AS blk_write_time,
+
+  stats.delay_time AS delay_time,
+  stats.total_time AS total_time
+
+FROM pg_class rel
+  JOIN pg_namespace ns ON ns.oid = rel.relnamespace,
+  LATERAL pg_stat_get_vacuum_tables(rel.oid) stats
+WHERE rel.relkind = 'r';
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 0528d1b6ecb..dd519447387 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -117,6 +117,9 @@ pg_atomic_uint32 *VacuumSharedCostBalance = NULL;
 pg_atomic_uint32 *VacuumActiveNWorkers = NULL;
 int			VacuumCostBalanceLocal = 0;
 
+/* Cumulative storage to report total vacuum delay time. */
+double		VacuumDelayTime = 0;	/* msec. */
+
 /* non-export function prototypes */
 static List *expand_vacuum_rel(VacuumRelation *vrel,
 							   MemoryContext vac_context, int options);
@@ -2536,6 +2539,7 @@ vacuum_delay_point(bool is_analyze)
 			exit(1);
 
 		VacuumCostBalance = 0;
+		VacuumDelayTime += msec;
 
 		/*
 		 * Balance and update limit values for autovacuum workers. We must do
diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c
index 8a37c08871a..114cd7c31d3 100644
--- a/src/backend/commands/vacuumparallel.c
+++ b/src/backend/commands/vacuumparallel.c
@@ -1054,6 +1054,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	/* Set cost-based vacuum delay */
 	VacuumUpdateCosts();
 	VacuumCostBalance = 0;
+	VacuumDelayTime = 0;
 	VacuumCostBalanceLocal = 0;
 	VacuumSharedCostBalance = &(shared->cost_balance);
 	VacuumActiveNWorkers = &(shared->active_nworkers);
diff --git a/src/backend/utils/activity/pgstat_relation.c b/src/backend/utils/activity/pgstat_relation.c
index 55a10c299db..361713479e8 100644
--- a/src/backend/utils/activity/pgstat_relation.c
+++ b/src/backend/utils/activity/pgstat_relation.c
@@ -47,6 +47,8 @@ static void add_tabstat_xact_level(PgStat_TableStatus *pgstat_info, int nest_lev
 static void ensure_tabstat_xact_level(PgStat_TableStatus *pgstat_info);
 static void save_truncdrop_counters(PgStat_TableXactStatus *trans, bool is_drop);
 static void restore_truncdrop_counters(PgStat_TableXactStatus *trans);
+static void pgstat_accumulate_extvac_stats(ExtVacReport * dst, ExtVacReport * src,
+										   bool accumulate_reltype_specific_info);
 
 
 /*
@@ -208,7 +210,7 @@ pgstat_drop_relation(Relation rel)
  */
 void
 pgstat_report_vacuum(Relation rel, PgStat_Counter livetuples,
-					 PgStat_Counter deadtuples, TimestampTz starttime)
+					 PgStat_Counter deadtuples, TimestampTz starttime, ExtVacReport * params)
 {
 	PgStat_EntryRef *entry_ref;
 	PgStatShared_Relation *shtabentry;
@@ -234,6 +236,8 @@ pgstat_report_vacuum(Relation rel, PgStat_Counter livetuples,
 	tabentry->live_tuples = livetuples;
 	tabentry->dead_tuples = deadtuples;
 
+	pgstat_accumulate_extvac_stats(&tabentry->vacuum_ext, params, true);
+
 	/*
 	 * It is quite possible that a non-aggressive VACUUM ended up skipping
 	 * various pages, however, we'll zero the insert counter here regardless.
@@ -880,6 +884,9 @@ pgstat_relation_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
 	tabentry->blocks_fetched += lstats->counts.blocks_fetched;
 	tabentry->blocks_hit += lstats->counts.blocks_hit;
 
+	tabentry->rev_all_frozen_pages += lstats->counts.rev_all_frozen_pages;
+	tabentry->rev_all_visible_pages += lstats->counts.rev_all_visible_pages;
+
 	/* Clamp live_tuples in case of negative delta_live_tuples */
 	tabentry->live_tuples = Max(tabentry->live_tuples, 0);
 	/* Likewise for dead_tuples */
@@ -1009,3 +1016,40 @@ restore_truncdrop_counters(PgStat_TableXactStatus *trans)
 		trans->tuples_deleted = trans->deleted_pre_truncdrop;
 	}
 }
+
+static void
+pgstat_accumulate_extvac_stats(ExtVacReport * dst, ExtVacReport * src,
+							   bool accumulate_reltype_specific_info)
+{
+	dst->total_blks_read += src->total_blks_read;
+	dst->total_blks_hit += src->total_blks_hit;
+	dst->total_blks_dirtied += src->total_blks_dirtied;
+	dst->total_blks_written += src->total_blks_written;
+	dst->wal_bytes += src->wal_bytes;
+	dst->wal_fpi += src->wal_fpi;
+	dst->wal_records += src->wal_records;
+	dst->blk_read_time += src->blk_read_time;
+	dst->blk_write_time += src->blk_write_time;
+	dst->delay_time += src->delay_time;
+	dst->total_time += src->total_time;
+
+	if (!accumulate_reltype_specific_info)
+		return;
+
+	dst->blks_fetched += src->blks_fetched;
+	dst->blks_hit += src->blks_hit;
+
+	dst->pages_scanned += src->pages_scanned;
+	dst->pages_removed += src->pages_removed;
+	dst->vm_new_frozen_pages += src->vm_new_frozen_pages;
+	dst->vm_new_visible_pages += src->vm_new_visible_pages;
+	dst->vm_new_visible_frozen_pages += src->vm_new_visible_frozen_pages;
+	dst->tuples_deleted += src->tuples_deleted;
+	dst->tuples_frozen += src->tuples_frozen;
+	dst->recently_dead_tuples += src->recently_dead_tuples;
+	dst->index_vacuum_count += src->index_vacuum_count;
+	dst->wraparound_failsafe_count += src->wraparound_failsafe_count;
+	dst->missed_dead_pages += src->missed_dead_pages;
+	dst->missed_dead_tuples += src->missed_dead_tuples;
+
+}
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index ef6fffe60b9..d7dfda0c1a7 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -106,6 +106,12 @@ PG_STAT_GET_RELENTRY_INT64(tuples_updated)
 /* pg_stat_get_vacuum_count */
 PG_STAT_GET_RELENTRY_INT64(vacuum_count)
 
+/* pg_stat_get_rev_frozen_pages */
+PG_STAT_GET_RELENTRY_INT64(rev_all_frozen_pages)
+
+/* pg_stat_get_rev_all_visible_pages */
+PG_STAT_GET_RELENTRY_INT64(rev_all_visible_pages)
+
 #define PG_STAT_GET_RELENTRY_FLOAT8(stat)						\
 Datum															\
 CppConcat(pg_stat_get_,stat)(PG_FUNCTION_ARGS)					\
@@ -2307,3 +2313,83 @@ pg_stat_have_stats(PG_FUNCTION_ARGS)
 
 	PG_RETURN_BOOL(pgstat_have_entry(kind, dboid, objid));
 }
+
+
+/*
+ * Get the vacuum statistics for the heap tables.
+ */
+Datum
+pg_stat_get_vacuum_tables(PG_FUNCTION_ARGS)
+{
+#define PG_STAT_GET_VACUUM_TABLES_STATS_COLS 26
+
+	Oid			relid = PG_GETARG_OID(0);
+	PgStat_StatTabEntry *tabentry;
+	ExtVacReport *extvacuum;
+	TupleDesc	tupdesc;
+	Datum		values[PG_STAT_GET_VACUUM_TABLES_STATS_COLS] = {0};
+	bool		nulls[PG_STAT_GET_VACUUM_TABLES_STATS_COLS] = {0};
+	char		buf[256];
+	int			i = 0;
+
+	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+		elog(ERROR, "return type must be a row type");
+
+	tabentry = pgstat_fetch_stat_tabentry(relid);
+
+	if (!tabentry)
+	{
+		InitMaterializedSRF(fcinfo, 0);
+		PG_RETURN_VOID();
+	}
+	else
+	{
+		extvacuum = &(tabentry->vacuum_ext);
+	}
+
+	i = 0;
+
+	values[i++] = ObjectIdGetDatum(relid);
+
+	values[i++] = Int64GetDatum(extvacuum->total_blks_read);
+	values[i++] = Int64GetDatum(extvacuum->total_blks_hit);
+	values[i++] = Int64GetDatum(extvacuum->total_blks_dirtied);
+	values[i++] = Int64GetDatum(extvacuum->total_blks_written);
+
+	values[i++] = Int64GetDatum(extvacuum->blks_fetched -
+								extvacuum->blks_hit);
+	values[i++] = Int64GetDatum(extvacuum->blks_hit);
+
+	values[i++] = Int64GetDatum(extvacuum->pages_scanned);
+	values[i++] = Int64GetDatum(extvacuum->pages_removed);
+	values[i++] = Int64GetDatum(extvacuum->vm_new_frozen_pages);
+	values[i++] = Int64GetDatum(extvacuum->vm_new_visible_pages);
+	values[i++] = Int64GetDatum(extvacuum->vm_new_visible_frozen_pages);
+	values[i++] = Int64GetDatum(extvacuum->missed_dead_pages);
+	values[i++] = Int64GetDatum(extvacuum->tuples_deleted);
+	values[i++] = Int64GetDatum(extvacuum->tuples_frozen);
+	values[i++] = Int64GetDatum(extvacuum->recently_dead_tuples);
+	values[i++] = Int64GetDatum(extvacuum->missed_dead_tuples);
+	values[i++] = Int32GetDatum(extvacuum->wraparound_failsafe_count);
+	values[i++] = Int64GetDatum(extvacuum->index_vacuum_count);
+
+	values[i++] = Int64GetDatum(extvacuum->wal_records);
+	values[i++] = Int64GetDatum(extvacuum->wal_fpi);
+
+	/* Convert to numeric, like pg_stat_statements */
+	snprintf(buf, sizeof buf, UINT64_FORMAT, extvacuum->wal_bytes);
+	values[i++] = DirectFunctionCall3(numeric_in,
+									  CStringGetDatum(buf),
+									  ObjectIdGetDatum(0),
+									  Int32GetDatum(-1));
+
+	values[i++] = Float8GetDatum(extvacuum->blk_read_time);
+	values[i++] = Float8GetDatum(extvacuum->blk_write_time);
+	values[i++] = Float8GetDatum(extvacuum->delay_time);
+	values[i++] = Float8GetDatum(extvacuum->total_time);
+
+	Assert(i == PG_STAT_GET_VACUUM_TABLES_STATS_COLS);
+
+	/* Returns the record as Datum */
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
+}
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index dc9e2255f8a..867638fe74b 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -669,6 +669,7 @@
 #track_wal_io_timing = off
 #track_functions = none                 # none, pl, all
 #stats_fetch_consistency = cache        # cache, none, snapshot
+#track_vacuum_statistics = off
 
 
 # - Monitoring -
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fd9448ec7b9..915a5a7822f 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12612,4 +12612,22 @@
   proargnames => '{pid,io_id,io_generation,state,operation,off,length,target,handle_data_len,raw_result,result,target_desc,f_sync,f_localmem,f_buffered}',
   prosrc => 'pg_get_aios' },
 
+{ oid => '8001',
+  descr => 'pg_stat_get_vacuum_tables returns vacuum stats values for table',
+  proname => 'pg_stat_get_vacuum_tables', prorows => 1000, provolatile => 's', prorettype => 'record',proisstrict => 'f',
+  proretset => 't',
+  proargtypes => 'oid',
+  proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int4,int8,int8,int8,numeric,float8,float8,float8,float8}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{reloid,relid,total_blks_read,total_blks_hit,total_blks_dirtied,total_blks_written,rel_blks_read,rel_blks_hit,pages_scanned,pages_removed,vm_new_frozen_pages,vm_new_visible_pages,vm_new_visible_frozen_pages,missed_dead_pages,tuples_deleted,tuples_frozen,recently_dead_tuples,missed_dead_tuples,wraparound_failsafe,index_vacuum_count,wal_records,wal_fpi,wal_bytes,blk_read_time,blk_write_time,delay_time,total_time}',
+  prosrc => 'pg_stat_get_vacuum_tables' },
+
+  { oid => '8002', descr => 'statistics: number of times the all-visible pages in the visibility map was removed for pages of table',
+  proname => 'pg_stat_get_rev_all_visible_pages', provolatile => 's',
+  proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+  prosrc => 'pg_stat_get_rev_all_visible_pages' },
+  { oid => '8003', descr => 'statistics: number of times the all-frozen pages in the visibility map was removed for pages of table',
+  proname => 'pg_stat_get_rev_all_frozen_pages', provolatile => 's',
+  proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+  prosrc => 'pg_stat_get_rev_all_frozen_pages' },
 ]
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 1f3290c7fbf..6b997bc7fb1 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -332,6 +332,7 @@ extern PGDLLIMPORT double vacuum_max_eager_freeze_failure_rate;
 extern PGDLLIMPORT pg_atomic_uint32 *VacuumSharedCostBalance;
 extern PGDLLIMPORT pg_atomic_uint32 *VacuumActiveNWorkers;
 extern PGDLLIMPORT int VacuumCostBalanceLocal;
+extern PGDLLIMPORT double VacuumDelayTime;
 
 extern PGDLLIMPORT bool VacuumFailsafeActive;
 extern PGDLLIMPORT double vacuum_cost_delay;
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 6714363144a..46d12fa3bd0 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -114,6 +114,66 @@ typedef struct PgStat_BackendSubEntry
 	PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
 } PgStat_BackendSubEntry;
 
+/* ----------
+ *
+ * ExtVacReport
+ *
+ * Additional statistics of vacuum processing over a heap relation.
+ * pages_removed is the amount by which the physically shrank,
+ * if any (ie the change in its total size on disk)
+ * pages_deleted refer to free space within the index file
+ * ----------
+ */
+typedef struct ExtVacReport
+{
+	/*
+	 * number of blocks missed, hit, dirtied and written during a vacuum of
+	 * specific relation
+	 */
+	int64		total_blks_read;
+	int64		total_blks_hit;
+	int64		total_blks_dirtied;
+	int64		total_blks_written;
+
+	/*
+	 * blocks missed and hit for just the heap during a vacuum of specific
+	 * relation
+	 */
+	int64		blks_fetched;
+	int64		blks_hit;
+
+	/* Vacuum WAL usage stats */
+	int64		wal_records;	/* wal usage: number of WAL records */
+	int64		wal_fpi;		/* wal usage: number of WAL full page images
+								 * produced */
+	uint64		wal_bytes;		/* wal usage: size of WAL records produced */
+
+	/* Time stats. */
+	double		blk_read_time;	/* time spent reading pages, in msec */
+	double		blk_write_time; /* time spent writing pages, in msec */
+	double		delay_time;		/* how long vacuum slept in vacuum delay
+								 * point, in msec */
+	double		total_time;		/* total time of a vacuum operation, in msec */
+
+	int64		pages_scanned;	/* heap pages examined (not skipped by VM) */
+	int64		pages_removed;	/* heap pages removed by vacuum "truncation" */
+	int64		vm_new_frozen_pages;	/* pages marked in VM as frozen */
+	int64		vm_new_visible_pages;	/* pages marked in VM as all-visible */
+	int64		vm_new_visible_frozen_pages;	/* pages marked in VM as
+												 * all-visible and frozen */
+	int64		missed_dead_tuples; /* tuples not pruned by vacuum due to
+									 * failure to get a cleanup lock */
+	int64		missed_dead_pages;	/* pages with missed dead tuples */
+	int64		tuples_deleted; /* tuples deleted by vacuum */
+	int64		tuples_frozen;	/* tuples frozen up by vacuum */
+	int64		recently_dead_tuples;	/* deleted tuples that are still
+										 * visible to some transaction */
+	int64		index_vacuum_count; /* the number of index vacuumings */
+	int32		wraparound_failsafe_count;	/* number of emergency vacuums to
+											 * prevent anti-wraparound
+											 * shutdown */
+}			ExtVacReport;
+
 /* ----------
  * PgStat_TableCounts			The actual per-table counts kept by a backend
  *
@@ -156,6 +216,15 @@ typedef struct PgStat_TableCounts
 
 	PgStat_Counter blocks_fetched;
 	PgStat_Counter blocks_hit;
+
+	PgStat_Counter rev_all_visible_pages;
+	PgStat_Counter rev_all_frozen_pages;
+
+	/*
+	 * Additional cumulative stat on vacuum operations. Use an expensive
+	 * structure as an abstraction for different types of relations.
+	 */
+	ExtVacReport vacuum_ext;
 } PgStat_TableCounts;
 
 /* ----------
@@ -214,7 +283,7 @@ typedef struct PgStat_TableXactStatus
  * ------------------------------------------------------------
  */
 
-#define PGSTAT_FILE_FORMAT_ID	0x01A5BCBB
+#define PGSTAT_FILE_FORMAT_ID	0x01A5BCBC
 
 typedef struct PgStat_ArchiverStats
 {
@@ -378,6 +447,8 @@ typedef struct PgStat_StatDBEntry
 	PgStat_Counter parallel_workers_launched;
 
 	TimestampTz stat_reset_timestamp;
+
+	ExtVacReport vacuum_ext;	/* extended vacuum statistics */
 } PgStat_StatDBEntry;
 
 typedef struct PgStat_StatFuncEntry
@@ -461,8 +532,12 @@ typedef struct PgStat_StatTabEntry
 	PgStat_Counter total_autovacuum_time;
 	PgStat_Counter total_analyze_time;
 	PgStat_Counter total_autoanalyze_time;
-
 	TimestampTz stat_reset_time;
+
+	PgStat_Counter rev_all_visible_pages;
+	PgStat_Counter rev_all_frozen_pages;
+
+	ExtVacReport vacuum_ext;
 } PgStat_StatTabEntry;
 
 /* ------
@@ -671,7 +746,7 @@ extern void pgstat_unlink_relation(Relation rel);
 
 extern void pgstat_report_vacuum(Relation rel, PgStat_Counter livetuples,
 								 PgStat_Counter deadtuples,
-								 TimestampTz starttime);
+								 TimestampTz starttime, ExtVacReport * params);
 extern void pgstat_report_analyze(Relation rel,
 								  PgStat_Counter livetuples, PgStat_Counter deadtuples,
 								  bool resetcounter, TimestampTz starttime);
@@ -722,6 +797,17 @@ extern void pgstat_report_analyze(Relation rel,
 		if (pgstat_should_count_relation(rel))						\
 			(rel)->pgstat_info->counts.blocks_hit++;				\
 	} while (0)
+/* accumulate unfrozen all-visible and all-frozen pages */
+#define pgstat_count_vm_rev_all_visible(rel)						\
+	do {															\
+		if (pgstat_should_count_relation(rel))						\
+			(rel)->pgstat_info->counts.rev_all_visible_pages++;	\
+	} while (0)
+#define pgstat_count_vm_rev_all_frozen(rel)						\
+	do {															\
+		if (pgstat_should_count_relation(rel))						\
+			(rel)->pgstat_info->counts.rev_all_frozen_pages++;	\
+	} while (0)
 
 extern void pgstat_count_heap_insert(Relation rel, PgStat_Counter n);
 extern void pgstat_count_heap_update(Relation rel, bool hot, bool newpage);
diff --git a/src/test/isolation/expected/vacuum-extending-in-repetable-read.out b/src/test/isolation/expected/vacuum-extending-in-repetable-read.out
new file mode 100644
index 00000000000..87f7e40b4a6
--- /dev/null
+++ b/src/test/isolation/expected/vacuum-extending-in-repetable-read.out
@@ -0,0 +1,53 @@
+unused step name: s2_delete
+Parsed test spec with 2 sessions
+
+starting permutation: s2_insert s2_print_vacuum_stats_table s1_begin_repeatable_read s2_update s2_insert_interrupt s2_vacuum s2_print_vacuum_stats_table s1_commit s2_checkpoint s2_vacuum s2_print_vacuum_stats_table
+step s2_insert: INSERT INTO test_vacuum_stat_isolation(id, ival) SELECT ival, ival%10 FROM generate_series(1,1000) As ival;
+step s2_print_vacuum_stats_table: 
+    SELECT
+    vt.relname, vt.tuples_deleted, vt.recently_dead_tuples, vt.missed_dead_tuples, vt.missed_dead_pages, vt.tuples_frozen
+    FROM pg_stat_vacuum_tables vt, pg_class c
+    WHERE vt.relname = 'test_vacuum_stat_isolation' AND vt.relid = c.oid;
+
+relname                   |tuples_deleted|recently_dead_tuples|missed_dead_tuples|missed_dead_pages|tuples_frozen
+--------------------------+--------------+--------------------+------------------+-----------------+-------------
+test_vacuum_stat_isolation|             0|                   0|                 0|                0|            0
+(1 row)
+
+step s1_begin_repeatable_read: 
+  BEGIN transaction ISOLATION LEVEL REPEATABLE READ;
+  select count(ival) from test_vacuum_stat_isolation where id>900;
+
+count
+-----
+  100
+(1 row)
+
+step s2_update: UPDATE test_vacuum_stat_isolation SET ival = ival + 2 where id > 900;
+step s2_insert_interrupt: INSERT INTO test_vacuum_stat_isolation values (1,1);
+step s2_vacuum: VACUUM test_vacuum_stat_isolation;
+step s2_print_vacuum_stats_table: 
+    SELECT
+    vt.relname, vt.tuples_deleted, vt.recently_dead_tuples, vt.missed_dead_tuples, vt.missed_dead_pages, vt.tuples_frozen
+    FROM pg_stat_vacuum_tables vt, pg_class c
+    WHERE vt.relname = 'test_vacuum_stat_isolation' AND vt.relid = c.oid;
+
+relname                   |tuples_deleted|recently_dead_tuples|missed_dead_tuples|missed_dead_pages|tuples_frozen
+--------------------------+--------------+--------------------+------------------+-----------------+-------------
+test_vacuum_stat_isolation|             0|                 100|                 0|                0|            0
+(1 row)
+
+step s1_commit: COMMIT;
+step s2_checkpoint: CHECKPOINT;
+step s2_vacuum: VACUUM test_vacuum_stat_isolation;
+step s2_print_vacuum_stats_table: 
+    SELECT
+    vt.relname, vt.tuples_deleted, vt.recently_dead_tuples, vt.missed_dead_tuples, vt.missed_dead_pages, vt.tuples_frozen
+    FROM pg_stat_vacuum_tables vt, pg_class c
+    WHERE vt.relname = 'test_vacuum_stat_isolation' AND vt.relid = c.oid;
+
+relname                   |tuples_deleted|recently_dead_tuples|missed_dead_tuples|missed_dead_pages|tuples_frozen
+--------------------------+--------------+--------------------+------------------+-----------------+-------------
+test_vacuum_stat_isolation|           100|                 100|                 0|                0|          101
+(1 row)
+
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index f2e067b1fbc..1c231418706 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -98,6 +98,7 @@ test: timeouts
 test: vacuum-concurrent-drop
 test: vacuum-conflict
 test: vacuum-skip-locked
+test: vacuum-extending-in-repetable-read
 test: stats
 test: horizons
 test: predicate-hash
diff --git a/src/test/isolation/specs/vacuum-extending-in-repetable-read.spec b/src/test/isolation/specs/vacuum-extending-in-repetable-read.spec
new file mode 100644
index 00000000000..5893d89573d
--- /dev/null
+++ b/src/test/isolation/specs/vacuum-extending-in-repetable-read.spec
@@ -0,0 +1,53 @@
+# Test for checking recently_dead_tuples, tuples_deleted and frozen tuples in pg_stat_vacuum_tables.
+# recently_dead_tuples values are counted when vacuum hasn't cleared tuples because they were deleted recently.
+# recently_dead_tuples aren't increased after releasing lock compared with tuples_deleted, which increased
+# by the value of the cleared tuples that the vacuum managed to clear.
+
+setup
+{
+    CREATE TABLE test_vacuum_stat_isolation(id int, ival int) WITH (autovacuum_enabled = off);
+    SET track_io_timing = on;
+    SET track_vacuum_statistics TO 'on';
+}
+
+teardown
+{
+    DROP TABLE test_vacuum_stat_isolation CASCADE;
+    RESET track_io_timing;
+    RESET track_vacuum_statistics;
+}
+
+session s1
+step s1_begin_repeatable_read   {
+  BEGIN transaction ISOLATION LEVEL REPEATABLE READ;
+  select count(ival) from test_vacuum_stat_isolation where id>900;
+  }
+step s1_commit                  { COMMIT; }
+
+session s2
+step s2_insert                  { INSERT INTO test_vacuum_stat_isolation(id, ival) SELECT ival, ival%10 FROM generate_series(1,1000) As ival; }
+step s2_update                  { UPDATE test_vacuum_stat_isolation SET ival = ival + 2 where id > 900; }
+step s2_delete                  { DELETE FROM test_vacuum_stat_isolation where id > 900; }
+step s2_insert_interrupt        { INSERT INTO test_vacuum_stat_isolation values (1,1); }
+step s2_vacuum                  { VACUUM test_vacuum_stat_isolation; }
+step s2_checkpoint              { CHECKPOINT; }
+step s2_print_vacuum_stats_table
+{
+    SELECT
+    vt.relname, vt.tuples_deleted, vt.recently_dead_tuples, vt.missed_dead_tuples, vt.missed_dead_pages, vt.tuples_frozen
+    FROM pg_stat_vacuum_tables vt, pg_class c
+    WHERE vt.relname = 'test_vacuum_stat_isolation' AND vt.relid = c.oid;
+}
+
+permutation
+    s2_insert
+    s2_print_vacuum_stats_table
+    s1_begin_repeatable_read
+    s2_update
+    s2_insert_interrupt
+    s2_vacuum
+    s2_print_vacuum_stats_table
+    s1_commit
+    s2_checkpoint
+    s2_vacuum
+    s2_print_vacuum_stats_table
\ No newline at end of file
diff --git a/src/test/recovery/t/050_vacuum_extending_basic_test.pl b/src/test/recovery/t/050_vacuum_extending_basic_test.pl
new file mode 100644
index 00000000000..7e25a3fe63f
--- /dev/null
+++ b/src/test/recovery/t/050_vacuum_extending_basic_test.pl
@@ -0,0 +1,571 @@
+# Copyright (c) 2025 PostgreSQL Global Development Group
+# Test cumulative vacuum stats system using TAP
+#
+# This test validates the accuracy and behavior of cumulative vacuum statistics
+# across tables using:
+#
+#   • pg_stat_vacuum_tables
+#
+# A polling helper function repeatedly checks the stats views until expected
+# deltas appear or a configurable timeout expires. This guarantees that
+# stats-collector propagation delays do not lead to flaky test behavior.
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+#------------------------------------------------------------------------------
+# Test harness setup
+#------------------------------------------------------------------------------
+
+my $node = PostgreSQL::Test::Cluster->new('stat_vacuum');
+$node->init;
+
+# Configure the server logging level for the test
+$node->append_conf('postgresql.conf', q{
+    log_min_messages = notice
+});
+
+my $stderr;
+my $base_stats;
+my $wals;
+my $ibase_stats;
+my $iwals;
+
+$node->start(
+    '>' => \$base_stats,
+	'2>' => \$stderr
+);
+
+#------------------------------------------------------------------------------
+# Database creation and initialization
+#------------------------------------------------------------------------------
+
+$node->safe_psql('postgres', q{
+    CREATE DATABASE statistic_vacuum_database_regression;
+});
+# Main test database name and number of rows to insert
+my $dbname   = 'statistic_vacuum_database_regression';
+my $size_tab = 1000;
+
+# Enable required session settings and force the stats collector to flush next
+$node->safe_psql($dbname, q{
+    SET track_functions = 'all';
+    SELECT pg_stat_force_next_flush();
+});
+
+#------------------------------------------------------------------------------
+# Create test table and populate it
+#------------------------------------------------------------------------------
+
+$node->safe_psql(
+    $dbname,
+    "CREATE TABLE vestat (x int)
+         WITH (autovacuum_enabled = off, fillfactor = 10);
+     INSERT INTO vestat SELECT x FROM generate_series(1, $size_tab) AS g(x);
+     ANALYZE vestat;"
+);
+
+#------------------------------------------------------------------------------
+# Timing parameters for polling loops
+#------------------------------------------------------------------------------
+
+my $timeout    = 30;     # overall wait timeout in seconds
+my $interval   = 0.015;  # poll interval in seconds (15 ms)
+my $start_time = time();
+my $updated    = 0;
+
+#------------------------------------------------------------------------------
+# wait_for_vacuum_stats
+#
+# Polls pg_stat_vacuum_tables until the table-level counters exceed
+# the provided baselines, or until the configured timeout elapses.
+#
+# Expected named args (baseline values):
+#   tab_tuples_deleted
+#   tab_wal_records
+#
+# Returns: 1 if the condition is met before timeout, 0 otherwise.
+#------------------------------------------------------------------------------
+
+sub wait_for_vacuum_stats {
+    my (%args) = @_;
+    my $tab_tuples_deleted = $args{tab_tuples_deleted} or 0;
+    my $tab_wal_records    = $args{tab_wal_records} or 0;
+
+    my $start = time();
+    while ((time() - $start) < $timeout) {
+
+        my $result_query = $node->safe_psql(
+            $dbname,
+            "VACUUM vestat;
+             SELECT tuples_deleted > $tab_tuples_deleted AND wal_records > $tab_wal_records
+                  FROM pg_stat_vacuum_tables
+                  WHERE relname = 'vestat';"
+        );
+
+        return 1 if ($result_query eq 't');
+
+        sleep($interval);
+    }
+
+    return 0;
+}
+
+#------------------------------------------------------------------------------
+# Variables to hold vacuum-stat snapshots for later comparisons
+#------------------------------------------------------------------------------
+
+my $pages_frozen = 0;
+my $tuples_deleted = 0;
+my $pages_scanned = 0;
+my $pages_removed = 0;
+my $wal_records = 0;
+my $wal_bytes = 0;
+my $wal_fpi = 0;
+
+my $pages_frozen_prev = 0;
+my $tuples_deleted_prev = 0;
+my $pages_scanned_prev = 0;
+my $pages_removed_prev = 0;
+my $wal_records_prev = 0;
+my $wal_bytes_prev = 0;
+my $wal_fpi_prev = 0;
+
+#------------------------------------------------------------------------------
+# fetch_vacuum_stats
+#
+# Reads current values of relevant vacuum counters for the test table,
+# storing them in package variables for subsequent comparisons.
+#------------------------------------------------------------------------------
+
+sub fetch_vacuum_stats {
+    # fetch actual base vacuum statistics
+    my $base_statistics = $node->safe_psql(
+        $dbname,
+        "SELECT vm_new_frozen_pages, tuples_deleted, pages_scanned, pages_removed, wal_records, wal_bytes, wal_fpi
+           FROM pg_stat_vacuum_tables
+          WHERE relname = 'vestat';"
+    );
+
+    $base_statistics =~ s/\s*\|\s*/ /g;   # transform " | " into space
+    ($pages_frozen, $tuples_deleted, $pages_scanned, $pages_removed, $wal_records, $wal_bytes, $wal_fpi)
+        = split /\s+/, $base_statistics;
+}
+
+#------------------------------------------------------------------------------
+# save_vacuum_stats
+#
+# Save current values (previously fetched by fetch_vacuum_stats) so that we
+# later fetch new values and compare them.
+#------------------------------------------------------------------------------
+sub save_vacuum_stats {
+    $pages_frozen_prev = $pages_frozen;
+    $tuples_deleted_prev = $tuples_deleted;
+    $pages_scanned_prev = $pages_scanned;
+    $pages_removed_prev = $pages_removed;
+    $wal_records_prev = $wal_records;
+    $wal_bytes_prev = $wal_bytes;
+    $wal_fpi_prev = $wal_fpi;
+}
+
+#------------------------------------------------------------------------------
+# print_vacuum_stats_on_error
+#
+# Print values in case of an error
+#------------------------------------------------------------------------------
+sub print_vacuum_stats_on_error {
+    diag(
+            "Statistics in the failed test\n" .
+            "Table statistics:\n" .
+            "  Before test:\n" .
+            "    pages_frozen      = $pages_frozen_prev\n" .
+            "    tuples_deleted    = $tuples_deleted_prev\n" .
+            "    pages_scanned     = $pages_scanned_prev\n" .
+            "    pages_removed     = $pages_removed_prev\n" .
+            "    wal_records       = $wal_records_prev\n" .
+            "    wal_bytes         = $wal_bytes_prev\n" .
+            "    wal_fpi           = $wal_fpi_prev\n" .
+            "  After test:\n" .
+            "    pages_frozen      = $pages_frozen\n" .
+            "    tuples_deleted    = $tuples_deleted\n" .
+            "    pages_scanned     = $pages_scanned\n" .
+            "    pages_removed     = $pages_removed\n" .
+            "    wal_records       = $wal_records\n" .
+            "    wal_bytes         = $wal_bytes\n" .
+            "    wal_fpi           = $wal_fpi\n"
+    );
+};
+
+#------------------------------------------------------------------------------
+# fetch_vacuum_stats during mismatch
+#
+# Print current values and old values of relevant vacuum counters for the test
+# table, storing them in package variables for subsequent comparisons.
+#------------------------------------------------------------------------------
+
+sub fetch_error_base_tab_vacuum_statistics {
+
+    # fetch actual base vacuum statistics
+    my $base_statistics = $node->safe_psql(
+    $dbname,
+    "SELECT vm_new_frozen_pages, tuples_deleted, pages_scanned, pages_removed
+       FROM pg_stat_vacuum_tables
+      WHERE relname = 'vestat';"
+    );
+    $base_statistics =~ s/\s*\|\s*/ /g;   # transform " | " in space
+    my ($cur_pages_frozen, $cur_tuples_deleted, $cur_pages_scanned, $cur_pages_removed) = split /\s+/, $base_statistics;
+
+    diag(
+            "BASE STATS MISMATCH FOR TABLE:\n" .
+            "  Baseline:\n" .
+            "    pages_frozen      = $pages_frozen\n" .
+            "    tuples_deleted    = $tuples_deleted\n" .
+            "    pages_scanned     = $pages_scanned\n" .
+            "    pages_removed     = $pages_removed\n" .
+            "  Current:\n" .
+            "    pages_frozen      = $cur_pages_frozen\n" .
+            "    tuples_deleted    = $cur_tuples_deleted\n" .
+            "    pages_scanned     = $cur_pages_scanned\n" .
+            "    pages_removed     = $cur_pages_removed\n"
+    );
+}
+
+sub fetch_error_wal_tab_vacuum_statistics {
+
+    my $wal_raw = $node->safe_psql(
+        $dbname,
+        "SELECT wal_records, wal_bytes, wal_fpi
+        FROM pg_stat_vacuum_tables
+        WHERE relname = 'vestat';"
+    );
+
+    $wal_raw =~ s/\s*\|\s*/ /g;   # transform " | " in space
+    my ($cur_wal_rec, $cur_wal_bytes, $cur_wal_fpi) = split /\s+/, $wal_raw;
+
+    diag(
+            "WAL STATS MISMATCH FOR TABLE:\n" .
+            "  Baseline:\n" .
+            "    wal_records = $wal_records\n" .
+            "    wal_bytes   = $wal_bytes\n" .
+            "    wal_fpi     = $wal_fpi\n" .
+            "  Current:\n" .
+            "    wal_records = $cur_wal_rec\n" .
+            "    wal_bytes   = $cur_wal_bytes\n" .
+            "    wal_fpi     = $cur_wal_fpi\n"
+    );
+}
+
+#------------------------------------------------------------------------------
+# Test 1: Delete half the rows, run VACUUM, and wait for stats to advance
+#------------------------------------------------------------------------------
+subtest 'Test 1: Delete half the rows, run VACUUM, and wait for stats to advance' => sub
+{
+
+$node->safe_psql($dbname, "DELETE FROM vestat WHERE x % 2 = 0;");
+$node->safe_psql($dbname, "VACUUM vestat;");
+
+# Poll the stats view until expected deltas appear or timeout
+$updated = wait_for_vacuum_stats(
+    tab_tuples_deleted => 0,
+    tab_wal_records => 0
+);
+ok($updated, 'vacuum stats updated after vacuuming half-deleted table (tuples_deleted and wal_fpi advanced)')
+  or diag "Timeout waiting for pg_stats_vacuum_* update after $timeout seconds after vacuuming half-deleted table";
+
+#------------------------------------------------------------------------------
+# Check statistics after half-table delete
+#------------------------------------------------------------------------------
+
+# Get current statistics
+fetch_vacuum_stats();
+
+ok($pages_frozen == $pages_frozen_prev, 'table pages_frozen stay the same');
+ok($tuples_deleted > $tuples_deleted_prev, 'table tuples_deleted has increased');
+ok($pages_scanned > $pages_scanned_prev, 'table pages_scanned has increased');
+ok($pages_removed == $pages_removed_prev, 'table pages_removed stay the same');
+ok($wal_records > $wal_records_prev, 'table wal_records has increased');
+ok($wal_bytes > $wal_bytes_prev, 'table wal_bytes has increased');
+ok($wal_fpi > $wal_fpi_prev, 'table wal_fpi has increased');
+
+} or print_vacuum_stats_on_error(); # End of subtest
+
+# Save statistics for the next test
+save_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Test 2: Delete all rows, run VACUUM, and wait for stats to advance
+#------------------------------------------------------------------------------
+subtest 'Test 2: Delete all rows, run VACUUM, and wait for stats to advance' => sub
+{
+
+$node->safe_psql($dbname, "DELETE FROM vestat;");
+$node->safe_psql($dbname, "VACUUM vestat;");
+
+$updated = wait_for_vacuum_stats(
+    tab_tuples_deleted => $tuples_deleted_prev,
+    tab_wal_records => $wal_records_prev,
+);
+
+ok($updated, 'vacuum stats updated after vacuuming all-deleted table (tuples_deleted and wal_records advanced)')
+  or diag "Timeout waiting for pg_stats_vacuum_* update after $timeout seconds after vacuuming all-deleted table";
+
+#------------------------------------------------------------------------------
+# Check statistics after full delete
+#------------------------------------------------------------------------------
+
+# Get current statistics
+fetch_vacuum_stats();
+
+ok($pages_frozen == $pages_frozen_prev, 'table pages_frozen stay the same');
+ok($tuples_deleted > $tuples_deleted_prev, 'table tuples_deleted has increased');
+ok($pages_scanned > $pages_scanned_prev, 'table pages_scanned has increased');
+ok($pages_removed > $pages_removed_prev, 'table pages_removed has increased');
+ok($wal_records > $wal_records_prev, 'table wal_records has increased');
+ok($wal_bytes > $wal_bytes_prev, 'table wal_bytes has increased');
+ok($wal_fpi == $wal_fpi_prev, 'table wal_fpi stay the same');
+
+} or print_vacuum_stats_on_error(); # End of subtest
+
+# Save statistics for the next test
+save_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Test 3: Test VACUUM FULL — it should not report to the stats collector
+#------------------------------------------------------------------------------
+subtest 'Test 3: Test VACUUM FULL — it should not report to the stats collector' => sub
+{
+
+$node->safe_psql(
+    $dbname,
+    "INSERT INTO vestat SELECT x FROM generate_series(1, $size_tab) AS g(x);
+     CHECKPOINT;
+     DELETE FROM vestat;
+     VACUUM FULL vestat;"
+);
+
+# Get current statistics
+fetch_vacuum_stats();
+
+ok($pages_frozen == $pages_frozen_prev, 'table pages_frozen stay the same');
+ok($tuples_deleted == $tuples_deleted_prev, 'table tuples_deleted stay the same');
+ok($pages_scanned == $pages_scanned_prev, 'table pages_scanned stay the same');
+ok($pages_removed == $pages_removed_prev, 'table pages_removed stay the same');
+ok($wal_records == $wal_records_prev, 'table wal_records stay the same');
+ok($wal_bytes == $wal_bytes_prev, 'table wal_bytes stay the same');
+ok($wal_fpi == $wal_fpi_prev, 'table wal_fpi stay the same');
+
+} or print_vacuum_stats_on_error(); # End of subtest
+
+# Save statistics for the next test
+save_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Test 4: Update table, checkpoint, and VACUUM to provoke WAL/FPI accounting
+#------------------------------------------------------------------------------
+subtest 'Test 4: Update table, checkpoint, and VACUUM to provoke WAL/FPI accounting' => sub
+{
+
+$node->safe_psql(
+    $dbname,
+    "INSERT INTO vestat SELECT x FROM generate_series(1, $size_tab) AS g(x);
+     CHECKPOINT;
+     UPDATE vestat SET x = x + 1000;
+     VACUUM vestat;"
+);
+
+$updated = wait_for_vacuum_stats(
+    tab_tuples_deleted => $tuples_deleted,
+    tab_wal_records => $wal_records,
+);
+
+ok($updated, 'vacuum stats updated after updating tuples in the table (tuples_deleted and wal_records advanced)')
+  or diag "Timeout waiting for pg_stats_vacuum_* update after $timeout seconds";
+
+#------------------------------------------------------------------------------
+# Verify statistics after updating tuples and vacuuming
+#------------------------------------------------------------------------------
+
+# Get current statistics
+fetch_vacuum_stats();
+
+ok($pages_frozen == $pages_frozen_prev, 'table pages_frozen stay the same');
+ok($tuples_deleted > $tuples_deleted_prev, 'table tuples_deleted has increased');
+ok($pages_scanned > $pages_scanned_prev, 'table pages_scanned has increased');
+ok($pages_removed == $pages_removed_prev, 'table pages_removed stay the same');
+ok($wal_records > $wal_records_prev, 'table wal_records has increased');
+ok($wal_bytes > $wal_bytes_prev, 'table wal_bytes has increased');
+ok($wal_fpi > $wal_fpi_prev, 'table wal_fpi has increased');
+
+} or print_vacuum_stats_on_error(); # End of subtest
+
+# Save statistics for the next test
+save_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Test 5: Update table, trancate and vacuuming
+#------------------------------------------------------------------------------
+subtest 'Test 5: Update table, trancate and vacuuming' => sub
+{
+
+$node->safe_psql(
+    $dbname,
+    "INSERT INTO vestat SELECT x FROM generate_series(1, $size_tab) AS g(x);
+     UPDATE vestat SET x = x + 1000;"
+);
+$node->safe_psql($dbname, "TRUNCATE vestat;");
+$node->safe_psql($dbname, "CHECKPOINT;");
+$node->safe_psql($dbname, "VACUUM vestat;");
+
+$updated = wait_for_vacuum_stats(
+    tab_tuples_deleted => 0,
+    tab_wal_records => $wal_records_prev
+);
+
+ok($updated, 'vacuum stats updated after updating tuples and trancation in the table (tuples_deleted and wal_records advanced)')
+  or diag "Timeout waiting for pg_stats_vacuum_* update after $timeout seconds";
+
+#------------------------------------------------------------------------------
+# Verify statistics after updating full table, vacuum and trancation
+#------------------------------------------------------------------------------
+
+# Get current statistics
+fetch_vacuum_stats();
+
+ok($pages_frozen == $pages_frozen_prev, 'table pages_frozen stay the same');
+ok($tuples_deleted == $tuples_deleted_prev, 'table tuples_deleted stay the same');
+ok($pages_scanned == $pages_scanned_prev, 'table pages_scanned stay the same');
+ok($pages_removed == $pages_removed_prev, 'table pages_removed stay the same');
+ok($wal_records > $wal_records_prev, 'table wal_records has increased');
+ok($wal_bytes > $wal_bytes_prev, 'table wal_bytes has increased');
+ok($wal_fpi == $wal_fpi_prev, 'table wal_fpi stay the same');
+
+} or print_vacuum_stats_on_error(); # End of subtest
+
+# Save statistics for the next test
+save_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Test 6: Delete all tuples from table, trancate, and vacuuming
+#------------------------------------------------------------------------------
+subtest 'Test 6: Delete all tuples from table, trancate, and vacuuming' => sub
+{
+
+$node->safe_psql(
+    $dbname,
+    "INSERT INTO vestat SELECT x FROM generate_series(1, $size_tab) AS g(x);
+     DELETE FROM vestat;
+     TRUNCATE vestat;
+     CHECKPOINT;
+     VACUUM vestat;"
+);
+
+$updated = wait_for_vacuum_stats(
+    tab_tuples_deleted => 0,
+    tab_wal_records => $wal_records
+);
+
+ok($updated, 'vacuum stats updated after deleting all tuples and trancation in the table (tuples_deleted and wal_records advanced)')
+  or diag "Timeout waiting for pg_stats_vacuum_* update after $timeout seconds";
+
+#------------------------------------------------------------------------------
+# Verify statistics after table vacuum and trancation
+#------------------------------------------------------------------------------
+
+# Get current statistics
+fetch_vacuum_stats();
+
+ok($pages_frozen == $pages_frozen_prev, 'table pages_frozen stay the same');
+ok($tuples_deleted == $tuples_deleted_prev, 'table tuples_deleted stay the same');
+ok($pages_scanned == $pages_scanned_prev, 'table pages_scanned stay the same');
+ok($pages_removed == $pages_removed_prev, 'table pages_removed stay the same');
+ok($wal_records > $wal_records_prev, 'table wal_records has increased');
+ok($wal_bytes > $wal_bytes_prev, 'table wal_bytes has increased');
+ok($wal_fpi == $wal_fpi_prev, 'table wal_fpi stay the same');
+
+} or print_vacuum_stats_on_error(); # End of subtest
+
+# Save statistics for the next test
+save_vacuum_stats();
+
+#-------------------------------------------------------------------------------------------------------
+# Test 8: Check if we return single vacuum statistics for particular relation from the current database
+#-------------------------------------------------------------------------------------------------------
+
+my $dboid = $node->safe_psql(
+    $dbname,
+    "SELECT oid FROM pg_database WHERE datname = current_database();"
+);
+
+my $reloid = $node->safe_psql(
+    $dbname,
+    q{
+        SELECT oid FROM pg_class WHERE relname = 'vestat';
+    }
+);
+
+# Check if we can get vacuum statistics of particular heap elation in the current database
+$base_stats = $node->safe_psql(
+    $dbname,
+    "SELECT count(*) = 1 FROM pg_stat_get_vacuum_tables($reloid);"
+);
+ok($base_stats eq 't', 'heap vacuum stats return from the current relation and database as expected');
+
+#------------------------------------------------------------------------------
+# Test 9: Check relation-level vacuum statistics from another database
+#------------------------------------------------------------------------------
+
+$base_stats = $node->safe_psql(
+    'postgres',
+    "SELECT count(*) = 0
+     FROM pg_stat_vacuum_tables
+     WHERE relname = 'vestat';"
+);
+ok($base_stats eq 't', 'check the printing heap vacuum extended statistics from another database are not available');
+
+$reloid = $node->safe_psql(
+    $dbname,
+    q{
+        SELECT oid FROM pg_class WHERE relname = 'pg_shdepend';
+    }
+);
+
+# Check if we can get vacuum statistics for cluster relations (dbid = 0)
+$base_stats = $node->safe_psql(
+    $dbname,
+    qq{
+        SELECT count(*) = 1
+        FROM pg_stat_get_vacuum_tables($reloid);
+    }
+);
+
+is($base_stats, 't', 'vacuum stats for common heap objects available');
+
+#------------------------------------------------------------------------------
+# Test 11: Cleanup checks: ensure functions return empty sets for OID = 0
+#------------------------------------------------------------------------------
+
+$node->safe_psql($dbname, q{
+    DROP TABLE vestat CASCADE;
+    VACUUM;
+});
+
+# Check that we don't print vacuum statistics for deleted objects
+$base_stats = $node->safe_psql(
+    $dbname,
+    q{
+        SELECT COUNT(*) = 0
+          FROM pg_stat_vacuum_tables WHERE relid = 0;
+    }
+);
+ok($base_stats eq 't', 'pg_stat_vacuum_tables correctly returns no rows for OID = 0');
+
+$node->safe_psql('postgres',
+    "DROP DATABASE $dbname;"
+);
+
+$node->stop;
+
+done_testing();
diff --git a/src/test/recovery/t/051_vacuum_extending_freeze_test.pl b/src/test/recovery/t/051_vacuum_extending_freeze_test.pl
new file mode 100644
index 00000000000..a9b5d6cb739
--- /dev/null
+++ b/src/test/recovery/t/051_vacuum_extending_freeze_test.pl
@@ -0,0 +1,395 @@
+# Copyright (c) 2025 PostgreSQL Global Development Group
+#
+# Test cumulative vacuum stats system using TAP
+#
+# In short, this test validates the correctness and stability of cumulative
+# vacuum statistics accounting around freezing, visibility, and revision
+# tracking across multiple VACUUMs and backend operations.
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+#------------------------------------------------------------------------------
+# Test cluster setup
+#------------------------------------------------------------------------------
+
+my $node = PostgreSQL::Test::Cluster->new('ext_stat_vacuum');
+$node->init;
+
+# Configure the server for aggressive freezing behavior used by the test
+# These settings ensure that VACUUM always freezes pages aggressively:
+# - vacuum_freeze_min_age = 0: freeze tuples as soon as possible (no age requirement)
+# - vacuum_freeze_table_age = 0: always perform aggressive scan (scan all pages)
+# - vacuum_multixact_freeze_min_age = 0: freeze multixacts as soon as possible
+# - vacuum_multixact_freeze_table_age = 0: always perform aggressive scan for multixacts
+# - vacuum_max_eager_freeze_failure_rate = 1.0: enable aggressive eager scanning (100% of pages)
+# - vacuum_failsafe_age = 0: disable failsafe (for testing)
+# - vacuum_multixact_failsafe_age = 0: disable multixact failsafe (for testing)
+$node->append_conf('postgresql.conf', q{
+	log_min_messages = notice
+	vacuum_freeze_min_age = 0
+	vacuum_freeze_table_age = 0
+	vacuum_multixact_freeze_min_age = 0
+	vacuum_multixact_freeze_table_age = 0
+	vacuum_max_eager_freeze_failure_rate = 1.0
+	vacuum_failsafe_age = 0
+	vacuum_multixact_failsafe_age = 0
+});
+
+$node->start();
+
+#------------------------------------------------------------------------------
+# Database creation and initialization
+#------------------------------------------------------------------------------
+
+$node->safe_psql('postgres', q{
+	CREATE DATABASE statistic_vacuum_database_regression;
+});
+
+# Main test database name
+my $dbname = 'statistic_vacuum_database_regression';
+
+# Enable necessary settings and force the stats collector to flush next
+$node->safe_psql($dbname, q{
+    SET track_functions = 'all';
+    SELECT pg_stat_force_next_flush();
+});
+
+#------------------------------------------------------------------------------
+# Timing parameters for polling loops
+#------------------------------------------------------------------------------
+
+my $timeout    = 30;     # overall wait timeout in seconds
+my $interval   = 0.015;  # poll interval in seconds (15 ms)
+my $start_time = time();
+my $updated    = 0;
+
+# wait_for_vacuum_stats
+#
+# Polls pg_stat_vacuum_tables until the named columns exceed the provided
+# baseline values or until timeout.  Callers should pass:
+#
+#   tab_frozen_column           => 'vm_new_frozen_pages'   # column name (string) or 'rev_all_frozen_pages'
+#   tab_visible_column          => 'vm_new_visible_pages'  # column name (string) or 'rev_all_visible_pages'
+#   tab_all_frozen_pages_count  => 0                       # baseline numeric
+#   tab_all_visible_pages_count => 0                       # baseline numeric
+#   run_vacuum                  => 0 or 1                  # if true, run vacuum_sql before polling
+#
+# Returns: 1 if the condition is met before timeout, 0 otherwise.
+sub wait_for_vacuum_stats {
+    my (%args) = @_;
+
+    my $tab_frozen_column           = $args{tab_frozen_column};
+    my $tab_visible_column          = $args{tab_visible_column};
+    my $tab_all_frozen_pages_count  = $args{tab_all_frozen_pages_count};
+    my $tab_all_visible_pages_count = $args{tab_all_visible_pages_count};
+    my $run_vacuum                  = $args{run_vacuum} ? 1 : 0;
+    my $result_query;
+
+    my $start = time();
+    my $sql;
+
+    while ((time() - $start) < $timeout) {
+
+        if ($run_vacuum) {
+            $node->safe_psql($dbname, 'VACUUM (FREEZE, VERBOSE) vestat');
+            $sql = "
+                SELECT ($tab_frozen_column > $tab_all_frozen_pages_count AND
+                        $tab_visible_column > $tab_all_visible_pages_count)
+                    FROM pg_stat_vacuum_tables
+                    WHERE relname = 'vestat'";
+        }
+        else {
+            $sql = "
+            SELECT (pg_stat_get_rev_all_frozen_pages(c.oid) > $tab_all_frozen_pages_count AND
+                     pg_stat_get_rev_all_visible_pages(c.oid) > $tab_all_visible_pages_count)
+                FROM pg_class c
+                WHERE relname = 'vestat'";
+        }
+
+        $result_query = $node->safe_psql($dbname, $sql);
+
+        return 1 if (defined $result_query && $result_query eq 't');
+
+        # sub-second sleep
+        sleep($interval);
+    }
+
+    return 0;
+}
+
+#------------------------------------------------------------------------------
+# Variables to hold vacuum statistics snapshots for comparisons
+#------------------------------------------------------------------------------
+
+my $vm_new_frozen_pages;
+my $vm_new_visible_pages;
+
+my $rev_all_frozen_pages;
+my $rev_all_visible_pages;
+
+my $res;
+
+#------------------------------------------------------------------------------
+# fetch_vacuum_stats
+#
+# Loads current values of the relevant vacuum counters for the test table
+# into the package-level variables above so tests can compare later.
+#------------------------------------------------------------------------------
+
+sub fetch_vacuum_stats {
+    # fetch actual base vacuum statistics
+    $vm_new_frozen_pages = $node->safe_psql(
+        $dbname,
+        "SELECT vt.vm_new_frozen_pages
+           FROM pg_stat_vacuum_tables vt
+          WHERE vt.relname = 'vestat';"
+    );
+
+    $vm_new_visible_pages = $node->safe_psql(
+        $dbname,
+        "SELECT vt.vm_new_visible_pages
+           FROM pg_stat_vacuum_tables vt
+          WHERE vt.relname = 'vestat';"
+    );
+
+    $rev_all_frozen_pages = $node->safe_psql(
+        $dbname,
+        "SELECT pg_stat_get_rev_all_frozen_pages(c.oid)
+           FROM pg_class c
+          WHERE c.relname = 'vestat';"
+    );
+
+    $rev_all_visible_pages = $node->safe_psql(
+        $dbname,
+        "SELECT pg_stat_get_rev_all_visible_pages(c.oid)
+           FROM pg_class c
+          WHERE c.relname = 'vestat';"
+    );
+}
+
+#------------------------------------------------------------------------------
+# fetch_vacuum_stats during mismatch
+#
+# Print current values and old values of relevant vacuum counters for the test
+# table, storing them in package variables for subsequent comparisons.
+#------------------------------------------------------------------------------
+
+sub fetch_error_tab_vacuum_statistics {
+    my (%args) = @_;
+
+    # Validate presence of required args (allow 0 as valid numeric baseline)
+    die "tab_column required"
+      unless exists $args{tab_column} && defined $args{tab_column};
+    die "tab_value required"
+      unless exists $args{tab_value};
+
+    my $tab_column = $args{tab_column};
+    my $tab_value  = $args{tab_value};
+
+    # fetch actual base vacuum statistics
+    my $cur_value = $node->safe_psql(
+    $dbname,
+    "SELECT $tab_column
+       FROM pg_stat_vacuum_tables
+      WHERE relname = 'vestat';"
+    );
+
+    diag("MISMATCH FOR $tab_column: the current value is $cur_value, while it should be $tab_value");
+}
+
+#------------------------------------------------------------------------------
+# Test 1: Create test table, populate it and run an initial vacuum to force freezing
+#------------------------------------------------------------------------------
+
+$node->safe_psql($dbname, q{
+	SELECT pg_stat_force_next_flush();
+	CREATE TABLE vestat (x int)
+		WITH (autovacuum_enabled = off, fillfactor = 10);
+	INSERT INTO vestat SELECT x FROM generate_series(1, 1000) AS g(x);
+	VACUUM (FREEZE, VERBOSE) vestat;
+});
+
+# Poll the stats view until the expected deltas appear or timeout.
+# We do not expect rev_all_* counters to change here, so we pass -1 for them.
+$updated = wait_for_vacuum_stats(
+			tab_frozen_column => 'vm_new_frozen_pages',
+			tab_visible_column => 'vm_new_visible_pages',
+			tab_all_frozen_pages_count => 0,
+			tab_all_visible_pages_count => 0,
+      run_vacuum => 1,
+);
+
+ok($updated,
+   'vacuum stats updated after vacuuming the table (vm_new_frozen_pages and vm_new_visible_pages advanced)')
+  or diag "Timeout waiting for pg_stat_vacuum_tables to update after $timeout seconds during vacuum";
+
+#------------------------------------------------------------------------------
+# Snapshot current statistics for later comparison
+#------------------------------------------------------------------------------
+
+fetch_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Verify initial statistics after vacuum
+#------------------------------------------------------------------------------
+
+$res = $node->safe_psql($dbname, q{
+    SELECT vm_new_frozen_pages > 0 FROM pg_stat_vacuum_tables WHERE relname = 'vestat';
+});
+ok($res eq 't', 'vacuum froze some pages, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'vm_new_frozen_pages', tab_value => $vm_new_frozen_pages,);
+
+$res =  $node->safe_psql($dbname, q{
+    SELECT vm_new_visible_pages > 0 FROM pg_stat_vacuum_tables WHERE relname = 'vestat';
+});
+ok($res eq 't', 'vacuum marked pages all-visible, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'vm_new_visible_pages', tab_value =>$vm_new_visible_pages,);
+
+$res =  $node->safe_psql($dbname, q{
+    SELECT pg_stat_get_rev_all_frozen_pages(c.oid) = 0
+    FROM pg_stat_vacuum_tables vt
+    JOIN pg_class c ON c.relname = vt.relname
+    WHERE vt.relname = 'vestat';
+});
+ok($res eq 't', 'vacuum did not increase frozen-page revision count, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'rev_all_frozen_pages', tab_value => 0,);
+
+$res =  $node->safe_psql($dbname, q{
+    SELECT pg_stat_get_rev_all_visible_pages(c.oid) = 0
+    FROM pg_stat_vacuum_tables vt
+    JOIN pg_class c ON c.relname = vt.relname
+    WHERE vt.relname = 'vestat';
+});
+ok($res eq 't', 'vacuum did not increase visible-page revision count, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'rev_all_visible_pages', tab_value => 0,);
+
+#------------------------------------------------------------------------------
+# Test 2: Trigger backend updates
+# Backend activity should reset per-page visibility/freeze marks and increment revision counters
+#------------------------------------------------------------------------------
+$node->safe_psql($dbname, q{
+    UPDATE vestat SET x = x + 1001;
+});
+
+# Poll until stats update or timeout.
+# We do not expect vm_new_frozen_pages or vm_new_visible_pages to change here,
+# so we pass -1 for those counters.
+$updated = wait_for_vacuum_stats(
+			tab_frozen_column => 'rev_all_frozen_pages',
+			tab_visible_column => 'rev_all_visible_pages',
+			tab_all_frozen_pages_count => 0,
+			tab_all_visible_pages_count => 0,
+      run_vacuum => 0,
+);
+ok($updated,
+   'vacuum stats updated after backend tuple updates (rev_all_frozen_pages and rev_all_visible_pages advanced)')
+  or diag "Timeout waiting for pg_stats_vacuum_* update after $timeout seconds";
+
+#------------------------------------------------------------------------------
+# Check updated statistics after backend activity
+#------------------------------------------------------------------------------
+
+$res = $node->safe_psql($dbname,
+	"SELECT vm_new_frozen_pages = $vm_new_frozen_pages FROM pg_stat_vacuum_tables WHERE relname = 'vestat';"
+);
+ok($res eq 't', 'backend activity did not increase the frozen-page count') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'vm_new_frozen_pages', tab_value => $vm_new_frozen_pages,);
+
+$res = $node->safe_psql($dbname,
+	"SELECT vm_new_visible_pages = $vm_new_visible_pages FROM pg_stat_vacuum_tables WHERE relname = 'vestat';"
+);
+ok($res eq 't', 'backend activity did not increase the all-visible page count') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'vm_new_visible_pages', tab_value => $vm_new_visible_pages,);
+
+$res = $node->safe_psql($dbname,
+	"SELECT pg_stat_get_rev_all_frozen_pages(c.oid) > $rev_all_frozen_pages
+	 FROM pg_stat_vacuum_tables vt
+	 JOIN pg_class c ON c.relname = vt.relname
+	 WHERE vt.relname = 'vestat';"
+);
+ok($res eq 't', 'backend activity increased frozen-page revision count') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'rev_all_frozen_pages', tab_value => $rev_all_frozen_pages,);
+
+$res = $node->safe_psql($dbname,
+	"SELECT pg_stat_get_rev_all_visible_pages(c.oid) > $rev_all_visible_pages
+	 FROM pg_stat_vacuum_tables vt
+	 JOIN pg_class c ON c.relname = vt.relname
+	 WHERE vt.relname = 'vestat';"
+);
+ok($res eq 't', 'backend activity increased visible-page revision count') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'rev_all_visible_pages', tab_value => $rev_all_visible_pages,);
+
+#------------------------------------------------------------------------------
+# Update saved snapshots
+#------------------------------------------------------------------------------
+
+fetch_vacuum_stats();
+
+#------------------------------------------------------------------------------
+# Test 3: Force another vacuum after backend modifications - vacuum should restore freeze/visibility
+#------------------------------------------------------------------------------
+
+$node->safe_psql($dbname, q{ VACUUM (FREEZE, VERBOSE) vestat; });
+
+# Poll until stats update or timeout.
+# We pass current snapshot values for vm_new_frozen_pages/vm_new_visible_pages and expect rev counters unchanged.
+$updated = wait_for_vacuum_stats(
+			tab_frozen_column => 'vm_new_frozen_pages',
+			tab_visible_column => 'vm_new_visible_pages',
+			tab_all_frozen_pages_count => $vm_new_frozen_pages,
+			tab_all_visible_pages_count => $vm_new_visible_pages,
+      run_vacuum => 1,
+);
+
+ok($updated,
+   'vacuum stats updated after vacuuming the all-updated table (vm_new_frozen_pages and vm_new_visible_pages advanced)')
+  or diag "Timeout waiting for pg_stat_vacuum_tables to update after $timeout seconds during vacuum";
+
+#------------------------------------------------------------------------------
+# Verify statistics after final vacuum
+# Check updated stats after backend work
+#------------------------------------------------------------------------------
+$res = $node->safe_psql($dbname,
+	"SELECT vm_new_frozen_pages > $vm_new_frozen_pages FROM pg_stat_vacuum_tables WHERE relname = 'vestat';"
+);
+ok($res eq 't', 'vacuum froze some pages after backend activity, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'vm_new_frozen_pages', tab_value => $vm_new_frozen_pages,);
+
+$res = $node->safe_psql($dbname,
+	"SELECT vm_new_visible_pages > $vm_new_visible_pages FROM pg_stat_vacuum_tables WHERE relname = 'vestat';"
+);
+ok($res eq 't', 'vacuum marked pages all-visible after backend activity, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'vm_new_visible_pages', tab_value => $vm_new_visible_pages,);
+
+$res = $node->safe_psql($dbname,
+	"SELECT pg_stat_get_rev_all_frozen_pages(c.oid) = $rev_all_frozen_pages
+	 FROM pg_stat_vacuum_tables vt
+	 JOIN pg_class c ON c.relname = vt.relname
+	 WHERE vt.relname = 'vestat';"
+);
+ok($res eq 't', 'vacuum did not increase frozen-page revision count after backend activity, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'rev_all_frozen_pages', tab_value => $rev_all_frozen_pages,);
+
+$res = $node->safe_psql($dbname,
+	"SELECT pg_stat_get_rev_all_visible_pages(c.oid) = $rev_all_visible_pages
+	 FROM pg_stat_vacuum_tables vt
+	 JOIN pg_class c ON c.relname = vt.relname
+	 WHERE vt.relname = 'vestat';"
+);
+ok($res eq 't', 'vacuum did not increase visible-page revision count after backend activity, as expected') or
+  fetch_error_tab_vacuum_statistics(tab_column => 'rev_all_visible_pages', tab_value => $rev_all_visible_pages,);
+
+#------------------------------------------------------------------------------
+# Cleanup
+#------------------------------------------------------------------------------
+
+$node->safe_psql('postgres', q{
+	DROP DATABASE statistic_vacuum_database_regression;
+});
+
+$node->stop;
+done_testing();
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 4286c266e17..e4a77878beb 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1844,7 +1844,9 @@ pg_stat_all_tables| SELECT c.oid AS relid,
     pg_stat_get_total_autovacuum_time(c.oid) AS total_autovacuum_time,
     pg_stat_get_total_analyze_time(c.oid) AS total_analyze_time,
     pg_stat_get_total_autoanalyze_time(c.oid) AS total_autoanalyze_time,
-    pg_stat_get_stat_reset_time(c.oid) AS stats_reset
+    pg_stat_get_stat_reset_time(c.oid) AS stats_reset,
+    pg_stat_get_rev_all_frozen_pages(c.oid) AS rev_all_frozen_pages,
+    pg_stat_get_rev_all_visible_pages(c.oid) AS rev_all_visible_pages
    FROM ((pg_class c
      LEFT JOIN pg_index i ON ((c.oid = i.indrelid)))
      LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace)))
@@ -2266,7 +2268,9 @@ pg_stat_sys_tables| SELECT relid,
     total_autovacuum_time,
     total_analyze_time,
     total_autoanalyze_time,
-    stats_reset
+    stats_reset,
+    rev_all_frozen_pages,
+    rev_all_visible_pages
    FROM pg_stat_all_tables
   WHERE ((schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (schemaname ~ '^pg_toast'::text));
 pg_stat_user_functions| SELECT p.oid AS funcid,
@@ -2321,9 +2325,43 @@ pg_stat_user_tables| SELECT relid,
     total_autovacuum_time,
     total_analyze_time,
     total_autoanalyze_time,
-    stats_reset
+    stats_reset,
+    rev_all_frozen_pages,
+    rev_all_visible_pages
    FROM pg_stat_all_tables
   WHERE ((schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (schemaname !~ '^pg_toast'::text));
+pg_stat_vacuum_tables| SELECT ns.nspname AS schemaname,
+    rel.relname,
+    stats.relid,
+    stats.total_blks_read,
+    stats.total_blks_hit,
+    stats.total_blks_dirtied,
+    stats.total_blks_written,
+    stats.rel_blks_read,
+    stats.rel_blks_hit,
+    stats.pages_scanned,
+    stats.pages_removed,
+    stats.vm_new_frozen_pages,
+    stats.vm_new_visible_pages,
+    stats.vm_new_visible_frozen_pages,
+    stats.missed_dead_pages,
+    stats.tuples_deleted,
+    stats.tuples_frozen,
+    stats.recently_dead_tuples,
+    stats.missed_dead_tuples,
+    stats.wraparound_failsafe,
+    stats.index_vacuum_count,
+    stats.wal_records,
+    stats.wal_fpi,
+    stats.wal_bytes,
+    stats.blk_read_time,
+    stats.blk_write_time,
+    stats.delay_time,
+    stats.total_time
+   FROM (pg_class rel
+     JOIN pg_namespace ns ON ((ns.oid = rel.relnamespace))),
+    LATERAL pg_stat_get_vacuum_tables(rel.oid) stats(relid, total_blks_read, total_blks_hit, total_blks_dirtied, total_blks_written, rel_blks_read, rel_blks_hit, pages_scanned, pages_removed, vm_new_frozen_pages, vm_new_visible_pages, vm_new_visible_frozen_pages, missed_dead_pages, tuples_deleted, tuples_frozen, recently_dead_tuples, missed_dead_tuples, wraparound_failsafe, index_vacuum_count, wal_records, wal_fpi, wal_bytes, blk_read_time, blk_write_time, delay_time, total_time)
+  WHERE (rel.relkind = 'r'::"char");
 pg_stat_wal| SELECT wal_records,
     wal_fpi,
     wal_bytes,
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 905f9bca959..62f2ac11659 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -139,4 +139,4 @@ test: fast_default
 
 # run tablespace test at the end because it drops the tablespace created during
 # setup that other tests may use.
-test: tablespace
+test: tablespace
\ No newline at end of file
-- 
2.39.5 (Apple Git-154)