v4-0002-Add-pg_stat_logical_replication_error-statistics-.patch

From 84752e871e0c46bbec1b02cdf87dfe092f67dfc2 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 v4 2/3] Add pg_stat_logical_replication_error 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.

It also adds SQL function pg_stat_reset_subscription_errror() to
reset the single subscription error.
---
 doc/src/sgml/monitoring.sgml                | 151 +++++
 src/backend/catalog/pg_subscription.c       |  23 +-
 src/backend/catalog/system_functions.sql    |   2 +
 src/backend/catalog/system_views.sql        |  27 +
 src/backend/commands/subscriptioncmds.c     |  30 +-
 src/backend/postmaster/pgstat.c             | 639 ++++++++++++++++++++
 src/backend/replication/logical/tablesync.c |  16 +-
 src/backend/replication/logical/worker.c    |  48 +-
 src/backend/utils/adt/pgstatfuncs.c         | 120 ++++
 src/backend/utils/error/elog.c              |  16 +
 src/include/catalog/pg_proc.dat             |  13 +
 src/include/catalog/pg_subscription_rel.h   |   3 +-
 src/include/pgstat.h                        | 112 ++++
 src/include/utils/elog.h                    |   1 +
 src/test/regress/expected/rules.out         |  22 +
 15 files changed, 1192 insertions(+), 31 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 74a58a916c..ca9eec5e22 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 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,126 @@ 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>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 +5301,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/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 25021e25a4..d76bdff36a 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -541,18 +541,27 @@ GetSubscriptionRelations(Oid subid)
 /*
  * Get all relations for subscription that are not in a ready state.
  *
- * Returned list is palloc'ed in current memory context.
+ * Returned HTAB is created in current memory context.
  */
-List *
+HTAB *
 GetSubscriptionNotReadyRelations(Oid subid)
 {
-	List	   *res = NIL;
+	HTAB	   *htab;
+	HASHCTL		hash_ctl;
 	Relation	rel;
 	HeapTuple	tup;
 	int			nkeys = 0;
 	ScanKeyData skey[2];
 	SysScanDesc scan;
 
+	hash_ctl.keysize = sizeof(Oid);
+	hash_ctl.entrysize = sizeof(SubscriptionRelState);
+	hash_ctl.hcxt = CurrentMemoryContext;
+	htab = hash_create("not ready relations in subscription",
+					   64,
+					   &hash_ctl,
+					   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+
 	rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
 	ScanKeyInit(&skey[nkeys++],
@@ -577,8 +586,8 @@ GetSubscriptionNotReadyRelations(Oid subid)
 
 		subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
 
-		relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
-		relstate->relid = subrel->srrelid;
+		relstate = (SubscriptionRelState *) hash_search(htab, (void *) &subrel->srrelid,
+														HASH_ENTER, NULL);
 		relstate->state = subrel->srsubstate;
 		d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
 							Anum_pg_subscription_rel_srsublsn, &isnull);
@@ -586,13 +595,11 @@ GetSubscriptionNotReadyRelations(Oid subid)
 			relstate->lsn = InvalidXLogRecPtr;
 		else
 			relstate->lsn = DatumGetLSN(d);
-
-		res = lappend(res, relstate);
 	}
 
 	/* Cleanup */
 	systable_endscan(scan);
 	table_close(rel, AccessShareLock);
 
-	return res;
+	return htab;
 }
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..cd07f2e02f 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/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 22ae982328..da02d3bbfa 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -86,7 +86,7 @@ typedef struct SubOpts
 static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
 static void check_duplicates_in_publist(List *publist, Datum *datums);
 static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
-static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
+static void ReportSlotConnectionError(HTAB *rstates, Oid subid, char *slotname, char *err);
 
 
 /*
@@ -1163,7 +1163,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	char	   *err = NULL;
 	WalReceiverConn *wrconn;
 	Form_pg_subscription form;
-	List	   *rstates;
+	HTAB	   *rstates;
+	HASH_SEQ_STATUS hstat;
+	SubscriptionRelState *rstate;
 
 	/*
 	 * Lock pg_subscription with AccessExclusiveLock to ensure that the
@@ -1286,9 +1288,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	 * exclusive lock on the subscription.
 	 */
 	rstates = GetSubscriptionNotReadyRelations(subid);
-	foreach(lc, rstates)
+	hash_seq_init(&hstat, rstates);
+	while ((rstate = (SubscriptionRelState *) hash_seq_search(&hstat)) != NULL)
 	{
-		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 		Oid			relid = rstate->relid;
 
 		/* Only cleanup resources of tablesync workers */
@@ -1321,8 +1323,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	 * If there is no slot associated with the subscription, we can finish
 	 * here.
 	 */
-	if (!slotname && rstates == NIL)
+	if (!slotname && hash_get_num_entries(rstates) == 0)
 	{
+		hash_destroy(rstates);
 		table_close(rel, NoLock);
 		return;
 	}
@@ -1346,7 +1349,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 		if (!slotname)
 		{
 			/* be tidy */
-			list_free(rstates);
+			hash_destroy(rstates);
 			table_close(rel, NoLock);
 			return;
 		}
@@ -1358,9 +1361,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 
 	PG_TRY();
 	{
-		foreach(lc, rstates)
+		hash_seq_init(&hstat, rstates);
+		while ((rstate = (SubscriptionRelState *) hash_seq_search(&hstat)) != NULL)
 		{
-			SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 			Oid			relid = rstate->relid;
 
 			/* Only cleanup resources of tablesync workers */
@@ -1389,7 +1392,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 			}
 		}
 
-		list_free(rstates);
+		hash_destroy(rstates);
 
 		/*
 		 * If there is a slot associated with the subscription, then drop the
@@ -1641,13 +1644,14 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
  * them manually, if required.
  */
 static void
-ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
+ReportSlotConnectionError(HTAB *rstates, Oid subid, char *slotname, char *err)
 {
-	ListCell   *lc;
+	HASH_SEQ_STATUS hstat;
+	SubscriptionRelState *rstate;
 
-	foreach(lc, rstates)
+	hash_seq_init(&hstat, rstates);
+	while ((rstate = (SubscriptionRelState *) hash_seq_search(&hstat)) != NULL)
 	{
-		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 		Oid			relid = rstate->relid;
 
 		/* Only cleanup resources of tablesync workers */
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 11702f2a80..4a35e6640f 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_ERR_HASH_SIZE	32
 
 
 /* ----------
@@ -279,6 +282,7 @@ static PgStat_GlobalStats globalStats;
 static PgStat_WalStats walStats;
 static PgStat_SLRUStats slruStats[SLRU_NUM_ELEMENTS];
 static HTAB *replSlotStatHash = NULL;
+static HTAB *subscriptionErrHash = NULL;
 
 /*
  * List of OIDs of databases we need to write out.  If an entry is InvalidOid,
@@ -320,6 +324,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_StatSubErrEntry *pgstat_get_subscription_error_entry(Oid subid,
+																   bool create);
+static PgStat_StatSubRelErrEntry *pgstat_get_subscription_rel_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);
@@ -358,6 +368,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
@@ -1134,6 +1148,133 @@ 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 (subscriptionErrHash)
+	{
+		PgStat_MsgSubscriptionPurge submsg;
+		PgStat_StatSubErrEntry *suberrent;
+		HTAB	   *htab;
+
+		htab = pgstat_collect_oids(SubscriptionRelationId, Anum_pg_subscription_oid);
+
+		submsg.m_nentries = 0;
+
+		hash_seq_init(&hstat, subscriptionErrHash);
+		while ((suberrent = (PgStat_StatSubErrEntry *) hash_seq_search(&hstat)) != NULL)
+		{
+			PgStat_MsgSubscriptionErrPurge errmsg;
+			PgStat_StatSubRelErrEntry *relerrent;
+			HASH_SEQ_STATUS hstat_rel;
+			HTAB	   *rstates;
+
+			CHECK_FOR_INTERRUPTS();
+
+			if (hash_search(htab, (void *) &(suberrent->subid), HASH_FIND, NULL) == NULL)
+			{
+				/* This subscription is dead, add subid to the message */
+				submsg.m_subids[submsg.m_nentries++] = suberrent->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 (suberrent->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.
+			 */
+			rstates = GetSubscriptionNotReadyRelations(suberrent->subid);
+
+			errmsg.m_nentries = 0;
+			errmsg.m_subid = suberrent->subid;
+
+			hash_seq_init(&hstat_rel, suberrent->suberrors);
+			while ((relerrent = (PgStat_StatSubRelErrEntry *) hash_seq_search(&hstat_rel)) != NULL)
+			{
+				CHECK_FOR_INTERRUPTS();
+
+				/* Skip the apply worker's error */
+				if (!OidIsValid(relerrent->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(rstates, (void *) &(relerrent->subrelid), HASH_FIND, NULL) == NULL)
+					errmsg.m_relids[errmsg.m_nentries++] = relerrent->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(rstates);
+		}
+
+		/* 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.
 	 */
@@ -1543,6 +1684,25 @@ pgstat_reset_replslot_counter(const char *name)
 	pgstat_send(&msg, sizeof(msg));
 }
 
+/* ----------
+ * pgstat_reset_subscription_error() -
+ *
+ *	Tell the collector about reset the error of subscription.
+ * ----------
+ */
+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() -
  *
@@ -1863,6 +2023,37 @@ pgstat_report_replslot_drop(const char *slotname)
 	pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
 }
 
+/* ----------
+ * pgstat_report_subscription_error() -
+ *
+ *	Tell the collector about error of subscription.
+ * ----------
+ */
+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_last_failure = GetCurrentTimestamp();
+	strlcpy(msg.m_errmsg, errmsg, PGSTAT_SUBSCRIPTIONERR_MSGLEN);
+
+	pgstat_send(&msg, len);
+}
+
 /* ----------
  * pgstat_ping() -
  *
@@ -2895,6 +3086,38 @@ 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 errors struct.
+ * ---------
+ */
+PgStat_StatSubErrEntry *
+pgstat_fetch_subscription_error(Oid subid)
+{
+	backend_read_statsfile();
+
+	return pgstat_get_subscription_error_entry(subid, false);
+}
+
+/*
+ * ---------
+ * pgstat_fetch_subscription_rel_error() -
+ *
+ *	Support function for the SQL-callable pgstat* functions. Returns
+ *	a pointer to the subscription error struct.
+ * ---------
+ */
+PgStat_StatSubRelErrEntry *
+pgstat_fetch_subscription_rel_error(Oid subid, Oid relid)
+{
+	backend_read_statsfile();
+
+	return pgstat_get_subscription_rel_error_entry(subid, relid, false);
+}
+
 /*
  * Shut down a single backend's statistics reporting at process exit.
  *
@@ -3424,6 +3647,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;
 			}
@@ -3725,6 +3961,50 @@ pgstat_write_statsfiles(bool permanent, bool allDbs)
 		}
 	}
 
+	/*
+	 * Write subscription error structs
+	 */
+	if (subscriptionErrHash)
+	{
+		PgStat_StatSubErrEntry *suberrent;
+
+		hash_seq_init(&hstat, subscriptionErrHash);
+		while ((suberrent = (PgStat_StatSubErrEntry *) hash_seq_search(&hstat)) != NULL)
+		{
+			PgStat_StatSubRelErrEntry *relerrent;
+			HASH_SEQ_STATUS relhstat;
+			long		nrels = hash_get_num_entries(suberrent->suberrors);
+
+			/* Skip this subscription if not have any errors */
+			if (suberrent->suberrors == NULL)
+				continue;
+
+			fputc('S', fpout);
+			rc = fwrite(suberrent, sizeof(PgStat_StatSubErrEntry), 1, fpout);
+			(void) rc;			/* we'll check for error with ferror */
+
+			/* The number of errors follows */
+			rc = fwrite(&nrels, sizeof(long), 1, fpout);
+			(void) rc;			/* we'll check for error with ferror */
+
+			/* Then, the error entries follow */
+			hash_seq_init(&relhstat, suberrent->suberrors);
+			while ((relerrent = (PgStat_StatSubRelErrEntry *) hash_seq_search(&relhstat)) != NULL)
+			{
+				/*
+				 * XXX we write the whole PgStat_StatSubRelErrEntry entry that
+				 * contains the fixed-legnth error message string which is
+				 * PGSTAT_SUBSCRIPTIONERR_MSGLEN in length, making the stats
+				 * file bloat.  We don't expect we have many error entries but
+				 * if the expectation became false we should write the string
+				 * and its length instead.
+				 */
+				rc = fwrite(relerrent, sizeof(PgStat_StatSubRelErrEntry), 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
@@ -4184,6 +4464,99 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep)
 					break;
 				}
 
+				/*
+				 * 'S'	A PgStat_StatSubErrEntry struct followed by the number
+				 * of errors and PgStat_StatSubRelErrEntry structs, describing
+				 * subscription errors.
+				 */
+			case 'S':
+				{
+					PgStat_StatSubErrEntry suberrbuf;
+					PgStat_StatSubErrEntry *suberrent;
+					long		nerrors;
+
+					/* Read the subscription entry */
+					if (fread(&suberrbuf, 1, sizeof(PgStat_StatSubErrEntry), fpin) !=
+						sizeof(PgStat_StatSubErrEntry))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						goto done;
+					}
+
+					/* Create hash table if we don't have it already. */
+					if (subscriptionErrHash == NULL)
+					{
+						HASHCTL		hash_ctl;
+
+						hash_ctl.keysize = sizeof(Oid);
+						hash_ctl.entrysize = sizeof(PgStat_StatSubErrEntry);
+						hash_ctl.hcxt = pgStatLocalContext;
+						subscriptionErrHash = hash_create("Subscription error hash",
+														  PGSTAT_SUBSCRIPTION_ERR_HASH_SIZE,
+														  &hash_ctl,
+														  HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+					}
+
+					/* Enter the subscription error entry */
+					suberrent =
+						(PgStat_StatSubErrEntry *) hash_search(subscriptionErrHash,
+															   (void *) &(suberrbuf.subid),
+															   HASH_ENTER, NULL);
+					suberrent->suberrors = NULL;
+
+					/* Read the number of errors in the subscription */
+					if (fread(&nerrors, 1, sizeof(long), fpin) != sizeof(long))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						goto done;
+					}
+
+					for (int i = 0; i < nerrors; i++)
+					{
+						PgStat_StatSubRelErrEntry *subrelent;
+						PgStat_StatSubRelErrEntry subrelbuf;
+
+						if (fread(&subrelbuf, 1, sizeof(PgStat_StatSubRelErrEntry), fpin) !=
+							sizeof(PgStat_StatSubRelErrEntry))
+						{
+							ereport(pgStatRunningInCollector ? LOG : WARNING,
+									(errmsg("corrupted statistics file \"%s\"",
+											statfile)));
+							goto done;
+						}
+
+						if (suberrent->suberrors == NULL)
+						{
+							HASHCTL		hash_ctl;
+
+							hash_ctl.keysize = sizeof(Oid);
+							hash_ctl.entrysize = sizeof(PgStat_StatSubRelErrEntry);
+							hash_ctl.hcxt = pgStatLocalContext;
+							suberrent->suberrors = hash_create("Subscription relation error hash",
+															   PGSTAT_SUBSCRIPTION_ERR_HASH_SIZE,
+															   &hash_ctl,
+															   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+						}
+
+						/*
+						 * Enter the error information to the subscription
+						 * hash
+						 */
+						subrelent =
+							(PgStat_StatSubRelErrEntry *) hash_search(suberrent->suberrors,
+																	  (void *) &(subrelbuf.subrelid),
+																	  HASH_ENTER, NULL);
+
+						memcpy(subrelent, &subrelbuf, sizeof(PgStat_StatSubRelErrEntry));
+					}
+
+					break;
+				}
+
 			case 'E':
 				goto done;
 
@@ -4526,6 +4899,50 @@ pgstat_read_db_statsfile_timestamp(Oid databaseid, bool permanent,
 				}
 				break;
 
+				/*
+				 * 'S'	A PgStat_StatSubErrEntry struct followed by the number
+				 * of errors and PgStat_StatSubRelErrEntry structs describing
+				 * a subscription errors.
+				 */
+			case 'S':
+				{
+					PgStat_StatSubErrEntry mySubErrs;
+					PgStat_StatSubRelErrEntry subrelbuf;
+					long		nrels;
+
+					if (fread(&mySubErrs, 1, sizeof(PgStat_StatSubErrEntry), fpin)
+						!= sizeof(PgStat_StatSubErrEntry))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						FreeFile(fpin);
+						return false;
+					}
+
+					if (fread(&nrels, 1, sizeof(long), fpin) != sizeof(long))
+					{
+						ereport(pgStatRunningInCollector ? LOG : WARNING,
+								(errmsg("corrupted statistics file \"%s\"",
+										statfile)));
+						goto done;
+					}
+
+					for (int i = 0; i < nrels; i++)
+					{
+						if (fread(&subrelbuf, 1, sizeof(PgStat_StatSubRelErrEntry), fpin) !=
+							sizeof(PgStat_StatSubRelErrEntry))
+						{
+							ereport(pgStatRunningInCollector ? LOG : WARNING,
+									(errmsg("corrupted statistics file \"%s\"",
+											statfile)));
+							goto done;
+						}
+					}
+				}
+
+				break;
+
 			case 'E':
 				goto done;
 
@@ -4716,6 +5133,7 @@ pgstat_clear_snapshot(void)
 	pgStatLocalContext = NULL;
 	pgStatDBHash = NULL;
 	replSlotStatHash = NULL;
+	subscriptionErrHash = NULL;
 
 	/*
 	 * Historically the backend_status.c facilities lived in this file, and
@@ -5650,6 +6068,122 @@ 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_StatSubRelErrEntry *relerrent;
+	bool create = !msg->m_reset;
+
+	/* Get subscription error */
+	relerrent = pgstat_get_subscription_rel_error_entry(msg->m_subid,
+														msg->m_subrelid,
+														create);
+
+	if (msg->m_reset)
+	{
+		if (relerrent)
+			return;
+
+		/* reset fields and set reset timestamp */
+		relerrent->relid = InvalidOid;
+		relerrent->command = 0;
+		relerrent->xid = InvalidTransactionId;
+		relerrent->failure_count = 0;
+		relerrent->last_failure = 0;
+		relerrent->errmsg[0] = '\0';
+		relerrent->stat_reset_timestamp = GetCurrentTimestamp();
+	}
+	else
+	{
+		Assert(relerrent);
+
+		/* update the error entry */
+		relerrent->databaseid = msg->m_databaseid;
+		relerrent->relid = msg->m_relid;
+		relerrent->command = msg->m_command;
+		relerrent->xid = msg->m_xid;
+		relerrent->failure_count++;
+		relerrent->last_failure = msg->m_last_failure;
+		strlcpy(relerrent->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 (subscriptionErrHash == NULL)
+		return;
+
+	for (int i = 0; i < msg->m_nentries; i++)
+	{
+		PgStat_StatSubErrEntry *suberrent;
+
+		suberrent = hash_search(subscriptionErrHash, (void *) &(msg->m_subids[i]),
+								HASH_FIND, NULL);
+
+		/*
+		 * 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 (suberrent == NULL)
+			continue;
+
+		/* Cleanup the hash table for errors */
+		if (suberrent->suberrors != NULL)
+			hash_destroy(suberrent->suberrors);
+
+		(void) hash_search(subscriptionErrHash, (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)
+{
+	if (subscriptionErrHash == NULL)
+		return;
+
+	for (int i = 0; i < msg->m_nentries; i++)
+	{
+		PgStat_StatSubErrEntry *suberrent;
+
+		suberrent = hash_search(subscriptionErrHash, (void *) &(msg->m_subid),
+								HASH_FIND, NULL);
+
+		/*
+		 * 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 (suberrent == NULL)
+			continue;
+
+		(void) hash_search(suberrent->suberrors, (void *) &(msg->m_relids[i]),
+						   HASH_REMOVE, NULL);
+	}
+}
+
 /* ----------
  * pgstat_write_statsfile_needed() -
  *
@@ -5747,6 +6281,111 @@ pgstat_get_replslot_entry(NameData name, bool create)
 	return slotent;
 }
 
+/* ----------
+ * pgstat_get_subscription_error_entry
+ *
+ * Return the entry of subscription errors entry 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_StatSubErrEntry *
+pgstat_get_subscription_error_entry(Oid subid, bool create)
+{
+	PgStat_StatSubErrEntry *suberrent;
+	HASHACTION	action = (create ? HASH_ENTER : HASH_FIND);
+	bool		found;
+
+	if (subscriptionErrHash == NULL)
+	{
+		HASHCTL		hash_ctl;
+
+		if (!create)
+			return NULL;
+
+		hash_ctl.keysize = sizeof(Oid);
+		hash_ctl.entrysize = sizeof(PgStat_StatSubErrEntry);
+		subscriptionErrHash = hash_create("Subscription error hash",
+										  PGSTAT_SUBSCRIPTION_ERR_HASH_SIZE,
+										  &hash_ctl,
+										  HASH_ELEM | HASH_BLOBS);
+	}
+
+	suberrent = (PgStat_StatSubErrEntry *) hash_search(subscriptionErrHash,
+													   (void *) &subid,
+													   action, &found);
+
+	/* initialize fields */
+	if (create && !found)
+		suberrent->suberrors = NULL;
+
+	return suberrent;
+}
+
+/* ----------
+ * pgstat_get_subscription_rel_error_entry
+ *
+ * Return the entry of subscription relation error entry with the subscription
+ * OID and relcation 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_StatSubRelErrEntry *
+pgstat_get_subscription_rel_error_entry(Oid subid, Oid subrelid, bool create)
+{
+	PgStat_StatSubErrEntry *suberrent;
+	PgStat_StatSubRelErrEntry *relerrent;
+	HASHACTION	action = (create ? HASH_ENTER : HASH_FIND);
+	bool		found;
+
+	suberrent = pgstat_get_subscription_error_entry(subid, create);
+
+	if (suberrent == NULL)
+	{
+		Assert(!create);
+		return NULL;
+	}
+
+	if (suberrent->suberrors == NULL)
+	{
+		HASHCTL		hash_ctl;
+
+		if (!create)
+			return NULL;
+
+		hash_ctl.keysize = sizeof(Oid);
+		hash_ctl.entrysize = sizeof(PgStat_StatSubRelErrEntry);
+		suberrent->suberrors = hash_create("Subscription relation error hash",
+										   PGSTAT_SUBSCRIPTION_ERR_HASH_SIZE,
+										   &hash_ctl,
+										   HASH_ELEM | HASH_BLOBS);
+	}
+
+	relerrent = (PgStat_StatSubRelErrEntry *) hash_search(suberrent->suberrors,
+														  (void *) &subrelid,
+														  action, &found);
+
+	/* initialize fields */
+	if (create && !found)
+	{
+		relerrent->databaseid = InvalidOid;
+		relerrent->relid = InvalidOid;
+		relerrent->command = 0;
+		relerrent->xid = InvalidTransactionId;
+		relerrent->failure_count = 0;
+		relerrent->last_failure = 0;
+		relerrent->errmsg[0] = '\0';
+		relerrent->stat_reset_timestamp = 0;
+	}
+
+	return relerrent;
+}
+
 /* ----------
  * pgstat_reset_replslot
  *
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a43c..8765396432 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1175,8 +1175,8 @@ FetchTableStates(bool *started_tx)
 	if (!table_states_valid)
 	{
 		MemoryContext oldctx;
-		List	   *rstates;
-		ListCell   *lc;
+		HTAB	   *rstates;
+		HASH_SEQ_STATUS hstat;
 		SubscriptionRelState *rstate;
 
 		/* Clean the old lists. */
@@ -1194,14 +1194,18 @@ FetchTableStates(bool *started_tx)
 
 		/* Allocate the tracking info in a permanent memory context. */
 		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
-		foreach(lc, rstates)
+		hash_seq_init(&hstat, rstates);
+		while ((rstate = (SubscriptionRelState *) hash_seq_search(&hstat)) != NULL)
 		{
-			rstate = palloc(sizeof(SubscriptionRelState));
-			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
-			table_states_not_ready = lappend(table_states_not_ready, rstate);
+			SubscriptionRelState *r = palloc(sizeof(SubscriptionRelState));
+
+			memcpy(r, rstate, sizeof(SubscriptionRelState));
+			table_states_not_ready = lappend(table_states_not_ready, r);
 		}
 		MemoryContextSwitchTo(oldctx);
 
+		hash_destroy(rstates);
+
 		/*
 		 * Does the subscription have tables?
 		 *
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index d346377b20..4f9c4e9014 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -227,6 +227,7 @@ typedef struct ApplyErrCallbackArg
 	LogicalRepMsgType command;	/* 0 if invalid */
 
 	/* Local relation information */
+	Oid			relid;			/* used for error reporting */
 	char	   *nspname;		/* used for error context */
 	char	   *relname;		/* used for error context */
 
@@ -236,6 +237,7 @@ typedef struct ApplyErrCallbackArg
 static ApplyErrCallbackArg apply_error_callback_arg =
 {
 	.command = 0,
+	.relid = InvalidOid,
 	.relname = NULL,
 	.nspname = NULL,
 	.remote_xid = InvalidTransactionId,
@@ -3522,8 +3524,26 @@ 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();
+		{
+			elog(NOTICE, "errmsg \"%s\"",
+				 geterrmessage());
+
+			/* 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);
@@ -3641,7 +3661,24 @@ 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.relid,
+											 apply_error_callback_arg.command,
+											 apply_error_callback_arg.remote_xid,
+											 geterrmessage());
+
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
 
 	proc_exit(0);
 }
@@ -3686,6 +3723,7 @@ apply_error_callback(void *arg)
 static void
 set_apply_error_context_rel(LogicalRepRelMapEntry *rel)
 {
+	apply_error_callback_arg.relid = rel->localreloid;
 	apply_error_callback_arg.nspname = rel->remoterel.nspname;
 	apply_error_callback_arg.relname = rel->remoterel.relname;
 }
@@ -3694,6 +3732,7 @@ set_apply_error_context_rel(LogicalRepRelMapEntry *rel)
 static void
 reset_apply_error_context_rel(void)
 {
+	apply_error_callback_arg.relid = InvalidOid;
 	apply_error_callback_arg.nspname = NULL;
 	apply_error_callback_arg.relname = NULL;
 }
@@ -3724,6 +3763,7 @@ set_logicalrep_error_context_rel(Relation rel)
 {
 	if (IsLogicalWorker())
 	{
+		apply_error_callback_arg.relid = RelationGetRelid(rel);
 		apply_error_callback_arg.nspname =
 			get_namespace_name(RelationGetNamespace(rel));
 		apply_error_callback_arg.relname =
@@ -3737,6 +3777,8 @@ reset_logicalrep_error_context_rel(void)
 {
 	if (IsLogicalWorker())
 	{
+		apply_error_callback_arg.relid = InvalidOid;
+
 		if (apply_error_callback_arg.nspname)
 			pfree(apply_error_callback_arg.nspname);
 		apply_error_callback_arg.nspname = NULL;
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index f0e09eae4d..f1348a415e 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -24,6 +24,8 @@
 #include "pgstat.h"
 #include "postmaster/bgworker_internals.h"
 #include "postmaster/postmaster.h"
+#include "replication/logicalproto.h"
+#include "replication/logicalworker.h"
 #include "replication/slot.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
@@ -2240,6 +2242,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)
 {
@@ -2380,3 +2399,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_StatSubRelErrEntry *relerrent;
+
+	/* 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 */
+	relerrent = pgstat_fetch_subscription_rel_error(subid, relid);
+
+	/* Return NULL if the subscription doesn't have any errors */
+	if (relerrent == NULL)
+		PG_RETURN_NULL();
+
+	values[0] = ObjectIdGetDatum(relerrent->databaseid);
+	values[1] = ObjectIdGetDatum(subid);
+
+	if (OidIsValid(relerrent->relid))
+		values[2] = ObjectIdGetDatum(relerrent->relid);
+	else
+		nulls[2] = true;
+
+	if (relerrent->command == 0)
+		nulls[3] = true;
+	else
+	{
+		/* should be apply worker */
+		Assert(!OidIsValid(relerrent->subrelid));
+
+		values[3] = CStringGetTextDatum(logicalrep_message_type(relerrent->command));
+	}
+
+	if (TransactionIdIsValid(relerrent->xid))
+		values[4] = TransactionIdGetDatum(relerrent->xid);
+	else
+		nulls[4] = true;
+
+	if (OidIsValid(relerrent->subrelid))
+		values[5] = CStringGetTextDatum("tablesync");
+	else
+		values[5] = CStringGetTextDatum("apply");
+
+
+	values[6] = Int64GetDatum(relerrent->failure_count);
+
+	if (relerrent->last_failure == 0)
+		nulls[7] = true;
+	else
+		values[7] = TimestampTzGetDatum(relerrent->last_failure);
+
+	values[8] = CStringGetTextDatum(relerrent->errmsg);
+
+	if (relerrent->stat_reset_timestamp == 0)
+		nulls[9] = true;
+	else
+		values[9] = TimestampTzGetDatum(relerrent->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 8cd0252082..044ff52227 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5321,6 +5321,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',
@@ -5708,6 +5716,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/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 632381b4e3..50053cdafc 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -22,6 +22,7 @@
 #include "catalog/genbki.h"
 #include "catalog/pg_subscription_rel_d.h"
 #include "nodes/pg_list.h"
+#include "utils/hsearch.h"
 
 /* ----------------
  *		pg_subscription_rel definition. cpp turns this into
@@ -89,6 +90,6 @@ extern void RemoveSubscriptionRel(Oid subid, Oid relid);
 
 extern bool HasSubscriptionRelations(Oid subid);
 extern List *GetSubscriptionRelations(Oid subid);
-extern List *GetSubscriptionNotReadyRelations(Oid subid);
+extern HTAB *GetSubscriptionNotReadyRelations(Oid subid);
 
 #endif							/* PG_SUBSCRIPTION_REL_H */
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 9612c0a6c2..1104886bef 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,
@@ -539,6 +543,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/clear 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 messages use below field */
+	bool		m_reset;		/* clear 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_last_failure;
+	char		m_errmsg[PGSTAT_SUBSCRIPTIONERR_MSGLEN];
+} PgStat_MsgSubscriptionErr;
+
+/* ----------
+ * 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_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_MsgRecoveryConflict	Sent by the backend upon recovery conflict
@@ -710,6 +776,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;
 
 
@@ -908,6 +977,42 @@ typedef struct PgStat_StatReplSlotEntry
 	TimestampTz stat_reset_timestamp;
 } PgStat_StatReplSlotEntry;
 
+/*
+ * Subscription error statistics kept in the stats collector.
+ *
+ * PgStat_StatSubErrEntry holds all errors associated with the subscription,
+ * reported by the apply worker and the table sync workers. This entry is
+ * created when the first error message with the subscription is reported
+ * and is dropped along with its errors when the subscription is dropped.
+ *
+ * PgStat_StatSubRelErrEntry represents a error 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 apply
+ * worker is dropped when the subscription is dropped whereas the error reported
+ * by the table sync worker is dropped when the table synchronization process
+ * completed.
+ */
+typedef struct PgStat_StatSubErrEntry
+{
+	Oid			subid;			/* hash table key */
+	HTAB	   *suberrors;
+} PgStat_StatSubErrEntry;
+
+typedef struct PgStat_StatSubRelErrEntry
+{
+	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		errmsg[PGSTAT_SUBSCRIPTIONERR_MSGLEN];
+	TimestampTz stat_reset_timestamp;
+} PgStat_StatSubRelErrEntry;
 
 /*
  * Working state needed to accumulate per-function-call timing statistics.
@@ -995,6 +1100,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,
@@ -1011,6 +1117,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);
 
@@ -1106,6 +1215,9 @@ 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);
+extern PgStat_StatSubRelErrEntry *pgstat_fetch_subscription_rel_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,
-- 
2.24.3 (Apple Git-128)