From 6f2d77c72a4e40e52ff83c54ec0cae8e29a962c4 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Fri, 16 Jul 2021 23:10:22 +0900
Subject: [PATCH v8 2/5] Add pg_stat_subscription_errors statistics view.

This commits adds new system view pg_stat_logical_replication_error,
showing errors happening during applying logical replication changes
as well as during performing initial table synchronization.

The subscription error entries are removed by autovacuum workers when
the table synchronization competed in table sync worker cases and when
dropping the subscription in apply worker cases.

It also adds SQL function pg_stat_reset_subscription_error() to
reset the single subscription error.
---
 doc/src/sgml/monitoring.sgml             | 160 ++++++
 src/backend/catalog/system_functions.sql |   2 +
 src/backend/catalog/system_views.sql     |  27 +
 src/backend/postmaster/pgstat.c          | 651 +++++++++++++++++++++++
 src/backend/replication/logical/worker.c |  40 +-
 src/backend/utils/adt/pgstatfuncs.c      | 119 +++++
 src/backend/utils/error/elog.c           |  16 +
 src/include/catalog/pg_proc.dat          |  13 +
 src/include/pgstat.h                     | 106 ++++
 src/include/utils/elog.h                 |   1 +
 src/test/regress/expected/rules.out      |  22 +
 src/tools/pgindent/typedefs.list         |   5 +
 12 files changed, 1159 insertions(+), 3 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 74a58a916c..0fd0bbfa1f 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -346,6 +346,15 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       </entry>
      </row>
 
+     <row>
+      <entry><structname>pg_stat_subscription_errors</structname><indexterm><primary>pg_stat_subscription_errors</primary></indexterm></entry>
+      <entry>One row per error that happened on subscription, showing information about
+       the subscription errors.
+       See <link linkend="monitoring-pg-stat-subscription-errors">
+       <structname>pg_stat_subscription_errors</structname></link> for details.
+      </entry>
+     </row>
+
      <row>
       <entry><structname>pg_stat_ssl</structname><indexterm><primary>pg_stat_ssl</primary></indexterm></entry>
       <entry>One row per connection (regular and replication), showing information about
@@ -3050,6 +3059,135 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
 
  </sect2>
 
+ <sect2 id="monitoring-pg-stat-subscription-errors">
+  <title><structname>pg_stat_subscription_errors</structname></title>
+
+  <indexterm>
+   <primary>pg_stat_subscription_errors</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_stat_subscription_errors</structname> view will contain one
+   row per subscription error reported by workers applying logical replication
+   changes and workers handling the initial data copy of the subscribed tables.
+  </para>
+
+  <table id="pg-stat-subscription-errors" xreflabel="pg_stat_subscription-errors">
+   <title><structname>pg_stat_subscription_errors</structname> View</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>datname</structfield> <type>name</type>
+      </para>
+      <para>
+        Name of the database in which the subscription is created.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>subid</structfield> <type>oid</type>
+      </para>
+      <para>
+       OID of the subscription.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>subname</structfield> <type>name</type>
+      </para>
+      <para>
+       Name of the subscription.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>relid</structfield> <type>oid</type>
+      </para>
+      <para>
+       OID of the relation that the worker is processing when the
+       error happened.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>command</structfield> <type>text</type>
+      </para>
+      <para>
+        Name of command being applied when the error happened.  This
+        field is always NULL if the error is reported by
+        <literal>tablesync</literal> worker.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>xid</structfield> <type>xid</type>
+      </para>
+      <para>
+        Transaction ID of publisher node being applied when the error
+        happened.  This field is always NULL if the error is reported
+        by <literal>tablesync</literal> worker.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>failure_source</structfield> <type>text</type>
+      </para>
+      <para>
+        Type of worker reported the error: <literal>apply</literal> or
+        <literal>tablesync</literal>.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>failure_count</structfield> <type>uint8</type>
+      </para>
+      <para>
+       Number of times error happened on the worker.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>last_failure</structfield> <type>timestamp with time zone</type>
+      </para>
+      <para>
+       Time at which the last error happened.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>last_failure_message</structfield> <type>text</type>
+      </para>
+      <para>
+       Error message which is reported last failure time.
+      </para></entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+
+ </sect2>
+
  <sect2 id="monitoring-pg-stat-ssl-view">
   <title><structname>pg_stat_ssl</structname></title>
 
@@ -5172,6 +5310,28 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
          can be granted EXECUTE to run the function.
        </para></entry>
       </row>
+
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+          <primary>pg_stat_reset_subscription_error</primary>
+        </indexterm>
+        <function>pg_stat_reset_subscription_error</function> ( <parameter>subid</parameter> <type>oid</type>, <parameter>relid</parameter> <type>oid</type> )
+        <returnvalue>void</returnvalue>
+       </para>
+       <para>
+        Resets statistics of a single subscription error.  If
+        the argument <parameter>relid</parameter> is not <literal>NULL</literal>,
+        resets error statistics of the <literal>tablesync</literal> worker for
+        the relation with <parameter>relid</parameter>.  Otherwise, resets the
+        error statistics of the <literal>apply</literal> worker running on the
+        subscription with <parameter>subid</parameter>.
+       </para>
+       <para>
+         This function is restricted to superusers by default, but other users
+         can be granted EXECUTE to run the function.
+       </para></entry>
+      </row>
      </tbody>
     </tgroup>
    </table>
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index a416e94d37..c9aa6f04d3 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -639,6 +639,8 @@ REVOKE EXECUTE ON FUNCTION pg_stat_reset_single_function_counters(oid) FROM publ
 
 REVOKE EXECUTE ON FUNCTION pg_stat_reset_replication_slot(text) FROM public;
 
+REVOKE EXECUTE ON FUNCTION pg_stat_reset_subscription_error(oid, oid) FROM public;
+
 REVOKE EXECUTE ON FUNCTION lo_import(text) FROM public;
 
 REVOKE EXECUTE ON FUNCTION lo_import(text, oid) FROM public;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 55f6e3711d..449692afa9 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1257,3 +1257,30 @@ REVOKE ALL ON pg_subscription FROM public;
 GRANT SELECT (oid, subdbid, subname, subowner, subenabled, subbinary,
               substream, subtwophasestate, subslotname, subsynccommit, subpublications)
     ON pg_subscription TO public;
+
+CREATE VIEW pg_stat_subscription_errors AS
+    SELECT
+	d.datname,
+	sr.subid,
+	s.subname,
+	e.relid,
+	e.command,
+	e.xid,
+	e.failure_source,
+	e.failure_count,
+	e.last_failure,
+	e.last_failure_message,
+	e.stats_reset
+    FROM (SELECT
+              oid as subid,
+              NULL as relid
+          FROM pg_subscription
+          UNION ALL
+          SELECT
+              srsubid as subid,
+              srrelid as relid
+          FROM pg_subscription_rel
+          WHERE srsubstate <> 'r') sr,
+          LATERAL pg_stat_get_subscription_error(sr.subid, sr.relid) e
+          JOIN pg_database d ON (e.datid = d.oid)
+          JOIN pg_subscription s ON (e.subid = s.oid);
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 1b54ef74eb..498ea63863 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -41,6 +41,8 @@
 #include "catalog/partition.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_proc.h"
+#include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_rel.h"
 #include "common/ip.h"
 #include "executor/instrument.h"
 #include "libpq/libpq.h"
@@ -106,6 +108,7 @@
 #define PGSTAT_TAB_HASH_SIZE	512
 #define PGSTAT_FUNCTION_HASH_SIZE	512
 #define PGSTAT_REPLSLOT_HASH_SIZE	32
+#define PGSTAT_SUBSCRIPTION_HASH_SIZE	32
 
 
 /* ----------
@@ -280,6 +283,7 @@ static PgStat_GlobalStats globalStats;
 static PgStat_WalStats walStats;
 static PgStat_SLRUStats slruStats[SLRU_NUM_ELEMENTS];
 static HTAB *replSlotStatHash = NULL;
+static HTAB *subscriptionHash = NULL;
 
 /*
  * List of OIDs of databases we need to write out.  If an entry is InvalidOid,
@@ -330,6 +334,12 @@ static bool pgstat_db_requested(Oid databaseid);
 static PgStat_StatReplSlotEntry *pgstat_get_replslot_entry(NameData name, bool create_it);
 static void pgstat_reset_replslot(PgStat_StatReplSlotEntry *slotstats, TimestampTz ts);
 
+static PgStat_StatSubEntry *pgstat_get_subscription_entry(Oid subid,
+														  bool create);
+static PgStat_StatSubErrEntry *pgstat_get_subscription_error_entry(Oid subid,
+																   Oid subrelid,
+																   bool create);
+
 static void pgstat_send_tabstat(PgStat_MsgTabstat *tsmsg);
 static void pgstat_send_funcstats(void);
 static void pgstat_send_slru(void);
@@ -370,6 +380,10 @@ static void pgstat_recv_checksum_failure(PgStat_MsgChecksumFailure *msg, int len
 static void pgstat_recv_connstat(PgStat_MsgConn *msg, int len);
 static void pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len);
 static void pgstat_recv_tempfile(PgStat_MsgTempFile *msg, int len);
+static void pgstat_recv_subscription_error(PgStat_MsgSubscriptionErr *msg, int len);
+static void pgstat_recv_subscription_error_purge(PgStat_MsgSubscriptionErrPurge *msg,
+												 int len);
+static void pgstat_recv_subscription_purge(PgStat_MsgSubscriptionPurge *msg, int len);
 
 /* ------------------------------------------------------------
  * Public functions called from postmaster follow
@@ -1148,6 +1162,166 @@ pgstat_vacuum_stat(void)
 		}
 	}
 
+	/*
+	 * Search for all the dead subscriptions and error entries in stats
+	 * hashtable and tell the stats collector to drop them.
+	 */
+	if (subscriptionHash)
+	{
+		PgStat_MsgSubscriptionPurge submsg;
+		PgStat_StatSubEntry *subent;
+		HTAB	   *htab;
+
+		htab = pgstat_collect_oids(SubscriptionRelationId, Anum_pg_subscription_oid);
+
+		submsg.m_nentries = 0;
+
+		hash_seq_init(&hstat, subscriptionHash);
+		while ((subent = (PgStat_StatSubEntry *) hash_seq_search(&hstat)) != NULL)
+		{
+			PgStat_MsgSubscriptionErrPurge errmsg;
+			PgStat_StatSubErrEntry *errent;
+			HASH_SEQ_STATUS hstat_rel;
+			List	   *not_ready_rels_list;
+			HTAB	   *not_ready_rels_htab;
+			ListCell   *lc;
+			HASHCTL		hash_ctl;
+
+			CHECK_FOR_INTERRUPTS();
+
+			if (hash_search(htab, (void *) &(subent->subid), HASH_FIND, NULL) == NULL)
+			{
+				/* This subscription is dead, add subid to the message */
+				submsg.m_subids[submsg.m_nentries++] = subent->subid;
+
+				/*
+				 * If the message is full, send it out and reinitialize to
+				 * empty
+				 */
+				if (submsg.m_nentries >= PGSTAT_NUM_SUBSCRIPTIONPURGE)
+				{
+					len = offsetof(PgStat_MsgSubscriptionPurge, m_subids[0])
+						+ submsg.m_nentries * sizeof(Oid);
+
+					pgstat_setheader(&submsg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONPURGE);
+					pgstat_send(&submsg, len);
+					submsg.m_nentries = 0;
+				}
+
+				continue;
+			}
+
+			/*
+			 * Nothing to do here if the subscription exists but has no error
+			 * entries.
+			 */
+			if (subent->suberrors == NULL)
+				continue;
+
+			/*
+			 * The subscription has error entries. We search errors of the
+			 * table sync workers who are already in sync state.  Those errors
+			 * should be removed.
+			 *
+			 * Note that the lifetime of error entries of the apply worker and
+			 * the table sync worker are different.  The former lives until
+			 * the subscription is dropped whereas the latter lives the table
+			 * synchronization is completed.
+			 */
+			not_ready_rels_list = GetSubscriptionNotReadyRelations(subent->subid);
+
+			hash_ctl.keysize = sizeof(Oid);
+			hash_ctl.entrysize = sizeof(SubscriptionRelState);
+			not_ready_rels_htab = hash_create("not ready relations in subscription",
+											  64,
+											  &hash_ctl,
+											  HASH_ELEM | HASH_BLOBS);
+
+			/*
+			 * The number of not-ready relations can be high for example right
+			 * after creating a subscription, so we load the list of
+			 * SubscriptionRelState into the hash table for a faster lookup.
+			 */
+			foreach(lc, not_ready_rels_list)
+			{
+				SubscriptionRelState *r_elem = (SubscriptionRelState *) lfirst(lc);
+				SubscriptionRelState *r_entry;
+
+				CHECK_FOR_INTERRUPTS();
+				r_entry = hash_search(not_ready_rels_htab, (void *) &(r_elem->relid),
+									  HASH_ENTER, NULL);
+				memcpy(r_entry, r_elem, sizeof(SubscriptionRelState));
+			}
+
+			list_free(not_ready_rels_list);
+
+			errmsg.m_nentries = 0;
+			errmsg.m_subid = subent->subid;
+
+			/*
+			 * Search for all error entries of which relation is already ready
+			 * state
+			 */
+			hash_seq_init(&hstat_rel, subent->suberrors);
+			while ((errent = (PgStat_StatSubErrEntry *) hash_seq_search(&hstat_rel)) != NULL)
+			{
+				CHECK_FOR_INTERRUPTS();
+
+				/* Skip the apply worker's error */
+				if (!OidIsValid(errent->subrelid))
+					continue;
+
+				/*
+				 * Add the relid to the message if the table synchronization
+				 * for this relation already complete or the table is no
+				 * longer subscribed.
+				 */
+				if (hash_search(not_ready_rels_htab, (void *) &(errent->subrelid),
+								HASH_FIND, NULL) == NULL)
+					errmsg.m_relids[errmsg.m_nentries++] = errent->subrelid;
+
+				/*
+				 * If the message is full, send it out and reinitialize to
+				 * empty
+				 */
+				if (errmsg.m_nentries >= PGSTAT_NUM_SUBSCRIPTIONERRPURGE)
+				{
+					len = offsetof(PgStat_MsgSubscriptionErrPurge, m_relids[0])
+						+ errmsg.m_nentries * sizeof(Oid);
+
+					pgstat_setheader(&errmsg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONERRPURGE);
+					pgstat_send(&errmsg, len);
+					errmsg.m_nentries = 0;
+				}
+			}
+
+			/* Send the rest of dead error entries */
+			if (errmsg.m_nentries > 0)
+			{
+				len = offsetof(PgStat_MsgSubscriptionErrPurge, m_relids[0])
+					+ errmsg.m_nentries * sizeof(Oid);
+
+				pgstat_setheader(&errmsg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONERRPURGE);
+				pgstat_send(&errmsg, len);
+				errmsg.m_nentries = 0;
+			}
+
+			hash_destroy(not_ready_rels_htab);
+		}
+
+		/* Send the rest of dead subscriptions */
+		if (submsg.m_nentries > 0)
+		{
+			len = offsetof(PgStat_MsgSubscriptionPurge, m_subids[0])
+				+ submsg.m_nentries * sizeof(Oid);
+
+			pgstat_setheader(&submsg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONPURGE);
+			pgstat_send(&submsg, len);
+		}
+
+		hash_destroy(htab);
+	}
+
 	/*
 	 * Lookup our own database entry; if not found, nothing more to do.
 	 */
@@ -1557,6 +1731,25 @@ pgstat_reset_replslot_counter(const char *name)
 	pgstat_send(&msg, sizeof(msg));
 }
 
+/* ----------
+ * pgstat_reset_subscription_error() -
+ *
+ *	Tell the collector about reset the subscription error.
+ * ----------
+ */
+void
+pgstat_reset_subscription_error(Oid subid, Oid subrelid)
+{
+	PgStat_MsgSubscriptionErr msg;
+
+	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONERR);
+	msg.m_subid = subid;
+	msg.m_subrelid = subrelid;
+	msg.m_reset = true;
+
+	pgstat_send(&msg, offsetof(PgStat_MsgSubscriptionErr, m_reset) + sizeof(bool));
+}
+
 /* ----------
  * pgstat_report_autovac() -
  *
@@ -1877,6 +2070,37 @@ pgstat_report_replslot_drop(const char *slotname)
 	pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
 }
 
+/* ----------
+ * pgstat_report_subscription_error() -
+ *
+ *	Tell the collector about the subscription error.
+ * ----------
+ */
+void
+pgstat_report_subscription_error(Oid subid, Oid subrelid, Oid relid,
+								 LogicalRepMsgType command, TransactionId xid,
+								 const char *errmsg)
+{
+	PgStat_MsgSubscriptionErr msg;
+	int			len;
+
+	len = offsetof(PgStat_MsgSubscriptionErr, m_errmsg[0]) + strlen(errmsg);
+	Assert(len < PGSTAT_MAX_MSG_SIZE);
+
+	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONERR);
+	msg.m_subid = subid;
+	msg.m_subrelid = subrelid;
+	msg.m_databaseid = MyDatabaseId;
+	msg.m_relid = relid;
+	msg.m_reset = false;
+	msg.m_command = command;
+	msg.m_xid = xid;
+	msg.m_failure_time = GetCurrentTimestamp();
+	strlcpy(msg.m_errmsg, errmsg, PGSTAT_SUBSCRIPTIONERR_MSGLEN);
+
+	pgstat_send(&msg, len);
+}
+
 /* ----------
  * pgstat_ping() -
  *
@@ -2944,6 +3168,22 @@ pgstat_fetch_replslot(NameData slotname)
 	return pgstat_get_replslot_entry(slotname, false);
 }
 
+/*
+ * ---------
+ * pgstat_fetch_subscription_error() -
+ *
+ *	Support function for the SQL-callable pgstat* functions. Returns
+ *	a pointer to the subscription error struct.
+ * ---------
+ */
+PgStat_StatSubErrEntry *
+pgstat_fetch_subscription_error(Oid subid, Oid relid)
+{
+	backend_read_statsfile();
+
+	return pgstat_get_subscription_error_entry(subid, relid, false);
+}
+
 /*
  * Shut down a single backend's statistics reporting at process exit.
  *
@@ -3525,6 +3765,19 @@ PgstatCollectorMain(int argc, char *argv[])
 					pgstat_recv_connstat(&msg.msg_conn, len);
 					break;
 
+				case PGSTAT_MTYPE_SUBSCRIPTIONERR:
+					pgstat_recv_subscription_error(&msg.msg_subscriptionerr, len);
+					break;
+
+				case PGSTAT_MTYPE_SUBSCRIPTIONERRPURGE:
+					pgstat_recv_subscription_error_purge(&msg.msg_subscriptionerrpurge,
+														 len);
+					break;
+
+				case PGSTAT_MTYPE_SUBSCRIPTIONPURGE:
+					pgstat_recv_subscription_purge(&msg.msg_subscriptionpurge, len);
+					break;
+
 				default:
 					break;
 			}
@@ -3826,6 +4079,50 @@ pgstat_write_statsfiles(bool permanent, bool allDbs)
 		}
 	}
 
+	/*
+	 * Write subscription error structs
+	 */
+	if (subscriptionHash)
+	{
+		PgStat_StatSubEntry *subent;
+
+		hash_seq_init(&hstat, subscriptionHash);
+		while ((subent = (PgStat_StatSubEntry *) hash_seq_search(&hstat)) != NULL)
+		{
+			PgStat_StatSubErrEntry *errent;
+			HASH_SEQ_STATUS relhstat;
+			int32		nerrors = hash_get_num_entries(subent->suberrors);
+
+			/* Skip this subscription if not have any errors */
+			if (subent->suberrors == NULL)
+				continue;
+
+			fputc('S', fpout);
+			rc = fwrite(subent, sizeof(PgStat_StatSubEntry), 1, fpout);
+			(void) rc;			/* we'll check for error with ferror */
+
+			/* The number of errors follows */
+			rc = fwrite(&nerrors, sizeof(int32), 1, fpout);
+			(void) rc;			/* we'll check for error with ferror */
+
+			/* Then, the error entries follow */
+			hash_seq_init(&relhstat, subent->suberrors);
+			while ((errent = (PgStat_StatSubErrEntry *) hash_seq_search(&relhstat)) != NULL)
+			{
+				/*
+				 * XXX we write the whole PgStat_StatSubErrEntry entry that
+				 * contains the fixed-legnth error message string which is
+				 * PGSTAT_SUBSCRIPTIONERR_MSGLEN in length, making the stats
+				 * file bloat.  It's okay since we assume that the number of
+				 * error entries is not high.  But if the expectation became
+				 * false we should write the string and its length instead.
+				 */
+				rc = fwrite(errent, sizeof(PgStat_StatSubErrEntry), 1, fpout);
+				(void) rc;		/* we'll check for error with ferror */
+			}
+		}
+	}
+
 	/*
 	 * No more output to be done. Close the temp file and replace the old
 	 * pgstat.stat with it.  The ferror() check replaces testing for error
@@ -4287,6 +4584,100 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep)
 					break;
 				}
 
+				/*
+				 * 'S'	A PgStat_StatSubEntry struct followed by the number of
+				 * errors and PgStat_StatSubErrEntry structs, describing
+				 * subscription errors.
+				 */
+			case 'S':
+				{
+					PgStat_StatSubEntry subbuf;
+					PgStat_StatSubEntry *subent;
+					int32		nerrors;
+
+					/* Read the subscription entry */
+					if (fread(&subbuf, 1, sizeof(PgStat_StatSubEntry), fpin) !=
+						sizeof(PgStat_StatSubEntry))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						goto done;
+					}
+
+					/* Create hash table if we don't have it already. */
+					if (subscriptionHash == NULL)
+					{
+						HASHCTL		hash_ctl;
+
+						hash_ctl.keysize = sizeof(Oid);
+						hash_ctl.entrysize = sizeof(PgStat_StatSubEntry);
+						hash_ctl.hcxt = pgStatLocalContext;
+						subscriptionHash = hash_create("Subscription stat entries",
+													   PGSTAT_SUBSCRIPTION_HASH_SIZE,
+													   &hash_ctl,
+													   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+					}
+
+					/* Enter the subscription error entry */
+					subent =
+						(PgStat_StatSubEntry *) hash_search(subscriptionHash,
+															(void *) &(subbuf.subid),
+															HASH_ENTER, NULL);
+					subent->suberrors = NULL;
+
+					/* Read the number of errors in the subscription */
+					if (fread(&nerrors, 1, sizeof(int32), fpin) != sizeof(int32))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						goto done;
+					}
+
+					/* Read error entries */
+					for (int i = 0; i < nerrors; i++)
+					{
+						PgStat_StatSubErrEntry errbuf;
+						PgStat_StatSubErrEntry *errent;
+
+						if (fread(&errbuf, 1, sizeof(PgStat_StatSubErrEntry), fpin) !=
+							sizeof(PgStat_StatSubErrEntry))
+						{
+							ereport(pgStatRunningInCollector ? LOG : WARNING,
+									(errmsg("corrupted statistics file \"%s\"",
+											statfile)));
+							goto done;
+						}
+
+						if (subent->suberrors == NULL)
+						{
+							HASHCTL		hash_ctl;
+
+							hash_ctl.keysize = sizeof(Oid);
+							hash_ctl.entrysize = sizeof(PgStat_StatSubErrEntry);
+							hash_ctl.hcxt = pgStatLocalContext;
+							subent->suberrors = hash_create("Subscription error hash",
+															PGSTAT_SUBSCRIPTION_HASH_SIZE,
+															&hash_ctl,
+															HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+						}
+
+						/*
+						 * Enter the error information to the subscription
+						 * hash
+						 */
+						errent =
+							(PgStat_StatSubErrEntry *) hash_search(subent->suberrors,
+																   (void *) &(errbuf.subrelid),
+																   HASH_ENTER, NULL);
+
+						memcpy(errent, &errbuf, sizeof(PgStat_StatSubErrEntry));
+					}
+
+					break;
+				}
+
 			case 'E':
 				goto done;
 
@@ -4629,6 +5020,50 @@ pgstat_read_db_statsfile_timestamp(Oid databaseid, bool permanent,
 				}
 				break;
 
+				/*
+				 * 'S'	A PgStat_StatSubEntry struct followed by the number of
+				 * errors and PgStat_StatSubErrEntry structs describing
+				 * subscription errors.
+				 */
+			case 'S':
+				{
+					PgStat_StatSubEntry subbuf;
+					PgStat_StatSubErrEntry errbuf;
+					int32		nerrors;
+
+					if (fread(&subbuf, 1, sizeof(PgStat_StatSubEntry), fpin)
+						!= sizeof(PgStat_StatSubEntry))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						FreeFile(fpin);
+						return false;
+					}
+
+					if (fread(&nerrors, 1, sizeof(int32), fpin) != sizeof(int32))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						goto done;
+					}
+
+					for (int i = 0; i < nerrors; i++)
+					{
+						if (fread(&errbuf, 1, sizeof(PgStat_StatSubErrEntry), fpin) !=
+							sizeof(PgStat_StatSubErrEntry))
+						{
+							ereport(pgStatRunningInCollector ? LOG : WARNING,
+									(errmsg("corrupted statistics file \"%s\"",
+											statfile)));
+							goto done;
+						}
+					}
+				}
+
+				break;
+
 			case 'E':
 				goto done;
 
@@ -4834,6 +5269,7 @@ pgstat_clear_snapshot(void)
 	pgStatLocalContext = NULL;
 	pgStatDBHash = NULL;
 	replSlotStatHash = NULL;
+	subscriptionHash = NULL;
 
 	/*
 	 * Historically the backend_status.c facilities lived in this file, and
@@ -5779,6 +6215,116 @@ pgstat_recv_funcpurge(PgStat_MsgFuncpurge *msg, int len)
 	}
 }
 
+/* ----------
+ * pgstat_recv_subscription_error() -
+ *
+ *	Process a SUBSCRIPTIONERR message.
+ * ----------
+ */
+static void
+pgstat_recv_subscription_error(PgStat_MsgSubscriptionErr *msg, int len)
+{
+	PgStat_StatSubErrEntry *errent;
+	bool		create = !msg->m_reset;
+
+	/* Get subscription error */
+	errent = pgstat_get_subscription_error_entry(msg->m_subid,
+												 msg->m_subrelid,
+												 create);
+
+	if (msg->m_reset)
+	{
+		if (errent == NULL)
+			return;
+
+		/* reset fields and set reset timestamp */
+		errent->relid = InvalidOid;
+		errent->command = 0;
+		errent->xid = InvalidTransactionId;
+		errent->failure_count = 0;
+		errent->last_failure = 0;
+		errent->last_errmsg[0] = '\0';
+		errent->stat_reset_timestamp = GetCurrentTimestamp();
+	}
+	else
+	{
+		Assert(errent);
+
+		/* update the error entry */
+		errent->databaseid = msg->m_databaseid;
+		errent->relid = msg->m_relid;
+		errent->command = msg->m_command;
+		errent->xid = msg->m_xid;
+		errent->failure_count++;
+		errent->last_failure = msg->m_failure_time;
+		strlcpy(errent->last_errmsg, msg->m_errmsg, PGSTAT_SUBSCRIPTIONERR_MSGLEN);
+	}
+}
+
+/* ----------
+ * pgstat_recv_subscription_purge() -
+ *
+ *	Process a SUBSCRIPTIONPURGE message.
+ * ----------
+ */
+static void
+pgstat_recv_subscription_purge(PgStat_MsgSubscriptionPurge *msg, int len)
+{
+	if (subscriptionHash == NULL)
+		return;
+
+	for (int i = 0; i < msg->m_nentries; i++)
+	{
+		PgStat_StatSubEntry *subent;
+
+		subent = pgstat_get_subscription_entry(msg->m_subids[i], false);
+
+		/*
+		 * Nothing to do if the subscription entry is not found.  This could
+		 * happen when the subscription is dropped and the message for
+		 * dropping subscription entry arrived before the message for
+		 * reporting the error.
+		 */
+		if (subent == NULL)
+			continue;
+
+		/* Cleanup the hash table for errors */
+		if (subent->suberrors != NULL)
+			hash_destroy(subent->suberrors);
+
+		/* Remote the subscription entry */
+		(void) hash_search(subscriptionHash, (void *) &(msg->m_subids[i]),
+						   HASH_REMOVE, NULL);
+	}
+}
+
+/* ----------
+ * pgstat_recv_subscription_error_purge() -
+ *
+ *	Process a SUBSCRIPTIONERRPURGE message.
+ * ----------
+ */
+static void
+pgstat_recv_subscription_error_purge(PgStat_MsgSubscriptionErrPurge *msg, int len)
+{
+	PgStat_StatSubEntry *subent;
+
+	subent = pgstat_get_subscription_entry(msg->m_subid, false);
+
+	/*
+	 * Nothing to do if the subscription entry is not found.  This could
+	 * happen when the subscription with msg->m_subid is removed and the
+	 * corresponding entry is also removed before receiving the error purge
+	 * message.
+	 */
+	if (subent == NULL)
+		return;
+
+	for (int i = 0; i < msg->m_nentries; i++)
+		(void) hash_search(subent->suberrors, (void *) &(msg->m_relids[i]),
+						   HASH_REMOVE, NULL);
+}
+
 /* ----------
  * pgstat_write_statsfile_needed() -
  *
@@ -5876,6 +6422,111 @@ pgstat_get_replslot_entry(NameData name, bool create)
 	return slotent;
 }
 
+/* ----------
+ * pgstat_get_subscription_entry
+ *
+ * Return the subscription statistics with the subscription OID. Return NULL
+ * if not found and the caller didn't request to create it.
+ *
+ * 'create' tells whether to create the new subscription entry if it is not
+ * found.
+ * ----------
+ */
+static PgStat_StatSubEntry *
+pgstat_get_subscription_entry(Oid subid, bool create)
+{
+	PgStat_StatSubEntry *subent;
+	HASHACTION	action = (create ? HASH_ENTER : HASH_FIND);
+	bool		found;
+
+	if (subscriptionHash == NULL)
+	{
+		HASHCTL		hash_ctl;
+
+		if (!create)
+			return NULL;
+
+		hash_ctl.keysize = sizeof(Oid);
+		hash_ctl.entrysize = sizeof(PgStat_StatSubEntry);
+		subscriptionHash = hash_create("Subscription stat entries",
+									   PGSTAT_SUBSCRIPTION_HASH_SIZE,
+									   &hash_ctl,
+									   HASH_ELEM | HASH_BLOBS);
+	}
+
+	subent = (PgStat_StatSubEntry *) hash_search(subscriptionHash,
+												 (void *) &subid,
+												 action, &found);
+
+	/* initialize field */
+	if (create && !found)
+		subent->suberrors = NULL;
+
+	return subent;
+}
+
+/* ----------
+ * pgstat_get_subscription_error_entry
+ *
+ * Return the entry of subscription error entry with the subscription
+ * OID and relation OID.  Return NULL if not found and the caller didn't
+ * request to create it.
+ *
+ * 'create' tells whether to create the new subscription relation entry if it is
+ * not found.
+ * ----------
+ */
+static PgStat_StatSubErrEntry *
+pgstat_get_subscription_error_entry(Oid subid, Oid subrelid, bool create)
+{
+	PgStat_StatSubEntry *subent;
+	PgStat_StatSubErrEntry *errent;
+	HASHACTION	action = (create ? HASH_ENTER : HASH_FIND);
+	bool		found;
+
+	subent = pgstat_get_subscription_entry(subid, create);
+
+	if (subent == NULL)
+	{
+		Assert(!create);
+		return NULL;
+	}
+
+	if (subent->suberrors == NULL)
+	{
+		HASHCTL		hash_ctl;
+
+		if (!create)
+			return NULL;
+
+		hash_ctl.keysize = sizeof(Oid);
+		hash_ctl.entrysize = sizeof(PgStat_StatSubErrEntry);
+		subent->suberrors = hash_create("Subscription error hash",
+										PGSTAT_SUBSCRIPTION_HASH_SIZE,
+										&hash_ctl,
+										HASH_ELEM | HASH_BLOBS);
+	}
+
+	errent = (PgStat_StatSubErrEntry *) hash_search(subent->suberrors,
+													(void *) &subrelid,
+													action, &found);
+
+	/* initialize fields */
+	if (create && !found)
+	{
+		errent->databaseid = InvalidOid;
+		errent->relid = InvalidOid;
+		errent->command = 0;
+		errent->xid = InvalidTransactionId;
+		errent->failure_count = 0;
+		errent->last_failure = 0;
+		errent->last_errmsg[0] = '\0';
+		errent->stat_reset_timestamp = 0;
+	}
+
+	return errent;
+}
+
 /* ----------
  * pgstat_reset_replslot
  *
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a74493b610..7ecabaacb8 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3556,8 +3556,23 @@ ApplyWorkerMain(Datum main_arg)
 	{
 		char	   *syncslotname;
 
-		/* This is table synchronization worker, call initial sync. */
-		syncslotname = LogicalRepSyncTableStart(&origin_startpos);
+		PG_TRY();
+		{
+			/* This is table synchronization worker, call initial sync. */
+			syncslotname = LogicalRepSyncTableStart(&origin_startpos);
+		}
+		PG_CATCH();
+		{
+			/* report the table sync error */
+			pgstat_report_subscription_error(MyLogicalRepWorker->subid,
+											 MyLogicalRepWorker->relid,
+											 MyLogicalRepWorker->relid,
+											 0,
+											 InvalidTransactionId,
+											 geterrmessage());
+			PG_RE_THROW();
+		}
+		PG_END_TRY();
 
 		/* allocate slot name in long-lived context */
 		myslotname = MemoryContextStrdup(ApplyContext, syncslotname);
@@ -3675,7 +3690,26 @@ ApplyWorkerMain(Datum main_arg)
 	}
 
 	/* Run the main loop. */
-	LogicalRepApplyLoop(origin_startpos);
+	PG_TRY();
+	{
+		LogicalRepApplyLoop(origin_startpos);
+	}
+	PG_CATCH();
+	{
+		/* report the apply error */
+		if (apply_error_callback_arg.command != 0)
+			pgstat_report_subscription_error(MySubscription->oid,
+											 InvalidOid,
+											 apply_error_callback_arg.rel != NULL
+											 ? apply_error_callback_arg.rel->localreloid
+											 : InvalidOid,
+											 apply_error_callback_arg.command,
+											 apply_error_callback_arg.remote_xid,
+											 geterrmessage());
+
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
 
 	proc_exit(0);
 }
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index ff5aedc99c..4048c99a9e 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -24,6 +24,7 @@
 #include "pgstat.h"
 #include "postmaster/bgworker_internals.h"
 #include "postmaster/postmaster.h"
+#include "replication/logicalproto.h"
 #include "replication/slot.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
@@ -2239,6 +2240,23 @@ pg_stat_reset_replication_slot(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+/* Reset a subscription error stats */
+Datum
+pg_stat_reset_subscription_error(PG_FUNCTION_ARGS)
+{
+	Oid			subid = PG_GETARG_OID(0);
+	Oid			relid;
+
+	if (PG_ARGISNULL(1))
+		relid = InvalidOid;		/* reset apply worker error stats */
+	else
+		relid = PG_GETARG_OID(1);	/* reset table sync worker error stats */
+
+	pgstat_reset_subscription_error(subid, relid);
+
+	PG_RETURN_VOID();
+}
+
 Datum
 pg_stat_get_archiver(PG_FUNCTION_ARGS)
 {
@@ -2379,3 +2397,104 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 	/* Returns the record as Datum */
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
 }
+
+/*
+ * Get the subscription error for the given subscription (and relation).
+ */
+Datum
+pg_stat_get_subscription_error(PG_FUNCTION_ARGS)
+{
+#define PG_STAT_GET_SUBSCRIPTION_ERROR_COLS 10
+	Oid			subid = PG_GETARG_OID(0);
+	Oid			relid;
+	TupleDesc	tupdesc;
+	Datum		values[PG_STAT_GET_SUBSCRIPTION_ERROR_COLS];
+	bool		nulls[PG_STAT_GET_SUBSCRIPTION_ERROR_COLS];
+	PgStat_StatSubErrEntry *errent;
+
+	/* Initialise values and NULL flags arrays */
+	MemSet(values, 0, sizeof(values));
+	MemSet(nulls, 0, sizeof(nulls));
+
+	/* Initialise attributes information in the tuple descriptor */
+	tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBSCRIPTION_ERROR_COLS);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "datid",
+					   OIDOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "subid",
+					   OIDOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "relid",
+					   OIDOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "command",
+					   TEXTOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "xid",
+					   XIDOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "failure_source",
+					   TEXTOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "failure_count",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "last_failure",
+					   TIMESTAMPTZOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "last_failure_message",
+					   TEXTOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "stats_reset",
+					   TIMESTAMPTZOID, -1, 0);
+	BlessTupleDesc(tupdesc);
+
+	if (PG_ARGISNULL(1))
+		relid = InvalidOid;
+	else
+		relid = PG_GETARG_OID(1);
+
+	/* Get subscription errors */
+	errent = pgstat_fetch_subscription_error(subid, relid);
+
+	/* Return NULL if the subscription doesn't have any errors */
+	if (errent == NULL)
+		PG_RETURN_NULL();
+
+	values[0] = ObjectIdGetDatum(errent->databaseid);
+	values[1] = ObjectIdGetDatum(subid);
+
+	if (OidIsValid(errent->relid))
+		values[2] = ObjectIdGetDatum(errent->relid);
+	else
+		nulls[2] = true;
+
+	if (errent->command == 0)
+		nulls[3] = true;
+	else
+	{
+		/* should be apply worker */
+		Assert(!OidIsValid(errent->subrelid));
+
+		values[3] = CStringGetTextDatum(logicalrep_message_type(errent->command));
+	}
+
+	if (TransactionIdIsValid(errent->xid))
+		values[4] = TransactionIdGetDatum(errent->xid);
+	else
+		nulls[4] = true;
+
+	if (OidIsValid(errent->subrelid))
+		values[5] = CStringGetTextDatum("tablesync");
+	else
+		values[5] = CStringGetTextDatum("apply");
+
+
+	values[6] = Int64GetDatum(errent->failure_count);
+
+	if (errent->last_failure == 0)
+		nulls[7] = true;
+	else
+		values[7] = TimestampTzGetDatum(errent->last_failure);
+
+	values[8] = CStringGetTextDatum(errent->last_errmsg);
+
+	if (errent->stat_reset_timestamp == 0)
+		nulls[9] = true;
+	else
+		values[9] = TimestampTzGetDatum(errent->stat_reset_timestamp);
+
+	/* Returns the record as Datum */
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
+}
diff --git a/src/backend/utils/error/elog.c b/src/backend/utils/error/elog.c
index a3e1c59a82..dd36850016 100644
--- a/src/backend/utils/error/elog.c
+++ b/src/backend/utils/error/elog.c
@@ -1441,6 +1441,22 @@ getinternalerrposition(void)
 	return edata->internalpos;
 }
 
+/*
+ * geterrmessage --- return the currently set error message
+ *
+ * This is only intended for use in error callback subroutines, since there
+ * is no other place outside elog.c where the concept is meaningful.
+ */
+const char *
+geterrmessage(void)
+{
+	ErrorData  *edata = &errordata[errordata_stack_depth];
+
+	/* we don't bother incrementing recursion_depth */
+	CHECK_STACK_DEPTH();
+
+	return (const char *) edata->message;
+}
 
 /*
  * Functions to allow construction of error message strings separately from
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index b603700ed9..7f9c27bdda 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5385,6 +5385,14 @@
   proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}',
   proargnames => '{slot_name,slot_name,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes,total_txns,total_bytes,stats_reset}',
   prosrc => 'pg_stat_get_replication_slot' },
+{ oid => '8523', descr => 'statistics: information about subscription error',
+  proname => 'pg_stat_get_subscription_error', prorows => '1', proisstrict => 'f',
+  proretset => 't', provolatile => 's', proparallel => 'r',
+  prorettype => 'record', proargtypes => 'oid oid',
+  proallargtypes => '{oid,oid,oid,oid,oid,text,xid,text,int8,timestamptz,text,timestamptz}',
+  proargmodes => '{i,i,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{subid,relid,datid,subid,relid,command,xid,failure_source,failure_count,last_failure,last_failure_message,stats_reset}',
+  prosrc => 'pg_stat_get_subscription_error' },
 { oid => '6118', descr => 'statistics: information about subscription',
   proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
@@ -5772,6 +5780,11 @@
   proname => 'pg_stat_reset_replication_slot', proisstrict => 'f',
   provolatile => 'v', prorettype => 'void', proargtypes => 'text',
   prosrc => 'pg_stat_reset_replication_slot' },
+{ oid => '8524',
+  descr => 'statistics: reset collected statistics for a single subscription error',
+  proname => 'pg_stat_reset_subscription_error', proisstrict => 'f',
+  provolatile => 'v', prorettype => 'void', proargtypes => 'oid oid',
+  prosrc => 'pg_stat_reset_subscription_error' },
 
 { oid => '3163', descr => 'current trigger depth',
   proname => 'pg_trigger_depth', provolatile => 's', proparallel => 'r',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 2068a68a5f..f66eb290bb 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -14,6 +14,7 @@
 #include "datatype/timestamp.h"
 #include "portability/instr_time.h"
 #include "postmaster/pgarch.h"	/* for MAX_XFN_CHARS */
+#include "replication/logicalproto.h"
 #include "utils/backend_progress.h" /* for backward compatibility */
 #include "utils/backend_status.h"	/* for backward compatibility */
 #include "utils/hsearch.h"
@@ -66,6 +67,9 @@ typedef enum StatMsgType
 	PGSTAT_MTYPE_RESETSINGLECOUNTER,
 	PGSTAT_MTYPE_RESETSLRUCOUNTER,
 	PGSTAT_MTYPE_RESETREPLSLOTCOUNTER,
+	PGSTAT_MTYPE_SUBSCRIPTIONERR,
+	PGSTAT_MTYPE_SUBSCRIPTIONERRPURGE,
+	PGSTAT_MTYPE_SUBSCRIPTIONPURGE,
 	PGSTAT_MTYPE_AUTOVAC_START,
 	PGSTAT_MTYPE_VACUUM,
 	PGSTAT_MTYPE_ANALYZE,
@@ -550,6 +554,68 @@ typedef struct PgStat_MsgReplSlot
 	PgStat_Counter m_total_bytes;
 } PgStat_MsgReplSlot;
 
+/* ----------
+ * PgStat_MsgSubscriptionErr	Sent by the apply worker or the table sync worker to
+ *								update/reset the error happening during logical
+ *								replication.
+ * ----------
+ */
+#define PGSTAT_SUBSCRIPTIONERR_MSGLEN 256
+typedef struct PgStat_MsgSubscriptionErr
+{
+	PgStat_MsgHdr m_hdr;
+
+	/*
+	 * m_subid and m_subrelid are used to determine the subscription and the
+	 * reporter of this error.  m_subrelid is InvalidOid if reported by the
+	 * apply worker, otherwise by the table sync worker.  In table sync worker
+	 * case, m_subrelid must be the same as m_relid.
+	 */
+	Oid			m_subid;
+	Oid			m_subrelid;
+
+	/* The clear message uses below field */
+	bool		m_reset;		/* Reset all fields and set reset_stats
+								 * timestamp */
+
+	/* The error report message uses below fields */
+	Oid			m_databaseid;
+	Oid			m_relid;
+	LogicalRepMsgType m_command;
+	TransactionId m_xid;
+	TimestampTz m_failure_time;
+	char		m_errmsg[PGSTAT_SUBSCRIPTIONERR_MSGLEN];
+} PgStat_MsgSubscriptionErr;
+
+/* ----------
+ * PgStat_MsgSubscriptionPurge	Sent by the autovacuum purge the subscriptions.
+ * ----------
+ */
+#define PGSTAT_NUM_SUBSCRIPTIONPURGE  \
+	((PGSTAT_MSG_PAYLOAD - sizeof(int)) / sizeof(Oid))
+
+typedef struct PgStat_MsgSubscriptionPurge
+{
+	PgStat_MsgHdr m_hdr;
+	int			m_nentries;
+	Oid			m_subids[PGSTAT_NUM_SUBSCRIPTIONPURGE];
+} PgStat_MsgSubscriptionPurge;
+
+/* ----------
+ * PgStat_MsgSubscriptionErrPurge	Sent by the autovacuum purge the subscription
+ *									errors.
+ * ----------
+ */
+#define PGSTAT_NUM_SUBSCRIPTIONERRPURGE  \
+	((PGSTAT_MSG_PAYLOAD - sizeof(Oid) - sizeof(int)) / sizeof(Oid))
+
+typedef struct PgStat_MsgSubscriptionErrPurge
+{
+	PgStat_MsgHdr m_hdr;
+	Oid			m_subid;
+	int			m_nentries;
+	Oid			m_relids[PGSTAT_NUM_SUBSCRIPTIONERRPURGE];
+} PgStat_MsgSubscriptionErrPurge;
 
 /* ----------
  * PgStat_MsgRecoveryConflict	Sent by the backend upon recovery conflict
@@ -722,6 +788,9 @@ typedef union PgStat_Msg
 	PgStat_MsgChecksumFailure msg_checksumfailure;
 	PgStat_MsgReplSlot msg_replslot;
 	PgStat_MsgConn msg_conn;
+	PgStat_MsgSubscriptionErr msg_subscriptionerr;
+	PgStat_MsgSubscriptionErrPurge msg_subscriptionerrpurge;
+	PgStat_MsgSubscriptionPurge msg_subscriptionpurge;
 } PgStat_Msg;
 
 
@@ -938,6 +1007,38 @@ typedef struct PgStat_StatReplSlotEntry
 	TimestampTz stat_reset_timestamp;
 } PgStat_StatReplSlotEntry;
 
+/*
+ * Subscription statistics kept in the stats collector.
+ */
+typedef struct PgStat_StatSubEntry
+{
+	Oid			subid;			/* hash table key */
+	HTAB	   *suberrors;
+} PgStat_StatSubEntry;
+
+/*
+ * Subscription error statistics kept in the stats collector.  One entry represents
+ * an error that happened during logical replication, reported by the apply worker
+ * (subrelid is InvalidOid) or by the table sync worker (subrelid is a valid OID).
+ * The error reported by the table sync worker is removed also when the table
+ * synchronization process completed.
+ */
+
+typedef struct PgStat_StatSubErrEntry
+{
+	Oid			subrelid;		/* InvalidOid if the apply worker, otherwise
+								 * the table sync worker. hash table key. */
+	Oid			databaseid;
+	Oid			relid;			/* OID of relation related to the error. Must
+								 * be the same as subrelid in the table sync
+								 * case. */
+	LogicalRepMsgType command;
+	TransactionId xid;
+	PgStat_Counter failure_count;
+	TimestampTz last_failure;
+	char		last_errmsg[PGSTAT_SUBSCRIPTIONERR_MSGLEN];
+	TimestampTz stat_reset_timestamp;
+} PgStat_StatSubErrEntry;
 
 /*
  * Working state needed to accumulate per-function-call timing statistics.
@@ -1031,6 +1132,7 @@ extern void pgstat_reset_shared_counters(const char *);
 extern void pgstat_reset_single_counter(Oid objectid, PgStat_Single_Reset_Type type);
 extern void pgstat_reset_slru_counter(const char *);
 extern void pgstat_reset_replslot_counter(const char *name);
+extern void pgstat_reset_subscription_error(Oid subid, Oid subrelid);
 
 extern void pgstat_report_autovac(Oid dboid);
 extern void pgstat_report_vacuum(Oid tableoid, bool shared,
@@ -1047,6 +1149,9 @@ extern void pgstat_report_checksum_failure(void);
 extern void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat);
 extern void pgstat_report_replslot_create(const char *slotname);
 extern void pgstat_report_replslot_drop(const char *slotname);
+extern void pgstat_report_subscription_error(Oid subid, Oid subrelid, Oid relid,
+											 LogicalRepMsgType command,
+											 TransactionId xid, const char *errmsg);
 
 extern void pgstat_initialize(void);
 
@@ -1145,6 +1250,7 @@ extern PgStat_GlobalStats *pgstat_fetch_global(void);
 extern PgStat_WalStats *pgstat_fetch_stat_wal(void);
 extern PgStat_SLRUStats *pgstat_fetch_slru(void);
 extern PgStat_StatReplSlotEntry *pgstat_fetch_replslot(NameData slotname);
+extern PgStat_StatSubErrEntry *pgstat_fetch_subscription_error(Oid subid, Oid relid);
 
 extern void pgstat_count_slru_page_zeroed(int slru_idx);
 extern void pgstat_count_slru_page_hit(int slru_idx);
diff --git a/src/include/utils/elog.h b/src/include/utils/elog.h
index f53607e12e..155145a77d 100644
--- a/src/include/utils/elog.h
+++ b/src/include/utils/elog.h
@@ -222,6 +222,7 @@ extern int	err_generic_string(int field, const char *str);
 extern int	geterrcode(void);
 extern int	geterrposition(void);
 extern int	getinternalerrposition(void);
+extern const char *geterrmessage(void);
 
 
 /*----------
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index e5ab11275d..ffad9790ae 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2094,6 +2094,28 @@ pg_stat_subscription| SELECT su.oid AS subid,
     st.latest_end_time
    FROM (pg_subscription su
      LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid = su.oid)));
+pg_stat_subscription_errors| SELECT d.datname,
+    sr.subid,
+    s.subname,
+    e.relid,
+    e.command,
+    e.xid,
+    e.failure_source,
+    e.failure_count,
+    e.last_failure,
+    e.last_failure_message,
+    e.stats_reset
+   FROM ( SELECT pg_subscription.oid AS subid,
+            NULL::oid AS relid
+           FROM pg_subscription
+        UNION ALL
+         SELECT pg_subscription_rel.srsubid AS subid,
+            pg_subscription_rel.srrelid AS relid
+           FROM pg_subscription_rel
+          WHERE (pg_subscription_rel.srsubstate <> 'r'::"char")) sr,
+    ((LATERAL pg_stat_get_subscription_error(sr.subid, sr.relid) e(datid, subid, relid, command, xid, failure_source, failure_count, last_failure, last_failure_message, stats_reset)
+     JOIN pg_database d ON ((e.datid = d.oid)))
+     JOIN pg_subscription s ON ((e.subid = s.oid)));
 pg_stat_sys_indexes| SELECT pg_stat_all_indexes.relid,
     pg_stat_all_indexes.indexrelid,
     pg_stat_all_indexes.schemaname,
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 2dea7b1ac7..05f5ab7a0b 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1938,6 +1938,9 @@ PgStat_MsgResetsharedcounter
 PgStat_MsgResetsinglecounter
 PgStat_MsgResetslrucounter
 PgStat_MsgSLRU
+PgStat_MsgSubscriptionErr
+PgStat_MsgSubscriptionErrPurge
+PgStat_MsgSubscriptionPurge
 PgStat_MsgTabpurge
 PgStat_MsgTabstat
 PgStat_MsgTempFile
@@ -1949,6 +1952,8 @@ PgStat_Single_Reset_Type
 PgStat_StatDBEntry
 PgStat_StatFuncEntry
 PgStat_StatReplSlotEntry
+PgStat_StatSubEntry
+PgStat_StatSubErrEntry
 PgStat_StatTabEntry
 PgStat_SubXactStatus
 PgStat_TableCounts
-- 
2.24.3 (Apple Git-128)

