From be48f4c72f381e3a9ce4d2b54c739c9da0d92e2e Mon Sep 17 00:00:00 2001
From: Scott Ray <scott@scottray.io>
Date: Sat, 8 Aug 2026 19:52:20 -0700
Subject: [PATCH v4] Add pg_xmin_horizon view showing per-input horizon pins

Add a pg_xmin_horizon system view with one row per horizon input.
An input is a snapshot-holding backend, replication slot, prepared transaction,
or slot-less hot-standby-feedback walsender.  The view shows each input's raw
per-class contribution (shared_xmin, catalog_xmin, data_xmin) alongside
identifying columns (pid, slot_name, gid, datid, datname, xact_start).  The rows
that pin nothing, which are VACUUM backends, logical decoding backends,
and idle or invalidated slots, appear with NULL xmin columns.  The SRF,
pg_get_xmin_horizon(), reads procs in a single pass while holding
ProcArrayLock and reads slots in a separate pass while holding
ReplicationSlotControlLock to get the effective xmins that
ReplicationSlotsComputeRequiredXmin() uses.  The view is readable by
pg_read_all_stats.
---
 doc/src/sgml/maintenance.sgml                |  38 ++-
 doc/src/sgml/system-views.sgml               | 226 +++++++++++++++
 src/backend/catalog/system_views.sql         |  25 ++
 src/backend/storage/ipc/Makefile             |   1 +
 src/backend/storage/ipc/meson.build          |   1 +
 src/backend/storage/ipc/procarray.c          |  50 +++-
 src/backend/storage/ipc/procarrayfuncs.c     | 268 ++++++++++++++++++
 src/include/catalog/pg_proc.dat              |   9 +
 src/include/storage/procarray.h              |  11 +
 src/test/recovery/meson.build                |   1 +
 src/test/recovery/t/056_xmin_horizon.pl      | 281 +++++++++++++++++++
 src/test/regress/expected/rules.out          |  18 ++
 src/test/regress/expected/xmin_horizon.out   | 118 ++++++++
 src/test/regress/expected/xmin_horizon_1.out |  21 ++
 src/test/regress/parallel_schedule           |   2 +-
 src/test/regress/sql/xmin_horizon.sql        |  90 ++++++
 src/tools/pgindent/typedefs.list             |   2 +
 17 files changed, 1143 insertions(+), 19 deletions(-)
 create mode 100644 src/backend/storage/ipc/procarrayfuncs.c
 create mode 100644 src/test/recovery/t/056_xmin_horizon.pl
 create mode 100644 src/test/regress/expected/xmin_horizon.out
 create mode 100644 src/test/regress/expected/xmin_horizon_1.out
 create mode 100644 src/test/regress/sql/xmin_horizon.sql

diff --git a/doc/src/sgml/maintenance.sgml b/doc/src/sgml/maintenance.sgml
index 33ab4edf87c..29cf1967879 100644
--- a/doc/src/sgml/maintenance.sgml
+++ b/doc/src/sgml/maintenance.sgml
@@ -701,25 +701,33 @@ HINT:  Execute a database-wide VACUUM in that database.
 
     <orderedlist>
      <listitem>
-      <simpara>Resolve old prepared transactions. You can find these by checking
-       <link linkend="view-pg-prepared-xacts"><structname>pg_prepared_xacts</structname></link> for rows where
-       <literal>age(transactionid)</literal> is large. Such transactions should be
-       committed or rolled back.</simpara>
+      <simpara>Identify the source.  Query
+       <link linkend="view-pg-xmin-horizon"><structname>pg_xmin_horizon</structname></link>,
+       which reports one row per snapshot-holding backend, replication slot,
+       prepared transaction, or hot-standby-feedback sender with each row's
+       per-class <structfield>xmin</structfield> contribution.  Sort by
+       <literal>age(data_xmin)</literal> or
+       <literal>age(catalog_xmin)</literal> to find the oldest pinner,
+       considering only rows whose <structfield>datid</structfield> is the
+       affected database or null.  A backend connected to another database
+       pins only that database's data and catalog horizons, though its
+       <structfield>shared_xmin</structfield> still applies to shared
+       catalogs everywhere.</simpara>
      </listitem>
      <listitem>
-      <simpara>End long-running open transactions. You can find these by checking
-       <link linkend="monitoring-pg-stat-activity-view"><structname>pg_stat_activity</structname></link> for rows where
-       <literal>age(backend_xid)</literal> or <literal>age(backend_xmin)</literal> is
-       large. Such transactions should be committed or rolled back, or the session
-       can be terminated using <literal>pg_terminate_backend</literal>.</simpara>
+      <simpara>Resolve old prepared transactions by committing or rolling
+       back.</simpara>
      </listitem>
      <listitem>
-      <simpara>Drop any old replication slots. Use
-       <link linkend="view-pg-replication-slots"><structname>pg_replication_slots</structname></link> to
-       find slots where <literal>age(xmin)</literal> or <literal>age(catalog_xmin)</literal>
-       is large. In many cases, such slots were created for replication to servers that no
-       longer exist, or that have been down for a long time. If you drop a slot for a server
-       that still exists and might still try to connect to that slot, that replica may
+      <simpara>End long-running open transactions by committing, rolling back,
+       or terminating the session using
+       <literal>pg_terminate_backend</literal>.</simpara>
+     </listitem>
+     <listitem>
+      <simpara>Drop any old replication slots. In many cases, such slots were
+       created for replication to servers that no longer exist, or that have
+       been down for a long time. If you drop a slot for a server that still
+       exists and might still try to connect to that slot, that replica may
        need to be rebuilt.</simpara>
      </listitem>
      <listitem>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 5ea19d68622..71559743451 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -246,6 +246,11 @@
       <entry>wait events</entry>
      </row>
 
+     <row>
+      <entry><link linkend="view-pg-xmin-horizon"><structname>pg_xmin_horizon</structname></link></entry>
+      <entry>sources holding back the cluster's VACUUM removal horizon</entry>
+     </row>
+
     </tbody>
    </tgroup>
   </table>
@@ -5783,4 +5788,225 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
   </table>
  </sect1>
 
+ <sect1 id="view-pg-xmin-horizon">
+  <title><structname>pg_xmin_horizon</structname></title>
+
+  <indexterm zone="view-pg-xmin-horizon">
+   <primary>pg_xmin_horizon</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_xmin_horizon</structname> view has one row per backend
+   currently holding a snapshot or transaction ID, plus one per
+   replication slot, prepared transaction, and hot-standby-feedback sender,
+   and reports each row's raw per-class <structfield>xmin</structfield>
+   contributions to the cluster's <command>VACUUM</command> removal horizon.
+   A row whose per-class <structfield>xmin</structfield> columns are all
+   null does not currently pin the horizon.
+  </para>
+
+  <table>
+   <title><structname>pg_xmin_horizon</structname> Columns</title>
+   <tgroup cols="1">
+    <thead>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       Column Type
+      </para>
+      <para>
+       Description
+      </para></entry>
+     </row>
+    </thead>
+
+    <tbody>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>kind</structfield> <type>text</type>
+      </para>
+      <para>
+       The row's kind: <literal>backend</literal>,
+       <literal>replication_slot</literal>, <literal>prepared_xact</literal>,
+       or <literal>standby_feedback</literal>.  A
+       <literal>standby_feedback</literal> row represents a physical-replication
+       sender propagating <link linkend="guc-hot-standby-feedback"><varname>hot_standby_feedback</varname></link>
+       without a replication slot.  A sender that does use a slot is reported
+       through that slot's <literal>replication_slot</literal> row instead.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>pid</structfield> <type>integer</type>
+      </para>
+      <para>
+       The process ID of the <literal>backend</literal> or
+       <literal>standby_feedback</literal> walsender that owns this row.
+       Null for <literal>replication_slot</literal> and
+       <literal>prepared_xact</literal> rows.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>slot_name</structfield> <type>name</type>
+      </para>
+      <para>
+       The replication slot name for <literal>replication_slot</literal>
+       rows.  Null for all other kinds.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>gid</structfield> <type>text</type>
+      </para>
+      <para>
+       The global transaction identifier for <literal>prepared_xact</literal>
+       rows, joined in from
+       <link linkend="view-pg-prepared-xacts"><structname>pg_prepared_xacts</structname></link>.
+       Null for all other kinds.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>datid</structfield> <type>oid</type>
+       (references <link linkend="catalog-pg-database"><structname>pg_database</structname></link>.<structfield>oid</structfield>)
+      </para>
+      <para>
+       For <literal>backend</literal> and <literal>prepared_xact</literal>
+       rows, the OID of the connected database.  For logical
+       <literal>replication_slot</literal> rows, the OID of the database
+       whose data is decoded.  Null for other rows.
+       <structfield>datid</structfield> scopes what a row pins.  A
+       <literal>backend</literal> or <literal>prepared_xact</literal> row's
+       <structfield>catalog_xmin</structfield> and
+       <structfield>data_xmin</structfield> hold back those horizons only
+       within its own database, while its
+       <structfield>shared_xmin</structfield> applies cluster-wide.
+       Slot and <literal>standby_feedback</literal> rows pin the horizon
+       cluster-wide regardless of
+       <structfield>datid</structfield>, and a query filtering
+       by <structfield>datid</structfield> will exclude them.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>datname</structfield> <type>name</type>
+       (references <link linkend="catalog-pg-database"><structname>pg_database</structname></link>.<structfield>datname</structfield>)
+      </para>
+      <para>
+       The name of the database referenced by
+       <structfield>datid</structfield>.
+       Null whenever <structfield>datid</structfield> is null.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>shared_xmin</structfield> <type>xid</type>
+      </para>
+      <para>
+       The transaction ID this row pins for the shared horizon class.
+       Null if it does not pin that class.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>catalog_xmin</structfield> <type>xid</type>
+      </para>
+      <para>
+       The transaction ID this row pins for the catalog horizon class.
+       Null if it does not pin that class.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>data_xmin</structfield> <type>xid</type>
+      </para>
+      <para>
+       The transaction ID this row pins for the data horizon class.
+       Null if it does not pin that class.  A backend running
+       <command>VACUUM</command>, or one performing logical decoding, pins none
+       of the three classes and shows null in all three
+       <structfield>*_xmin</structfield> columns.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>xact_start</structfield> <type>timestamptz</type>
+      </para>
+      <para>
+       Timestamp for the source row's transaction.  For
+       <literal>backend</literal> rows this is the transaction's start
+       time; for <literal>prepared_xact</literal> rows it is the time of
+       <command>PREPARE TRANSACTION</command>.  Null for
+       <literal>replication_slot</literal> and
+       <literal>standby_feedback</literal> rows.
+      </para></entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+
+  <para>
+   For <literal>replication_slot</literal> rows,
+   <structfield>shared_xmin</structfield> and
+   <structfield>catalog_xmin</structfield> both report the older of the
+   slot's <structfield>xmin</structfield> and
+   <structfield>catalog_xmin</structfield>.  The
+   <structfield>data_xmin</structfield> column reports the slot's
+   <structfield>xmin</structfield> alone, because catalog pinning does
+   not restrict data-heap pruning.  The reported values are the slot's
+   effective xmins, meaning the inputs from which the server computes the
+   slots' combined contribution to the horizon.  These values can briefly
+   differ from
+   <link linkend="view-pg-replication-slots"><structname>pg_replication_slots</structname></link>
+   while a slot is being created.  The slots' combined contribution to the
+   horizon is maintained separately from the per-slot values shown here and
+   can transiently be older than all of them, until a slot next advances.
+   A horizon can therefore be held back briefly by a slot contribution that
+   no row reports.  An
+   invalidated slot no longer pins the horizon and is reported with all
+   per-class columns null.
+  </para>
+
+  <para>
+   To find the row binding a horizon class, order the applicable rows by
+   descending <function>age()</function> of the class column.
+   <type>xid</type> has no ordering operators, so
+   <function>age()</function> serves as the wraparound-safe sort key.
+   Which rows are applicable depends on the class and on
+   <structfield>datid</structfield> scoping, as described above.  When
+   several rows tie for the oldest value, removing one of them does not
+   advance the horizon.  When no row contributes to a class at all, the
+   horizon advances to an internal baseline just past the newest completed
+   transaction ID, which this view does not display.  The data is not
+   guaranteed to be entirely consistent.  Backend rows and replication slot
+   rows are gathered at slightly different times, without freezing the state
+   of the whole system, so a single query does not represent the cluster at
+   one instant.  See
+   <xref linkend="vacuum-for-wraparound"/> for steps to take when the horizon
+   is held back.
+  </para>
+
+  <note>
+   <para>
+    Querying <structname>pg_xmin_horizon</structname> on a hot standby is not
+    supported and will raise an error.
+   </para>
+  </note>
+
+  <para>
+   By default, the <structname>pg_xmin_horizon</structname> view can be
+   read only by superusers or roles with privileges of the
+   <literal>pg_read_all_stats</literal> role.
+  </para>
+ </sect1>
+
 </chapter>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 090281a03dd..4bf17923289 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1127,6 +1127,31 @@ CREATE VIEW pg_replication_slots AS
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
+CREATE VIEW pg_xmin_horizon AS
+    SELECT
+        c.kind,
+        c.pid,
+        c.slot_name,
+        p.gid,
+        c.datid,
+        d.datname,
+        c.shared_xmin,
+        c.catalog_xmin,
+        c.data_xmin,
+        CASE c.kind
+            WHEN 'backend'       THEN a.xact_start
+            WHEN 'prepared_xact' THEN p.prepared
+            ELSE NULL
+        END AS xact_start
+    FROM pg_get_xmin_horizon() c
+        LEFT JOIN pg_stat_activity a ON a.pid = c.pid
+        LEFT JOIN pg_prepared_xacts p
+            ON c.kind = 'prepared_xact' AND p.transaction = c.data_xmin
+        LEFT JOIN pg_database d ON d.oid = c.datid;
+
+REVOKE ALL ON pg_xmin_horizon FROM PUBLIC;
+GRANT SELECT ON pg_xmin_horizon TO pg_read_all_stats;
+
 CREATE VIEW pg_stat_replication_slots AS
     SELECT
             s.slot_name,
diff --git a/src/backend/storage/ipc/Makefile b/src/backend/storage/ipc/Makefile
index f71653bbe48..b6522cc12fb 100644
--- a/src/backend/storage/ipc/Makefile
+++ b/src/backend/storage/ipc/Makefile
@@ -18,6 +18,7 @@ OBJS = \
 	latch.o \
 	pmsignal.o \
 	procarray.o \
+	procarrayfuncs.o \
 	procsignal.o \
 	shm_mq.o \
 	shm_toc.o \
diff --git a/src/backend/storage/ipc/meson.build b/src/backend/storage/ipc/meson.build
index b8c31e29967..e5c83d75108 100644
--- a/src/backend/storage/ipc/meson.build
+++ b/src/backend/storage/ipc/meson.build
@@ -10,6 +10,7 @@ backend_sources += files(
   'latch.c',
   'pmsignal.c',
   'procarray.c',
+  'procarrayfuncs.c',
   'procsignal.c',
   'shm_mq.c',
   'shm_toc.c',
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 60336b31803..59947a93393 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -1765,7 +1765,8 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 		/*
 		 * Skip over backends either vacuuming (which is ok with rows being
 		 * removed, as long as pg_subtrans is not truncated) or doing logical
-		 * decoding (which manages xmin separately, check below).
+		 * decoding (which manages xmin separately, check below).  Keep this
+		 * filter in sync with pg_get_xmin_horizon()'s classifier.
 		 */
 		if (statusFlags & (PROC_IN_VACUUM | PROC_IN_LOGICAL_DECODING))
 			continue;
@@ -1845,7 +1846,8 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 	 * catalog xmin is applied to the catalog one (so catalogs can be accessed
 	 * for logical decoding). Initialize with data horizon, and then back up
 	 * further if necessary. Have to back up the shared horizon as well, since
-	 * that also can contain catalogs.
+	 * that also can contain catalogs.  Keep how slot xmins feed each horizon
+	 * class in sync with pg_get_xmin_horizon()'s per-slot classification.
 	 */
 	h->shared_oldest_nonremovable_raw = h->shared_oldest_nonremovable;
 	h->shared_oldest_nonremovable =
@@ -2258,7 +2260,8 @@ GetSnapshotData(Snapshot snapshot)
 
 			/*
 			 * Skip over backends doing logical decoding which manages xmin
-			 * separately (check below) and ones running LAZY VACUUM.
+			 * separately (check below) and ones running LAZY VACUUM.  Keep
+			 * this filter in sync with ComputeXidHorizons().
 			 */
 			statusFlags = allStatusFlags[pgxactoff];
 			if (statusFlags & (PROC_IN_LOGICAL_DECODING | PROC_IN_VACUUM))
@@ -3145,6 +3148,47 @@ ProcNumberGetTransactionIds(ProcNumber procNumber, TransactionId *xid,
 	LWLockRelease(ProcArrayLock);
 }
 
+/*
+ * GetXidHorizonProcs -- snapshot every active proc's horizon inputs
+ *
+ * Returns a palloc'd array with one entry per active proc, all captured in
+ * a single ProcArrayLock SHARED pass so the rows are mutually consistent.  The
+ * number of valid entries is returned into *n.
+ */
+XidHorizonProc *
+GetXidHorizonProcs(int *n)
+{
+	XidHorizonProc *result;
+	ProcArrayStruct *arrayP = procArray;
+	int			count = 0;
+	int			index;
+
+	result = palloc_array(XidHorizonProc, arrayP->maxProcs);
+
+	LWLockAcquire(ProcArrayLock, LW_SHARED);
+
+	for (index = 0; index < arrayP->numProcs; index++)
+	{
+		int			pgprocno = arrayP->pgprocnos[index];
+		PGPROC	   *proc = &allProcs[pgprocno];
+		XidHorizonProc *p = &result[count++];
+
+		p->pid = proc->pid;
+		p->databaseId = proc->databaseId;
+		p->statusFlags = ProcGlobal->statusFlags[index];
+
+		/* Fetch xid just once - see GetNewTransactionId */
+		p->xid = UINT32_ACCESS_ONCE(ProcGlobal->xids[index]);
+		p->xmin = UINT32_ACCESS_ONCE(proc->xmin);
+	}
+
+	LWLockRelease(ProcArrayLock);
+
+	*n = count;
+
+	return result;
+}
+
 /*
  * BackendPidGetProc -- get a backend's PGPROC given its PID
  *
diff --git a/src/backend/storage/ipc/procarrayfuncs.c b/src/backend/storage/ipc/procarrayfuncs.c
new file mode 100644
index 00000000000..f5dae7d92ad
--- /dev/null
+++ b/src/backend/storage/ipc/procarrayfuncs.c
@@ -0,0 +1,268 @@
+/*-------------------------------------------------------------------------
+ *
+ * procarrayfuncs.c
+ *	  SQL-callable functions exposing ProcArray-derived diagnostics.
+ *
+ * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/storage/ipc/procarrayfuncs.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/transam.h"
+#include "access/twophase.h"
+#include "access/xlog.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "replication/slot.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/procarray.h"
+#include "utils/builtins.h"
+#include "utils/tuplestore.h"
+
+/*
+ * One gathered row, with the per-class xmin values and the columns that
+ * identify it.  Fields that may be SQL NULL use a sentinel: 0 for pid,
+ * empty NameData for slot_name, InvalidOid for datid, and
+ * InvalidTransactionId for shared/catalog/data.
+ */
+typedef struct XidHorizonRow
+{
+	Datum		kind;			/* text Datum (never NULL) */
+	int			pid;			/* backend / walsender PID; 0 = NULL */
+	NameData	slot_name;		/* replication slot name; empty = NULL */
+	Oid			datid;			/* InvalidOid = NULL */
+	TransactionId shared;		/* per-class xmin values, InvalidTransactionId
+								 * = NULL */
+	TransactionId catalog;
+	TransactionId data;
+} XidHorizonRow;
+
+/*
+ * pg_get_xmin_horizon - SQL SRF reporting, per row, the raw per-class xmin
+ * contributions to the cluster's xmin horizons.
+ *
+ * Emits one row per in-use replication slot, prepared transaction, slot-less
+ * hot-standby-feedback walsender, and backend that holds a snapshot.
+ */
+Datum
+pg_get_xmin_horizon(PG_FUNCTION_ARGS)
+{
+#define PG_GET_XMIN_HORIZON_COLS 7
+	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	XidHorizonProc *procs;
+	int			nprocs;
+	XidHorizonRow *rows;
+	int			nrows;
+	int			maxrows;
+
+	/*
+	 * On a hot standby, KnownAssignedXids carries the primary's running xacts
+	 * and is folded into the standby's data horizon by GetSnapshotData() but
+	 * is not surfaced as procarray rows.  Error instead of reporting
+	 * incomplete data, which might be mistakenly interpreted as complete.
+	 */
+	if (RecoveryInProgress())
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("recovery is in progress"),
+				 errhint("pg_get_xmin_horizon() cannot be executed during recovery.")));
+
+	InitMaterializedSRF(fcinfo, 0);
+
+	/* MaxBackends covers walsenders (standby_feedback rows) too. */
+	maxrows = MaxBackends + max_prepared_xacts +
+		max_replication_slots + max_repack_replication_slots;
+	rows = palloc_array(XidHorizonRow, maxrows);
+	nrows = 0;
+
+	procs = GetXidHorizonProcs(&nprocs);
+
+	for (int i = 0; i < nprocs; i++)
+	{
+		XidHorizonProc *p = &procs[i];
+		TransactionId effective_xmin;
+		TransactionId shared;
+		TransactionId catalog;
+		TransactionId data;
+		int			pid;
+		XidHorizonRow *r;
+
+		/*
+		 * Within a backend xmin is set before xid is assigned, so the only
+		 * xmin > xid window is the cross-backend read race; see
+		 * ComputeXidHorizons().  A backend with neither set holds no snapshot
+		 * and does not influence the horizon, so skip it.
+		 */
+		effective_xmin = TransactionIdOlder(p->xid, p->xmin);
+		if (!TransactionIdIsValid(effective_xmin))
+			continue;
+
+		/*
+		 * Classify this proc's per-class xmin.  VACUUM backends do not pin
+		 * xmin, and logical decoding backends pin xmin indirectly through
+		 * slots.  Emit them with NULLs so the view lists them without
+		 * claiming they pin.  Every other proc pins all three classes.
+		 * VISHORIZON_TEMP is omitted because the temp horizon is per-backend
+		 * and has no cross-backend diagnostic value.  Keep in sync with
+		 * ComputeXidHorizons() and GetSnapshotData().
+		 */
+		if (p->statusFlags & (PROC_IN_VACUUM | PROC_IN_LOGICAL_DECODING))
+			shared = catalog = data = InvalidTransactionId;
+		else
+			shared = catalog = data = effective_xmin;
+
+		Assert(nrows < maxrows);
+		r = &rows[nrows++];
+		memset(r, 0, sizeof(*r));
+		r->datid = p->databaseId;
+
+		pid = p->pid;
+		if (pid == 0)
+		{
+			/*
+			 * A prepared xact's dummy PGPROC never carries an xmin (see
+			 * MarkAsPreparingGuts()), so this row's per-class values are
+			 * always its xid.  The pg_xmin_horizon view's join to
+			 * pg_prepared_xacts relies on that to attach the gid.
+			 */
+			r->kind = CStringGetTextDatum("prepared_xact");
+		}
+		else if (p->statusFlags & PROC_AFFECTS_ALL_HORIZONS)
+		{
+			r->kind = CStringGetTextDatum("standby_feedback");
+			r->pid = pid;
+			r->datid = InvalidOid;
+		}
+		else
+		{
+			r->kind = CStringGetTextDatum("backend");
+			r->pid = pid;
+		}
+
+		r->shared = shared;
+		r->catalog = catalog;
+		r->data = data;
+	}
+
+	pfree(procs);
+
+	/*
+	 * Two-phase gather: ProcArrayLock has been released inside
+	 * GetXidHorizonProcs() before we acquire ReplicationSlotControlLock here.
+	 * Do not "fix" this by holding both locks at once.  The lock-ordering
+	 * convention in this subsystem takes ReplicationSlotControlLock without
+	 * ProcArrayLock held; pg_get_replication_slots() and
+	 * ReplicationSlotsComputeRequiredXmin() both follow it.  Holding
+	 * ProcArrayLock across that acquisition would establish a new
+	 * lock-ordering and risk deadlocks.
+	 *
+	 * This view does not guarantee cross-phase consistency: each row reports
+	 * a per-source xmin observed under its own lock.
+	 *
+	 * The effective aggregate contribution of all slots to the xmin used by
+	 * vacuum may be temporarily older than any individual slot xmin reported
+	 * by this view.  See the comment in
+	 * ReplicationSlotsComputeRequiredXmin().
+	 */
+
+	/*
+	 * Gather replication slots.  Unlike a backend, which is only listed while
+	 * it holds a snapshot, a slot is reported as soon as it exists, even when
+	 * it currently pins no xmin, e.g. a freshly created physical slot with no
+	 * standby attached, or an invalidated slot.  Such a slot appears with all
+	 * per-class columns NULL.
+	 */
+	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+	for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
+	{
+		ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+		TransactionId xmin;
+		TransactionId catalog_xmin;
+		ReplicationSlotInvalidationCause invalidated;
+		XidHorizonRow *r;
+
+		if (!s->in_use)
+			continue;
+
+		/* r->kind is palloc'd here so no allocation happens under the mutex */
+		Assert(nrows < maxrows);
+		r = &rows[nrows++];
+		memset(r, 0, sizeof(*r));
+		r->kind = CStringGetTextDatum("replication_slot");
+
+		SpinLockAcquire(&s->mutex);
+		xmin = s->effective_xmin;
+		catalog_xmin = s->effective_catalog_xmin;
+		invalidated = s->data.invalidated;
+		r->datid = s->data.database;
+		namestrcpy(&r->slot_name, NameStr(s->data.name));
+		SpinLockRelease(&s->mutex);
+
+		/* Invalidated slots do not affect the horizon */
+		if (invalidated != RS_INVAL_NONE)
+			xmin = catalog_xmin = InvalidTransactionId;
+
+		r->shared = TransactionIdOlder(xmin, catalog_xmin);
+		r->catalog = r->shared;
+		r->data = xmin;
+	}
+	LWLockRelease(ReplicationSlotControlLock);
+
+	for (int i = 0; i < nrows; i++)
+	{
+		XidHorizonRow *r = &rows[i];
+		Datum		values[PG_GET_XMIN_HORIZON_COLS];
+		bool		nulls[PG_GET_XMIN_HORIZON_COLS];
+		int			col;
+
+		memset(values, 0, sizeof(values));
+		memset(nulls, 0, sizeof(nulls));
+		col = 0;
+
+		values[col++] = r->kind;
+
+		if (r->pid == 0)
+			nulls[col++] = true;
+		else
+			values[col++] = Int32GetDatum(r->pid);
+
+		if (r->slot_name.data[0] == '\0')
+			nulls[col++] = true;
+		else
+			values[col++] = NameGetDatum(&r->slot_name);
+
+		if (r->datid == InvalidOid)
+			nulls[col++] = true;
+		else
+			values[col++] = ObjectIdGetDatum(r->datid);
+
+		if (TransactionIdIsValid(r->shared))
+			values[col++] = TransactionIdGetDatum(r->shared);
+		else
+			nulls[col++] = true;
+
+		if (TransactionIdIsValid(r->catalog))
+			values[col++] = TransactionIdGetDatum(r->catalog);
+		else
+			nulls[col++] = true;
+
+		if (TransactionIdIsValid(r->data))
+			values[col++] = TransactionIdGetDatum(r->data);
+		else
+			nulls[col++] = true;
+
+		Assert(col == PG_GET_XMIN_HORIZON_COLS);
+
+		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
+							 values, nulls);
+	}
+
+	return (Datum) 0;
+}
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index f8a021987b5..00d4ef62704 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11676,6 +11676,15 @@
   proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
   proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,two_phase_at,inactive_since,conflicting,invalidation_reason,failover,synced,slotsync_skip_reason}',
   prosrc => 'pg_get_replication_slots' },
+{ oid => '8054',
+  descr => 'per-source inputs to the cluster xmin horizons',
+  proname => 'pg_get_xmin_horizon', prorows => '100', proisstrict => 'f',
+  proretset => 't', provolatile => 'v', proparallel => 'r', prorettype => 'record',
+  proargtypes => '',
+  proallargtypes => '{text,int4,name,oid,xid,xid,xid}',
+  proargmodes => '{o,o,o,o,o,o,o}',
+  proargnames => '{kind,pid,slot_name,datid,shared_xmin,catalog_xmin,data_xmin}',
+  prosrc => 'pg_get_xmin_horizon', proacl => '{POSTGRES=X,pg_read_all_stats=X}' },
 { oid => '3786', descr => 'set up a logical replication slot',
   proname => 'pg_create_logical_replication_slot', provolatile => 'v',
   proparallel => 'u', prorettype => 'record',
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index d718a5b542f..8313d3cffef 100644
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -65,6 +65,17 @@ extern PGPROC *ProcNumberGetProc(int procNumber);
 extern void ProcNumberGetTransactionIds(int procNumber, TransactionId *xid,
 										TransactionId *xmin, int *nsubxid,
 										bool *overflowed);
+
+typedef struct XidHorizonProc
+{
+	int			pid;			/* OS pid; 0 for prepared-xact dummy procs */
+	Oid			databaseId;
+	uint8		statusFlags;
+	TransactionId xid;
+	TransactionId xmin;
+} XidHorizonProc;
+
+extern XidHorizonProc *GetXidHorizonProcs(int *n);
 extern PGPROC *BackendPidGetProc(int pid);
 extern PGPROC *BackendPidGetProcWithLock(int pid);
 extern int	BackendXidGetPid(TransactionId xid);
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 39ec8c4946d..65e779770c5 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -64,6 +64,7 @@ tests += {
       't/053_standby_login_event_trigger.pl',
       't/054_unlogged_sequence_promotion.pl',
       't/055_cascade_reconnect.pl',
+      't/056_xmin_horizon.pl',
     ],
   },
 }
diff --git a/src/test/recovery/t/056_xmin_horizon.pl b/src/test/recovery/t/056_xmin_horizon.pl
new file mode 100644
index 00000000000..2570f05f798
--- /dev/null
+++ b/src/test/recovery/t/056_xmin_horizon.pl
@@ -0,0 +1,281 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Tests pg_xmin_horizon's replication_slot and standby_feedback rows.
+
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+$node_primary->init(allows_streaming => 'logical');
+$node_primary->start;
+
+$node_primary->safe_psql('postgres',
+	"SELECT pg_create_physical_replication_slot('xmin_horizon_phys');");
+
+my $backup_name = 'xmin_horizon_backup';
+$node_primary->backup($backup_name);
+
+# Slotted standby drives the replication_slot row.
+my $node_slotted = PostgreSQL::Test::Cluster->new('standby_slotted');
+$node_slotted->init_from_backup($node_primary, $backup_name,
+	has_streaming => 1);
+$node_slotted->append_conf(
+	'postgresql.conf', qq[
+primary_slot_name = 'xmin_horizon_phys'
+hot_standby_feedback = on
+wal_receiver_status_interval = 1
+]);
+$node_slotted->start;
+
+# Slot-less standby drives the standby_feedback row.
+my $node_slotless = PostgreSQL::Test::Cluster->new('standby_slotless');
+$node_slotless->init_from_backup($node_primary, $backup_name,
+	has_streaming => 1);
+$node_slotless->append_conf(
+	'postgresql.conf', qq[
+hot_standby_feedback = on
+wal_receiver_status_interval = 1
+]);
+$node_slotless->start;
+
+$node_primary->wait_for_catchup($node_slotted);
+$node_primary->wait_for_catchup($node_slotless);
+
+# (0) pg_xmin_horizon errors during recovery.  KnownAssignedXids
+# is not surfaced as procarray rows, so the standby-local view alone would
+# misattribute the horizon.
+my $stderr;
+$node_slotted->psql(
+	'postgres',
+	'SELECT * FROM pg_xmin_horizon LIMIT 1;',
+	stderr => \$stderr);
+ok( $stderr =~ /recovery is in progress/,
+	'pg_xmin_horizon errors on a hot standby');
+
+# Pin a stable xmin on each standby for hot_standby_feedback to relay.
+my $psql_slotted = $node_slotted->background_psql('postgres');
+$psql_slotted->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT 1;");
+
+my $psql_slotless = $node_slotless->background_psql('postgres');
+$psql_slotless->query_safe(
+	"BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT 1;");
+
+# (1) replication_slot row.
+$node_primary->poll_query_until(
+	'postgres',
+	"SELECT xmin IS NOT NULL FROM pg_replication_slots WHERE slot_name = 'xmin_horizon_phys'",
+	't')
+  or die "timed out waiting for slot 'xmin_horizon_phys' to acquire an xmin";
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'replication_slot'
+		   AND slot_name = 'xmin_horizon_phys'
+		   AND data_xmin IS NOT NULL"),
+	'1',
+	'replication_slot row for xmin_horizon_phys carries an xmin');
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'replication_slot'
+		   AND slot_name = 'xmin_horizon_phys'
+		   AND shared_xmin IS NOT NULL
+		   AND shared_xmin = catalog_xmin
+		   AND shared_xmin = data_xmin"),
+	'1',
+	'slot row: shared_xmin = catalog_xmin = data_xmin (non-null)');
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'replication_slot'
+		   AND slot_name = 'xmin_horizon_phys'
+		   AND datid IS NULL
+		   AND xact_start IS NULL"),
+	'1',
+	'slot row has NULL datid and xact_start');
+
+# (2) standby_feedback row.  Exactly one row: the slot-less standby's
+# walsender.  The slotted standby itself has no xmin (the slot carries it) and
+# must NOT appear here.
+$node_primary->poll_query_until(
+	'postgres',
+	"SELECT count(*) > 0 FROM pg_xmin_horizon WHERE kind = 'standby_feedback'",
+	't') or die "timed out waiting for a standby_feedback row to appear";
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'standby_feedback'
+		   AND data_xmin IS NOT NULL
+		   AND datid IS NULL
+		   AND xact_start IS NULL"),
+	'1',
+	'exactly one standby_feedback row: non-null data_xmin, NULL datid, NULL xact_start'
+);
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon h
+		 JOIN pg_stat_activity a ON a.pid = h.pid
+		 WHERE h.kind = 'standby_feedback'
+		   AND a.backend_type = 'walsender'"),
+	'1',
+	'standby_feedback pid is the walsender PID');
+
+# (3) logical replication slot row: catalog_xmin non-null, data_xmin NULL.
+$node_primary->safe_psql('postgres',
+	"SELECT pg_create_logical_replication_slot('xmin_horizon_log', 'test_decoding');"
+);
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'replication_slot'
+		   AND slot_name = 'xmin_horizon_log'
+		   AND catalog_xmin IS NOT NULL
+		   AND shared_xmin = catalog_xmin
+		   AND data_xmin IS NULL"),
+	'1',
+	'logical slot row: shared_xmin = catalog_xmin (non-null), data_xmin NULL'
+);
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT datid = (SELECT oid FROM pg_database WHERE datname = 'postgres')
+		        AND datname = 'postgres'
+		 FROM pg_xmin_horizon
+		 WHERE kind = 'replication_slot' AND slot_name = 'xmin_horizon_log'"),
+	't',
+	'logical slot row: datid and datname are the decoded database');
+
+$node_primary->safe_psql('postgres',
+	"SELECT pg_drop_replication_slot('xmin_horizon_log');");
+
+# (4) per-database scoping: backend row's datid reflects its connection.
+$node_primary->safe_psql('postgres', 'CREATE DATABASE xmin_horizon_db2;');
+
+my $psql_db2 = $node_primary->background_psql('xmin_horizon_db2');
+$psql_db2->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT 1;");
+
+$node_primary->poll_query_until(
+	'postgres',
+	"SELECT count(*) > 0 FROM pg_xmin_horizon
+	 WHERE kind = 'backend'
+	   AND datid = (SELECT oid FROM pg_database WHERE datname = 'xmin_horizon_db2')
+	   AND data_xmin IS NOT NULL",
+	't') or die "timed out waiting for xmin_horizon_db2 backend xmin";
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'backend'
+		   AND datid = (SELECT oid FROM pg_database WHERE datname = 'xmin_horizon_db2')
+		   AND datname = 'xmin_horizon_db2'
+		   AND data_xmin IS NOT NULL"),
+	'1',
+	'backend in xmin_horizon_db2 appears with the right datid and datname');
+
+$psql_db2->quit;
+$node_primary->safe_psql('postgres', 'DROP DATABASE xmin_horizon_db2;');
+
+# (5) Tied backend rows via shared exported snapshot.  Tear down older-xmin
+# holders first so the tied pair is the global min.
+$psql_slotted->quit;
+$psql_slotless->quit;
+$node_slotless->stop;
+$node_slotted->stop;
+$node_primary->safe_psql('postgres',
+	"SELECT pg_drop_replication_slot('xmin_horizon_phys')");
+
+my $psql_tie_a = $node_primary->background_psql('postgres');
+my $psql_tie_b = $node_primary->background_psql('postgres');
+
+$psql_tie_a->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ");
+my $snap_id = $psql_tie_a->query_safe("SELECT pg_export_snapshot()");
+chomp $snap_id;
+my $tie_pid_a = $psql_tie_a->query_safe("SELECT pg_backend_pid()");
+chomp $tie_pid_a;
+
+$psql_tie_b->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ");
+$psql_tie_b->query_safe("SET TRANSACTION SNAPSHOT '$snap_id'");
+my $tie_pid_b = $psql_tie_b->query_safe("SELECT pg_backend_pid()");
+chomp $tie_pid_b;
+
+$node_primary->poll_query_until(
+	'postgres',
+	"SELECT count(*) = 2 FROM pg_xmin_horizon
+	 WHERE kind = 'backend' AND pid IN ($tie_pid_a, $tie_pid_b)
+	   AND data_xmin IS NOT NULL",
+	't') or die "timed out waiting for tied backends' xmin";
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT (SELECT data_xmin FROM pg_xmin_horizon
+		         WHERE kind = 'backend' AND pid = $tie_pid_a)
+		      = (SELECT data_xmin FROM pg_xmin_horizon
+		         WHERE kind = 'backend' AND pid = $tie_pid_b)"),
+	't',
+	'tied backends share data_xmin');
+
+$psql_tie_a->quit;
+$psql_tie_b->quit;
+
+# (6) VACUUM backend row: PROC_IN_VACUUM is set with all per-class xmins
+# NULL.  Parking VACUUM on a conflicting lock holds it in that state while
+# the assertion runs.
+$node_primary->safe_psql('postgres',
+	"CREATE UNLOGGED TABLE xmin_horizon_vac_t (id int);");
+
+# Locker holds SHARE on the target so VACUUM's ShareUpdateExclusiveLock
+# acquisition blocks.
+my $psql_locker = $node_primary->background_psql('postgres');
+$psql_locker->query_safe(
+	"BEGIN; LOCK TABLE xmin_horizon_vac_t IN SHARE MODE;");
+
+# VACUUM session: capture pid first, then fire VACUUM via query_until so
+# the main thread proceeds while VACUUM is still blocked.
+my $psql_vacuum = $node_primary->background_psql('postgres');
+my $vac_pid = $psql_vacuum->query_safe("SELECT pg_backend_pid()");
+chomp $vac_pid;
+
+$psql_vacuum->query_until(
+	qr/vacuum_started/, q(
+\echo vacuum_started
+VACUUM xmin_horizon_vac_t;
+));
+
+# The VACUUM backend's row appears once its snapshot is pushed, which sets
+# its xmin.  By then, PROC_IN_VACUUM is already set and the proc is blocked
+# acquiring the ShareUpdateExclusiveLock.
+$node_primary->poll_query_until(
+	'postgres',
+	"SELECT count(*) = 1 FROM pg_xmin_horizon
+	 WHERE kind = 'backend' AND pid = $vac_pid",
+	't') or die "timed out waiting for VACUUM backend row";
+
+is( $node_primary->safe_psql(
+		'postgres',
+		"SELECT count(*) FROM pg_xmin_horizon
+		 WHERE kind = 'backend' AND pid = $vac_pid
+		   AND shared_xmin IS NULL
+		   AND catalog_xmin IS NULL
+		   AND data_xmin IS NULL"),
+	'1',
+	'VACUUM backend row: shared_xmin, catalog_xmin, data_xmin all NULL');
+
+# Release the lock so VACUUM completes, then drain both sessions.
+$psql_locker->query_safe("COMMIT");
+$psql_locker->quit;
+$psql_vacuum->quit;
+
+$node_primary->safe_psql('postgres', "DROP TABLE xmin_horizon_vac_t");
+
+$node_primary->stop;
+
+done_testing();
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 6a3341356da..207c532f865 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2906,6 +2906,24 @@ pg_wait_events| SELECT type,
     name,
     description
    FROM pg_get_wait_events() pg_get_wait_events(type, name, description);
+pg_xmin_horizon| SELECT c.kind,
+    c.pid,
+    c.slot_name,
+    p.gid,
+    c.datid,
+    d.datname,
+    c.shared_xmin,
+    c.catalog_xmin,
+    c.data_xmin,
+        CASE c.kind
+            WHEN 'backend'::text THEN a.xact_start
+            WHEN 'prepared_xact'::text THEN p.prepared
+            ELSE NULL::timestamp with time zone
+        END AS xact_start
+   FROM (((pg_get_xmin_horizon() c(kind, pid, slot_name, datid, shared_xmin, catalog_xmin, data_xmin)
+     LEFT JOIN pg_stat_activity a ON ((a.pid = c.pid)))
+     LEFT JOIN pg_prepared_xacts p ON (((c.kind = 'prepared_xact'::text) AND (p.transaction = c.data_xmin))))
+     LEFT JOIN pg_database d ON ((d.oid = c.datid)));
 SELECT tablename, rulename, definition FROM pg_rules
 WHERE schemaname = 'pg_catalog'
 ORDER BY tablename, rulename;
diff --git a/src/test/regress/expected/xmin_horizon.out b/src/test/regress/expected/xmin_horizon.out
new file mode 100644
index 00000000000..499e778482b
--- /dev/null
+++ b/src/test/regress/expected/xmin_horizon.out
@@ -0,0 +1,118 @@
+SELECT current_setting('max_prepared_transactions')::integer < 2 AS skip_test \gset
+-- ACL coverage: pg_xmin_horizon is REVOKE'd from PUBLIC and GRANT'd to
+-- pg_read_all_stats.
+CREATE ROLE regress_pxh_unpriv;
+CREATE ROLE regress_pxh_priv;
+GRANT pg_read_all_stats TO regress_pxh_priv;
+SET ROLE regress_pxh_unpriv;
+SELECT count(*) FROM pg_xmin_horizon;
+ERROR:  permission denied for view pg_xmin_horizon
+RESET ROLE;
+SET ROLE regress_pxh_priv;
+SELECT count(*) >= 0 AS readable FROM pg_xmin_horizon;
+ readable 
+----------
+ t
+(1 row)
+
+RESET ROLE;
+DROP ROLE regress_pxh_unpriv, regress_pxh_priv;
+\if :skip_test
+\quit
+\endif
+--
+-- XMIN_HORIZON
+--
+-- Structure probe.
+SELECT * FROM pg_xmin_horizon WHERE false;
+ kind | pid | slot_name | gid | datid | datname | shared_xmin | catalog_xmin | data_xmin | xact_start 
+------+-----+-----------+-----+-------+---------+-------------+--------------+-----------+------------
+(0 rows)
+
+-- A REPEATABLE READ backend with a snapshot but no xid contributes a row with
+-- all three classes equal to its backend_xmin, and xact_start joined from
+-- pg_stat_activity.
+BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+SELECT 1;
+ ?column? 
+----------
+        1
+(1 row)
+
+SELECT
+    data_xmin = catalog_xmin AND catalog_xmin = shared_xmin AS classes_equal,
+    data_xmin = (SELECT backend_xmin
+                 FROM pg_stat_activity
+                 WHERE pid = pg_backend_pid()) AS matches_backend_xmin,
+    xact_start = (SELECT xact_start
+                  FROM pg_stat_activity
+                  WHERE pid = pg_backend_pid()) AS xact_start_correct
+FROM pg_xmin_horizon
+WHERE kind = 'backend'
+  AND pid = pg_backend_pid();
+ classes_equal | matches_backend_xmin | xact_start_correct 
+---------------+----------------------+--------------------
+ t             | t                    | t
+(1 row)
+
+COMMIT;
+-- Writing transaction with assigned xid: data_xmin = backend_xmin still holds.
+BEGIN ISOLATION LEVEL REPEATABLE READ;
+SELECT pg_current_xact_id() IS NOT NULL AS xid_assigned;
+ xid_assigned 
+--------------
+ t
+(1 row)
+
+SELECT
+    data_xmin = catalog_xmin AND catalog_xmin = shared_xmin AS classes_equal,
+    data_xmin = (SELECT backend_xmin
+                 FROM pg_stat_activity
+                 WHERE pid = pg_backend_pid()) AS matches_backend_xmin
+FROM pg_xmin_horizon
+WHERE kind = 'backend'
+  AND pid = pg_backend_pid();
+ classes_equal | matches_backend_xmin 
+---------------+----------------------
+ t             | t
+(1 row)
+
+COMMIT;
+-- A prepared transaction's dummy PGPROC must be reported as kind =
+-- 'prepared_xact' with a non-null data_xmin, and it must not leak as
+-- a backend row with pid = 0.
+CREATE TABLE xmin_horizon_t(i int);
+BEGIN;
+INSERT INTO xmin_horizon_t VALUES (1);
+PREPARE TRANSACTION 'xmin_horizon_p';
+SELECT count(*) AS prepared_xact_rows
+FROM pg_xmin_horizon
+WHERE kind = 'prepared_xact' AND gid = 'xmin_horizon_p'
+  AND data_xmin IS NOT NULL;
+ prepared_xact_rows 
+--------------------
+                  1
+(1 row)
+
+SELECT count(*) AS leaked_backend_rows
+FROM pg_xmin_horizon
+WHERE kind = 'backend' AND pid = 0;
+ leaked_backend_rows 
+---------------------
+                   0
+(1 row)
+
+-- gid joined in from pg_prepared_xacts and xact_start from
+-- pg_prepared_xacts.prepared.
+SELECT gid = 'xmin_horizon_p' AS gid_matches,
+       xact_start = (SELECT prepared FROM pg_prepared_xacts
+                     WHERE gid = 'xmin_horizon_p') AS xact_start_matches_prepared
+FROM pg_xmin_horizon
+WHERE kind = 'prepared_xact' AND gid = 'xmin_horizon_p';
+ gid_matches | xact_start_matches_prepared 
+-------------+-----------------------------
+ t           | t
+(1 row)
+
+ROLLBACK PREPARED 'xmin_horizon_p';
+DROP TABLE xmin_horizon_t;
diff --git a/src/test/regress/expected/xmin_horizon_1.out b/src/test/regress/expected/xmin_horizon_1.out
new file mode 100644
index 00000000000..320791392d8
--- /dev/null
+++ b/src/test/regress/expected/xmin_horizon_1.out
@@ -0,0 +1,21 @@
+SELECT current_setting('max_prepared_transactions')::integer < 2 AS skip_test \gset
+-- ACL coverage: pg_xmin_horizon is REVOKE'd from PUBLIC and GRANT'd to
+-- pg_read_all_stats.
+CREATE ROLE regress_pxh_unpriv;
+CREATE ROLE regress_pxh_priv;
+GRANT pg_read_all_stats TO regress_pxh_priv;
+SET ROLE regress_pxh_unpriv;
+SELECT count(*) FROM pg_xmin_horizon;
+ERROR:  permission denied for view pg_xmin_horizon
+RESET ROLE;
+SET ROLE regress_pxh_priv;
+SELECT count(*) >= 0 AS readable FROM pg_xmin_horizon;
+ readable 
+----------
+ t
+(1 row)
+
+RESET ROLE;
+DROP ROLE regress_pxh_unpriv, regress_pxh_priv;
+\if :skip_test
+\quit
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 8fa0a6c47fb..1c16ac2d2ee 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -128,7 +128,7 @@ test: partition_merge partition_split partition_join partition_prune reloptions
 # ----------
 # Another group of parallel tests (compression)
 # ----------
-test: compression compression_lz4 compression_pglz cluster
+test: compression compression_lz4 compression_pglz cluster xmin_horizon
 
 # event_trigger depends on create_am and cannot run concurrently with
 # any test that runs DDL
diff --git a/src/test/regress/sql/xmin_horizon.sql b/src/test/regress/sql/xmin_horizon.sql
new file mode 100644
index 00000000000..95128750e68
--- /dev/null
+++ b/src/test/regress/sql/xmin_horizon.sql
@@ -0,0 +1,90 @@
+SELECT current_setting('max_prepared_transactions')::integer < 2 AS skip_test \gset
+
+-- ACL coverage: pg_xmin_horizon is REVOKE'd from PUBLIC and GRANT'd to
+-- pg_read_all_stats.
+CREATE ROLE regress_pxh_unpriv;
+CREATE ROLE regress_pxh_priv;
+GRANT pg_read_all_stats TO regress_pxh_priv;
+
+SET ROLE regress_pxh_unpriv;
+SELECT count(*) FROM pg_xmin_horizon;
+RESET ROLE;
+
+SET ROLE regress_pxh_priv;
+SELECT count(*) >= 0 AS readable FROM pg_xmin_horizon;
+RESET ROLE;
+
+DROP ROLE regress_pxh_unpriv, regress_pxh_priv;
+
+\if :skip_test
+\quit
+\endif
+--
+-- XMIN_HORIZON
+--
+
+-- Structure probe.
+SELECT * FROM pg_xmin_horizon WHERE false;
+
+-- A REPEATABLE READ backend with a snapshot but no xid contributes a row with
+-- all three classes equal to its backend_xmin, and xact_start joined from
+-- pg_stat_activity.
+BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+SELECT 1;
+
+SELECT
+    data_xmin = catalog_xmin AND catalog_xmin = shared_xmin AS classes_equal,
+    data_xmin = (SELECT backend_xmin
+                 FROM pg_stat_activity
+                 WHERE pid = pg_backend_pid()) AS matches_backend_xmin,
+    xact_start = (SELECT xact_start
+                  FROM pg_stat_activity
+                  WHERE pid = pg_backend_pid()) AS xact_start_correct
+FROM pg_xmin_horizon
+WHERE kind = 'backend'
+  AND pid = pg_backend_pid();
+
+COMMIT;
+
+-- Writing transaction with assigned xid: data_xmin = backend_xmin still holds.
+BEGIN ISOLATION LEVEL REPEATABLE READ;
+SELECT pg_current_xact_id() IS NOT NULL AS xid_assigned;
+
+SELECT
+    data_xmin = catalog_xmin AND catalog_xmin = shared_xmin AS classes_equal,
+    data_xmin = (SELECT backend_xmin
+                 FROM pg_stat_activity
+                 WHERE pid = pg_backend_pid()) AS matches_backend_xmin
+FROM pg_xmin_horizon
+WHERE kind = 'backend'
+  AND pid = pg_backend_pid();
+
+COMMIT;
+
+-- A prepared transaction's dummy PGPROC must be reported as kind =
+-- 'prepared_xact' with a non-null data_xmin, and it must not leak as
+-- a backend row with pid = 0.
+CREATE TABLE xmin_horizon_t(i int);
+BEGIN;
+INSERT INTO xmin_horizon_t VALUES (1);
+PREPARE TRANSACTION 'xmin_horizon_p';
+
+SELECT count(*) AS prepared_xact_rows
+FROM pg_xmin_horizon
+WHERE kind = 'prepared_xact' AND gid = 'xmin_horizon_p'
+  AND data_xmin IS NOT NULL;
+
+SELECT count(*) AS leaked_backend_rows
+FROM pg_xmin_horizon
+WHERE kind = 'backend' AND pid = 0;
+
+-- gid joined in from pg_prepared_xacts and xact_start from
+-- pg_prepared_xacts.prepared.
+SELECT gid = 'xmin_horizon_p' AS gid_matches,
+       xact_start = (SELECT prepared FROM pg_prepared_xacts
+                     WHERE gid = 'xmin_horizon_p') AS xact_start_matches_prepared
+FROM pg_xmin_horizon
+WHERE kind = 'prepared_xact' AND gid = 'xmin_horizon_p';
+
+ROLLBACK PREPARED 'xmin_horizon_p';
+DROP TABLE xmin_horizon_t;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 85d989f395d..b0ab53ec131 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3548,6 +3548,8 @@ XactLockTableWaitInfo
 XidBoundsViolation
 XidCacheStatus
 XidCommitStatus
+XidHorizonProc
+XidHorizonRow
 XidStatus
 XmlExpr
 XmlExprOp
-- 
2.50.1 (Apple Git-155)

