pg_stat_replication.v1.patch

diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 346eaaf..75419b7 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -488,6 +488,21 @@ CREATE VIEW pg_stat_activity AS
     WHERE S.datid = D.oid AND
             S.usesysid = U.oid;
 
+CREATE VIEW pg_stat_replication_activity AS
+    SELECT
+            S.procpid,
+            S.usesysid,
+            U.rolname AS usename,
+            S.application_name,
+            S.client_addr,
+            S.client_port,
+            S.backend_start,
+            R.sent_location
+    FROM pg_stat_get_activity(NULL) AS S, pg_authid U,
+            pg_stat_get_replication_activity() AS R
+    WHERE S.usesysid = U.oid AND
+            S.procpid = R.procpid;
+
 CREATE VIEW pg_stat_database AS
     SELECT
             D.oid AS datid,
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index e9d8847..4422f5a 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -37,6 +37,7 @@
 #include <signal.h>
 #include <unistd.h>
 
+#include "funcapi.h"
 #include "access/xlog_internal.h"
 #include "catalog/pg_type.h"
 #include "libpq/libpq.h"
@@ -49,6 +50,7 @@
 #include "storage/ipc.h"
 #include "storage/pmsignal.h"
 #include "tcop/tcopprot.h"
+#include "utils/builtins.h"
 #include "utils/guc.h"
 #include "utils/memutils.h"
 #include "utils/ps_status.h"
@@ -1122,6 +1124,91 @@ WalSndWakeup(void)
 }
 
 /*
+ * Returns the Send position of walsenders with given pid.
+ * Or InvalidXLogRecPtr if none.
+ */
+Datum
+pg_stat_get_replication_activity(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx;
+
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext oldcontext;
+		TupleDesc	tupdesc;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+#define PG_STAT_GET_REP_ACTIVITY_COLS 	2
+		tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_REP_ACTIVITY_COLS, false);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "procpid", INT4OID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "sent_location", TEXTOID, -1, 0);
+
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		funcctx->user_fctx = palloc0(sizeof(int));
+		if (superuser())
+			funcctx->max_calls = max_wal_senders;
+		else
+			funcctx->max_calls = 0;
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	/* stuff done on every call of the function */
+	funcctx = SRF_PERCALL_SETUP();
+
+	if (funcctx->call_cntr < funcctx->max_calls)
+	{
+		/* for each row */
+		Datum		values[PG_STAT_GET_REP_ACTIVITY_COLS];
+		bool		nulls[PG_STAT_GET_REP_ACTIVITY_COLS];
+		HeapTuple	tuple;
+
+		MemSet(values, 0, sizeof(values));
+		MemSet(nulls, 0, sizeof(nulls));
+
+		{
+			/* use volatile pointer to prevent code rearrangement */
+			volatile WalSnd *walsnd = &WalSndCtl->walsnds[funcctx->call_cntr];
+			char		sendlocation[MAXFNAMELEN];
+			XLogRecPtr	recptr;
+
+			if (walsnd->pid == 0)
+			{
+				nulls[0] = true;
+				nulls[1] = true;
+			}
+			else
+			{
+				SpinLockAcquire(&walsnd->mutex);
+				recptr = walsnd->sentPtr;
+				SpinLockRelease(&walsnd->mutex);
+
+				snprintf(sendlocation, sizeof(sendlocation), "%X/%X",
+						 recptr.xlogid, recptr.xrecoff);
+
+				nulls[0] = false;
+				values[0] = Int32GetDatum(walsnd->pid);
+
+				nulls[1] = false;
+				values[1] = CStringGetTextDatum(sendlocation);
+			}
+
+			tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+	else
+	{
+		/* nothing left */
+		SRF_RETURN_DONE(funcctx);
+	}
+}
+
+/*
  * This isn't currently used for anything. Monitoring tools might be
  * interested in the future, and we'll need something like this in the
  * future for synchronous replication.
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 1e6e75f..3218b87 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -3067,6 +3067,8 @@ DATA(insert OID = 1936 (  pg_stat_get_backend_idset		PGNSP PGUID 12 1 100 0 f f
 DESCR("statistics: currently active backend IDs");
 DATA(insert OID = 2022 (  pg_stat_get_activity			PGNSP PGUID 12 1 100 0 f f f f t s 1 0 2249 "23" "{23,26,23,26,25,25,16,1184,1184,1184,869,23}" "{i,o,o,o,o,o,o,o,o,o,o,o}" "{pid,datid,procpid,usesysid,application_name,current_query,waiting,xact_start,query_start,backend_start,client_addr,client_port}" _null_ pg_stat_get_activity _null_ _null_ _null_ ));
 DESCR("statistics: information about currently active backends");
+DATA(insert OID = 3099 (  pg_stat_get_replication_activity	PGNSP PGUID 12 1 10 0 f f f f t s 0 0 2249 "" "{23,25}" "{o,o}" "{pid,sent_location}" _null_ pg_stat_get_replication_activity _null_ _null_ _null_ ));
+DESCR("statistics: information about currently active replication");
 DATA(insert OID = 2026 (  pg_backend_pid				PGNSP PGUID 12 1 0 0 f f f t f s 0 0 23 "" _null_ _null_ _null_ _null_ pg_backend_pid _null_ _null_ _null_ ));
 DESCR("statistics: current backend PID");
 DATA(insert OID = 1937 (  pg_stat_get_backend_pid		PGNSP PGUID 12 1 0 0 f f f t f s 1 0 23 "23" _null_ _null_ _null_ _null_ pg_stat_get_backend_pid _null_ _null_ _null_ ));