From e513f9b2e1e8c5b05c1aafe08a2794445e280501 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 v2 2/3] Add pg_stat_logical_replication_error statistics view.

---
 doc/src/sgml/monitoring.sgml             | 126 +++++++
 src/backend/catalog/system_views.sql     |  15 +
 src/backend/postmaster/pgstat.c          | 451 +++++++++++++++++++++++
 src/backend/replication/logical/worker.c |  48 ++-
 src/backend/utils/adt/pgstatfuncs.c      | 106 ++++++
 src/backend/utils/error/elog.c           |  16 +
 src/include/catalog/pg_proc.dat          |   8 +
 src/include/pgstat.h                     |  73 ++++
 src/include/utils/elog.h                 |   1 +
 src/test/regress/expected/rules.out      |  12 +
 10 files changed, 853 insertions(+), 3 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 74a58a916c..06c4b0c8a5 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,123 @@ 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, and additional rows for errors reported by 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.
+      </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.
+      </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>
 
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 55f6e3711d..6031f063d2 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1257,3 +1257,18 @@ 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,
+        s.subname,
+        e.relid,
+        e.command,
+        e.xid,
+        e.failure_source,
+        e.failure_count,
+        e.last_failure,
+        e.last_failure_message
+    FROM pg_subscription as s,
+        LATERAL pg_stat_get_subscription_error(s.oid) as e
+	JOIN pg_database as d ON (e.datid = d.oid);
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 11702f2a80..fc79b724e8 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -41,6 +41,7 @@
 #include "catalog/partition.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_proc.h"
+#include "catalog/pg_subscription.h"
 #include "common/ip.h"
 #include "executor/instrument.h"
 #include "libpq/libpq.h"
@@ -106,6 +107,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 +281,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 +323,11 @@ 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(PgStat_StatSubErrEntry *suberrent,
+																		   Oid subrelid);
+
 static void pgstat_send_tabstat(PgStat_MsgTabstat *tsmsg);
 static void pgstat_send_funcstats(void);
 static void pgstat_send_slru(void);
@@ -358,6 +366,8 @@ 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_purge(PgStat_MsgSubscriptionPurge *msg, int len);
 
 /* ------------------------------------------------------------
  * Public functions called from postmaster follow
@@ -1134,6 +1144,52 @@ pgstat_vacuum_stat(void)
 		}
 	}
 
+	/*
+	 * Search for all the dead subscriptions in stats hashtable and tell the
+	 * stats collector to drop them.
+	 */
+	if (subscriptionErrHash)
+	{
+		PgStat_MsgSubscriptionPurge s_msg;
+		PgStat_StatSubErrEntry *suberrent;
+		HTAB	*htab;
+
+		htab = pgstat_collect_oids(SubscriptionRelationId, Anum_pg_subscription_oid);
+
+		s_msg.m_nentries = 0;
+		hash_seq_init(&hstat, subscriptionErrHash);
+		while ((suberrent = (PgStat_StatSubErrEntry *) hash_seq_search(&hstat)) != NULL)
+		{
+			CHECK_FOR_INTERRUPTS();
+
+			if (hash_search(htab, (void *) &(suberrent->subid), HASH_FIND, NULL) == NULL)
+				s_msg.m_subids[s_msg.m_nentries++] = suberrent->subid;
+
+			/* If the message is full, send it out and reinitialize to empty */
+			if (msg.m_nentries >= PGSTAT_NUM_SUBSCRIPTIONPURGE)
+			{
+				len = offsetof(PgStat_MsgSubscriptionPurge, m_subids[0])
+					+ s_msg.m_nentries * sizeof(Oid);
+
+				pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONPURGE);
+				pgstat_send(&msg, len);
+				s_msg.m_nentries = 0;
+			}
+		}
+
+		/* Send the rest */
+		if (s_msg.m_nentries > 0)
+		{
+			len = offsetof(PgStat_MsgSubscriptionPurge, m_subids[0])
+				+ s_msg.m_nentries * sizeof(Oid);
+
+			pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONPURGE);
+			pgstat_send(&msg, len);
+		}
+
+		hash_destroy(htab);
+	}
+
 	/*
 	 * Lookup our own database entry; if not found, nothing more to do.
 	 */
@@ -1863,6 +1919,56 @@ 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_clear = 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_clear_subscription_error() -
+ *
+ *	Tell the collector about clear the error of subscription.
+ * ----------
+ */
+void
+pgstat_clear_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_clear = true;
+
+	pgstat_send(&msg, offsetof(PgStat_MsgSubscriptionErr, m_clear) + sizeof(bool));
+}
+
 /* ----------
  * pgstat_ping() -
  *
@@ -2895,6 +3001,23 @@ 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 logical replication error struct.
+ * ---------
+ */
+PgStat_StatSubErrEntry *
+pgstat_fetch_subscription_error(Oid subid)
+{
+	backend_read_statsfile();
+
+	return pgstat_get_subscription_error_entry(subid, false);
+}
+
+
 /*
  * Shut down a single backend's statistics reporting at process exit.
  *
@@ -3424,6 +3547,14 @@ 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_SUBSCRIPTIONPURGE:
+					pgstat_recv_subscription_purge(&msg.msg_subscriptionpurge, len);
+					break;
+
 				default:
 					break;
 			}
@@ -3725,6 +3856,41 @@ 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 */
+
+			hash_seq_init(&relhstat, suberrent->suberrors);
+			while ((relerrent = (PgStat_StatSubRelErrEntry *) hash_seq_search(&relhstat)) != NULL)
+			{
+				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 +4350,96 @@ 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 a
+				 * subscription errors.
+				 */
+			case 'S':
+				{
+					PgStat_StatSubErrEntry suberrbuf;
+					PgStat_StatSubErrEntry *suberrent;
+					long nrels;
+
+					/* 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(&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++)
+					{
+						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 +4782,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 +5016,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 +5951,76 @@ 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 *suberrent;
+	PgStat_StatSubRelErrEntry *relerrent;
+
+	/* Get subscription errors */
+	suberrent = pgstat_get_subscription_error_entry(msg->m_subid, true);
+	Assert(suberrent);
+
+	/* Get the error entry of the relation */
+	relerrent = pgstat_get_subscription_rel_error_entry(suberrent,
+														msg->m_subrelid);
+	Assert(relerrent);
+
+	if (msg->m_clear)
+	{
+		/* reset all fields except for databaseid and failure_count */
+		relerrent->relid = InvalidOid;
+		relerrent->command = 0;
+		relerrent->xid = InvalidTransactionId;
+		relerrent->last_failure = 0;
+		relerrent->errmsg[0] = '\0';
+	}
+	else
+	{
+		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);
+
+		/* 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_write_statsfile_needed() -
  *
@@ -5747,6 +6118,86 @@ pgstat_get_replslot_entry(NameData name, bool create)
 	return slotent;
 }
 
+/* ----------
+ * pgstat_get_subscription_error_entry
+ *
+ * Return the entry of subscription error 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 slot 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;
+
+		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);
+
+	if (create && !found)
+		suberrent->suberrors = NULL;
+
+	return suberrent;
+}
+
+/* ----------
+ * pgstat_get_subscription_rel_error_entry
+ *
+ * Lookup and create the subscription error entry with 'relid' from 'suberrent'.
+ * ----------
+ */
+static PgStat_StatSubRelErrEntry *
+pgstat_get_subscription_rel_error_entry(PgStat_StatSubErrEntry *suberrent,
+										Oid subrelid)
+{
+	PgStat_StatSubRelErrEntry *relerrent;
+	bool		found;
+
+	if (suberrent->suberrors == NULL)
+	{
+		HASHCTL		hash_ctl;
+
+		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,
+														  HASH_ENTER, &found);
+
+	/* initialize fields */
+	if (!found)
+	{
+		relerrent->command = 0;
+		relerrent->xid = InvalidTransactionId;
+		relerrent->failure_count = 0;
+		relerrent->last_failure = 0;
+		relerrent->errmsg[0] = '\0';
+	}
+
+	return relerrent;
+}
+
 /* ----------
  * pgstat_reset_replslot
  *
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c23713468c..7c2ec983bb 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,
@@ -3524,8 +3526,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);
@@ -3643,7 +3663,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);
 }
@@ -3688,6 +3725,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;
 }
@@ -3696,6 +3734,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;
 }
@@ -3725,6 +3764,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 =
@@ -3738,6 +3778,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..b155c30fcf 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"
@@ -2380,3 +2382,107 @@ 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 logical replication error for the given subscription.
+ */
+Datum
+pg_stat_get_subscription_errors(PG_FUNCTION_ARGS)
+{
+#define PG_STAT_GET_SUBSCRIPTION_ERROR_COLS 9
+	Oid			subid = PG_GETARG_OID(0);
+	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	TupleDesc	tupdesc;
+	Tuplestorestate *tupstore;
+	MemoryContext per_query_ctx;
+	MemoryContext oldcontext;
+	PgStat_StatSubErrEntry *suberrent;
+	PgStat_StatSubRelErrEntry *relerrent;
+	HASH_SEQ_STATUS hstat;
+
+	/* check to see if caller supports us returning a tuplestore */
+	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("set-valued function called in context that cannot accept a set")));
+	if (!(rsinfo->allowedModes & SFRM_Materialize))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("materialize mode required, but it is not allowed in this context")));
+
+	/* Build a tuple descriptor for our result type */
+	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+		elog(ERROR, "return type must be a row type");
+
+	per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
+	oldcontext = MemoryContextSwitchTo(per_query_ctx);
+
+	tupstore = tuplestore_begin_heap(true, false, work_mem);
+	rsinfo->returnMode = SFRM_Materialize;
+	rsinfo->setResult = tupstore;
+	rsinfo->setDesc = tupdesc;
+
+	MemoryContextSwitchTo(oldcontext);
+
+	/* Get subscription errors */
+	suberrent = pgstat_fetch_subscription_error(subid);
+
+	/* Return NULL if the subscription doesn't have any errors */
+	if (suberrent == NULL || suberrent->suberrors == NULL)
+		PG_RETURN_NULL();
+
+	hash_seq_init(&hstat, suberrent->suberrors);
+	while ((relerrent = (PgStat_StatSubRelErrEntry *) hash_seq_search(&hstat)) != NULL)
+	{
+		Datum	values[PG_STAT_GET_SUBSCRIPTION_ERROR_COLS];
+		bool	nulls[PG_STAT_GET_SUBSCRIPTION_ERROR_COLS];
+
+		MemSet(nulls, 0, sizeof(nulls));
+
+		/* databaseid */
+		values[0] = ObjectIdGetDatum(relerrent->databaseid);
+
+		/* subid */
+		values[1] = ObjectIdGetDatum(subid);
+
+		/* relid */
+		if (OidIsValid(relerrent->relid))
+			values[2] = ObjectIdGetDatum(relerrent->relid);
+		else
+			nulls[2] = true;
+
+		/* command */
+		if (OidIsValid(relerrent->subrelid))
+			nulls[3] = true;
+		else
+			values[3] = CStringGetTextDatum(logicalrep_message_type(relerrent->command));
+
+		/* xid */
+		if (TransactionIdIsValid(relerrent->xid))
+			values[4] = TransactionIdGetDatum(relerrent->xid);
+		else
+			nulls[4] = true;
+
+		/* failure_source */
+		if (OidIsValid(relerrent->subrelid))
+			values[5] = CStringGetTextDatum("tablesync");
+		else
+			values[5] = CStringGetTextDatum("apply");
+
+		/* failure_count */
+		values[6] = Int64GetDatum(relerrent->failure_count);
+
+		/* last_failure */
+		values[7] = TimestampTzGetDatum(relerrent->last_failure);
+
+		/* failure_message */
+		values[8] = CStringGetTextDatum(relerrent->errmsg);
+
+		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+	}
+
+	/* clean up and return the tuplestore */
+	tuplestore_donestoring(tupstore);
+
+	return (Datum) 0;
+}
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..92297d60d1 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 logical replication error',
+  proname => 'pg_stat_get_subscription_error', prorows => '10', proisstrict => 'f',
+  proretset => 't', provolatile => 's', proparallel => 'r',
+  prorettype => 'record', proargtypes => 'oid',
+  proallargtypes => '{oid,oid,oid,oid,text,xid,text,int8,timestamptz,text}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{subid,datid,subid,relid,command,xid,failure_source,failure_count,last_failure,last_failure_message}',
+  prosrc => 'pg_stat_get_subscription_errors' },
 { oid => '6118', descr => 'statistics: information about subscription',
   proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 9612c0a6c2..215ac3abd5 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,8 @@ typedef enum StatMsgType
 	PGSTAT_MTYPE_RESETSINGLECOUNTER,
 	PGSTAT_MTYPE_RESETSLRUCOUNTER,
 	PGSTAT_MTYPE_RESETREPLSLOTCOUNTER,
+	PGSTAT_MTYPE_SUBSCRIPTIONERR,
+	PGSTAT_MTYPE_SUBSCRIPTIONPURGE,
 	PGSTAT_MTYPE_AUTOVAC_START,
 	PGSTAT_MTYPE_VACUUM,
 	PGSTAT_MTYPE_ANALYZE,
@@ -539,6 +542,47 @@ 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 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;
+
+	bool	m_clear;
+	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_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 +754,8 @@ typedef union PgStat_Msg
 	PgStat_MsgChecksumFailure msg_checksumfailure;
 	PgStat_MsgReplSlot msg_replslot;
 	PgStat_MsgConn msg_conn;
+	PgStat_MsgSubscriptionErr msg_subscriptionerr;
+	PgStat_MsgSubscriptionPurge msg_subscriptionpurge;
 } PgStat_Msg;
 
 
@@ -908,6 +954,28 @@ typedef struct PgStat_StatReplSlotEntry
 	TimestampTz stat_reset_timestamp;
 } PgStat_StatReplSlotEntry;
 
+/*
+ * Subscription error statistics kept in the stats collector
+ */
+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];
+} PgStat_StatSubRelErrEntry;
 
 /*
  * Working state needed to accumulate per-function-call timing statistics.
@@ -1011,6 +1079,10 @@ 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_clear_subscription_error(Oid subid, Oid subrelid);
 
 extern void pgstat_initialize(void);
 
@@ -1106,6 +1178,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);
 
 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..9cde7c2d09 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2094,6 +2094,18 @@ 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,
+    s.subname,
+    e.relid,
+    e.command,
+    e.xid,
+    e.failure_source,
+    e.failure_count,
+    e.last_failure,
+    e.last_failure_message
+   FROM pg_subscription s,
+    (LATERAL pg_stat_get_subscription_error(s.oid) e(datid, subid, relid, command, xid, failure_source, failure_count, last_failure, last_failure_message)
+     JOIN pg_database d ON ((e.datid = d.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)

