1 files changed, 59 insertions, 97 deletions

diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 30e8dfa7c12..fd993d0d5fb 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -2163,7 +2163,7 @@ pg_stat_reset_single_table_counters(PG_FUNCTION_ARGS)
 {
 	Oid			taboid = PG_GETARG_OID(0);
 
-	pgstat_reset_single_counter(taboid, InvalidOid, RESET_TABLE);
+	pgstat_reset_single_counter(taboid, RESET_TABLE);
 
 	PG_RETURN_VOID();
 }
@@ -2173,38 +2173,11 @@ pg_stat_reset_single_function_counters(PG_FUNCTION_ARGS)
 {
 	Oid			funcoid = PG_GETARG_OID(0);
 
-	pgstat_reset_single_counter(funcoid, InvalidOid, RESET_FUNCTION);
+	pgstat_reset_single_counter(funcoid, RESET_FUNCTION);
 
 	PG_RETURN_VOID();
 }
 
-Datum
-pg_stat_reset_subscription_worker_subrel(PG_FUNCTION_ARGS)
-{
-	Oid			subid = PG_GETARG_OID(0);
-	Oid			relid = PG_ARGISNULL(1) ? InvalidOid : PG_GETARG_OID(1);
-
-	pgstat_reset_single_counter(subid, relid, RESET_SUBWORKER);
-
-	PG_RETURN_VOID();
-}
-
-/* Reset all subscription worker stats associated with the given subscription */
-Datum
-pg_stat_reset_subscription_worker_sub(PG_FUNCTION_ARGS)
-{
-	Oid			subid = PG_GETARG_OID(0);
-
-	/*
-	 * Use subscription drop message to remove statistics of all subscription
-	 * workers.
-	 */
-	pgstat_report_subscription_drop(subid);
-
-	PG_RETURN_VOID();
-}
-
-
 /* Reset SLRU counters (a specific one or all of them). */
 Datum
 pg_stat_reset_slru(PG_FUNCTION_ARGS)
@@ -2258,6 +2231,32 @@ pg_stat_reset_replication_slot(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+/* Reset subscription stats (a specific one or all of them) */
+Datum
+pg_stat_reset_subscription_stats(PG_FUNCTION_ARGS)
+{
+	Oid			subid;
+
+	if (PG_ARGISNULL(0))
+	{
+		/* Clear all subscription stats */
+		subid = InvalidOid;
+	}
+	else
+	{
+		subid = PG_GETARG_OID(0);
+
+		if (!OidIsValid(subid))
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("invalid subscription OID %u", subid)));
+	}
+
+	pgstat_reset_subscription_counter(subid);
+
+	PG_RETURN_VOID();
+}
+
 Datum
 pg_stat_get_archiver(PG_FUNCTION_ARGS)
 {
@@ -2400,50 +2399,32 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 }
 
 /*
- * Get the subscription worker statistics for the given subscription
- * (and relation).
+ * Get the subscription statistics for the given subscription. If the
+ * subscription statistics is not available, return all-zeros stats.
  */
 Datum
-pg_stat_get_subscription_worker(PG_FUNCTION_ARGS)
+pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_SUBSCRIPTION_WORKER_COLS	8
+#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS	4
 	Oid			subid = PG_GETARG_OID(0);
-	Oid			subrelid;
 	TupleDesc	tupdesc;
-	Datum		values[PG_STAT_GET_SUBSCRIPTION_WORKER_COLS];
-	bool		nulls[PG_STAT_GET_SUBSCRIPTION_WORKER_COLS];
-	PgStat_StatSubWorkerEntry *wentry;
-	int			i;
-
-	if (PG_ARGISNULL(1))
-		subrelid = InvalidOid;
-	else
-		subrelid = PG_GETARG_OID(1);
+	Datum		values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS];
+	bool		nulls[PG_STAT_GET_SUBSCRIPTION_STATS_COLS];
+	PgStat_StatSubEntry *subentry;
+	PgStat_StatSubEntry allzero;
 
-	/* Get subscription worker stats */
-	wentry = pgstat_fetch_stat_subworker_entry(subid, subrelid);
-
-	/* Return NULL if there is no worker statistics */
-	if (wentry == NULL)
-		PG_RETURN_NULL();
+	/* Get subscription stats */
+	subentry = pgstat_fetch_stat_subscription(subid);
 
 	/* Initialise attributes information in the tuple descriptor */
-	tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBSCRIPTION_WORKER_COLS);
+	tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBSCRIPTION_STATS_COLS);
 	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "subid",
 					   OIDOID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "subrelid",
-					   OIDOID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "last_error_relid",
-					   OIDOID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "last_error_command",
-					   TEXTOID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "last_error_xid",
-					   XIDOID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "last_error_count",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "apply_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "last_error_message",
-					   TEXTOID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "last_error_time",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "stats_reset",
 					   TIMESTAMPTZOID, -1, 0);
 	BlessTupleDesc(tupdesc);
 
@@ -2451,46 +2432,27 @@ pg_stat_get_subscription_worker(PG_FUNCTION_ARGS)
 	MemSet(values, 0, sizeof(values));
 	MemSet(nulls, 0, sizeof(nulls));
 
-	i = 0;
-	/* subid */
-	values[i++] = ObjectIdGetDatum(subid);
-
-	/* subrelid */
-	if (OidIsValid(subrelid))
-		values[i++] = ObjectIdGetDatum(subrelid);
-	else
-		nulls[i++] = true;
-
-	/* last_error_relid */
-	if (OidIsValid(wentry->last_error_relid))
-		values[i++] = ObjectIdGetDatum(wentry->last_error_relid);
-	else
-		nulls[i++] = true;
-
-	/* last_error_command */
-	if (wentry->last_error_command != 0)
-		values[i++] =
-			CStringGetTextDatum(logicalrep_message_type(wentry->last_error_command));
-	else
-		nulls[i++] = true;
+	if (!subentry)
+	{
+		/* If the subscription is not found, initialise its stats */
+		memset(&allzero, 0, sizeof(PgStat_StatSubEntry));
+		subentry = &allzero;
+	}
 
-	/* last_error_xid */
-	if (TransactionIdIsValid(wentry->last_error_xid))
-		values[i++] = TransactionIdGetDatum(wentry->last_error_xid);
-	else
-		nulls[i++] = true;
+	/* subid */
+	values[0] = ObjectIdGetDatum(subid);
 
-	/* last_error_count */
-	values[i++] = Int64GetDatum(wentry->last_error_count);
+	/* apply_error_count */
+	values[1] = Int64GetDatum(subentry->apply_error_count);
 
-	/* last_error_message */
-	values[i++] = CStringGetTextDatum(wentry->last_error_message);
+	/* sync_error_count */
+	values[2] = Int64GetDatum(subentry->sync_error_count);
 
-	/* last_error_time */
-	if (wentry->last_error_time != 0)
-		values[i++] = TimestampTzGetDatum(wentry->last_error_time);
+	/* stats_reset */
+	if (subentry->stat_reset_timestamp == 0)
+		nulls[3] = true;
 	else
-		nulls[i++] = true;
+		values[3] = TimestampTzGetDatum(subentry->stat_reset_timestamp);
 
 	/* Returns the record as Datum */
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));