pg_stat_walsender-20110104.patch

diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 262b372..3e758f6 100644
*** a/src/backend/catalog/system_views.sql
--- b/src/backend/catalog/system_views.sql
*************** CREATE VIEW pg_stat_activity AS
*** 492,497 ****
--- 492,509 ----
      WHERE S.datid = D.oid AND
              S.usesysid = U.oid;
  
+ CREATE VIEW pg_stat_walsender AS
+     SELECT
+             S.procpid,
+             S.usesysid,
+             U.rolname AS usename,
+             S.client_addr,
+             S.client_port,
+             S.backend_start,
+             S.xlog_sending
+     FROM pg_stat_get_walsender(NULL) AS S, pg_authid U
+     WHERE S.usesysid = U.oid;
+ 
  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 269d200..62422d4 100644
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
*************** WalSndWakeup(void)
*** 943,948 ****
--- 943,979 ----
  }
  
  /*
+  * Retrieve status of a WAL sender. Return true if a WAL sender found.
+  */
+ bool
+ WALSndStatus(pid_t pid, XLogRecPtr *sending)
+ {
+ 	int			i;
+ 
+ 	Assert(sending != NULL);
+ 
+ 	for (i = 0; i < max_wal_senders; i++)
+ 	{
+ 		/* use volatile pointer to prevent code rearrangement */
+ 		volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
+ 
+ 		if (walsnd->pid == pid)
+ 		{
+ 			SpinLockAcquire(&walsnd->mutex);
+ 			*sending = walsnd->sentPtr;
+ 			SpinLockRelease(&walsnd->mutex);
+ 			elog(INFO, "WALSndStatus %X/%X", sending->xlogid, sending->xrecoff);
+ 			return true;
+ 		}
+ 	}
+ 
+ 	/* not found */
+ 	elog(INFO, "WALSndStatus NOT FOUND: %d", pid);
+ 	memset(sending, 0, sizeof(XLogRecPtr));
+ 	return false;
+ }
+ 
+ /*
   * 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/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 31df9ee..05826a8 100644
*** a/src/backend/utils/adt/pgstatfuncs.c
--- b/src/backend/utils/adt/pgstatfuncs.c
***************
*** 18,23 ****
--- 18,24 ----
  #include "miscadmin.h"
  #include "pgstat.h"
  #include "catalog/pg_type.h"
+ #include "replication/walsender.h"
  #include "utils/builtins.h"
  #include "utils/inet.h"
  #include "libpq/ip.h"
*************** extern Datum pg_stat_get_function_self_t
*** 49,54 ****
--- 50,56 ----
  
  extern Datum pg_stat_get_backend_idset(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_activity(PG_FUNCTION_ARGS);
+ extern Datum pg_stat_get_walsender(PG_FUNCTION_ARGS);
  extern Datum pg_backend_pid(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_backend_pid(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_backend_dbid(PG_FUNCTION_ARGS);
*************** pg_stat_get_activity(PG_FUNCTION_ARGS)
*** 675,680 ****
--- 677,799 ----
  	}
  }
  
+ Datum
+ pg_stat_get_walsender(PG_FUNCTION_ARGS)
+ {
+ #define PG_STAT_WALSENDERS_COLS		6
+ 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+ 	TupleDesc	tupdesc;
+ 	Tuplestorestate *tupstore;
+ 	MemoryContext per_query_ctx;
+ 	MemoryContext oldcontext;
+ 	int			num_backends;
+ 	int			beid;
+ 	int			pid = (PG_ARGISNULL(0) ? 0 : PG_GETARG_INT32(0));
+ 	Oid			userid = GetUserId();
+ 	bool		is_superuser = superuser();
+ 
+ 	/* 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);
+ 
+ 	num_backends = pgstat_fetch_stat_numbackends();
+ 	for (beid = 0; beid < num_backends; beid++)
+ 	{
+ 		/* for each row */
+ 		Datum			values[PG_STAT_WALSENDERS_COLS];
+ 		bool			nulls[PG_STAT_WALSENDERS_COLS];
+ 		int				i;
+ 		PgBackendStatus *beentry;
+ 
+ 		beentry = pgstat_fetch_stat_beentry(beid + 1);		/* 1-based index */
+ 
+ 		/* Skip if not active, normal backends, or non-target processes */
+ 		if (!beentry || beentry->st_databaseid != InvalidOid ||
+ 			(pid != 0 && beentry->st_procpid != pid))
+ 			continue;
+ 
+ 		memset(values, 0, sizeof(values));
+ 		memset(nulls, 0, sizeof(nulls));
+ 		i = 0;
+ 
+ 		/* Values available to all callers */
+ 		values[i++] = Int32GetDatum(beentry->st_procpid);
+ 		values[i++] = ObjectIdGetDatum(beentry->st_userid);
+ 
+ 		/* Values only available to same user or superuser */
+ 		if (is_superuser || beentry->st_userid == userid)
+ 		{
+ 			inet	   *client_addr;
+ 			int			client_port;
+ 			XLogRecPtr	xlog_sending;
+ 
+ 			if (beentry->st_proc_start_timestamp != 0)
+ 				values[i++] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
+ 			else
+ 				nulls[i++] = true;
+ 
+ 			get_host_and_port(&beentry->st_clientaddr, &client_addr, &client_port);
+ 			if (client_addr != NULL)
+ 				values[i++] = PointerGetDatum(client_addr);
+ 			else
+ 				nulls[i++] = true;
+ 			if (client_port != 0)
+ 				values[i++] = DatumGetInt32(client_port);
+ 			else
+ 				nulls[i++] = true;
+ 
+ 			if (WALSndStatus(beentry->st_procpid, &xlog_sending))
+ 			{
+ 				char	location[18];
+ 
+ 				snprintf(location, sizeof(location), "%X/%X",
+ 						 xlog_sending.xlogid, xlog_sending.xrecoff);
+ 				values[i++] = CStringGetTextDatum(location);
+ 			}
+ 			else
+ 				nulls[i++] = true;
+ 		}
+ 		else
+ 		{
+ 			/* No permissions to view data about this session */
+ 			nulls[i++] = true;	/* backend_start */
+ 			nulls[i++] = true;	/* client_addr */
+ 			nulls[i++] = true;	/* client_port */
+ 			nulls[i++] = true;	/* xlog_sending */
+ 		}
+ 
+ 		Assert(i == PG_STAT_WALSENDERS_COLS);
+ 
+ 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+ 	}
+ 
+ 	/* clean up and return the tuplestore */
+ 	tuplestore_donestoring(tupstore);
+ 
+ 	return (Datum) 0;
+ }
+ 
  
  Datum
  pg_backend_pid(PG_FUNCTION_ARGS)
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index d6ed60a..27389fe 100644
*** a/src/include/catalog/pg_proc.h
--- b/src/include/catalog/pg_proc.h
*************** DATA(insert OID = 1936 (  pg_stat_get_ba
*** 3075,3080 ****
--- 3075,3082 ----
  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 = 3831 (  pg_stat_get_walsender			PGNSP PGUID 12 1 100 0 f f f f t s 1 0 2249 "23" "{23,23,26,1184,869,23,25}" "{i,o,o,o,o,o,o}" "{pid,procpid,usesysid,backend_start,client_addr,client_port,xlog_sending}" _null_ pg_stat_get_walsender _null_ _null_ _null_ ));
+ DESCR("statistics: information about wal senders");
  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_ ));
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 6f543ae..87906bf 100644
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
*************** extern void WalSndSignals(void);
*** 53,57 ****
--- 53,58 ----
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);
  extern void WalSndWakeup(void);
+ extern bool WALSndStatus(pid_t pid, XLogRecPtr *sending);
  
  #endif   /* _WALSENDER_H */
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index a6baae2..0558c13 100644
*** a/src/test/regress/expected/rules.out
--- b/src/test/regress/expected/rules.out
*************** SELECT viewname, definition FROM pg_view
*** 1301,1306 ****
--- 1301,1307 ----
   pg_stat_user_functions      | SELECT p.oid AS funcid, n.nspname AS schemaname, p.proname AS funcname, pg_stat_get_function_calls(p.oid) AS calls, (pg_stat_get_function_time(p.oid) / 1000) AS total_time, (pg_stat_get_function_self_time(p.oid) / 1000) AS self_time FROM (pg_proc p LEFT JOIN pg_namespace n ON ((n.oid = p.pronamespace))) WHERE ((p.prolang <> (12)::oid) AND (pg_stat_get_function_calls(p.oid) IS NOT NULL));
   pg_stat_user_indexes        | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_indexes.schemaname !~ '^pg_toast'::text));
   pg_stat_user_tables         | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze, pg_stat_all_tables.vacuum_count, pg_stat_all_tables.autovacuum_count, pg_stat_all_tables.analyze_count, pg_stat_all_tables.autoanalyze_count FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_tables.schemaname !~ '^pg_toast'::text));
+  pg_stat_walsender           | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.client_addr, s.client_port, s.backend_start, s.xlog_sending FROM pg_stat_get_walsender(NULL::integer) s(procpid, usesysid, backend_start, client_addr, client_port, xlog_sending), pg_authid u WHERE (s.usesysid = u.oid);
   pg_stat_xact_all_tables     | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, pg_stat_get_xact_numscans(c.oid) AS seq_scan, pg_stat_get_xact_tuples_returned(c.oid) AS seq_tup_read, (sum(pg_stat_get_xact_numscans(i.indexrelid)))::bigint AS idx_scan, ((sum(pg_stat_get_xact_tuples_fetched(i.indexrelid)))::bigint + pg_stat_get_xact_tuples_fetched(c.oid)) AS idx_tup_fetch, pg_stat_get_xact_tuples_inserted(c.oid) AS n_tup_ins, pg_stat_get_xact_tuples_updated(c.oid) AS n_tup_upd, pg_stat_get_xact_tuples_deleted(c.oid) AS n_tup_del, pg_stat_get_xact_tuples_hot_updated(c.oid) AS n_tup_hot_upd FROM ((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname;
   pg_stat_xact_sys_tables     | SELECT pg_stat_xact_all_tables.relid, pg_stat_xact_all_tables.schemaname, pg_stat_xact_all_tables.relname, pg_stat_xact_all_tables.seq_scan, pg_stat_xact_all_tables.seq_tup_read, pg_stat_xact_all_tables.idx_scan, pg_stat_xact_all_tables.idx_tup_fetch, pg_stat_xact_all_tables.n_tup_ins, pg_stat_xact_all_tables.n_tup_upd, pg_stat_xact_all_tables.n_tup_del, pg_stat_xact_all_tables.n_tup_hot_upd FROM pg_stat_xact_all_tables WHERE ((pg_stat_xact_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_xact_all_tables.schemaname ~ '^pg_toast'::text));
   pg_stat_xact_user_functions | SELECT p.oid AS funcid, n.nspname AS schemaname, p.proname AS funcname, pg_stat_get_xact_function_calls(p.oid) AS calls, (pg_stat_get_xact_function_time(p.oid) / 1000) AS total_time, (pg_stat_get_xact_function_self_time(p.oid) / 1000) AS self_time FROM (pg_proc p LEFT JOIN pg_namespace n ON ((n.oid = p.pronamespace))) WHERE ((p.prolang <> (12)::oid) AND (pg_stat_get_xact_function_calls(p.oid) IS NOT NULL));
*************** SELECT viewname, definition FROM pg_view
*** 1335,1341 ****
   shoelace_obsolete           | SELECT shoelace.sl_name, shoelace.sl_avail, shoelace.sl_color, shoelace.sl_len, shoelace.sl_unit, shoelace.sl_len_cm FROM shoelace WHERE (NOT (EXISTS (SELECT shoe.shoename FROM shoe WHERE (shoe.slcolor = shoelace.sl_color))));
   street                      | SELECT r.name, r.thepath, c.cname FROM ONLY road r, real_city c WHERE (c.outline ## r.thepath);
   toyemp                      | SELECT emp.name, emp.age, emp.location, (12 * emp.salary) AS annualsal FROM emp;
! (57 rows)
  
  SELECT tablename, rulename, definition FROM pg_rules
  	ORDER BY tablename, rulename;
--- 1336,1342 ----
   shoelace_obsolete           | SELECT shoelace.sl_name, shoelace.sl_avail, shoelace.sl_color, shoelace.sl_len, shoelace.sl_unit, shoelace.sl_len_cm FROM shoelace WHERE (NOT (EXISTS (SELECT shoe.shoename FROM shoe WHERE (shoe.slcolor = shoelace.sl_color))));
   street                      | SELECT r.name, r.thepath, c.cname FROM ONLY road r, real_city c WHERE (c.outline ## r.thepath);
   toyemp                      | SELECT emp.name, emp.age, emp.location, (12 * emp.salary) AS annualsal FROM emp;
! (58 rows)
  
  SELECT tablename, rulename, definition FROM pg_rules
  	ORDER BY tablename, rulename;