agora inbox for pgsql-hackers@postgresql.org  

help / color / mirror / Atom feed

[PATCH 2/2] Add monitoring aid for max_replication_slots.
13+ messages / 2 participants
[nested] [flat]

* [PATCH 2/2] Add monitoring aid for max_replication_slots.
@ 2017-09-07 10:13 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-09-07 10:13 UTC (permalink / raw)

Adds two columns "live" and "distance" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows how long a slot can live on or how many bytes a
slot have lost if max_slot_wal_keep_size is set.
---
 src/backend/access/transam/xlog.c    | 128 ++++++++++++++++++++++++++++++++++-
 src/backend/catalog/system_views.sql |   4 +-
 src/backend/replication/slotfuncs.c  |  25 ++++++-
 src/include/access/xlog.h            |   1 +
 src/include/catalog/pg_proc.h        |   2 +-
 src/test/regress/expected/rules.out  |   6 +-
 6 files changed, 160 insertions(+), 6 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index cfdae39..be53e0f 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9402,6 +9402,128 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minSecureLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo currSeg;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+	uint64 keepSegs;
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(currpos, currSeg, wal_segment_size);
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minSecureLSN)
+	{
+		if (max_slot_wal_keep_size_mb > 0)
+		{
+			uint64 slotlimitbytes = 1024 * 1024 * max_slot_wal_keep_size_mb;
+			uint64 slotlimitfragment = XLogSegmentOffset(slotlimitbytes,
+														 wal_segment_size);
+			uint64 currposoff = XLogSegmentOffset(currpos, wal_segment_size);
+
+			/* Calculate keep segments. Must be in sync with KeepLogSeg. */
+			Assert(wal_keep_segments >= 0);
+			Assert(max_slot_wal_keep_size_mb >= 0);
+
+			keepSegs = wal_keep_segments +
+				ConvertToXSegs(max_slot_wal_keep_size_mb, wal_segment_size);
+			if (currposoff < slotlimitfragment)
+				keepSegs++;
+
+			/*
+			 * calculate the oldest segment that will be kept by
+			 * wal_keep_segments and max_slot_wal_keep_size_mb
+			 */
+			if (currSeg < keepSegs)
+				tailSeg = 0;
+			else
+				tailSeg = currSeg - keepSegs;
+
+		}
+		else
+		{
+			/* all requred segments are secured in this case */
+			XLogRecPtr keep = XLogGetReplicationSlotMinimumLSN();
+			XLByteToSeg(keep, tailSeg, wal_segment_size);
+		}
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minSecureLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Retreat *logSegNo to the last segment that we need to retain because of
  * either wal_keep_segments or replication slots.
@@ -9429,7 +9551,11 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)
 			segno = segno - wal_keep_segments;
 	}
 
-	/* then check whether slots limit removal further */
+	/*
+	 * then check whether slots limit removal further
+	 * should be consistent with IsLsnStillAvaiable().
+	 */
+
 	if (max_replication_slots > 0 && keep != InvalidXLogRecPtr)
 	{
 		XLogSegNo	slotSegNo;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index dc40cde..6512ac3 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.status,
+			L.min_secure_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index ab776e8..200a478 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -304,6 +304,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_secure_lsn;
+			char *status = "borken";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_secure_lsn)))
+			{
+				if (min_secure_lsn <= restart_lsn)
+					status = "secured";
+				else
+					status = "insecured";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_secure_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index f0c0255..a316ead 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 93c031a..d03fd6f 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5340,7 +5340,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,status,min_secure_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index f1c1b44..d9d74a3 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.status,
+    l.min_secure_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, status, min_secure_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Thu_Nov_09_17_31_28_2017_905)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0


-- 
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

----Next_Part(Thu_Nov_09_17_31_28_2017_905)----





^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/2] Add monitoring aid for max_replication_slots.
@ 2017-09-07 10:13 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-09-07 10:13 UTC (permalink / raw)

Adds two columns "live" and "distance" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows how long a slot can live on or how many bytes a
slot have lost if max_slot_wal_keep_size is set.
---
 src/backend/access/transam/xlog.c    | 137 ++++++++++++++++++++++++++++++++++-
 src/backend/catalog/system_views.sql |   4 +-
 src/backend/replication/slotfuncs.c  |  16 +++-
 src/include/access/xlog.h            |   1 +
 src/include/catalog/pg_proc.h        |   2 +-
 src/test/regress/expected/rules.out  |   6 +-
 6 files changed, 160 insertions(+), 6 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f79cefb..a9203ff 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9402,6 +9402,140 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If distance is given, it receives the
+ * distance to the point where the margin defined by max_slot_wal_keep_size_mb
+ * and wal_keep_segments will be exhausted or how many bytes we have lost
+ * after restartLSN.
+ *
+ * true and distance = 0 means that restartLSN will be lost by at most two
+ * checkpoints.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, uint64 *distance)
+{
+	XLogRecPtr currpos;
+	XLogSegNo currSeg;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+	uint64 keepSegs;
+	uint64 restbytes;
+
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(currpos, currSeg, wal_segment_size);
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+	/* no need to calculate distance. very easy. */
+	if (!distance)
+		return	oldestSeg <= restartSeg;
+
+	/* This must perform the same thing as KeepLogSeg. */
+	Assert(wal_keep_segments >= 0);
+	Assert(max_slot_wal_keep_size_mb >= 0);
+	keepSegs = wal_keep_segments +
+		ConvertToXSegs(max_slot_wal_keep_size_mb, wal_segment_size);
+
+	/*
+	 * calculate the oldest segment that will be kept by wal_keep_segments and
+	 * max_slot_wal_keep_size_mb
+	 */
+	if (currSeg < keepSegs)
+		tailSeg = 0;
+	else
+		tailSeg = currSeg - keepSegs;
+
+	if (restartSeg < oldestSeg)
+	{
+		/*
+		 * restartSeg has been removed. Calculate how many bytes from the
+		 * restartLSN have lost.
+		 */
+		restbytes = (restartSeg + 1) * wal_segment_size - restartLSN;
+		*distance =
+			(oldestSeg - (restartSeg + 1)) * wal_segment_size + restbytes;
+
+		return false;
+	}
+
+	if (tailSeg <= restartSeg)
+	{
+		/* Return how many bytes we can advance before the slot loses margin */
+		restbytes = wal_segment_size - (currpos % wal_segment_size);
+		*distance = (restartSeg - tailSeg) * wal_segment_size + restbytes;
+	}
+	else
+	{
+		/* the margin ran out */
+		*distance = 0;
+	}
+
+	return true;
+}
+
 /*
  * Retreat *logSegNo to the last segment that we need to retain because of
  * either wal_keep_segments or replication slots.
@@ -9433,7 +9567,8 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)
 	if (max_replication_slots > 0 && keep != InvalidXLogRecPtr)
 	{
 		XLogSegNo	slotSegNo;
-		int			slotlimitsegs = ConvertToXSegs(max_slot_wal_keep_size_mb);
+		int			slotlimitsegs =
+			ConvertToXSegs(max_slot_wal_keep_size_mb, wal_segment_size);
 
 		XLByteToSeg(keep, slotSegNo, wal_segment_size);
 
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index dc40cde..c55c88b 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.live,
+			L.distance
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index ab776e8..107da1a 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -304,6 +304,20 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (max_slot_wal_keep_size_mb > 0 && restart_lsn != InvalidXLogRecPtr)
+		{
+			uint64 distance;
+
+			values[i++] = BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+														  &distance));
+			values[i++] = Int64GetDatum(distance);
+		}
+		else
+		{
+			values[i++] = BoolGetDatum(true);
+			nulls[i++] = true;
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index f0c0255..a7a1e4d 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, uint64 *distance);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 93c031a..0913e56 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5340,7 +5340,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,16,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,live,distance}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index f1c1b44..16a99d8 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.live,
+    l.distance
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, live, distance)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Tue_Oct_31_18_43_10_2017_912)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0


-- 
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

----Next_Part(Tue_Oct_31_18_43_10_2017_912)----





^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/2] Add monitoring aid for max_replication_slots.
@ 2017-09-07 10:13 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-09-07 10:13 UTC (permalink / raw)

Adds two columns "live" and "distance" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows how long a slot can live on or how many bytes a
slot have lost if max_slot_wal_keep_size is set.
---
 src/backend/access/transam/xlog.c    | 122 ++++++++++++++++++++++++++++++++++-
 src/backend/catalog/system_views.sql |   4 +-
 src/backend/replication/slotfuncs.c  |  27 +++++++-
 src/include/access/xlog.h            |   1 +
 src/include/catalog/pg_proc.h        |   2 +-
 src/test/regress/expected/rules.out  |   6 +-
 6 files changed, 156 insertions(+), 6 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index cfdae39..8ce7044 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9402,6 +9402,122 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If distance is given, it receives the
+ * distance to the point where the margin defined by max_slot_wal_keep_size_mb
+ * and wal_keep_segments will be exhausted or how many bytes we have lost
+ * after restartLSN.
+ *
+ * true and distance = 0 means that restartLSN will be lost by at most two
+ * checkpoints.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo currSeg;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+	uint64 keepSegs;
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(currpos, currSeg, wal_segment_size);
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minSecureLSN)
+	{
+		uint64 slotlimitbytes = 1024 * 1024 * max_slot_wal_keep_size_mb;
+		uint64 slotlimitfragment = XLogSegmentOffset(slotlimitbytes,
+													 wal_segment_size);
+		uint64 currposoff = XLogSegmentOffset(currpos, wal_segment_size);
+
+		/* Calculate keep segments. Must be in sync with KeepLogSeg. */
+		Assert(wal_keep_segments >= 0);
+		Assert(max_slot_wal_keep_size_mb >= 0);
+		keepSegs = wal_keep_segments +
+			ConvertToXSegs(max_slot_wal_keep_size_mb, wal_segment_size);
+		if (currposoff < slotlimitfragment)
+			keepSegs++;
+
+		/*
+		 * calculate the oldest segment that will be kept by wal_keep_segments and
+		 * max_slot_wal_keep_size_mb
+		 */
+		if (currSeg < keepSegs)
+			tailSeg = 0;
+		else
+			tailSeg = currSeg - keepSegs;
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minSecureLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Retreat *logSegNo to the last segment that we need to retain because of
  * either wal_keep_segments or replication slots.
@@ -9429,7 +9545,11 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)
 			segno = segno - wal_keep_segments;
 	}
 
-	/* then check whether slots limit removal further */
+	/*
+	 * then check whether slots limit removal further
+	 * should be consistent with IsLsnStillAvaiable().
+	 */
+
 	if (max_replication_slots > 0 && keep != InvalidXLogRecPtr)
 	{
 		XLogSegNo	slotSegNo;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index dc40cde..6512ac3 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.status,
+			L.min_secure_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index ab776e8..0dca618 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -304,6 +304,31 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (max_slot_wal_keep_size_mb > 0 && restart_lsn != InvalidXLogRecPtr)
+		{
+			XLogRecPtr	min_secure_lsn;
+			char *status = "unknown";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_secure_lsn)))
+			{
+				if (min_secure_lsn <= restart_lsn)
+					status = "secured";
+				else
+					status = "insecured";
+			}
+			else
+				status = "broken";
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_secure_lsn);
+		}
+		else
+		{
+			values[i++] = BoolGetDatum(true);
+			nulls[i++] = true;
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index f0c0255..a316ead 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 93c031a..d03fd6f 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5340,7 +5340,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,status,min_secure_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index f1c1b44..d9d74a3 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.status,
+    l.min_secure_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, status, min_secure_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Wed_Nov_08_13_14_31_2017_781)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0


-- 
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

----Next_Part(Wed_Nov_08_13_14_31_2017_781)----





^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/2] Add monitoring aid for max_replication_slots.
@ 2017-09-07 10:13 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-09-07 10:13 UTC (permalink / raw)

Adds two columns "live" and "distance" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows how long a slot can live on or how many bytes a
slot have lost if max_slot_wal_keep_size is set.
---
 src/backend/access/transam/xlog.c    | 80 ++++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql |  4 +-
 src/backend/replication/slotfuncs.c  | 16 +++++++-
 src/include/access/xlog.h            |  1 +
 src/include/catalog/pg_proc.h        |  2 +-
 5 files changed, 100 insertions(+), 3 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index ae70d7d..c4c8307 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9324,6 +9324,86 @@ CreateRestartPoint(int flags)
 }
 
 /*
+ * Check if the record on the given lsn will be preserved at the next
+ * checkpoint.
+ *
+ * Returns true if it will be preserved. If distance is given, the distance
+ * from origin to the beginning of the first segment kept at the next
+ * checkpoint. It means margin when this function returns true and gap of lost
+ * records when false.
+ *
+ * This function should return the consistent result with KeepLogSeg.
+ */
+bool
+GetMarginToSlotSegmentLimit(XLogRecPtr restartLSN, uint64 *distance)
+{
+	XLogRecPtr currpos;
+	XLogRecPtr tailpos;
+	uint64 currSeg;
+	uint64 restByteInSeg;
+	uint64 restartSeg;
+	uint64 tailSeg;
+	uint64 keepSegs;
+
+	currpos = GetXLogWriteRecPtr();
+
+	LWLockAcquire(ControlFileLock, LW_SHARED);
+	tailpos = ControlFile->checkPointCopy.redo;
+	LWLockRelease(ControlFileLock);
+
+	/* Move the pointer to the beginning of the segment*/
+	XLByteToSeg(currpos, currSeg);
+	XLByteToSeg(restartLSN, restartSeg);
+	XLByteToSeg(tailpos, tailSeg);
+	restByteInSeg = 0;
+
+	Assert(wal_keep_segments >= 0);
+	Assert(max_slot_wal_keep_size_mb >= 0);
+
+	/*
+	 * WAL are removed by the unit of segment.
+	 */
+	keepSegs = wal_keep_segments + ConvertToXSegs(max_slot_wal_keep_size_mb);
+
+	/*
+	 * If the latest checkpoint's redo point is older than the current head
+	 * minus keep segments, the next checkpoint keeps the redo point's
+	 * segment. Elsewise use current head minus number of segments to keep.
+	 */
+	if (currSeg < tailSeg + keepSegs)
+	{
+		if (currSeg < keepSegs)
+			tailSeg = 0;
+		else
+			tailSeg = currSeg - keepSegs;
+
+		/* In this case, the margin will be the bytes to the next segment */
+		restByteInSeg = XLogSegSize - (currpos % XLogSegSize);
+	}
+
+	elog(LOG, "rest %lx, curr %lx, tail %lx", restartSeg, currSeg, tailSeg);
+	/* Required sements will be removed at the next checkpoint */
+	if (restartSeg < tailSeg)
+	{
+		/* Calculate how may bytes the slot have lost */
+		if (distance)
+		{
+			uint64 restbytes = (restartSeg + 1) * XLogSegSize - restartLSN;
+			*distance =
+				(tailSeg - restartSeg - 1) * XLogSegSize
+				+ restbytes;
+		}
+		return false;
+	}
+
+	/* Margin at the next checkpoint before the slot lose sync  */
+	if (distance)
+		*distance = (restartSeg - tailSeg) * XLogSegSize + restByteInSeg;
+
+	return true;
+}
+
+/*
  * Retreat *logSegNo to the last segment that we need to retain because of
  * either wal_keep_segments or replication slots.
  *
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index dc40cde..c55c88b 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.live,
+			L.distance
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index d4cbd83..52a9d26 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -304,6 +304,20 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (max_slot_wal_keep_size_mb > 0 && restart_lsn != InvalidXLogRecPtr)
+		{
+			uint64 distance;
+
+			values[i++] = BoolGetDatum(GetMarginToSlotSegmentLimit(restart_lsn,
+																   &distance));
+			values[i++] = Int64GetDatum(distance);
+		}
+		else
+		{
+			values[i++] = BoolGetDatum(true);
+			nulls[i++] = true;
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index f596794..30507eb 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -267,6 +267,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool GetMarginToSlotSegmentLimit(XLogRecPtr restartLSN, uint64 *distance);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 8b33b4e..4bad345 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5293,7 +5293,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,16,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,live,distance}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
-- 
2.9.2


----Next_Part(Thu_Sep_07_21_59_56_2017_119)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0


-- 
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

----Next_Part(Thu_Sep_07_21_59_56_2017_119)----





^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 contrib/test_decoding/expected/ddl.out |  4 +-
 src/backend/access/transam/xlog.c      | 93 ++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql   |  4 +-
 src/backend/replication/slotfuncs.c    | 25 ++++++++-
 src/include/access/xlog.h              |  1 +
 src/include/catalog/pg_proc.h          |  2 +-
 src/test/regress/expected/rules.out    |  6 ++-
 7 files changed, 128 insertions(+), 7 deletions(-)

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index 1e22c1eefc..92cd56a5f0 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -702,7 +702,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
+ slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | min_keep_lsn 
+-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
 (0 rows)
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 542e1d78fe..529aee9014 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9346,6 +9346,99 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetOldestKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 5e6e8a64f6..9284175f7d 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -795,7 +795,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index e873dd1f81..16575fa411 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -185,7 +185,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -307,6 +307,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1d10..52e64f392d 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 0fdb42f639..e8e32c1a97 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5385,7 +5385,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 5e0597e091..3944e36681 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.16.2


----Next_Part(Mon_Mar_19_17_09_48_2018_369)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/3] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 src/backend/access/transam/xlog.c    | 93 ++++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql |  4 +-
 src/backend/replication/slotfuncs.c  | 25 +++++++++-
 src/include/access/xlog.h            |  1 +
 src/include/catalog/pg_proc.h        |  2 +-
 src/test/regress/expected/rules.out  |  6 ++-
 6 files changed, 126 insertions(+), 5 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 723a983..b630224 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9349,6 +9349,99 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minSecureLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minSecureLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetMinSecuredSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minSecureLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 394aea8..4167146 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.status,
+			L.min_secure_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index ab776e8..3880807 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -304,6 +304,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_secure_lsn;
+			char *status = "broken";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_secure_lsn)))
+			{
+				if (min_secure_lsn <= restart_lsn)
+					status = "secured";
+				else
+					status = "insecured";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_secure_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 45eb51a..542df28 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index c969375..1157438 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5340,7 +5340,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,status,min_secure_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index f1c1b44..d9d74a3 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.status,
+    l.min_secure_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, status, min_secure_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Fri_Dec_22_15_03_20_2017_403)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 src/backend/access/transam/xlog.c    | 93 ++++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql |  4 +-
 src/backend/replication/slotfuncs.c  | 25 +++++++++-
 src/include/access/xlog.h            |  1 +
 src/include/catalog/pg_proc.h        |  2 +-
 src/test/regress/expected/rules.out  |  6 ++-
 6 files changed, 126 insertions(+), 5 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index bdb7156..a8423f7 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9349,6 +9349,99 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetMinKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 5652e9e..cd714cc 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index b02df59..84f4154 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -304,6 +304,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1..52e64f3 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 298e0ae..93f62d7 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5345,7 +5345,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index f1c1b44..75d44af 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Thu_Jan_11_15_59_10_2018_946)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 contrib/test_decoding/expected/ddl.out |  4 +-
 src/backend/access/transam/xlog.c      | 93 ++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql   |  4 +-
 src/backend/replication/slotfuncs.c    | 25 ++++++++-
 src/include/access/xlog.h              |  1 +
 src/include/catalog/pg_proc.dat        |  6 +--
 src/test/regress/expected/rules.out    |  6 ++-
 7 files changed, 130 insertions(+), 9 deletions(-)

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index b7c76469fc..276f7f6efd 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -706,7 +706,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
+ slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | min_keep_lsn 
+-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
 (0 rows)
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index cf48ed06af..048f55ab77 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9379,6 +9379,99 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetOldestKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8cd8bf40ac..1664a086e9 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -797,7 +797,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 2806e1076c..f13aa4d455 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -185,7 +185,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -307,6 +307,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1d10..52e64f392d 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 40d54ed030..ef8f0eab91 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -9796,9 +9796,9 @@
   proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', prorettype => 'record',
   proargtypes => '',
-  proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn}',
-  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}',
+  proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,pg_lsn}',
+  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}',
   prosrc => 'pg_get_replication_slots' },
 { oid => '3786', descr => 'set up a logical replication slot',
   proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index ae0cd253d5..93f6bff77e 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.16.3


----Next_Part(Tue_Jun_26_16_26_59_2018_841)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v2-0001-Add-WAL-releaf-vent-for-replication-slots.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 contrib/test_decoding/expected/ddl.out |  4 +-
 src/backend/access/transam/xlog.c      | 95 ++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql   |  4 +-
 src/backend/replication/slotfuncs.c    | 25 ++++++++-
 src/include/access/xlog.h              |  1 +
 src/include/catalog/pg_proc.dat        |  6 +--
 src/test/regress/expected/rules.out    |  6 ++-
 7 files changed, 132 insertions(+), 9 deletions(-)

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index b7c76469fc..276f7f6efd 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -706,7 +706,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
+ slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | min_keep_lsn 
+-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
 (0 rows)
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 959d18e029..5c16750e89 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9379,6 +9379,101 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	FreeDir(xldir);
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetOldestKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8cd8bf40ac..1664a086e9 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -797,7 +797,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 2806e1076c..f13aa4d455 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -185,7 +185,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -307,6 +307,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1d10..52e64f392d 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 40d54ed030..ef8f0eab91 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -9796,9 +9796,9 @@
   proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', prorettype => 'record',
   proargtypes => '',
-  proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn}',
-  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}',
+  proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,pg_lsn}',
+  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}',
   prosrc => 'pg_get_replication_slots' },
 { oid => '3786', descr => 'set up a logical replication slot',
   proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index ae0cd253d5..93f6bff77e 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.16.3


----Next_Part(Wed_Jul_04_17_28_38_2018_173)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v3-0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 contrib/test_decoding/expected/ddl.out |  4 +-
 src/backend/access/transam/xlog.c      | 89 ++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql   |  4 +-
 src/backend/replication/slotfuncs.c    | 25 +++++++++-
 src/include/access/xlog.h              |  1 +
 src/include/catalog/pg_proc.dat        |  6 +--
 src/test/regress/expected/rules.out    |  6 ++-
 7 files changed, 126 insertions(+), 9 deletions(-)

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index b7c76469fc..276f7f6efd 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -706,7 +706,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
+ slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | min_keep_lsn 
+-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
 (0 rows)
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index c27a589e89..577be4ecd9 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9426,6 +9426,95 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	FreeDir(xldir);
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetOldestKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8cd8bf40ac..1664a086e9 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -797,7 +797,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 2806e1076c..f13aa4d455 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -185,7 +185,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -307,6 +307,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1d10..52e64f392d 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 40d54ed030..ef8f0eab91 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -9796,9 +9796,9 @@
   proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', prorettype => 'record',
   proargtypes => '',
-  proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn}',
-  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}',
+  proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,pg_lsn}',
+  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}',
   prosrc => 'pg_get_replication_slots' },
 { oid => '3786', descr => 'set up a logical replication slot',
   proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index ae0cd253d5..93f6bff77e 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.16.3


----Next_Part(Mon_Jul_09_14_47_06_2018_788)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v4-0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 contrib/test_decoding/expected/ddl.out |  4 +-
 src/backend/access/transam/xlog.c      | 93 ++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql   |  4 +-
 src/backend/replication/slotfuncs.c    | 25 ++++++++-
 src/include/access/xlog.h              |  1 +
 src/include/catalog/pg_proc.h          |  2 +-
 src/test/regress/expected/rules.out    |  6 ++-
 7 files changed, 128 insertions(+), 7 deletions(-)

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index 1e22c1e..92cd56a 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -702,7 +702,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
+ slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | min_keep_lsn 
+-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
 (0 rows)
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 55694cb..cbece1b 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9349,6 +9349,99 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetMinKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 5652e9e..cd714cc 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index cf2195b..e6e0386 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -185,7 +185,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -307,6 +307,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1..52e64f3 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index f01648c..0712e60 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5345,7 +5345,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 5433944..c5240c9 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Mon_Jan_29_19_26_34_2018_950)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 2/4] Add monitoring aid for max_replication_slots.
@ 2017-12-21 12:23 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 13+ messages in thread

From: Kyotaro Horiguchi @ 2017-12-21 12:23 UTC (permalink / raw)

Adds two columns "status" and "min_secure_lsn" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows that a slot can be reconnected or not, or about
to lose required WAL segments. And the LSN back to where the next
checkpoint will secure.
---
 contrib/test_decoding/expected/ddl.out |  4 +-
 src/backend/access/transam/xlog.c      | 93 ++++++++++++++++++++++++++++++++++
 src/backend/catalog/system_views.sql   |  4 +-
 src/backend/replication/slotfuncs.c    | 25 ++++++++-
 src/include/access/xlog.h              |  1 +
 src/include/catalog/pg_proc.h          |  2 +-
 src/test/regress/expected/rules.out    |  6 ++-
 7 files changed, 128 insertions(+), 7 deletions(-)

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index 1e22c1e..92cd56a 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -702,7 +702,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
+ slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | min_keep_lsn 
+-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
 (0 rows)
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index c01d0b3..099d89f 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9349,6 +9349,99 @@ CreateRestartPoint(int flags)
 	return true;
 }
 
+
+/*
+ * Returns the segment number of the oldest file in XLOG directory.
+ */
+static XLogSegNo
+GetOldestXLogFileSegNo(void)
+{
+	DIR		*xldir;
+	struct dirent *xlde;
+	XLogSegNo segno = 0;
+
+	xldir = AllocateDir(XLOGDIR);
+	if (xldir == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open write-ahead log directory \"%s\": %m",
+						XLOGDIR)));
+
+	while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+	{
+		TimeLineID tli;
+		XLogSegNo fsegno;
+
+		/* Ignore files that are not XLOG segments */
+		if (!IsXLogFileName(xlde->d_name) &&
+			!IsPartialXLogFileName(xlde->d_name))
+			continue;
+
+		XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+		/* get minimum segment ignorig timeline ID */
+		if (segno == 0 || fsegno < segno)
+			segno = fsegno;
+	}
+
+	return segno;
+}
+
+/*
+ * Check if the record on the given restartLSN is present in XLOG files.
+ *
+ * Returns true if it is present. If minKeepLSN is given, it receives the
+ * LSN at the beginning of the oldest existing WAL segment.
+ */
+bool
+IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minKeepLSN)
+{
+	XLogRecPtr currpos;
+	XLogSegNo restartSeg;
+	XLogSegNo tailSeg;
+	XLogSegNo oldestSeg;
+
+	Assert(!XLogRecPtrIsInvalid(restartLSN));
+
+	currpos = GetXLogWriteRecPtr();
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	oldestSeg = XLogCtl->lastRemovedSegNo;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * oldestSeg is zero before at least one segment has been removed since
+	 * startup. Use oldest segno taken from file names.
+	 */
+	if (oldestSeg == 0)
+	{
+		static XLogSegNo oldestFileSeg = 0;
+
+		if (oldestFileSeg == 0)
+			oldestFileSeg = GetOldestXLogFileSegNo();
+		/* let it have the same meaning with lastRemovedSegNo here */
+		oldestSeg = oldestFileSeg - 1;
+	}
+
+	/* oldest segment is just after the last removed segment */
+	oldestSeg++;
+
+	XLByteToSeg(restartLSN, restartSeg, wal_segment_size);
+
+
+	if (minKeepLSN)
+	{
+		XLogRecPtr slotPtr = XLogGetReplicationSlotMinimumLSN();
+		Assert(!XLogRecPtrIsInvalid(slotPtr));
+
+		tailSeg = GetOldestKeepSegment(currpos, slotPtr);
+
+		XLogSegNoOffsetToRecPtr(tailSeg, 0, *minKeepLSN, wal_segment_size);
+	}
+
+	return	oldestSeg <= restartSeg;
+}
+
 /*
  * Returns minimum segment number the next checktpoint must leave considering
  * wal_keep_segments, replication slots and max_slot_wal_keep_size.
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 5652e9e..cd714cc 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn,
-            L.confirmed_flush_lsn
+            L.confirmed_flush_lsn,
+			L.wal_status,
+			L.min_keep_lsn
     FROM pg_get_replication_slots() AS L
             LEFT JOIN pg_database D ON (L.datoid = D.oid);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index cf2195b..e6e0386 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -185,7 +185,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 11
+#define PG_GET_REPLICATION_SLOTS_COLS 13
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -307,6 +307,29 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
+		if (restart_lsn == InvalidXLogRecPtr)
+		{
+			values[i++] = CStringGetTextDatum("unknown");
+			values[i++] = LSNGetDatum(InvalidXLogRecPtr);
+		}
+		else
+		{
+			XLogRecPtr	min_keep_lsn;
+			char *status = "lost";
+
+			if (BoolGetDatum(IsLsnStillAvaiable(restart_lsn,
+												&min_keep_lsn)))
+			{
+				if (min_keep_lsn <= restart_lsn)
+					status = "streaming";
+				else
+					status = "keeping";
+			}
+
+			values[i++] = CStringGetTextDatum(status);
+			values[i++] = LSNGetDatum(min_keep_lsn);
+		}
+
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 	}
 	LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 12cd0d1..52e64f3 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -269,6 +269,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
+extern bool IsLsnStillAvaiable(XLogRecPtr restartLSN, XLogRecPtr *minSecureLSN);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);
 extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index f01648c..0712e60 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5345,7 +5345,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots	PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,25,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,min_keep_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 5433944..c5240c9 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1451,8 +1451,10 @@ pg_replication_slots| SELECT l.slot_name,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn,
-    l.confirmed_flush_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn)
+    l.confirmed_flush_lsn,
+    l.wal_status,
+    l.min_keep_lsn
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, min_keep_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,
-- 
2.9.2


----Next_Part(Mon_Jan_29_19_40_23_2018_721)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0003-TAP-test-for-the-slot-limit-feature.patch"



^ permalink  raw  reply  [nested|flat] 13+ messages in thread

* [PATCH 5/5] Serialize decoded tuples without flattening.
@ 2026-02-27 18:01 Antonin Houska <ah@cybertec.at>
  0 siblings, 0 replies; 13+ messages in thread

From: Antonin Houska @ 2026-02-27 18:01 UTC (permalink / raw)

So far we used to flatten the tuples produced by logical decoding in order to
write them to a temporary file. That means that all the TOASTed attributes
ended up in a single chunk of memory. The problem is that even if individual
attributes do fit into the maximum chunk size (currently 1 GB), the whole
tuple is not guaranteed to fit.

The flattening might be a problem in the future anyway, since there's an
effort to restrict the impact of REPACK (CONCURRENTLY) on the VACUUM xmin
horizon. However, as the flattening assumes scans of the TOAST relaation, the
less we restrict VACUUM, the more likely it is that the TOAST attributes
needed for the flattening won't be available at decoding time.

(This patch also removes unnecessary field 'tupdesc' from RepackDecodingState
and allocates it in the appropriate memory context.)
---
 src/backend/commands/cluster.c                | 260 +++++++++++++++---
 .../pgoutput_repack/pgoutput_repack.c         | 114 ++++++--
 src/include/commands/cluster.h                |   5 +-
 .../expected/repack_toast.out                 |   1 +
 .../injection_points/specs/repack_toast.spec  |   7 +
 5 files changed, 318 insertions(+), 69 deletions(-)

diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index db710259eeb..e4de410e733 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -35,6 +35,7 @@
 #include "postgres.h"
 
 #include "access/amapi.h"
+#include "access/detoast.h"
 #include "access/heapam.h"
 #include "access/multixact.h"
 #include "access/relscan.h"
@@ -285,6 +286,10 @@ static void apply_concurrent_update(Relation rel, HeapTuple tup,
 									IndexInsertState *iistate,
 									TupleTableSlot *index_slot);
 static void apply_concurrent_delete(Relation rel, HeapTuple tup_target);
+static HeapTuple restore_tuple(BufFile *file, Relation relation,
+							   MemoryContext cxt);
+static HeapTuple adjust_toast_pointers(Relation relation, HeapTuple tup_dst,
+									   HeapTuple tup_src, MemoryContext cxt);
 static HeapTuple find_target_tuple(Relation rel, ChangeDest *dest,
 								   HeapTuple tup_key,
 								   TupleTableSlot *ident_slot);
@@ -2574,6 +2579,7 @@ setup_logical_decoding(Oid relid)
 	LogicalDecodingContext *ctx;
 	NameData	slotname;
 	RepackDecodingState *dstate;
+	MemoryContext oldcxt;
 
 	/*
 	 * REPACK CONCURRENTLY is not allowed in a transaction block, so this
@@ -2624,14 +2630,6 @@ setup_logical_decoding(Oid relid)
 	 */
 	ctx->reader->routine.page_read = read_local_xlog_page_no_wait;
 
-	/*
-	 * read_local_xlog_page_no_wait() needs to be able to indicate the end of
-	 * WAL.
-	 */
-	ctx->reader->private_data = MemoryContextAllocZero(ctx->context,
-													   sizeof(ReadLocalXLogPageNoWaitPrivate));
-
-
 	/* Some WAL records should have been read. */
 	Assert(ctx->reader->EndRecPtr != InvalidXLogRecPtr);
 
@@ -2642,18 +2640,23 @@ setup_logical_decoding(Oid relid)
 	XLByteToSeg(ctx->reader->EndRecPtr, repack_current_segment,
 				wal_segment_size);
 
-	dstate = palloc0_object(RepackDecodingState);
-	dstate->relid = relid;
+	/* Our private state belongs to the decoding context. */
+	oldcxt = MemoryContextSwitchTo(ctx->context);
 
 	/*
-	 * Tuple descriptor may be needed to flatten a tuple before we write it to
-	 * a file. A copy is needed because the decoding worker invalidates system
-	 * caches before it starts to do the actual work.
+	 * read_local_xlog_page_no_wait() needs to be able to indicate the end of
+	 * WAL.
 	 */
-	rel = table_open(relid, AccessShareLock);
-	dstate->tupdesc = CreateTupleDescCopy(RelationGetDescr(rel));
+	ctx->reader->private_data = palloc0_object(ReadLocalXLogPageNoWaitPrivate);
+	dstate = palloc0_object(RepackDecodingState);
+	MemoryContextSwitchTo(oldcxt);
+
+#ifdef	USE_ASSERT_CHECKING
+	dstate->relid = relid;
+#endif
 
 	/* Avoid logical decoding of other relations. */
+	rel = table_open(relid, AccessShareLock);
 	repacked_rel_locator = rel->rd_locator;
 	toastrelid = rel->rd_rel->reltoastrelid;
 	if (OidIsValid(toastrelid))
@@ -2822,10 +2825,10 @@ static void
 apply_concurrent_changes(BufFile *file, ChangeDest *dest)
 {
 	char		kind;
-	uint32		t_len;
 	Relation	rel = dest->rel;
 	TupleTableSlot *index_slot,
 			   *ident_slot;
+	MemoryContext apply_cxt;
 	HeapTuple	tup_old = NULL;
 
 	/* TupleTableSlot is needed to pass the tuple to ExecInsertIndexTuples(). */
@@ -2835,6 +2838,14 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
 	/* A slot to fetch tuples from identity index. */
 	ident_slot = table_slot_create(rel, NULL);
 
+	/*
+	 * Use a separate memory context for the tuples and any memory needed to
+	 * process them.
+	 */
+	apply_cxt = AllocSetContextCreate(TopTransactionContext,
+									  "REPACK - apply",
+									  ALLOCSET_DEFAULT_SIZES);
+
 	while (true)
 	{
 		size_t		nread;
@@ -2849,14 +2860,7 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
 			break;
 
 		/* Read the tuple. */
-		BufFileReadExact(file, &t_len, sizeof(t_len));
-		tup = (HeapTuple) palloc(HEAPTUPLESIZE + t_len);
-		tup->t_data = (HeapTupleHeader) ((char *) tup + HEAPTUPLESIZE);
-		BufFileReadExact(file, tup->t_data, t_len);
-		tup->t_len = t_len;
-		ItemPointerSetInvalid(&tup->t_self);
-		tup->t_tableOid = RelationGetRelid(dest->rel);
-
+		tup = restore_tuple(file, rel, apply_cxt);
 		if (kind == CHANGE_UPDATE_OLD)
 		{
 			Assert(tup_old == NULL);
@@ -2868,7 +2872,7 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
 
 			apply_concurrent_insert(rel, tup, dest->iistate, index_slot);
 
-			pfree(tup);
+			MemoryContextReset(apply_cxt);
 		}
 		else if (kind == CHANGE_UPDATE_NEW || kind == CHANGE_DELETE)
 		{
@@ -2892,18 +2896,28 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
 				elog(ERROR, "failed to find target tuple");
 
 			if (kind == CHANGE_UPDATE_NEW)
+			{
+				/*
+				 * If 'tup' contains TOAST pointers, they point to the old
+				 * relation's toast. Copy the corresponding TOAST pointers for
+				 * the new relation from the existing tuple. (The fact that we
+				 * received a TOAST pointer here implies that the attribute
+				 * hasn't changed.)
+				 */
+				if (HeapTupleHasExternal(tup))
+					tup = adjust_toast_pointers(rel, tup, tup_exist,
+												apply_cxt);
+
 				apply_concurrent_update(rel, tup, tup_exist, dest->iistate,
 										index_slot);
+			}
 			else
 				apply_concurrent_delete(rel, tup_exist);
 
 			if (tup_old != NULL)
-			{
-				pfree(tup_old);
 				tup_old = NULL;
-			}
 
-			pfree(tup);
+			MemoryContextReset(apply_cxt);
 		}
 		else
 			elog(ERROR, "unrecognized kind of change: %d", kind);
@@ -2920,6 +2934,7 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
 	}
 
 	/* Cleanup. */
+	MemoryContextDelete(apply_cxt);
 	ExecDropSingleTupleTableSlot(index_slot);
 	ExecDropSingleTupleTableSlot(ident_slot);
 }
@@ -3030,6 +3045,186 @@ apply_concurrent_delete(Relation rel, HeapTuple tup_target)
 	pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_DELETED, 1);
 }
 
+/*
+ * Read tuple from file and store it in given memory context.
+ *
+ * External attributes are stored in separate memory chunks, in order to avoid
+ * exceeding MaxAllocSize - that could happen if the individual attributes are
+ * smaller than MaxAllocSize but the whole tuple is bigger.
+ */
+static HeapTuple
+restore_tuple(BufFile *file, Relation relation, MemoryContext cxt)
+{
+	uint32		t_len;
+	HeapTuple	tup;
+	MemoryContext oldcxt;
+	uint32		natt_ext;
+	List	   *attrs_ext = NIL;
+
+	oldcxt = MemoryContextSwitchTo(cxt);
+
+	/* First, read the external attributes if there are some. */
+	BufFileReadExact(file, &natt_ext, sizeof(natt_ext));
+	for (int i = 0; i < natt_ext; i++)
+	{
+		varlena		varhdr;
+		char	   *ext_val;
+		Size		ext_val_size;
+
+		/*
+		 * Read the header into properly aligned memory before accessing it.
+		 */
+		BufFileReadExact(file, &varhdr, VARHDRSZ);
+		ext_val_size = VARSIZE_ANY(&varhdr);
+		ext_val = palloc0(ext_val_size);
+		memcpy(ext_val, &varhdr, VARHDRSZ);
+		BufFileReadExact(file, ext_val + VARHDRSZ, ext_val_size - VARHDRSZ);
+
+		attrs_ext = lappend(attrs_ext, ext_val);
+	}
+
+	/* Read the tuple. */
+	BufFileReadExact(file, &t_len, sizeof(t_len));
+	tup = (HeapTuple) palloc(HEAPTUPLESIZE + t_len);
+	tup->t_data = (HeapTupleHeader) ((char *) tup + HEAPTUPLESIZE);
+	BufFileReadExact(file, tup->t_data, t_len);
+	tup->t_len = t_len;
+	ItemPointerSetInvalid(&tup->t_self);
+	tup->t_tableOid = RelationGetRelid(relation);
+
+	/*
+	 * If there are "external indirect" attributes, make the tuple reference
+	 * them.
+	 */
+	if (natt_ext > 0)
+	{
+		TupleDesc	desc;
+		Datum	   *attrs;
+		bool	   *isnull;
+		ListCell   *lc;
+
+		desc = RelationGetDescr(relation);
+		attrs = palloc0_array(Datum, desc->natts);
+		isnull = palloc0_array(bool, desc->natts);
+
+		heap_deform_tuple(tup, desc, attrs, isnull);
+
+		lc = list_head(attrs_ext);
+		for (int i = 0; i < desc->natts; i++)
+		{
+			CompactAttribute *attr = TupleDescCompactAttr(desc, i);
+			varlena    *varlena_pointer,
+					   *new_datum;
+			varatt_indirect redirect_pointer;
+
+			/* Find the external attributes, like in store_change(). */
+			if (attr->attisdropped)
+				continue;
+			if (attr->attlen != -1)
+				continue;
+			if (isnull[i])
+				continue;
+			varlena_pointer = (varlena *) DatumGetPointer(attrs[i]);
+			if (!VARATT_IS_EXTERNAL_INDIRECT(varlena_pointer))
+				continue;
+
+			/* Update the attribute. */
+			memset(&redirect_pointer, 0, sizeof(redirect_pointer));
+			redirect_pointer.pointer = lfirst(lc);
+
+			new_datum = (varlena *) palloc0(INDIRECT_POINTER_SIZE);
+			SET_VARTAG_EXTERNAL(new_datum, VARTAG_INDIRECT);
+			memcpy(VARDATA_EXTERNAL(new_datum), &redirect_pointer,
+				   sizeof(redirect_pointer));
+
+			attrs[i] = PointerGetDatum(new_datum);
+
+			lc = lnext(attrs_ext, lc);
+		}
+		Assert(lc == NULL);
+
+		tup = heap_form_tuple(desc, attrs, isnull);
+	}
+
+	MemoryContextSwitchTo(oldcxt);
+
+	return tup;
+}
+
+/*
+ * Copy TOAST pointers from 'tup_src' to 'tup_dst' and return the modified
+ * 'tup_dst'. Use 'cxt' for all memory allocations.
+ */
+static HeapTuple
+adjust_toast_pointers(Relation relation, HeapTuple tup_dst,
+					  HeapTuple tup_src, MemoryContext cxt)
+{
+	TupleDesc	desc;
+	Datum	   *attrs_dst;
+	bool	   *isnull_dst;
+	Datum	   *attrs_src = NULL;
+	bool	   *isnull_src = NULL;
+	MemoryContext oldcxt;
+	HeapTuple	result;
+
+	oldcxt = MemoryContextSwitchTo(cxt);
+
+	desc = RelationGetDescr(relation);
+	attrs_dst = palloc0_array(Datum, desc->natts);
+	isnull_dst = palloc0_array(bool, desc->natts);
+
+	heap_deform_tuple(tup_dst, desc, attrs_dst, isnull_dst);
+
+	for (int i = 0; i < desc->natts; i++)
+	{
+		CompactAttribute *attr = TupleDescCompactAttr(desc, i);
+		varlena    *varlena_dst,
+				   *varlena_src;
+
+		if (attr->attisdropped)
+			continue;
+		if (attr->attlen != -1)
+			continue;
+		if (isnull_dst[i])
+			continue;
+
+		varlena_dst = (varlena *) DatumGetPointer(attrs_dst[i]);
+		if (!VARATT_IS_EXTERNAL_ONDISK(varlena_dst))
+			continue;
+
+		/*
+		 * The first TOAST pointer? The caller can easily check if the tuple
+		 * has external attributes(s) in general, but is not supposed to check
+		 * specifically for TOAST.
+		 */
+		if (attrs_src == NULL)
+		{
+			attrs_src = palloc0_array(Datum, desc->natts);
+			isnull_src = palloc0_array(bool, desc->natts);
+
+			heap_deform_tuple(tup_src, desc, attrs_src, isnull_src);
+		}
+
+		varlena_src = (varlena *) DatumGetPointer(attrs_src[i]);
+
+		/*
+		 * Logical decoding only uses TOAST pointer for attributes that
+		 * haven't changed.
+		 */
+		Assert(!isnull_src[i]);
+		Assert(VARATT_IS_EXTERNAL_ONDISK(varlena_src));
+
+		/* Copy the pointer. */
+		memcpy(VARDATA_EXTERNAL(varlena_dst), VARDATA_EXTERNAL(varlena_src),
+			   sizeof(varatt_external));
+	}
+	result = heap_form_tuple(desc, attrs_dst, isnull_dst);
+
+	MemoryContextSwitchTo(oldcxt);
+
+	return result;
+}
+
 /*
  * Find the tuple to be updated or deleted.
  *
@@ -3263,15 +3458,8 @@ free_index_insert_state(IndexInsertState *iistate)
 static void
 cleanup_logical_decoding(LogicalDecodingContext *ctx)
 {
-	RepackDecodingState *dstate;
-
-	dstate = (RepackDecodingState *) ctx->output_writer_private;
-
-	FreeTupleDesc(dstate->tupdesc);
 	FreeDecodingContext(ctx);
-
 	ReplicationSlotDropAcquired();
-	pfree(dstate);
 }
 
 /*
diff --git a/src/backend/replication/pgoutput_repack/pgoutput_repack.c b/src/backend/replication/pgoutput_repack/pgoutput_repack.c
index d0c464a98d5..8e22aeb163b 100644
--- a/src/backend/replication/pgoutput_repack/pgoutput_repack.c
+++ b/src/backend/replication/pgoutput_repack/pgoutput_repack.c
@@ -12,7 +12,7 @@
  */
 #include "postgres.h"
 
-#include "access/heaptoast.h"
+#include "access/detoast.h"
 #include "commands/cluster.h"
 #include "replication/snapbuild.h"
 #include "utils/memutils.h"
@@ -28,7 +28,7 @@ static void plugin_commit_txn(LogicalDecodingContext *ctx,
 							  ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
 static void plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 						  Relation rel, ReorderBufferChange *change);
-static void store_change(LogicalDecodingContext *ctx,
+static void store_change(LogicalDecodingContext *ctx, Relation relation,
 						 ConcurrentChangeKind kind, HeapTuple tuple);
 
 void
@@ -97,9 +97,8 @@ plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 
 	dstate = (RepackDecodingState *) ctx->output_writer_private;
 
-	/* Only interested in one particular relation. */
-	if (relation->rd_id != dstate->relid)
-		return;
+	/* Changes of other relation should not have been decoded. */
+	Assert(RelationGetRelid(relation) == dstate->relid);
 
 	/* Decode entry depending on its type */
 	switch (change->action)
@@ -117,7 +116,7 @@ plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				if (newtuple == NULL)
 					elog(ERROR, "incomplete insert info.");
 
-				store_change(ctx, CHANGE_INSERT, newtuple);
+				store_change(ctx, relation, CHANGE_INSERT, newtuple);
 			}
 			break;
 		case REORDER_BUFFER_CHANGE_UPDATE:
@@ -132,9 +131,9 @@ plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 					elog(ERROR, "incomplete update info.");
 
 				if (oldtuple != NULL)
-					store_change(ctx, CHANGE_UPDATE_OLD, oldtuple);
+					store_change(ctx, relation, CHANGE_UPDATE_OLD, oldtuple);
 
-				store_change(ctx, CHANGE_UPDATE_NEW, newtuple);
+				store_change(ctx, relation, CHANGE_UPDATE_NEW, newtuple);
 			}
 			break;
 		case REORDER_BUFFER_CHANGE_DELETE:
@@ -146,7 +145,7 @@ plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				if (oldtuple == NULL)
 					elog(ERROR, "incomplete delete info.");
 
-				store_change(ctx, CHANGE_DELETE, oldtuple);
+				store_change(ctx, relation, CHANGE_DELETE, oldtuple);
 			}
 			break;
 		default:
@@ -164,39 +163,94 @@ plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 
 /* Store concurrent data change. */
 static void
-store_change(LogicalDecodingContext *ctx, ConcurrentChangeKind kind,
-			 HeapTuple tuple)
+store_change(LogicalDecodingContext *ctx, Relation relation,
+			 ConcurrentChangeKind kind, HeapTuple tuple)
 {
 	RepackDecodingState *dstate;
+	BufFile    *file;
 	char		kind_byte = (char) kind;
-	bool		flattened = false;
+	List	   *attrs_ext = NIL;
+	uint32		natt_ext;
 
 	dstate = (RepackDecodingState *) ctx->output_writer_private;
+	file = dstate->file;
 
 	/* Store the change kind. */
-	BufFileWrite(dstate->file, &kind_byte, 1);
+	BufFileWrite(file, &kind_byte, 1);
 
 	/*
-	 * ReorderBufferCommit() stores the TOAST chunks in its private memory
-	 * context and frees them after having called apply_change().  Therefore
-	 * we need flat copy (including TOAST) that we eventually copy into the
-	 * memory context which is available to decode_concurrent_changes().
+	 * If the tuple contains "external indirect" attributes, we need to write
+	 * the contents to the file because we have no control over that memory.
 	 */
 	if (HeapTupleHasExternal(tuple))
 	{
-		/*
-		 * toast_flatten_tuple_to_datum() might be more convenient but we
-		 * don't want the decompression it does.
-		 */
-		tuple = toast_flatten_tuple(tuple, dstate->tupdesc);
-		flattened = true;
+		TupleDesc	desc;
+		Datum	   *attrs;
+		bool	   *isnull;
+
+		desc = RelationGetDescr(relation);
+		attrs = palloc0_array(Datum, desc->natts);
+		isnull = palloc0_array(bool, desc->natts);
+
+		heap_deform_tuple(tuple, desc, attrs, isnull);
+
+		/* First, gather and count the "external indirect" attributes. */
+		for (int i = 0; i < desc->natts; i++)
+		{
+			CompactAttribute *attr = TupleDescCompactAttr(desc, i);
+			varlena    *varlena_pointer;
+
+			if (attr->attisdropped)
+				continue;
+
+			/* not a varlena datatype */
+			if (attr->attlen != -1)
+				continue;
+
+			/* no data */
+			if (isnull[i])
+				continue;
+
+			/* ok, we know we have a toast datum */
+			varlena_pointer = (varlena *) DatumGetPointer(attrs[i]);
+
+			if (!VARATT_IS_EXTERNAL(varlena_pointer))
+				continue;
+
+			if (VARATT_IS_EXTERNAL_INDIRECT(varlena_pointer))
+				attrs_ext = lappend(attrs_ext, varlena_pointer);
+			else
+			{
+				/*
+				 * Logical decoding should not produce "external expanded"
+				 * attributes (those actually should never appear on disk), so
+				 * only TOASTed attribute can be seen here.
+				 */
+				Assert(VARATT_IS_EXTERNAL_ONDISK(varlena_pointer));
+			}
+		}
+		natt_ext = list_length(attrs_ext);
 	}
-	/* Store the tuple size ... */
-	BufFileWrite(dstate->file, &tuple->t_len, sizeof(tuple->t_len));
-	/* ... and the tuple itself. */
-	BufFileWrite(dstate->file, tuple->t_data, tuple->t_len);
+	else
+		natt_ext = 0;
 
-	/* Free the flat copy if created above. */
-	if (flattened)
-		pfree(tuple);
+	/* Write the number of external attributes. */
+	BufFileWrite(file, &natt_ext, sizeof(natt_ext));
+	/* ... and the attributes themselves, if there are some. */
+	foreach_ptr(varlena, attr_val, attrs_ext)
+	{
+		varlena    *ext_val;
+		Size		ext_val_size;
+
+		ext_val = detoast_external_attr(attr_val);
+		ext_val_size = VARSIZE_ANY(ext_val);
+		BufFileWrite(file, ext_val, ext_val_size);
+		pfree(ext_val);
+	}
+	list_free(attrs_ext);
+
+	/* Finally write the tuple size ... */
+	BufFileWrite(file, &tuple->t_len, sizeof(tuple->t_len));
+	/* ... and the tuple itself. */
+	BufFileWrite(file, tuple->t_data, tuple->t_len);
 }
diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h
index 1b05d5d418b..a32813e8530 100644
--- a/src/include/commands/cluster.h
+++ b/src/include/commands/cluster.h
@@ -65,11 +65,10 @@ typedef enum
  */
 typedef struct RepackDecodingState
 {
+#ifdef	USE_ASSERT_CHECKING
 	/* The relation whose changes we're decoding. */
 	Oid			relid;
-
-	/* Tuple descriptor of the relation being processed. */
-	TupleDesc	tupdesc;
+#endif
 
 	/* The current output file. */
 	BufFile    *file;
diff --git a/src/test/modules/injection_points/expected/repack_toast.out b/src/test/modules/injection_points/expected/repack_toast.out
index 4f866a74e32..b56dde134f8 100644
--- a/src/test/modules/injection_points/expected/repack_toast.out
+++ b/src/test/modules/injection_points/expected/repack_toast.out
@@ -13,6 +13,7 @@ step change:
 	UPDATE repack_test SET j=get_long_string() where i=2;
 	DELETE FROM repack_test WHERE i=3;
 	INSERT INTO repack_test(i, j) VALUES (4, get_long_string());
+	UPDATE repack_test SET i=3 where i=1;
 
 step check2: 
 	INSERT INTO relfilenodes(node)
diff --git a/src/test/modules/injection_points/specs/repack_toast.spec b/src/test/modules/injection_points/specs/repack_toast.spec
index b48abf21450..f84917daca2 100644
--- a/src/test/modules/injection_points/specs/repack_toast.spec
+++ b/src/test/modules/injection_points/specs/repack_toast.spec
@@ -74,10 +74,17 @@ teardown
 
 session s2
 step change
+# Separately test UPDATE of both plain ("i") and TOASTed ("j") attribute. In
+# the first case, the new tuple we get from reorderbuffer.c contains "j" as a
+# TOAST pointer, wich we need to update so it points to the new heap. In the
+# latter case, we receive "j" as "external indirect" value - here we test that
+# the decoding worker writes the tuple to a file correctly and that the
+# backend executing REPACK manages to restore it.
 {
 	UPDATE repack_test SET j=get_long_string() where i=2;
 	DELETE FROM repack_test WHERE i=3;
 	INSERT INTO repack_test(i, j) VALUES (4, get_long_string());
+	UPDATE repack_test SET i=3 where i=1;
 }
 # Check the table from the perspective of s2.
 step check2
-- 
2.47.3


--=-=-=--





^ permalink  raw  reply  [nested|flat] 13+ messages in thread

end of thread, other threads:[~2026-02-27 18:01 UTC | newest]

Thread overview: 13+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2017-09-07 10:13 [PATCH 2/2] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-09-07 10:13 [PATCH 2/2] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-09-07 10:13 [PATCH 2/2] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-09-07 10:13 [PATCH 2/2] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/3] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2017-12-21 12:23 [PATCH 2/4] Add monitoring aid for max_replication_slots. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2026-02-27 18:01 [PATCH 5/5] Serialize decoded tuples without flattening. Antonin Houska <ah@cybertec.at>

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox