diff --git a/.github/workflows/build-gpdb.yml b/.github/workflows/build-gpdb.yml
index 08df4037f73..4d369ae2af1 100644
--- a/.github/workflows/build-gpdb.yml
+++ b/.github/workflows/build-gpdb.yml
@@ -121,6 +121,9 @@ jobs:
               {"test":"ic-mdb",
                "make_configs":["src/test/isolation2:installcheck-mdb"]
               },
+              {"test":"ic-isolation",
+               "make_configs":["src/test/isolation:installcheck"]
+              },
               {"test":"ic-isolation2",
                "make_configs":["src/test/isolation2:installcheck"]
               },
@@ -153,6 +156,7 @@ jobs:
                                "contrib/try_convert:generate-tests,installcheck",
                                "contrib/tsearch2:installcheck",
                                "contrib/unaccent:installcheck",
+                               "contrib/vacuum_stats:installcheck,installcheck-tap",
                                "contrib/xml2:installcheck"]
               },
               {"test":"ic-gpcontrib",
diff --git a/contrib/Makefile b/contrib/Makefile
index 63d47aac5ea..5bdad45f5ad 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -46,7 +46,8 @@ SUBDIRS = \
 		try_convert \
 		tsearch2	\
 		tablefunc	\
-		unaccent
+		unaccent	\
+		vacuum_stats
 # GPDB_92_MERGE_FIXME: Why some files (especially Makefile) are missing under
 # these directories?
 #               seg         \
diff --git a/contrib/vacuum_stats/.gitignore b/contrib/vacuum_stats/.gitignore
new file mode 100644
index 00000000000..5dcb3ff9723
--- /dev/null
+++ b/contrib/vacuum_stats/.gitignore
@@ -0,0 +1,4 @@
+# Generated subdirectories
+/log/
+/results/
+/tmp_check/
diff --git a/contrib/vacuum_stats/Makefile b/contrib/vacuum_stats/Makefile
new file mode 100644
index 00000000000..3334d1cee31
--- /dev/null
+++ b/contrib/vacuum_stats/Makefile
@@ -0,0 +1,29 @@
+# contrib/vacuum_stats/Makefile
+
+MODULE_big = vacuum_stats
+OBJS = vacuum_stats.o
+
+EXTENSION = vacuum_stats
+DATA = vacuum_stats--1.0.sql
+PGFILEDESC = "vacuum_stats - per-relation and per-database vacuum statistics"
+
+REGRESS = vacuum_stats
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/vacuum_stats
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
+
+check-tap:
+	$(prove_check)
+
+installcheck-tap:
+	$(prove_installcheck)
+
+.PHONY: check-tap installcheck-tap
diff --git a/contrib/vacuum_stats/README.md b/contrib/vacuum_stats/README.md
new file mode 100644
index 00000000000..df0e40b6a61
--- /dev/null
+++ b/contrib/vacuum_stats/README.md
@@ -0,0 +1,159 @@
+# vacuum_stats
+
+Exposes vacuum counters accumulated by the statistics collector for
+relations (tables and indexes) and databases, without any system catalog
+changes: the counters live in the statistics collector's per-relation and
+per-database entries, and all SQL objects are created by this extension.
+
+## Why
+
+Vacuum computes all of this while it works, and then throws it away.  The
+numbers reach the server log — one block of text per run, from `VACUUM
+VERBOSE` or `log_autovacuum_min_duration` — and are not available to SQL
+afterwards.  What the server does keep about vacuum in
+`pg_stat_all_tables` is `vacuum_count`, `last_vacuum` and the current
+`n_dead_tup` estimate: that tells you how often vacuum ran, not what work
+it had to do or what the work cost.  There is no `pg_stat_progress_vacuum`
+in this release, and progress views only describe a vacuum that is running
+right now anyway, not the picture over a period.
+
+The gap matters because an administrator has to balance the useful effect
+of vacuum against the overhead it puts on the system, and that balance is
+different for every relation.  It is near zero for an append-only table
+and highest for a frequently updated one.  Indexes have no visibility map,
+so vacuum scans them in full: the cost grows with the number and the size
+of the indexes, and the worst case is a bloating index on a small table.
+In Greenplum this is entirely the administrator's problem — autovacuum is
+disabled here except for the anti-wraparound vacuum of `template0` and
+other non-connectable databases, so all VACUUMing of user tables is done
+by hand or on a schedule, with no feedback loop to correct a bad guess.
+
+Questions the counters answer:
+
+- **Where does the vacuum budget actually go?**  `total_time`,
+  `tuples_deleted` and `pages_deleted` per table and per index show which
+  relations the maintenance window is being spent on, instead of a
+  cluster-wide impression.
+- **Is vacuum running but unable to do its job?**  A `dead_tuples` value
+  that stays high means vacuum did visit the table but the dead rows are
+  still visible to an old snapshot — a long-running transaction, an
+  idle-in-transaction session, an old distributed snapshot.  Vacuuming
+  more often will not help; the blocker has to be found and removed.
+- **How quickly is the work undone?**  Comparing `rev_all_visible_pages`
+  with `pages_all_visible` shows whether the pages vacuum marks
+  all-visible keep the mark or lose it again immediately.  A table that
+  constantly revokes it makes the next vacuum redo the same scan, and is
+  a candidate for a lower `fillfactor` (to get HOT updates instead of
+  page-spanning ones) or for being vacuumed on a different schedule.
+- **When did a routine VACUUM turn into a full-table scan?**
+  `wraparound_vacuum_count` counts the runs that crossed
+  `vacuum_freeze_table_age` and therefore could not skip a single page
+  through the visibility map.  That escalation is invisible otherwise and
+  is usually the reason a nightly VACUUM suddenly takes hours.  It also
+  measures the freezing pressure: this release has no autovacuum for user
+  tables, so a manual VACUUM is the only thing that ever advances
+  `relfrozenxid`, and a steadily growing counter means the schedule is
+  only just keeping up with the wraparound horizon.  The counter reports
+  only genuine freeze-age escalation: on a young cluster the freeze limits
+  are clamped to their minimum values, which makes every relation match
+  them, and those full-table scans are deliberately not counted.
+- **Is vacuum freezing anything at all?**  `pages_frozen` says on how many
+  pages the run actually froze tuples.  A table that accumulates vacuum
+  runs with `pages_frozen` staying at zero is never getting its tuples
+  frozen — the freezing is being deferred to a later, much more expensive
+  full-table run, and `vacuum_freeze_min_age` is worth revisiting.
+- **Is the load skewed?**  The `gp_segment_id` breakdown in the
+  `gp_stat_vacuum_*` views shows a segment doing much more vacuum work
+  than its peers, which usually means unevenly distributed data rather
+  than a vacuum problem.
+
+## Counters
+
+For every heap relation, index and database the following counters are
+accumulated by each (auto)vacuum run since the last statistics reset:
+
+| column                  | meaning                                                                 |
+|-------------------------|-------------------------------------------------------------------------|
+| `tuples_deleted`        | tuples (index entries) removed by vacuum                                |
+| `dead_tuples`           | dead tuples found but not yet removable (visible to old snapshots)      |
+| `pages_deleted`         | pages truncated from a heap / pages deleted in an index                 |
+| `dead_pages`            | pages left with unremovable dead tuples (heap) or deleted-but-not-yet-reusable pages (index) |
+| `pages_frozen`          | pages on which vacuum froze at least one tuple — shows whether vacuum is freezing at all |
+| `pages_all_visible`     | pages vacuum marked all-visible in the visibility map                   |
+| `rev_all_frozen_pages`  | pages whose all-frozen status was revoked (always 0: the visibility map has no all-frozen bit in this release) |
+| `rev_all_visible_pages` | pages whose all-visible bit was cleared by ordinary DML (delivered with the regular relation statistics, not the vacuum report) |
+| `wraparound_vacuum_count` | vacuum runs that had to scan the whole relation because its `relfrozenxid`/`relminmxid` reached the freeze table age |
+| `total_time`            | total vacuum time spent on the relation, in milliseconds                |
+
+For indexes only the tuple-deletion, page and timing counters are
+meaningful; the dead-tuple counter, the visibility-map counters and
+`wraparound_vacuum_count` stay zero, since those describe the heap
+relation the index belongs to.
+
+Like the rest of the collected statistics, the counters are written to
+the permanent statistics files when the statistics collector exits,
+which happens on a clean shutdown and also when the postmaster dies
+unexpectedly, so they survive a restart of the cluster.  After crash
+recovery the server resets all collected statistics (including these),
+as PostgreSQL considers them invalid after a crash.
+
+## Tests
+
+- `sql/vacuum_stats.sql` — a pg_regress smoke test of the extension
+  objects and the cluster-wide views (`make installcheck`).
+- `t/001_vacuum_statistics.pl` — a TAP test with the table, index and
+  database scenarios of the upstream vacuum statistics patch, plus the
+  restart-persistence and crash-reset checks (`make installcheck-tap`,
+  requires a build with `--enable-tap-tests`).
+- `src/test/isolation/specs/vacuum-extending-in-repeatable-read.spec` —
+  an isolation test checking that dead tuples held back by a repeatable
+  read snapshot show up in `dead_tuples` and move to `tuples_deleted`
+  once the snapshot is released.
+
+## Views
+
+Local (current node) views:
+
+- `pg_stat_vacuum_tables`
+- `pg_stat_vacuum_indexes`
+- `pg_stat_vacuum_database`
+
+Cluster-wide views (master plus every segment, with a `gp_segment_id`
+column; in Greenplum vacuum does its real work on the segments, so these
+are usually the interesting ones):
+
+- `gp_stat_vacuum_tables`
+- `gp_stat_vacuum_indexes`
+- `gp_stat_vacuum_database`
+
+## Usage
+
+```sql
+CREATE EXTENSION vacuum_stats;
+
+VACUUM my_table;
+
+SELECT gp_segment_id, tuples_deleted, dead_tuples, pages_deleted,
+       pages_all_visible, rev_all_visible_pages,
+       wraparound_vacuum_count, total_time
+FROM gp_stat_vacuum_tables
+WHERE relname = 'my_table';
+```
+
+Summed over the segments, to rank the tables by the vacuum time they
+cost:
+
+```sql
+SELECT schemaname, relname,
+       round(sum(total_time), 2) AS total_time_ms,
+       sum(tuples_deleted) AS tuples_deleted,
+       sum(dead_tuples) AS dead_tuples
+FROM gp_stat_vacuum_tables
+GROUP BY schemaname, relname
+HAVING sum(total_time) > 0
+ORDER BY total_time_ms DESC
+LIMIT 10;
+```
+
+The counters are reset together with the rest of the collected statistics
+(`pg_stat_reset()`, `pg_stat_reset_single_table_counters()`).
diff --git a/contrib/vacuum_stats/expected/vacuum_stats.out b/contrib/vacuum_stats/expected/vacuum_stats.out
new file mode 100644
index 00000000000..251e5535773
--- /dev/null
+++ b/contrib/vacuum_stats/expected/vacuum_stats.out
@@ -0,0 +1,128 @@
+CREATE EXTENSION vacuum_stats;
+SET client_min_messages = warning;
+CREATE TABLE vstat_heap (id int, val text) DISTRIBUTED BY (id);
+CREATE INDEX vstat_heap_idx ON vstat_heap (val);
+INSERT INTO vstat_heap SELECT g, 'val_' || g FROM generate_series(1, 1000) g;
+DELETE FROM vstat_heap WHERE id % 2 = 0;
+VACUUM vstat_heap;
+-- The statistics collector is fed asynchronously (UDP), and every QE
+-- caches its stats snapshot for the transaction, so poll until the
+-- counters become visible, resetting the snapshots on each iteration.
+CREATE FUNCTION vstat_clear_segment_snapshots() RETURNS SETOF void AS
+$$ SELECT pg_catalog.pg_stat_clear_snapshot() $$
+LANGUAGE SQL EXECUTE ON ALL SEGMENTS;
+CREATE FUNCTION wait_for_vacuum_stats(cond text) RETURNS bool AS
+$$
+DECLARE
+    ok bool;
+BEGIN
+    FOR i IN 1..120 LOOP
+        PERFORM pg_stat_clear_snapshot();
+        PERFORM * FROM vstat_clear_segment_snapshots();
+        EXECUTE cond INTO ok;
+        IF ok THEN
+            RETURN true;
+        END IF;
+        PERFORM pg_sleep(0.25);
+    END LOOP;
+    RETURN false;
+END
+$$ LANGUAGE plpgsql;
+-- table counters: the deleted tuples must show up
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(tuples_deleted) > 0
+    FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap' $$);
+ wait_for_vacuum_stats 
+-----------------------
+ t
+(1 row)
+
+SELECT sum(tuples_deleted) > 0 AS tuples_deleted_ok,
+       sum(dead_tuples) >= 0 AS dead_tuples_ok,
+       sum(pages_deleted) >= 0 AS pages_deleted_ok,
+       sum(dead_pages) >= 0 AS dead_pages_ok,
+       sum(pages_frozen) >= 0 AS pages_frozen_ok,
+       sum(wraparound_vacuum_count) >= 0 AS wraparound_ok,
+       sum(rev_all_frozen_pages) = 0 AS rev_all_frozen_pages_ok,
+       sum(total_time) > 0 AS total_time_ok
+FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap';
+ tuples_deleted_ok | dead_tuples_ok | pages_deleted_ok | dead_pages_ok | pages_frozen_ok | wraparound_ok | rev_all_frozen_pages_ok | total_time_ok 
+-------------------+----------------+------------------+---------------+-----------------+---------------+-------------------------+---------------
+ t                 | t              | t                | t             | t               | t             | t                       | t
+(1 row)
+
+-- vacuum marks the remaining pages all-visible
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(pages_all_visible) > 0
+    FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap' $$);
+ wait_for_vacuum_stats 
+-----------------------
+ t
+(1 row)
+
+-- index counters: the index entries removed by vacuum must show up
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(tuples_deleted) > 0
+    FROM gp_stat_vacuum_indexes WHERE indexrelname = 'vstat_heap_idx' $$);
+ wait_for_vacuum_stats 
+-----------------------
+ t
+(1 row)
+
+SELECT sum(tuples_deleted) > 0 AS tuples_deleted_ok,
+       sum(pages_deleted) >= 0 AS pages_deleted_ok,
+       sum(dead_pages) >= 0 AS dead_pages_ok,
+       sum(total_time) >= 0 AS total_time_ok
+FROM gp_stat_vacuum_indexes WHERE indexrelname = 'vstat_heap_idx';
+ tuples_deleted_ok | pages_deleted_ok | dead_pages_ok | total_time_ok 
+-------------------+------------------+---------------+---------------
+ t                 | t                | t             | t
+(1 row)
+
+-- inserting into all-visible pages revokes their all-visible status,
+-- which is delivered with the regular relation statistics
+INSERT INTO vstat_heap SELECT g, 'again_' || g FROM generate_series(1, 30) g;
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(rev_all_visible_pages) > 0
+    FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap' $$);
+ wait_for_vacuum_stats 
+-----------------------
+ t
+(1 row)
+
+-- database counters accumulate the per-relation ones
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(tuples_deleted) > 0
+    FROM gp_stat_vacuum_database WHERE datname = current_database() $$);
+ wait_for_vacuum_stats 
+-----------------------
+ t
+(1 row)
+
+-- local (per-node) views also work; on the master the table is empty,
+-- so just check that the relations are visible there
+SELECT count(*) = 1 AS heap_visible FROM pg_stat_vacuum_tables
+WHERE relname = 'vstat_heap';
+ heap_visible 
+--------------
+ t
+(1 row)
+
+SELECT count(*) = 1 AS index_visible FROM pg_stat_vacuum_indexes
+WHERE indexrelname = 'vstat_heap_idx';
+ index_visible 
+---------------
+ t
+(1 row)
+
+SELECT count(*) = 1 AS db_visible FROM pg_stat_vacuum_database
+WHERE datname = current_database();
+ db_visible 
+------------
+ t
+(1 row)
+
+DROP FUNCTION wait_for_vacuum_stats(text);
+DROP FUNCTION vstat_clear_segment_snapshots();
+DROP TABLE vstat_heap;
+DROP EXTENSION vacuum_stats;
diff --git a/contrib/vacuum_stats/sql/vacuum_stats.sql b/contrib/vacuum_stats/sql/vacuum_stats.sql
new file mode 100644
index 00000000000..bd13e8bc350
--- /dev/null
+++ b/contrib/vacuum_stats/sql/vacuum_stats.sql
@@ -0,0 +1,92 @@
+CREATE EXTENSION vacuum_stats;
+
+SET client_min_messages = warning;
+
+CREATE TABLE vstat_heap (id int, val text) DISTRIBUTED BY (id);
+CREATE INDEX vstat_heap_idx ON vstat_heap (val);
+INSERT INTO vstat_heap SELECT g, 'val_' || g FROM generate_series(1, 1000) g;
+DELETE FROM vstat_heap WHERE id % 2 = 0;
+VACUUM vstat_heap;
+
+-- The statistics collector is fed asynchronously (UDP), and every QE
+-- caches its stats snapshot for the transaction, so poll until the
+-- counters become visible, resetting the snapshots on each iteration.
+CREATE FUNCTION vstat_clear_segment_snapshots() RETURNS SETOF void AS
+$$ SELECT pg_catalog.pg_stat_clear_snapshot() $$
+LANGUAGE SQL EXECUTE ON ALL SEGMENTS;
+
+CREATE FUNCTION wait_for_vacuum_stats(cond text) RETURNS bool AS
+$$
+DECLARE
+    ok bool;
+BEGIN
+    FOR i IN 1..120 LOOP
+        PERFORM pg_stat_clear_snapshot();
+        PERFORM * FROM vstat_clear_segment_snapshots();
+        EXECUTE cond INTO ok;
+        IF ok THEN
+            RETURN true;
+        END IF;
+        PERFORM pg_sleep(0.25);
+    END LOOP;
+    RETURN false;
+END
+$$ LANGUAGE plpgsql;
+
+-- table counters: the deleted tuples must show up
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(tuples_deleted) > 0
+    FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap' $$);
+
+SELECT sum(tuples_deleted) > 0 AS tuples_deleted_ok,
+       sum(dead_tuples) >= 0 AS dead_tuples_ok,
+       sum(pages_deleted) >= 0 AS pages_deleted_ok,
+       sum(dead_pages) >= 0 AS dead_pages_ok,
+       sum(pages_frozen) >= 0 AS pages_frozen_ok,
+       sum(wraparound_vacuum_count) >= 0 AS wraparound_ok,
+       sum(rev_all_frozen_pages) = 0 AS rev_all_frozen_pages_ok,
+       sum(total_time) > 0 AS total_time_ok
+FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap';
+
+-- vacuum marks the remaining pages all-visible
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(pages_all_visible) > 0
+    FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap' $$);
+
+-- index counters: the index entries removed by vacuum must show up
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(tuples_deleted) > 0
+    FROM gp_stat_vacuum_indexes WHERE indexrelname = 'vstat_heap_idx' $$);
+
+SELECT sum(tuples_deleted) > 0 AS tuples_deleted_ok,
+       sum(pages_deleted) >= 0 AS pages_deleted_ok,
+       sum(dead_pages) >= 0 AS dead_pages_ok,
+       sum(total_time) >= 0 AS total_time_ok
+FROM gp_stat_vacuum_indexes WHERE indexrelname = 'vstat_heap_idx';
+
+-- inserting into all-visible pages revokes their all-visible status,
+-- which is delivered with the regular relation statistics
+INSERT INTO vstat_heap SELECT g, 'again_' || g FROM generate_series(1, 30) g;
+
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(rev_all_visible_pages) > 0
+    FROM gp_stat_vacuum_tables WHERE relname = 'vstat_heap' $$);
+
+-- database counters accumulate the per-relation ones
+SELECT wait_for_vacuum_stats($$
+    SELECT sum(tuples_deleted) > 0
+    FROM gp_stat_vacuum_database WHERE datname = current_database() $$);
+
+-- local (per-node) views also work; on the master the table is empty,
+-- so just check that the relations are visible there
+SELECT count(*) = 1 AS heap_visible FROM pg_stat_vacuum_tables
+WHERE relname = 'vstat_heap';
+SELECT count(*) = 1 AS index_visible FROM pg_stat_vacuum_indexes
+WHERE indexrelname = 'vstat_heap_idx';
+SELECT count(*) = 1 AS db_visible FROM pg_stat_vacuum_database
+WHERE datname = current_database();
+
+DROP FUNCTION wait_for_vacuum_stats(text);
+DROP FUNCTION vstat_clear_segment_snapshots();
+DROP TABLE vstat_heap;
+DROP EXTENSION vacuum_stats;
diff --git a/contrib/vacuum_stats/t/001_vacuum_statistics.pl b/contrib/vacuum_stats/t/001_vacuum_statistics.pl
new file mode 100644
index 00000000000..19fb78355ec
--- /dev/null
+++ b/contrib/vacuum_stats/t/001_vacuum_statistics.pl
@@ -0,0 +1,187 @@
+# Test the vacuum statistics collected for tables, indexes and databases.
+#
+# The scenarios follow the regression tests of the upstream "Vacuum
+# statistics" patch
+# (https://www.postgresql.org/message-id/70e1cca9-ff89-4e76-a611-d38bcc0e14ad%40yandex.ru),
+# adapted to the asynchronous statistics collector by polling for the
+# expected state.  The node runs in utility mode, so the local
+# pg_stat_vacuum_* views are used.  On top of the upstream scenarios this
+# also checks that the statistics survive a clean restart of the cluster
+# and are reset by crash recovery, like the rest of the collected
+# statistics.
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 13;
+
+my $node = get_new_node('vacuum_stats');
+$node->init;
+$node->append_conf('postgresql.conf', 'autovacuum = off');
+$node->append_conf('postgresql.conf', 'track_counts = on');
+# The node runs standalone in utility mode, so there is no dispatcher to
+# advance the distributed oldest xmin; run in maintenance mode, where
+# GetOldestXmin() does not consult the distributed log and vacuum can
+# remove tuples.
+$node->append_conf('postgresql.conf', 'maintenance_mode = on');
+# freeze aggressively, so that every vacuum has to scan the whole
+# relation and is counted as a wraparound-driven one
+$node->append_conf('postgresql.conf', 'vacuum_freeze_min_age = 0');
+$node->append_conf('postgresql.conf', 'vacuum_freeze_table_age = 0');
+$node->start;
+
+$node->safe_psql('postgres', 'CREATE EXTENSION vacuum_stats');
+$node->safe_psql('postgres',
+	'CREATE TABLE vestat (x int primary key) '
+	  . 'WITH (autovacuum_enabled = off, fillfactor = 10)');
+$node->safe_psql('postgres',
+	'INSERT INTO vestat SELECT g FROM generate_series(1, 10000) g');
+$node->safe_psql('postgres', 'ANALYZE vestat');
+
+my $tab_counters =
+	'SELECT tuples_deleted, dead_tuples, pages_deleted, dead_pages, '
+  . 'pages_frozen, pages_all_visible, rev_all_frozen_pages, '
+  . 'rev_all_visible_pages, '
+  . 'wraparound_vacuum_count, total_time '
+  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'";
+my $idx_counters =
+	'SELECT tuples_deleted, pages_deleted, dead_pages, total_time '
+  . "FROM pg_stat_vacuum_indexes WHERE indexrelname = 'vestat_pkey'";
+
+# Before the first vacuum execution the extended stats view is empty.
+my $result = $node->safe_psql('postgres',
+	'SELECT tuples_deleted = 0 AND pages_deleted = 0 AND pages_frozen = 0 '
+	  . 'AND pages_all_visible = 0 AND wraparound_vacuum_count = 0 '
+	  . 'AND total_time = 0 '
+	  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'");
+is($result, 't', 'table statistics are empty before the first vacuum');
+
+$node->safe_psql('postgres', 'DELETE FROM vestat WHERE x % 2 = 0');
+$node->safe_psql('postgres', 'VACUUM vestat');
+
+# Vacuum removed tuples, but the table was not truncated.
+ok( $node->poll_query_until(
+		'postgres',
+		'SELECT tuples_deleted > 0 AND pages_frozen > 0 '
+		  . 'AND pages_all_visible > 0 '
+		  . 'AND pages_deleted = 0 AND total_time > 0 '
+		  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'"),
+	'vacuum removed tuples of the table and froze pages');
+
+# vacuum_freeze_table_age is 0, so every run scans the whole relation and
+# is counted as a wraparound-driven vacuum.
+ok( $node->poll_query_until(
+		'postgres',
+		'SELECT wraparound_vacuum_count > 0 '
+		  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'"),
+	'the full-table vacuum is counted as a wraparound one');
+
+# Vacuum removed index entries, but no index pages could be deleted yet
+# because every other key remains.
+ok( $node->poll_query_until(
+		'postgres',
+		'SELECT tuples_deleted > 0 AND pages_deleted = 0 AND total_time > 0 '
+		  . "FROM pg_stat_vacuum_indexes WHERE indexrelname = 'vestat_pkey'"),
+	'vacuum removed index entries');
+
+# Take a baseline once the counters have settled.
+$node->safe_psql('postgres', 'SELECT pg_sleep(1)');
+my ($tab_deleted, $tab_pages_deleted) = split /\|/,
+  $node->safe_psql('postgres',
+	'SELECT tuples_deleted, pages_deleted '
+	  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'");
+my ($idx_deleted, $idx_pages_deleted) = split /\|/,
+  $node->safe_psql('postgres',
+	'SELECT tuples_deleted, pages_deleted '
+	  . "FROM pg_stat_vacuum_indexes WHERE indexrelname = 'vestat_pkey'");
+
+$node->safe_psql('postgres', 'DELETE FROM vestat');
+$node->safe_psql('postgres', 'VACUUM vestat');
+
+# The emptied pages must have been truncated away from the heap and
+# deleted from the index.
+ok( $node->poll_query_until(
+		'postgres',
+		"SELECT pages_deleted > $tab_pages_deleted "
+		  . "AND tuples_deleted > $tab_deleted "
+		  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'"),
+	'the emptied heap pages were truncated away');
+
+ok( $node->poll_query_until(
+		'postgres',
+		"SELECT pages_deleted > $idx_pages_deleted "
+		  . "AND tuples_deleted > $idx_deleted "
+		  . "FROM pg_stat_vacuum_indexes WHERE indexrelname = 'vestat_pkey'"),
+	'the emptied index pages were deleted');
+
+# VACUUM FULL doesn't report to the stats collector, so no advancement of
+# the statistics is detected here.
+$node->safe_psql('postgres', 'SELECT pg_sleep(1)');
+my $tab_before = $node->safe_psql('postgres', $tab_counters);
+my $idx_before = $node->safe_psql('postgres', $idx_counters);
+
+$node->safe_psql('postgres',
+	'INSERT INTO vestat SELECT g FROM generate_series(1, 10000) g');
+$node->safe_psql('postgres', 'DELETE FROM vestat WHERE x % 2 = 0');
+$node->safe_psql('postgres', 'VACUUM FULL vestat');
+$node->safe_psql('postgres', 'SELECT pg_sleep(1)');
+
+is($node->safe_psql('postgres', $tab_counters),
+	$tab_before, 'VACUUM FULL does not advance the table statistics');
+is($node->safe_psql('postgres', $idx_counters),
+	$idx_before, 'VACUUM FULL does not advance the index statistics');
+
+# A plain vacuum marks the rewritten pages all-visible again.
+my ($all_visible, $rev_all_visible) = split /\|/,
+  $node->safe_psql('postgres',
+	'SELECT pages_all_visible, rev_all_visible_pages '
+	  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'");
+
+$node->safe_psql('postgres', 'VACUUM vestat');
+
+ok( $node->poll_query_until(
+		'postgres',
+		"SELECT pages_all_visible > $all_visible "
+		  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'"),
+	'vacuum marked the pages all-visible again');
+
+# Ordinary DML revokes the all-visible status of the pages it touches;
+# this is delivered with the regular relation statistics.  There is no
+# all-frozen bit in the visibility map in this release, so the
+# corresponding counter stays zero.
+$node->safe_psql('postgres', 'UPDATE vestat SET x = x + 20000');
+
+ok( $node->poll_query_until(
+		'postgres',
+		"SELECT rev_all_visible_pages > $rev_all_visible "
+		  . 'AND rev_all_frozen_pages = 0 '
+		  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'"),
+	'the backend revoked the all-visible status of the updated pages');
+
+# The database-level counters accumulate the per-relation ones.
+ok( $node->poll_query_until(
+		'postgres',
+		'SELECT tuples_deleted > 0 AND wraparound_vacuum_count > 0 '
+		  . "AND total_time > 0 "
+		  . "FROM pg_stat_vacuum_database WHERE datname = 'postgres'"),
+	'the database-level statistics accumulate the vacuum counters');
+
+# The statistics survive a clean restart: the collector saves them to the
+# permanent statistics files when it exits.
+$node->safe_psql('postgres', 'SELECT pg_sleep(1)');
+$tab_before = $node->safe_psql('postgres', $tab_counters);
+$node->restart;
+is($node->safe_psql('postgres', $tab_counters),
+	$tab_before, 'vacuum statistics survive a clean restart');
+
+# Like the rest of the collected statistics, they are reset by crash
+# recovery.
+$node->stop('immediate');
+$node->start;
+ok( $node->poll_query_until(
+		'postgres',
+		'SELECT tuples_deleted = 0 AND total_time = 0 '
+		  . "FROM pg_stat_vacuum_tables WHERE relname = 'vestat'"),
+	'vacuum statistics are reset by crash recovery');
+
+$node->stop;
diff --git a/contrib/vacuum_stats/vacuum_stats--1.0.sql b/contrib/vacuum_stats/vacuum_stats--1.0.sql
new file mode 100644
index 00000000000..71cb29e0b0a
--- /dev/null
+++ b/contrib/vacuum_stats/vacuum_stats--1.0.sql
@@ -0,0 +1,337 @@
+/* contrib/vacuum_stats/vacuum_stats--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION vacuum_stats" to load this file. \quit
+
+--
+-- Per-relation accessor functions (tables and indexes).
+--
+CREATE FUNCTION pg_stat_get_vacuum_tuples_deleted(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_tuples_deleted'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_dead_tuples(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_dead_tuples'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_pages_deleted(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_pages_deleted'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_dead_pages(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_dead_pages'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_pages_frozen(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_pages_frozen'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_pages_all_visible(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_pages_all_visible'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_wraparound_count(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_wraparound_count'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_rev_all_frozen_pages(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_rev_all_frozen_pages'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_rev_all_visible_pages(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_rev_all_visible_pages'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_vacuum_total_time(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_vacuum_total_time'
+LANGUAGE C STABLE STRICT;
+
+--
+-- Per-database accessor functions.
+--
+CREATE FUNCTION pg_stat_get_db_vacuum_tuples_deleted(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_tuples_deleted'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_dead_tuples(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_dead_tuples'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_pages_deleted(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_pages_deleted'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_dead_pages(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_dead_pages'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_pages_frozen(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_pages_frozen'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_pages_all_visible(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_pages_all_visible'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_wraparound_count(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_wraparound_count'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_rev_all_frozen_pages(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_rev_all_frozen_pages'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_rev_all_visible_pages(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_rev_all_visible_pages'
+LANGUAGE C STABLE STRICT;
+
+CREATE FUNCTION pg_stat_get_db_vacuum_total_time(oid) RETURNS int8
+AS 'MODULE_PATHNAME', 'pg_stat_get_db_vacuum_total_time'
+LANGUAGE C STABLE STRICT;
+
+--
+-- Local (per-node) views.  total_time is exposed in milliseconds, like
+-- the other timing columns of the cumulative statistics views.
+--
+CREATE VIEW pg_stat_vacuum_tables AS
+    SELECT
+        c.oid AS relid,
+        n.nspname AS schemaname,
+        c.relname AS relname,
+        pg_stat_get_vacuum_tuples_deleted(c.oid) AS tuples_deleted,
+        pg_stat_get_vacuum_dead_tuples(c.oid) AS dead_tuples,
+        pg_stat_get_vacuum_pages_deleted(c.oid) AS pages_deleted,
+        pg_stat_get_vacuum_dead_pages(c.oid) AS dead_pages,
+        pg_stat_get_vacuum_pages_frozen(c.oid) AS pages_frozen,
+        pg_stat_get_vacuum_pages_all_visible(c.oid) AS pages_all_visible,
+        pg_stat_get_vacuum_rev_all_frozen_pages(c.oid) AS rev_all_frozen_pages,
+        pg_stat_get_vacuum_rev_all_visible_pages(c.oid) AS rev_all_visible_pages,
+        pg_stat_get_vacuum_wraparound_count(c.oid) AS wraparound_vacuum_count,
+        pg_stat_get_vacuum_total_time(c.oid) / 1000.0 AS total_time
+    FROM pg_catalog.pg_class c
+        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+    WHERE c.relkind IN ('r', 't', 'm');
+
+CREATE VIEW pg_stat_vacuum_indexes AS
+    SELECT
+        c.oid AS relid,
+        i.oid AS indexrelid,
+        n.nspname AS schemaname,
+        c.relname AS relname,
+        i.relname AS indexrelname,
+        pg_stat_get_vacuum_tuples_deleted(i.oid) AS tuples_deleted,
+        pg_stat_get_vacuum_dead_tuples(i.oid) AS dead_tuples,
+        pg_stat_get_vacuum_pages_deleted(i.oid) AS pages_deleted,
+        pg_stat_get_vacuum_dead_pages(i.oid) AS dead_pages,
+        pg_stat_get_vacuum_pages_frozen(i.oid) AS pages_frozen,
+        pg_stat_get_vacuum_pages_all_visible(i.oid) AS pages_all_visible,
+        pg_stat_get_vacuum_rev_all_frozen_pages(i.oid) AS rev_all_frozen_pages,
+        pg_stat_get_vacuum_rev_all_visible_pages(i.oid) AS rev_all_visible_pages,
+        pg_stat_get_vacuum_wraparound_count(i.oid) AS wraparound_vacuum_count,
+        pg_stat_get_vacuum_total_time(i.oid) / 1000.0 AS total_time
+    FROM pg_catalog.pg_class c
+        JOIN pg_catalog.pg_index x ON c.oid = x.indrelid
+        JOIN pg_catalog.pg_class i ON i.oid = x.indexrelid
+        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+    WHERE c.relkind IN ('r', 't', 'm') AND i.relkind = 'i';
+
+CREATE VIEW pg_stat_vacuum_database AS
+    SELECT
+        d.oid AS datid,
+        d.datname AS datname,
+        pg_stat_get_db_vacuum_tuples_deleted(d.oid) AS tuples_deleted,
+        pg_stat_get_db_vacuum_dead_tuples(d.oid) AS dead_tuples,
+        pg_stat_get_db_vacuum_pages_deleted(d.oid) AS pages_deleted,
+        pg_stat_get_db_vacuum_dead_pages(d.oid) AS dead_pages,
+        pg_stat_get_db_vacuum_pages_frozen(d.oid) AS pages_frozen,
+        pg_stat_get_db_vacuum_pages_all_visible(d.oid) AS pages_all_visible,
+        pg_stat_get_db_vacuum_rev_all_frozen_pages(d.oid) AS rev_all_frozen_pages,
+        pg_stat_get_db_vacuum_rev_all_visible_pages(d.oid) AS rev_all_visible_pages,
+        pg_stat_get_db_vacuum_wraparound_count(d.oid) AS wraparound_vacuum_count,
+        pg_stat_get_db_vacuum_total_time(d.oid) / 1000.0 AS total_time
+    FROM pg_catalog.pg_database d;
+
+--
+-- Cluster-wide views: the same information from the master and from
+-- every segment, following the gp_stat_replication pattern.
+--
+-- NB: the queries are spelled out instead of referencing the local views,
+-- because a function executing on a QE slice may only access catalog
+-- relations.
+--
+CREATE FUNCTION gp_stat_get_master_vacuum_tables() RETURNS SETOF RECORD AS
+$$
+    SELECT pg_catalog.gp_execution_segment() AS gp_segment_id,
+        c.oid, n.nspname, c.relname,
+        @extschema@.pg_stat_get_vacuum_tuples_deleted(c.oid),
+        @extschema@.pg_stat_get_vacuum_dead_tuples(c.oid),
+        @extschema@.pg_stat_get_vacuum_pages_deleted(c.oid),
+        @extschema@.pg_stat_get_vacuum_dead_pages(c.oid),
+        @extschema@.pg_stat_get_vacuum_pages_frozen(c.oid),
+        @extschema@.pg_stat_get_vacuum_pages_all_visible(c.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_frozen_pages(c.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_visible_pages(c.oid),
+        @extschema@.pg_stat_get_vacuum_wraparound_count(c.oid),
+        @extschema@.pg_stat_get_vacuum_total_time(c.oid) / 1000.0
+    FROM pg_catalog.pg_class c
+        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+    WHERE c.relkind IN ('r', 't', 'm')
+$$
+LANGUAGE SQL EXECUTE ON MASTER;
+
+CREATE FUNCTION gp_stat_get_segment_vacuum_tables() RETURNS SETOF RECORD AS
+$$
+    SELECT pg_catalog.gp_execution_segment() AS gp_segment_id,
+        c.oid, n.nspname, c.relname,
+        @extschema@.pg_stat_get_vacuum_tuples_deleted(c.oid),
+        @extschema@.pg_stat_get_vacuum_dead_tuples(c.oid),
+        @extschema@.pg_stat_get_vacuum_pages_deleted(c.oid),
+        @extschema@.pg_stat_get_vacuum_dead_pages(c.oid),
+        @extschema@.pg_stat_get_vacuum_pages_frozen(c.oid),
+        @extschema@.pg_stat_get_vacuum_pages_all_visible(c.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_frozen_pages(c.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_visible_pages(c.oid),
+        @extschema@.pg_stat_get_vacuum_wraparound_count(c.oid),
+        @extschema@.pg_stat_get_vacuum_total_time(c.oid) / 1000.0
+    FROM pg_catalog.pg_class c
+        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+    WHERE c.relkind IN ('r', 't', 'm')
+$$
+LANGUAGE SQL EXECUTE ON ALL SEGMENTS;
+
+CREATE VIEW gp_stat_vacuum_tables AS
+    SELECT * FROM gp_stat_get_master_vacuum_tables() AS T
+        (gp_segment_id int, relid oid, schemaname name, relname name,
+         tuples_deleted int8, dead_tuples int8,
+         pages_deleted int8, dead_pages int8,
+         pages_frozen int8, pages_all_visible int8,
+         rev_all_frozen_pages int8, rev_all_visible_pages int8,
+         wraparound_vacuum_count int8, total_time numeric)
+    UNION ALL
+    SELECT * FROM gp_stat_get_segment_vacuum_tables() AS T
+        (gp_segment_id int, relid oid, schemaname name, relname name,
+         tuples_deleted int8, dead_tuples int8,
+         pages_deleted int8, dead_pages int8,
+         pages_frozen int8, pages_all_visible int8,
+         rev_all_frozen_pages int8, rev_all_visible_pages int8,
+         wraparound_vacuum_count int8, total_time numeric);
+
+CREATE FUNCTION gp_stat_get_master_vacuum_indexes() RETURNS SETOF RECORD AS
+$$
+    SELECT pg_catalog.gp_execution_segment() AS gp_segment_id,
+        c.oid, i.oid, n.nspname, c.relname, i.relname,
+        @extschema@.pg_stat_get_vacuum_tuples_deleted(i.oid),
+        @extschema@.pg_stat_get_vacuum_dead_tuples(i.oid),
+        @extschema@.pg_stat_get_vacuum_pages_deleted(i.oid),
+        @extschema@.pg_stat_get_vacuum_dead_pages(i.oid),
+        @extschema@.pg_stat_get_vacuum_pages_frozen(i.oid),
+        @extschema@.pg_stat_get_vacuum_pages_all_visible(i.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_frozen_pages(i.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_visible_pages(i.oid),
+        @extschema@.pg_stat_get_vacuum_wraparound_count(i.oid),
+        @extschema@.pg_stat_get_vacuum_total_time(i.oid) / 1000.0
+    FROM pg_catalog.pg_class c
+        JOIN pg_catalog.pg_index x ON c.oid = x.indrelid
+        JOIN pg_catalog.pg_class i ON i.oid = x.indexrelid
+        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+    WHERE c.relkind IN ('r', 't', 'm') AND i.relkind = 'i'
+$$
+LANGUAGE SQL EXECUTE ON MASTER;
+
+CREATE FUNCTION gp_stat_get_segment_vacuum_indexes() RETURNS SETOF RECORD AS
+$$
+    SELECT pg_catalog.gp_execution_segment() AS gp_segment_id,
+        c.oid, i.oid, n.nspname, c.relname, i.relname,
+        @extschema@.pg_stat_get_vacuum_tuples_deleted(i.oid),
+        @extschema@.pg_stat_get_vacuum_dead_tuples(i.oid),
+        @extschema@.pg_stat_get_vacuum_pages_deleted(i.oid),
+        @extschema@.pg_stat_get_vacuum_dead_pages(i.oid),
+        @extschema@.pg_stat_get_vacuum_pages_frozen(i.oid),
+        @extschema@.pg_stat_get_vacuum_pages_all_visible(i.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_frozen_pages(i.oid),
+        @extschema@.pg_stat_get_vacuum_rev_all_visible_pages(i.oid),
+        @extschema@.pg_stat_get_vacuum_wraparound_count(i.oid),
+        @extschema@.pg_stat_get_vacuum_total_time(i.oid) / 1000.0
+    FROM pg_catalog.pg_class c
+        JOIN pg_catalog.pg_index x ON c.oid = x.indrelid
+        JOIN pg_catalog.pg_class i ON i.oid = x.indexrelid
+        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+    WHERE c.relkind IN ('r', 't', 'm') AND i.relkind = 'i'
+$$
+LANGUAGE SQL EXECUTE ON ALL SEGMENTS;
+
+CREATE VIEW gp_stat_vacuum_indexes AS
+    SELECT * FROM gp_stat_get_master_vacuum_indexes() AS T
+        (gp_segment_id int, relid oid, indexrelid oid, schemaname name,
+         relname name, indexrelname name,
+         tuples_deleted int8, dead_tuples int8,
+         pages_deleted int8, dead_pages int8,
+         pages_frozen int8, pages_all_visible int8,
+         rev_all_frozen_pages int8, rev_all_visible_pages int8,
+         wraparound_vacuum_count int8, total_time numeric)
+    UNION ALL
+    SELECT * FROM gp_stat_get_segment_vacuum_indexes() AS T
+        (gp_segment_id int, relid oid, indexrelid oid, schemaname name,
+         relname name, indexrelname name,
+         tuples_deleted int8, dead_tuples int8,
+         pages_deleted int8, dead_pages int8,
+         pages_frozen int8, pages_all_visible int8,
+         rev_all_frozen_pages int8, rev_all_visible_pages int8,
+         wraparound_vacuum_count int8, total_time numeric);
+
+CREATE FUNCTION gp_stat_get_master_vacuum_database() RETURNS SETOF RECORD AS
+$$
+    SELECT pg_catalog.gp_execution_segment() AS gp_segment_id,
+        d.oid, d.datname,
+        @extschema@.pg_stat_get_db_vacuum_tuples_deleted(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_dead_tuples(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_pages_deleted(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_dead_pages(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_pages_frozen(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_pages_all_visible(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_rev_all_frozen_pages(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_rev_all_visible_pages(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_wraparound_count(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_total_time(d.oid) / 1000.0
+    FROM pg_catalog.pg_database d
+$$
+LANGUAGE SQL EXECUTE ON MASTER;
+
+CREATE FUNCTION gp_stat_get_segment_vacuum_database() RETURNS SETOF RECORD AS
+$$
+    SELECT pg_catalog.gp_execution_segment() AS gp_segment_id,
+        d.oid, d.datname,
+        @extschema@.pg_stat_get_db_vacuum_tuples_deleted(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_dead_tuples(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_pages_deleted(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_dead_pages(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_pages_frozen(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_pages_all_visible(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_rev_all_frozen_pages(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_rev_all_visible_pages(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_wraparound_count(d.oid),
+        @extschema@.pg_stat_get_db_vacuum_total_time(d.oid) / 1000.0
+    FROM pg_catalog.pg_database d
+$$
+LANGUAGE SQL EXECUTE ON ALL SEGMENTS;
+
+CREATE VIEW gp_stat_vacuum_database AS
+    SELECT * FROM gp_stat_get_master_vacuum_database() AS T
+        (gp_segment_id int, datid oid, datname name,
+         tuples_deleted int8, dead_tuples int8,
+         pages_deleted int8, dead_pages int8,
+         pages_frozen int8, pages_all_visible int8,
+         rev_all_frozen_pages int8, rev_all_visible_pages int8,
+         wraparound_vacuum_count int8, total_time numeric)
+    UNION ALL
+    SELECT * FROM gp_stat_get_segment_vacuum_database() AS T
+        (gp_segment_id int, datid oid, datname name,
+         tuples_deleted int8, dead_tuples int8,
+         pages_deleted int8, dead_pages int8,
+         pages_frozen int8, pages_all_visible int8,
+         rev_all_frozen_pages int8, rev_all_visible_pages int8,
+         wraparound_vacuum_count int8, total_time numeric);
+
+GRANT SELECT ON pg_stat_vacuum_tables, pg_stat_vacuum_indexes,
+    pg_stat_vacuum_database, gp_stat_vacuum_tables,
+    gp_stat_vacuum_indexes, gp_stat_vacuum_database TO PUBLIC;
diff --git a/contrib/vacuum_stats/vacuum_stats.c b/contrib/vacuum_stats/vacuum_stats.c
new file mode 100644
index 00000000000..80a9273880d
--- /dev/null
+++ b/contrib/vacuum_stats/vacuum_stats.c
@@ -0,0 +1,124 @@
+/*-------------------------------------------------------------------------
+ *
+ * vacuum_stats.c
+ *		Expose the vacuum counters accumulated by the statistics collector
+ *		for relations (tables and indexes) and databases.
+ *
+ * The counters themselves are gathered by (auto)vacuum and delivered to
+ * the statistics collector via PGSTAT_MTYPE_VACSTATS messages; here we
+ * only read them back through the regular pgstat fetch API, so no system
+ * catalog changes are required.
+ *
+ * contrib/vacuum_stats/vacuum_stats.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "fmgr.h"
+#include "pgstat.h"
+
+PG_MODULE_MAGIC;
+
+/*
+ * Fetch the vacuum counters for a relation (table or index), or NULL if
+ * the statistics collector has no entry for it.
+ */
+static PgStat_VacuumStats *
+fetch_rel_vacuum_stats(Oid relid)
+{
+	PgStat_StatTabEntry *tabentry;
+
+	tabentry = pgstat_fetch_stat_tabentry(relid);
+	if (tabentry == NULL)
+		return NULL;
+
+	return &tabentry->vacuum_stats;
+}
+
+/*
+ * Fetch the per-database vacuum counters, or NULL if the statistics
+ * collector has no entry for the database.
+ */
+static PgStat_VacuumStats *
+fetch_db_vacuum_stats(Oid dbid)
+{
+	PgStat_StatDBEntry *dbentry;
+
+	dbentry = pgstat_fetch_stat_dbentry(dbid);
+	if (dbentry == NULL)
+		return NULL;
+
+	return &dbentry->n_vacuum_stats;
+}
+
+#define DEFINE_REL_VACSTAT_FUNC(funcname, field) \
+PG_FUNCTION_INFO_V1(funcname); \
+Datum \
+funcname(PG_FUNCTION_ARGS) \
+{ \
+	Oid			relid = PG_GETARG_OID(0); \
+	PgStat_VacuumStats *stats = fetch_rel_vacuum_stats(relid); \
+\
+	PG_RETURN_INT64(stats ? (int64) stats->field : 0); \
+}
+
+#define DEFINE_DB_VACSTAT_FUNC(funcname, field) \
+PG_FUNCTION_INFO_V1(funcname); \
+Datum \
+funcname(PG_FUNCTION_ARGS) \
+{ \
+	Oid			dbid = PG_GETARG_OID(0); \
+	PgStat_VacuumStats *stats = fetch_db_vacuum_stats(dbid); \
+\
+	PG_RETURN_INT64(stats ? (int64) stats->field : 0); \
+}
+
+/*
+ * The "rev" counters live directly in the relation/database entries, not
+ * in the embedded PgStat_VacuumStats, since they are fed from the regular
+ * relation statistics rather than from the vacuum report.
+ */
+#define DEFINE_REL_ENTRY_FUNC(funcname, field) \
+PG_FUNCTION_INFO_V1(funcname); \
+Datum \
+funcname(PG_FUNCTION_ARGS) \
+{ \
+	Oid			relid = PG_GETARG_OID(0); \
+	PgStat_StatTabEntry *tabentry = pgstat_fetch_stat_tabentry(relid); \
+\
+	PG_RETURN_INT64(tabentry ? (int64) tabentry->field : 0); \
+}
+
+#define DEFINE_DB_ENTRY_FUNC(funcname, field) \
+PG_FUNCTION_INFO_V1(funcname); \
+Datum \
+funcname(PG_FUNCTION_ARGS) \
+{ \
+	Oid			dbid = PG_GETARG_OID(0); \
+	PgStat_StatDBEntry *dbentry = pgstat_fetch_stat_dbentry(dbid); \
+\
+	PG_RETURN_INT64(dbentry ? (int64) dbentry->field : 0); \
+}
+
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_tuples_deleted, tuples_deleted)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_dead_tuples, dead_tuples)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_pages_deleted, pages_deleted)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_dead_pages, dead_pages)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_pages_frozen, pages_frozen)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_pages_all_visible, pages_all_visible)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_wraparound_count, wraparound_vacuum_count)
+DEFINE_REL_ENTRY_FUNC(pg_stat_get_vacuum_rev_all_frozen_pages, rev_all_frozen_pages)
+DEFINE_REL_ENTRY_FUNC(pg_stat_get_vacuum_rev_all_visible_pages, rev_all_visible_pages)
+DEFINE_REL_VACSTAT_FUNC(pg_stat_get_vacuum_total_time, total_time)
+
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_tuples_deleted, tuples_deleted)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_dead_tuples, dead_tuples)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_pages_deleted, pages_deleted)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_dead_pages, dead_pages)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_pages_frozen, pages_frozen)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_pages_all_visible, pages_all_visible)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_wraparound_count, wraparound_vacuum_count)
+DEFINE_DB_ENTRY_FUNC(pg_stat_get_db_vacuum_rev_all_frozen_pages, n_rev_all_frozen_pages)
+DEFINE_DB_ENTRY_FUNC(pg_stat_get_db_vacuum_rev_all_visible_pages, n_rev_all_visible_pages)
+DEFINE_DB_VACSTAT_FUNC(pg_stat_get_db_vacuum_total_time, total_time)
diff --git a/contrib/vacuum_stats/vacuum_stats.control b/contrib/vacuum_stats/vacuum_stats.control
new file mode 100644
index 00000000000..5f80449984d
--- /dev/null
+++ b/contrib/vacuum_stats/vacuum_stats.control
@@ -0,0 +1,5 @@
+# vacuum_stats extension
+comment = 'per-relation and per-database vacuum statistics'
+default_version = '1.0'
+module_pathname = '$libdir/vacuum_stats'
+relocatable = false
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 942f87d6088..faa9116772f 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2490,6 +2490,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 		visibilitymap_clear(relation,
 							ItemPointerGetBlockNumber(&(heaptup->t_self)),
 							vmbuffer);
+		pgstat_count_rev_all_visible(relation);
 	}
 
 	/*
@@ -2831,6 +2832,7 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
 			visibilitymap_clear(relation,
 								BufferGetBlockNumber(buffer),
 								vmbuffer);
+			pgstat_count_rev_all_visible(relation);
 		}
 
 		/*
@@ -3392,6 +3394,7 @@ heap_delete(Relation relation, ItemPointer tid,
 		PageClearAllVisible(page);
 		visibilitymap_clear(relation, BufferGetBlockNumber(buffer),
 							vmbuffer);
+		pgstat_count_rev_all_visible(relation);
 	}
 
 	/* store transaction information of xact deleting the tuple */
@@ -4347,6 +4350,7 @@ heap_update_internal(Relation relation, ItemPointer otid, HeapTuple newtup,
 		PageClearAllVisible(BufferGetPage(buffer));
 		visibilitymap_clear(relation, BufferGetBlockNumber(buffer),
 							vmbuffer);
+		pgstat_count_rev_all_visible(relation);
 	}
 	if (newbuf != buffer && PageIsAllVisible(BufferGetPage(newbuf)))
 	{
@@ -4354,6 +4358,7 @@ heap_update_internal(Relation relation, ItemPointer otid, HeapTuple newtup,
 		PageClearAllVisible(BufferGetPage(newbuf));
 		visibilitymap_clear(relation, BufferGetBlockNumber(newbuf),
 							vmbuffer_new);
+		pgstat_count_rev_all_visible(relation);
 	}
 
 	if (newbuf != buffer)
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index a2ec3737a2a..4314217fcb1 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -127,6 +127,9 @@ typedef struct LVRelStats
 	double		new_dead_tuples;	/* new estimated total # of dead tuples */
 	BlockNumber pages_removed;
 	double		tuples_deleted;
+	BlockNumber dead_pages;		/* pages left with unremovable dead tuples */
+	BlockNumber pages_frozen;	/* pages where we froze tuples */
+	BlockNumber pages_all_visible;	/* pages we marked all-visible */
 	BlockNumber nonempty_pages; /* actually, last nonempty page + 1 */
 	/* List of TIDs of tuples we intend to delete */
 	/* NB: this list is ordered by TID address */
@@ -157,10 +160,12 @@ static void lazy_vacuum_heap(Relation onerel, LVRelStats *vacrelstats);
 static bool lazy_check_needs_freeze(Buffer buf);
 static void lazy_vacuum_index(Relation indrel,
 				  IndexBulkDeleteResult **stats,
-				  LVRelStats *vacrelstats);
+				  LVRelStats *vacrelstats,
+				  uint64 *elapsed_us);
 static void lazy_cleanup_index(Relation indrel,
 				   IndexBulkDeleteResult *stats,
-				   LVRelStats *vacrelstats);
+				   LVRelStats *vacrelstats,
+				   uint64 *elapsed_us);
 static int lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 				 int tupindex, LVRelStats *vacrelstats, Buffer *vmbuffer);
 static void lazy_truncate_heap(Relation onerel, LVRelStats *vacrelstats);
@@ -198,12 +203,15 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 	BlockNumber possibly_freeable;
 	PGRUsage	ru0;
 	TimestampTz starttime = 0;
+	instr_time	vacstart;
+	instr_time	vacend;
 	long		secs;
 	int			usecs;
 	double		read_rate,
 				write_rate;
 	bool		scan_all;		/* should we scan all pages? */
 	bool		scanned_all;	/* did we actually scan all pages? */
+	bool		wraparound;		/* did the freeze table age force it? */
 	TransactionId xidFullScanLimit;
 	MultiXactId mxactFullScanLimit;
 	BlockNumber new_rel_pages;
@@ -213,6 +221,9 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 	TransactionId new_frozen_xid;
 	MultiXactId new_min_multi;
 
+	/* measure elapsed time for the vacuum statistics */
+	INSTR_TIME_SET_CURRENT(vacstart);
+
 	/* measure elapsed time iff autovacuum logging requires it */
 	if (IsAutoVacuumWorkerProcess() && Log_autovacuum_min_duration >= 0)
 	{
@@ -272,6 +283,21 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 	scan_all |= MultiXactIdPrecedesOrEquals(onerel->rd_rel->relminmxid,
 											mxactFullScanLimit);
 
+	/*
+	 * Remember whether the relation really reached the freeze table age, for
+	 * the vacuum statistics.  Note this is not the same as scan_all: on a
+	 * young cluster the limits above are clamped to their minimum values, so
+	 * every relation matches them and every vacuum ends up scanning the whole
+	 * relation.  Such runs are not driven by wraparound pressure, so don't
+	 * report them as such.
+	 */
+	wraparound = (TransactionIdPrecedesOrEquals(onerel->rd_rel->relfrozenxid,
+												xidFullScanLimit) &&
+				  xidFullScanLimit != FirstNormalTransactionId) ||
+		(MultiXactIdPrecedesOrEquals(onerel->rd_rel->relminmxid,
+									 mxactFullScanLimit) &&
+		 mxactFullScanLimit != FirstMultiXactId);
+
 	/*
 	 * Execute the various vacuum operations. Appendonly tables are treated
 	 * differently.
@@ -388,6 +414,29 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 						 new_live_tuples,
 						 vacrelstats->new_dead_tuples);
 
+	/* report the per-vacuum counters as well */
+	{
+		PgStat_VacuumStats vacstats;
+
+		MemSet(&vacstats, 0, sizeof(vacstats));
+		vacstats.tuples_deleted = (PgStat_Counter) vacrelstats->tuples_deleted;
+		vacstats.dead_tuples = (PgStat_Counter) vacrelstats->new_dead_tuples;
+		vacstats.pages_deleted = (PgStat_Counter) vacrelstats->pages_removed;
+		vacstats.dead_pages = (PgStat_Counter) vacrelstats->dead_pages;
+		vacstats.pages_frozen = (PgStat_Counter) vacrelstats->pages_frozen;
+		vacstats.pages_all_visible =
+			(PgStat_Counter) vacrelstats->pages_all_visible;
+		vacstats.wraparound_vacuum_count = wraparound ? 1 : 0;
+
+		INSTR_TIME_SET_CURRENT(vacend);
+		INSTR_TIME_SUBTRACT(vacend, vacstart);
+		vacstats.total_time = (PgStat_Counter) INSTR_TIME_GET_MICROSEC(vacend);
+
+		pgstat_report_vacstats(RelationGetRelid(onerel),
+							   onerel->rd_rel->relisshared,
+							   &vacstats);
+	}
+
 	if (gp_indexcheck_vacuum == INDEX_CHECK_ALL ||
 		(gp_indexcheck_vacuum == INDEX_CHECK_SYSTEM &&
 		 PG_CATALOG_NAMESPACE == RelationGetNamespace(onerel)))
@@ -747,6 +796,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 				nkeep,
 				nunused;
 	IndexBulkDeleteResult **indstats;
+	uint64	   *indelapsed;
 	int			i;
 	PGRUsage	ru0;
 	Buffer		vmbuffer = InvalidBuffer;
@@ -767,6 +817,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 
 	indstats = (IndexBulkDeleteResult **)
 		palloc0(nindexes * sizeof(IndexBulkDeleteResult *));
+	indelapsed = (uint64 *) palloc0(nindexes * sizeof(uint64));
 
 	nblocks = RelationGetNumberOfBlocks(onerel);
 	vacrelstats->rel_pages = nblocks;
@@ -831,6 +882,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 		bool		tupgone,
 					hastup;
 		int			prev_dead_count;
+		double		prev_nkeep;
 		int			nfrozen;
 		Size		freespace;
 		bool		all_visible_according_to_vm;
@@ -898,7 +950,8 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			for (i = 0; i < nindexes; i++)
 				lazy_vacuum_index(Irel[i],
 								  &indstats[i],
-								  vacrelstats);
+								  vacrelstats,
+								  &indelapsed[i]);
 			/* Remove tuples from heap */
 			lazy_vacuum_heap(onerel, vacrelstats);
 
@@ -1038,6 +1091,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 				PageSetAllVisible(page);
 				visibilitymap_set(onerel, blkno, buf, InvalidXLogRecPtr,
 								  vmbuffer, InvalidTransactionId);
+				vacrelstats->pages_all_visible++;
 				END_CRIT_SECTION();
 			}
 
@@ -1063,6 +1117,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 		nfrozen = 0;
 		hastup = false;
 		prev_dead_count = vacrelstats->num_dead_tuples;
+		prev_nkeep = nkeep;
 		maxoff = PageGetMaxOffsetNumber(page);
 
 		/*
@@ -1238,6 +1293,10 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			}
 		}						/* scan along page */
 
+		/* Remember pages that keep dead tuples we could not remove yet */
+		if (nkeep > prev_nkeep)
+			vacrelstats->dead_pages++;
+
 		/*
 		 * If we froze any tuples, mark the buffer dirty, and write a WAL
 		 * record recording the changes.  We must log the changes to be
@@ -1245,6 +1304,8 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 		 */
 		if (nfrozen > 0)
 		{
+			vacrelstats->pages_frozen++;
+
 			START_CRIT_SECTION();
 
 			MarkBufferDirty(buf);
@@ -1316,6 +1377,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			MarkBufferDirty(buf);
 			visibilitymap_set(onerel, blkno, buf, InvalidXLogRecPtr,
 							  vmbuffer, visibility_cutoff_xid);
+			vacrelstats->pages_all_visible++;
 		}
 
 		/*
@@ -1331,6 +1393,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			elog(WARNING, "page is not marked all-visible but visibility map bit is set in relation \"%s\" page %u",
 				 relname, blkno);
 			visibilitymap_clear(onerel, blkno, vmbuffer);
+			pgstat_count_rev_all_visible(onerel);
 		}
 
 		/*
@@ -1353,6 +1416,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			PageClearAllVisible(page);
 			MarkBufferDirty(buf);
 			visibilitymap_clear(onerel, blkno, vmbuffer);
+			pgstat_count_rev_all_visible(onerel);
 		}
 
 		UnlockReleaseBuffer(buf);
@@ -1408,7 +1472,8 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 		for (i = 0; i < nindexes; i++)
 			lazy_vacuum_index(Irel[i],
 							  &indstats[i],
-							  vacrelstats);
+							  vacrelstats,
+							  &indelapsed[i]);
 		/* Remove tuples from heap */
 		lazy_vacuum_heap(onerel, vacrelstats);
 		vacrelstats->num_index_scans++;
@@ -1416,7 +1481,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 
 	/* Do post-vacuum cleanup and statistics update for each index */
 	for (i = 0; i < nindexes; i++)
-		lazy_cleanup_index(Irel[i], indstats[i], vacrelstats);
+		lazy_cleanup_index(Irel[i], indstats[i], vacrelstats, &indelapsed[i]);
 
 	/* If no indexes, make log report that lazy_vacuum_heap would've made */
 	if (vacuumed_pages)
@@ -1590,6 +1655,7 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 		Assert(BufferIsValid(*vmbuffer));
 		visibilitymap_set(onerel, blkno, buffer, InvalidXLogRecPtr, *vmbuffer,
 						  visibility_cutoff_xid);
+		vacrelstats->pages_all_visible++;
 	}
 
 	return tupindex;
@@ -1649,12 +1715,16 @@ lazy_check_needs_freeze(Buffer buf)
 static void
 lazy_vacuum_index(Relation indrel,
 				  IndexBulkDeleteResult **stats,
-				  LVRelStats *vacrelstats)
+				  LVRelStats *vacrelstats,
+				  uint64 *elapsed_us)
 {
 	IndexVacuumInfo ivinfo;
 	PGRUsage	ru0;
+	instr_time	starttime;
+	instr_time	endtime;
 
 	pg_rusage_init(&ru0);
+	INSTR_TIME_SET_CURRENT(starttime);
 
 	ivinfo.index = indrel;
 	ivinfo.analyze_only = false;
@@ -1667,6 +1737,10 @@ lazy_vacuum_index(Relation indrel,
 	*stats = index_bulk_delete(&ivinfo, *stats,
 							   lazy_tid_reaped, (void *) vacrelstats);
 
+	INSTR_TIME_SET_CURRENT(endtime);
+	INSTR_TIME_SUBTRACT(endtime, starttime);
+	*elapsed_us += (uint64) INSTR_TIME_GET_MICROSEC(endtime);
+
 	ereport(elevel,
 			(errmsg("scanned index \"%s\" to remove %d row versions",
 					RelationGetRelationName(indrel),
@@ -1680,12 +1754,16 @@ lazy_vacuum_index(Relation indrel,
 static void
 lazy_cleanup_index(Relation indrel,
 				   IndexBulkDeleteResult *stats,
-				   LVRelStats *vacrelstats)
+				   LVRelStats *vacrelstats,
+				   uint64 *elapsed_us)
 {
 	IndexVacuumInfo ivinfo;
 	PGRUsage	ru0;
+	instr_time	starttime;
+	instr_time	endtime;
 
 	pg_rusage_init(&ru0);
+	INSTR_TIME_SET_CURRENT(starttime);
 
 	ivinfo.index = indrel;
 	ivinfo.analyze_only = false;
@@ -1696,6 +1774,10 @@ lazy_cleanup_index(Relation indrel,
 
 	stats = index_vacuum_cleanup(&ivinfo, stats);
 
+	INSTR_TIME_SET_CURRENT(endtime);
+	INSTR_TIME_SUBTRACT(endtime, starttime);
+	*elapsed_us += (uint64) INSTR_TIME_GET_MICROSEC(endtime);
+
 	if (!stats)
 		return;
 
@@ -1714,6 +1796,24 @@ lazy_cleanup_index(Relation indrel,
 							false,
 							true /* isvacuum */);
 
+	/* report the counters accumulated while vacuuming this index */
+	{
+		PgStat_VacuumStats vacstats;
+
+		MemSet(&vacstats, 0, sizeof(vacstats));
+		vacstats.tuples_deleted = (PgStat_Counter) stats->tuples_removed;
+		vacstats.pages_deleted = (PgStat_Counter) stats->pages_deleted;
+		/* deleted pages that are not yet reusable still hold dead entries */
+		if (stats->pages_deleted > stats->pages_free)
+			vacstats.dead_pages =
+				(PgStat_Counter) (stats->pages_deleted - stats->pages_free);
+		vacstats.total_time = (PgStat_Counter) *elapsed_us;
+
+		pgstat_report_vacstats(RelationGetRelid(indrel),
+							   indrel->rd_rel->relisshared,
+							   &vacstats);
+	}
+
 	ereport(elevel,
 			(errmsg("index \"%s\" now contains %.0f row versions in %u pages",
 					RelationGetRelationName(indrel),
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index c6e4524e40f..0ef13b0416d 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -309,6 +309,7 @@ static void pgstat_recv_resetsharedcounter(PgStat_MsgResetsharedcounter *msg, in
 static void pgstat_recv_resetsinglecounter(PgStat_MsgResetsinglecounter *msg, int len);
 static void pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len);
 static void pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len);
+static void pgstat_recv_vacstats(PgStat_MsgVacstats *msg, int len);
 static void pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len);
 static void pgstat_recv_archiver(PgStat_MsgArchiver *msg, int len);
 static void pgstat_recv_queuestat(PgStat_MsgQueuestat *msg, int len); /* GPDB */
@@ -1399,6 +1400,29 @@ pgstat_report_vacuum(Oid tableoid, bool shared,
 	pgstat_send(&msg, sizeof(msg));
 }
 
+/* ---------
+ * pgstat_report_vacstats() -
+ *
+ *	Tell the collector about the counters accumulated while vacuuming a
+ *	relation (a heap relation or an index).
+ * ---------
+ */
+void
+pgstat_report_vacstats(Oid tableoid, bool shared,
+					   const PgStat_VacuumStats *stats)
+{
+	PgStat_MsgVacstats msg;
+
+	if (pgStatSock == PGINVALID_SOCKET || !pgstat_track_counts)
+		return;
+
+	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_VACSTATS);
+	msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
+	msg.m_tableoid = tableoid;
+	msg.m_stats = *stats;
+	pgstat_send(&msg, sizeof(msg));
+}
+
 /* --------
  * pgstat_report_analyze() -
  *
@@ -3540,6 +3564,10 @@ PgstatCollectorMain(int argc, char *argv[])
 					pgstat_recv_vacuum((PgStat_MsgVacuum *) &msg, len);
 					break;
 
+				case PGSTAT_MTYPE_VACSTATS:
+					pgstat_recv_vacstats((PgStat_MsgVacstats *) &msg, len);
+					break;
+
 				case PGSTAT_MTYPE_ANALYZE:
 					pgstat_recv_analyze((PgStat_MsgAnalyze *) &msg, len);
 					break;
@@ -3676,6 +3704,9 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry)
 	dbentry->n_deadlocks = 0;
 	dbentry->n_block_read_time = 0;
 	dbentry->n_block_write_time = 0;
+	MemSet(&dbentry->n_vacuum_stats, 0, sizeof(dbentry->n_vacuum_stats));
+	dbentry->n_rev_all_frozen_pages = 0;
+	dbentry->n_rev_all_visible_pages = 0;
 
 	dbentry->stat_reset_timestamp = GetCurrentTimestamp();
 	dbentry->stats_timestamp = 0;
@@ -3772,6 +3803,9 @@ pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create)
 		result->analyze_count = 0;
 		result->autovac_analyze_timestamp = 0;
 		result->autovac_analyze_count = 0;
+		MemSet(&result->vacuum_stats, 0, sizeof(result->vacuum_stats));
+		result->rev_all_frozen_pages = 0;
+		result->rev_all_visible_pages = 0;
 	}
 
 	return result;
@@ -4985,6 +5019,11 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 			tabentry->analyze_count = 0;
 			tabentry->autovac_analyze_timestamp = 0;
 			tabentry->autovac_analyze_count = 0;
+			MemSet(&tabentry->vacuum_stats, 0, sizeof(tabentry->vacuum_stats));
+			tabentry->rev_all_frozen_pages =
+				tabmsg->t_counts.t_rev_all_frozen_pages;
+			tabentry->rev_all_visible_pages =
+				tabmsg->t_counts.t_rev_all_visible_pages;
 		}
 		else
 		{
@@ -5003,6 +5042,10 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 			tabentry->changes_since_analyze += tabmsg->t_counts.t_changed_tuples;
 			tabentry->blocks_fetched += tabmsg->t_counts.t_blocks_fetched;
 			tabentry->blocks_hit += tabmsg->t_counts.t_blocks_hit;
+			tabentry->rev_all_frozen_pages +=
+				tabmsg->t_counts.t_rev_all_frozen_pages;
+			tabentry->rev_all_visible_pages +=
+				tabmsg->t_counts.t_rev_all_visible_pages;
 		}
 
 		/* Clamp n_live_tuples in case of negative delta_live_tuples */
@@ -5020,6 +5063,10 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 		dbentry->n_tuples_deleted += tabmsg->t_counts.t_tuples_deleted;
 		dbentry->n_blocks_fetched += tabmsg->t_counts.t_blocks_fetched;
 		dbentry->n_blocks_hit += tabmsg->t_counts.t_blocks_hit;
+		dbentry->n_rev_all_frozen_pages +=
+			tabmsg->t_counts.t_rev_all_frozen_pages;
+		dbentry->n_rev_all_visible_pages +=
+			tabmsg->t_counts.t_rev_all_visible_pages;
 	}
 }
 
@@ -5247,6 +5294,44 @@ pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
 	}
 }
 
+/* ----------
+ * pgstat_recv_vacstats() -
+ *
+ *	Process a VACSTATS message: accumulate the vacuum counters into the
+ *	relation's entry and into the per-database totals.
+ * ----------
+ */
+static void
+pgstat_recv_vacstats(PgStat_MsgVacstats *msg, int len)
+{
+	PgStat_StatDBEntry *dbentry;
+	PgStat_StatTabEntry *tabentry;
+
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
+
+	tabentry = pgstat_get_tab_entry(dbentry, msg->m_tableoid, true);
+
+	tabentry->vacuum_stats.tuples_deleted += msg->m_stats.tuples_deleted;
+	tabentry->vacuum_stats.dead_tuples += msg->m_stats.dead_tuples;
+	tabentry->vacuum_stats.pages_deleted += msg->m_stats.pages_deleted;
+	tabentry->vacuum_stats.dead_pages += msg->m_stats.dead_pages;
+	tabentry->vacuum_stats.pages_frozen += msg->m_stats.pages_frozen;
+	tabentry->vacuum_stats.pages_all_visible += msg->m_stats.pages_all_visible;
+	tabentry->vacuum_stats.wraparound_vacuum_count +=
+		msg->m_stats.wraparound_vacuum_count;
+	tabentry->vacuum_stats.total_time += msg->m_stats.total_time;
+
+	dbentry->n_vacuum_stats.tuples_deleted += msg->m_stats.tuples_deleted;
+	dbentry->n_vacuum_stats.dead_tuples += msg->m_stats.dead_tuples;
+	dbentry->n_vacuum_stats.pages_deleted += msg->m_stats.pages_deleted;
+	dbentry->n_vacuum_stats.dead_pages += msg->m_stats.dead_pages;
+	dbentry->n_vacuum_stats.pages_frozen += msg->m_stats.pages_frozen;
+	dbentry->n_vacuum_stats.pages_all_visible += msg->m_stats.pages_all_visible;
+	dbentry->n_vacuum_stats.wraparound_vacuum_count +=
+		msg->m_stats.wraparound_vacuum_count;
+	dbentry->n_vacuum_stats.total_time += msg->m_stats.total_time;
+}
+
 /* ----------
  * pgstat_recv_analyze() -
  *
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 05691474b57..21c0218763b 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -63,7 +63,8 @@ typedef enum StatMsgType
 	PGSTAT_MTYPE_FUNCPURGE,
 	PGSTAT_MTYPE_RECOVERYCONFLICT,
 	PGSTAT_MTYPE_TEMPFILE,
-	PGSTAT_MTYPE_DEADLOCK
+	PGSTAT_MTYPE_DEADLOCK,
+	PGSTAT_MTYPE_VACSTATS
 } StatMsgType;
 
 /* ----------
@@ -111,6 +112,9 @@ typedef struct PgStat_TableCounts
 
 	PgStat_Counter t_blocks_fetched;
 	PgStat_Counter t_blocks_hit;
+
+	PgStat_Counter t_rev_all_frozen_pages;
+	PgStat_Counter t_rev_all_visible_pages;
 } PgStat_TableCounts;
 
 /* Possible targets for resetting cluster-wide shared values */
@@ -366,6 +370,47 @@ typedef struct PgStat_MsgVacuum
 } PgStat_MsgVacuum;
 
 
+/* ----------
+ * PgStat_VacuumStats			Counters accumulated by (auto)vacuum for a
+ *								single relation.  Tuple counters cover heap
+ *								relations, page counters both heap relations
+ *								and indexes.
+ * ----------
+ */
+typedef struct PgStat_VacuumStats
+{
+	PgStat_Counter tuples_deleted;		/* tuples removed by vacuum */
+	PgStat_Counter dead_tuples;			/* dead tuples left unremoved */
+	PgStat_Counter pages_deleted;		/* pages removed/deleted by vacuum */
+	PgStat_Counter dead_pages;			/* pages with unremoved dead tuples */
+	PgStat_Counter pages_frozen;		/* pages where vacuum froze tuples */
+	PgStat_Counter pages_all_visible;	/* pages marked all-visible by vacuum */
+
+	/*
+	 * Number of vacuum runs that had to scan the whole relation because its
+	 * relfrozenxid/relminmxid reached the freeze table age.  Such a run
+	 * cannot skip pages using the visibility map, so it is much more
+	 * expensive than an ordinary one.
+	 */
+	PgStat_Counter wraparound_vacuum_count;
+
+	PgStat_Counter total_time;	/* total vacuum time, in microseconds */
+} PgStat_VacuumStats;
+
+/* ----------
+ * PgStat_MsgVacstats			Sent by the backend or autovacuum daemon
+ *								after vacuuming a heap relation or an index
+ *								to report per-relation vacuum counters.
+ * ----------
+ */
+typedef struct PgStat_MsgVacstats
+{
+	PgStat_MsgHdr m_hdr;
+	Oid			m_databaseid;
+	Oid			m_tableoid;
+	PgStat_VacuumStats m_stats;
+} PgStat_MsgVacstats;
+
 /* ----------
  * PgStat_MsgAnalyze			Sent by the backend or autovacuum daemon
  *								after ANALYZE
@@ -557,6 +602,7 @@ typedef union PgStat_Msg
 	PgStat_MsgResetsinglecounter msg_resetsinglecounter;
 	PgStat_MsgAutovacStart msg_autovacuum;
 	PgStat_MsgVacuum msg_vacuum;
+	PgStat_MsgVacstats msg_vacstats;
 	PgStat_MsgAnalyze msg_analyze;
 	PgStat_MsgArchiver msg_archiver;
 	PgStat_MsgQueuestat msg_queuestat;  /* GPDB */
@@ -576,7 +622,7 @@ typedef union PgStat_Msg
  * ------------------------------------------------------------
  */
 
-#define PGSTAT_FILE_FORMAT_ID	0x01A5BC9C
+#define PGSTAT_FILE_FORMAT_ID	0x01A5BC9D
 
 /* ----------
  * PgStat_StatDBEntry			The collector's data per database
@@ -605,6 +651,10 @@ typedef struct PgStat_StatDBEntry
 	PgStat_Counter n_deadlocks;
 	PgStat_Counter n_block_read_time;	/* times in microseconds */
 	PgStat_Counter n_block_write_time;
+	PgStat_VacuumStats n_vacuum_stats;	/* summed over the db's relations */
+	/* likewise; fed from the relation statistics, see PgStat_StatTabEntry */
+	PgStat_Counter n_rev_all_frozen_pages;
+	PgStat_Counter n_rev_all_visible_pages;
 
 	TimestampTz stat_reset_timestamp;
 	TimestampTz stats_timestamp;	/* time of db stats file update */
@@ -651,6 +701,20 @@ typedef struct PgStat_StatTabEntry
 	PgStat_Counter analyze_count;
 	TimestampTz autovac_analyze_timestamp;		/* autovacuum initiated */
 	PgStat_Counter autovac_analyze_count;
+
+	PgStat_VacuumStats vacuum_stats;
+
+	/*
+	 * "rev" counters track how quickly the work done by vacuum is undone:
+	 * pages that lost their all-frozen/all-visible status.  Unlike
+	 * vacuum_stats above, they are fed from the regular relation statistics
+	 * (PgStat_TableCounts), not from the vacuum report, since the bits are
+	 * cleared by ordinary DML.  rev_all_frozen_pages is always zero in this
+	 * release, since the visibility map has no all-frozen bit yet; it is
+	 * kept for compatibility with the upstream vacuum statistics layout.
+	 */
+	PgStat_Counter rev_all_frozen_pages;
+	PgStat_Counter rev_all_visible_pages;
 } PgStat_StatTabEntry;
 
 
@@ -949,6 +1013,22 @@ extern void pgstat_reset_shared_counters(const char *);
 extern void pgstat_reset_single_counter(Oid objectid, PgStat_Single_Reset_Type type);
 
 extern void pgstat_report_autovac(Oid dboid);
+extern void pgstat_report_vacstats(Oid tableoid, bool shared,
+					   const PgStat_VacuumStats *stats);
+
+/* count a page whose all-visible bit is being cleared */
+#define pgstat_count_rev_all_visible(rel)							\
+	do {															\
+		if ((rel)->pgstat_info != NULL)								\
+			(rel)->pgstat_info->t_counts.t_rev_all_visible_pages++;	\
+	} while (0)
+/* count a page whose all-frozen bit is being cleared */
+#define pgstat_count_rev_all_frozen(rel)							\
+	do {															\
+		if ((rel)->pgstat_info != NULL)								\
+			(rel)->pgstat_info->t_counts.t_rev_all_frozen_pages++;	\
+	} while (0)
+
 extern void pgstat_report_vacuum(Oid tableoid, bool shared,
 					 PgStat_Counter livetuples, PgStat_Counter deadtuples);
 extern void pgstat_report_analyze(Relation rel,
diff --git a/src/test/isolation/expected/vacuum-extending-in-repeatable-read.out b/src/test/isolation/expected/vacuum-extending-in-repeatable-read.out
new file mode 100644
index 00000000000..3a69666f3ab
--- /dev/null
+++ b/src/test/isolation/expected/vacuum-extending-in-repeatable-read.out
@@ -0,0 +1,59 @@
+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_wait_dead_tuples s2_print_vacuum_stats_table s1_commit s2_checkpoint s2_vacuum s2_wait_tuples_deleted 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 sum(tuples_deleted) AS tuples_deleted,
+           sum(dead_tuples) AS dead_tuples,
+           sum(pages_frozen) AS pages_frozen
+    FROM gp_stat_vacuum_tables
+    WHERE relname = 'test_vacuum_stat_isolation';
+
+tuples_deleted dead_tuples    pages_frozen   
+
+0              0              0              
+step s1_begin_repeatable_read: 
+    BEGIN transaction ISOLATION LEVEL REPEATABLE READ;
+    select count(ival) from test_vacuum_stat_isolation where id>900;
+
+count          
+
+100            
+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_wait_dead_tuples: 
+    SELECT wait_vacuum_stats('SELECT sum(dead_tuples) = 100 FROM gp_stat_vacuum_tables WHERE relname = ''test_vacuum_stat_isolation''');
+
+wait_vacuum_stats
+
+t              
+step s2_print_vacuum_stats_table: 
+    SELECT sum(tuples_deleted) AS tuples_deleted,
+           sum(dead_tuples) AS dead_tuples,
+           sum(pages_frozen) AS pages_frozen
+    FROM gp_stat_vacuum_tables
+    WHERE relname = 'test_vacuum_stat_isolation';
+
+tuples_deleted dead_tuples    pages_frozen   
+
+0              100            0              
+step s1_commit: COMMIT;
+step s2_checkpoint: CHECKPOINT;
+step s2_vacuum: VACUUM test_vacuum_stat_isolation;
+step s2_wait_tuples_deleted: 
+    SELECT wait_vacuum_stats('SELECT sum(tuples_deleted) = 100 FROM gp_stat_vacuum_tables WHERE relname = ''test_vacuum_stat_isolation''');
+
+wait_vacuum_stats
+
+t              
+step s2_print_vacuum_stats_table: 
+    SELECT sum(tuples_deleted) AS tuples_deleted,
+           sum(dead_tuples) AS dead_tuples,
+           sum(pages_frozen) AS pages_frozen
+    FROM gp_stat_vacuum_tables
+    WHERE relname = 'test_vacuum_stat_isolation';
+
+tuples_deleted dead_tuples    pages_frozen   
+
+100            100            0              
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index b64c8195dff..4d0d93ccc2a 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -49,5 +49,6 @@ test: heap-repeatable-read-vacuum-freeze
 test: heap-repeatable-read-vacuum
 test: ao-repeatable-read-vacuum
 test: vacuum-full-permissions
+test: vacuum-extending-in-repeatable-read
 
 test: ao-insert-eof create_index_hot udf-insert-deadlock heap-repeatable-read ao-repeatable-read
diff --git a/src/test/isolation/specs/vacuum-extending-in-repeatable-read.spec b/src/test/isolation/specs/vacuum-extending-in-repeatable-read.spec
new file mode 100644
index 00000000000..93c017de357
--- /dev/null
+++ b/src/test/isolation/specs/vacuum-extending-in-repeatable-read.spec
@@ -0,0 +1,94 @@
+# Test for checking dead_tuples, tuples_deleted and pages_frozen in
+# pg_stat_vacuum_tables (provided by the vacuum_stats extension).
+#
+# Dead_tuples values are counted when vacuum cannot clean up unused tuples
+# while a snapshot is held by another transaction.  Dead_tuples aren't
+# increased after releasing the snapshot, unlike tuples_deleted, which is
+# increased by the number of tuples that the vacuum managed to remove.
+# Pages_frozen stays zero here: with the default vacuum_freeze_min_age
+# these tuples are far too young to be frozen.
+#
+# The counters are summed over the whole cluster with the
+# gp_stat_vacuum_tables view, since in Greenplum the data lives on the
+# segments.  The statistics collector is fed asynchronously, so before
+# printing the counters the wait steps poll until the vacuum report
+# becomes visible, resetting the cached statistics snapshots on the
+# master and the segments on every iteration.
+
+setup
+{
+    CREATE EXTENSION vacuum_stats;
+    CREATE TABLE test_vacuum_stat_isolation(id int, ival int) WITH (autovacuum_enabled = off);
+
+    CREATE FUNCTION clear_stats_snapshot_on_segments() RETURNS SETOF void AS
+    'SELECT pg_catalog.pg_stat_clear_snapshot()' LANGUAGE SQL EXECUTE ON ALL SEGMENTS;
+
+    CREATE FUNCTION wait_vacuum_stats(cond text) RETURNS bool AS '
+    DECLARE
+        ok bool;
+    BEGIN
+        FOR i IN 1..120 LOOP
+            PERFORM pg_stat_clear_snapshot();
+            PERFORM * FROM clear_stats_snapshot_on_segments();
+            EXECUTE cond INTO ok;
+            IF ok THEN
+                RETURN true;
+            END IF;
+            PERFORM pg_sleep(0.25);
+        END LOOP;
+        RETURN false;
+    END' LANGUAGE plpgsql;
+}
+
+teardown
+{
+    DROP TABLE test_vacuum_stat_isolation CASCADE;
+    DROP FUNCTION wait_vacuum_stats(text);
+    DROP FUNCTION clear_stats_snapshot_on_segments();
+    DROP EXTENSION vacuum_stats;
+}
+
+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_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_wait_dead_tuples"
+{
+    SELECT wait_vacuum_stats('SELECT sum(dead_tuples) = 100 FROM gp_stat_vacuum_tables WHERE relname = ''test_vacuum_stat_isolation''');
+}
+step "s2_wait_tuples_deleted"
+{
+    SELECT wait_vacuum_stats('SELECT sum(tuples_deleted) = 100 FROM gp_stat_vacuum_tables WHERE relname = ''test_vacuum_stat_isolation''');
+}
+step "s2_print_vacuum_stats_table"
+{
+    SELECT sum(tuples_deleted) AS tuples_deleted,
+           sum(dead_tuples) AS dead_tuples,
+           sum(pages_frozen) AS pages_frozen
+    FROM gp_stat_vacuum_tables
+    WHERE relname = 'test_vacuum_stat_isolation';
+}
+
+permutation
+    "s2_insert"
+    "s2_print_vacuum_stats_table"
+    "s1_begin_repeatable_read"
+    "s2_update"
+    "s2_insert_interrupt"
+    "s2_vacuum"
+    "s2_wait_dead_tuples"
+    "s2_print_vacuum_stats_table"
+    "s1_commit"
+    "s2_checkpoint"
+    "s2_vacuum"
+    "s2_wait_tuples_deleted"
+    "s2_print_vacuum_stats_table"