From 0f4e04e61e6a12ca1e7d3397164c06adc61c8e79 Mon Sep 17 00:00:00 2001
From: Ian Barwick <barwick@gmail.com>
Date: Thu, 22 Nov 2018 19:29:56 +0900
Subject: [PATCH] Add function get_current_lsn()

This is a somewhat convoluted attempt to retrieve the current LSN
of any node, regardless of whether in recovery or not, and if in
recovery, independent of whether streaming or recovering from
archive.
---
 dbutils.c | 129 +++++++++++++++++++++++++++++++++++++++++++++++++++++-
 dbutils.h |   1 +
 2 files changed, 129 insertions(+), 1 deletion(-)

diff --git a/dbutils.c b/dbutils.c
index 71962395..b27a6804 100644
--- a/dbutils.c
+++ b/dbutils.c
@@ -4787,7 +4787,6 @@ get_last_wal_receive_location(PGconn *conn)
 	PGresult   *res = NULL;
 	XLogRecPtr	ptr = InvalidXLogRecPtr;
 
-
 	if (server_version_num >= 100000)
 	{
 		res = PQexec(conn, "SELECT pg_catalog.pg_last_wal_receive_lsn()");
@@ -4807,6 +4806,134 @@ get_last_wal_receive_location(PGconn *conn)
 	return ptr;
 }
 
+XLogRecPtr
+get_current_lsn(PGconn *conn)
+{
+	PQExpBufferData query;
+	PGresult   *res = NULL;
+	XLogRecPtr	ptr = InvalidXLogRecPtr;
+
+	initPQExpBuffer(&query);
+
+	if (server_version_num == UNKNOWN_SERVER_VERSION_NUM)
+		server_version_num = get_server_version(conn, NULL);
+
+/*
+WITH lsn_states AS (
+  SELECT
+    CASE WHEN pg_catalog.pg_is_in_recovery() IS FALSE
+      THEN pg_catalog.pg_current_wal_lsn()
+      ELSE NULL
+    END
+      AS current_wal_lsn,
+    CASE WHEN pg_catalog.pg_is_in_recovery() IS TRUE
+	   THEN pg_catalog.pg_last_wal_receive_lsn()
+       ELSE NULL
+    END
+      AS last_wal_receive_lsn,
+    CASE WHEN pg_catalog.pg_is_in_recovery() IS TRUE
+	   THEN pg_catalog.pg_last_wal_replay_lsn()
+       ELSE NULL
+    END
+      AS last_wal_replay_lsn
+)
+SELECT
+  CASE WHEN pg_catalog.pg_is_in_recovery() IS FALSE
+    THEN current_wal_lsn
+    ELSE
+      CASE WHEN last_wal_receive_lsn IS NULL
+         THEN last_wal_replay_lsn
+         ELSE
+           CASE WHEN last_wal_replay_lsn > last_wal_receive_lsn
+             THEN last_wal_replay_lsn
+             ELSE last_wal_receive_lsn
+           END
+       END
+  END
+    AS current_lsn
+   FROM lsn_states
+
+
+*/
+	if (server_version_num >= 100000)
+	{
+		appendPQExpBufferStr(&query,
+							 " WITH lsn_states AS ( "
+							 "  SELECT "
+							 "    CASE WHEN pg_catalog.pg_is_in_recovery() IS FALSE "
+							 "      THEN pg_catalog.pg_current_wal_lsn() "
+							 "      ELSE NULL "
+							 "    END "
+							 "      AS current_wal_lsn, "
+							 "    CASE WHEN pg_catalog.pg_is_in_recovery() IS TRUE "
+							 "      THEN pg_catalog.pg_last_wal_receive_lsn() "
+							 "      ELSE NULL "
+							 "    END "
+							 "      AS last_wal_receive_lsn, "
+							 "    CASE WHEN pg_catalog.pg_is_in_recovery() IS TRUE "
+							 "      THEN pg_catalog.pg_last_wal_replay_lsn() "
+							 "      ELSE NULL "
+							 "     END "
+							 "       AS last_wal_replay_lsn "
+							 " ) ");
+	}
+	else
+	{
+		appendPQExpBufferStr(&query,
+							 " WITH lsn_states AS ( "
+							 "  SELECT "
+							 "    CASE WHEN pg_catalog.pg_is_in_recovery() IS FALSE "
+							 "      THEN pg_catalog.pg_current_xlog_location() "
+							 "      ELSE NULL "
+							 "    END "
+							 "      AS current_wal_lsn, "
+							 "    CASE WHEN pg_catalog.pg_is_in_recovery() IS TRUE "
+							 "      THEN pg_catalog.pg_last_xlog_receive_location() "
+							 "      ELSE NULL "
+							 "    END "
+							 "      AS last_wal_receive_lsn, "
+							 "    CASE WHEN pg_catalog.pg_is_in_recovery() IS TRUE "
+							 "      THEN pg_catalog.pg_last_xlog_replay_location()) "
+							 "      ELSE NULL "
+							 "     END "
+							 "       AS last_wal_replay_lsn "
+							 " ) ");
+	}
+
+	appendPQExpBufferStr(&query,
+						 " SELECT "
+						 "   CASE WHEN pg_catalog.pg_is_in_recovery() IS FALSE "
+						 "     THEN current_wal_lsn "
+						 "     ELSE "
+						 "       CASE WHEN last_wal_receive_lsn IS NULL "
+						 "       THEN last_wal_replay_lsn "
+						 "         ELSE "
+						 "           CASE WHEN last_wal_replay_lsn > last_wal_receive_lsn "
+						 "             THEN last_wal_replay_lsn "
+						 "             ELSE last_wal_receive_lsn "
+						 "           END "
+						 "       END "
+						 "   END "
+						 "     AS current_lsn "
+						 "   FROM lsn_states ");
+
+	res = PQexec(conn, query.data);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		log_db_error(conn, query.data, _("unable to execute get_current_lsn()"));
+	}
+	else if (!PQgetisnull(res, 0, 0))
+	{
+		ptr = parse_lsn(PQgetvalue(res, 0, 0));
+	}
+
+	termPQExpBuffer(&query);
+	PQclear(res);
+
+	return ptr;
+}
+
 /* ============= */
 /* BDR functions */
 /* ============= */
diff --git a/dbutils.h b/dbutils.h
index 9b745258..9737fe87 100644
--- a/dbutils.h
+++ b/dbutils.h
@@ -545,6 +545,7 @@ void		reset_voting_status(PGconn *conn);
 /* replication status functions */
 XLogRecPtr	get_current_wal_lsn(PGconn *conn);
 XLogRecPtr	get_last_wal_receive_location(PGconn *conn);
+XLogRecPtr	get_current_lsn(PGconn *conn);
 bool		get_replication_info(PGconn *conn, ReplInfo *replication_info);
 int			get_replication_lag_seconds(PGconn *conn);
 void		get_node_replication_stats(PGconn *conn, int server_version_num, t_node_info *node_info);