public inbox for [email protected]  
help / color / mirror / Atom feed
From: [email protected] <[email protected]>
To: [email protected] <[email protected]>
Subject: Failed transaction statistics to measure the logical replication progress
Date: Thu, 8 Jul 2021 06:54:45 +0000
Message-ID: <OSBPR01MB48887CA8F40C8D984A6DC00CED199@OSBPR01MB4888.jpnprd01.prod.outlook.com> (raw)

Hello, hackers


When the current HEAD fails during logical decoding, the failure
increments txns count in pg_stat_replication_slots - [1] and adds
the transaction size to the sum of bytes in the same repeatedly
on the publisher, until the problem is solved.
One of the good examples is duplication error on the subscriber side
and this applies to both streaming and spill cases as well.

This update prevents users from grasping the exact number and size of
successful and unsuccessful transactions. Accordingly, we need to
have new columns of failed transactions that will work to differentiate
both of them for all types, which means spill, streaming and normal
transactions. This will help users to measure the exact status of
logical replication.

Attached file is the POC patch for this.
Current design is to save failed stats data in the ReplicationSlot struct.
This is because after the error, I'm not able to access the ReorderBuffer object.
Thus, I chose the object where I can interact with at the ReplicationSlotRelease timing.

Below is one example that I can get on the publisher,
after the duplication error on the subscriber caused by insert is solved.

postgres=# select * from pg_stat_replication_slots;
-[ RECORD 1 ]-------+------
slot_name           | mysub
spill_txns          | 0
spill_count         | 0
spill_bytes         | 0
failed_spill_txns   | 0
failed_spill_bytes  | 0
stream_txns         | 0
stream_count        | 0
stream_bytes        | 0
failed_stream_txns  | 0
failed_stream_bytes | 0
total_txns          | 4
total_bytes         | 528
failed_total_txns   | 3
failed_total_bytes  | 396
stats_reset         | 


Any ideas and comments are welcome.

[1] - https://www.postgresql.org/docs/devel/monitoring-stats.html#MONITORING-PG-STAT-REPLICATION-SLOTS-VIE...


Best Regards,
	Takamichi Osumi



Attachments:

  [application/octet-stream] failed_transaction_stats_POC_v01.patch (17.1K, ../OSBPR01MB48887CA8F40C8D984A6DC00CED199@OSBPR01MB4888.jpnprd01.prod.outlook.com/2-failed_transaction_stats_POC_v01.patch)
  download | inline diff:
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 999d984..b35b667 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -962,11 +962,17 @@ CREATE VIEW pg_stat_replication_slots AS
             s.spill_txns,
             s.spill_count,
             s.spill_bytes,
+            s.failed_spill_txns,
+            s.failed_spill_bytes,
             s.stream_txns,
             s.stream_count,
             s.stream_bytes,
+            s.failed_stream_txns,
+            s.failed_stream_bytes,
             s.total_txns,
             s.total_bytes,
+            s.failed_total_txns,
+            s.failed_total_bytes,
             s.stats_reset
     FROM pg_replication_slots as r,
         LATERAL pg_stat_get_replication_slot(slot_name) as s
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index ce8888c..de98eb7 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -1819,11 +1819,17 @@ pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat)
 	msg.m_spill_txns = repSlotStat->spill_txns;
 	msg.m_spill_count = repSlotStat->spill_count;
 	msg.m_spill_bytes = repSlotStat->spill_bytes;
+	msg.m_failed_spill_txns = repSlotStat->failed_spill_txns;
+	msg.m_failed_spill_bytes = repSlotStat->failed_spill_bytes;
 	msg.m_stream_txns = repSlotStat->stream_txns;
 	msg.m_stream_count = repSlotStat->stream_count;
 	msg.m_stream_bytes = repSlotStat->stream_bytes;
+	msg.m_failed_stream_txns = repSlotStat->failed_stream_txns;
+	msg.m_failed_stream_bytes = repSlotStat->failed_stream_bytes;
 	msg.m_total_txns = repSlotStat->total_txns;
 	msg.m_total_bytes = repSlotStat->total_bytes;
+	msg.m_failed_total_txns = repSlotStat->failed_total_txns;
+	msg.m_failed_total_bytes = repSlotStat->failed_total_bytes;
 	pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
 }
 
@@ -5509,11 +5515,17 @@ pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len)
 			slotent->spill_txns += msg->m_spill_txns;
 			slotent->spill_count += msg->m_spill_count;
 			slotent->spill_bytes += msg->m_spill_bytes;
+			slotent->failed_spill_txns += msg->m_failed_spill_txns;
+			slotent->failed_spill_bytes += msg->m_failed_spill_bytes;
 			slotent->stream_txns += msg->m_stream_txns;
 			slotent->stream_count += msg->m_stream_count;
 			slotent->stream_bytes += msg->m_stream_bytes;
+			slotent->failed_stream_txns += msg->m_failed_stream_txns;
+			slotent->failed_stream_bytes += msg->m_failed_stream_bytes;
 			slotent->total_txns += msg->m_total_txns;
 			slotent->total_bytes += msg->m_total_bytes;
+			slotent->failed_total_txns += msg->m_failed_total_txns;
+			slotent->failed_total_bytes += msg->m_failed_total_bytes;
 		}
 	}
 }
@@ -5760,11 +5772,17 @@ pgstat_reset_replslot(PgStat_StatReplSlotEntry *slotent, TimestampTz ts)
 	slotent->spill_txns = 0;
 	slotent->spill_count = 0;
 	slotent->spill_bytes = 0;
+	slotent->failed_spill_txns = 0;
+	slotent->failed_spill_bytes = 0;
 	slotent->stream_txns = 0;
 	slotent->stream_count = 0;
 	slotent->stream_bytes = 0;
+	slotent->failed_stream_txns = 0;
+	slotent->failed_stream_bytes = 0;
 	slotent->total_txns = 0;
 	slotent->total_bytes = 0;
+	slotent->failed_total_txns = 0;
+	slotent->failed_total_bytes = 0;
 	slotent->stat_reset_timestamp = ts;
 }
 
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index d536a5f..159223f 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -196,6 +196,12 @@ StartupDecodingContext(List *output_plugin_options,
 		LWLockRelease(ProcArrayLock);
 	}
 
+	slot->failedSpillTxns = 0;
+	slot->failedSpillBytes = 0;
+	slot->failedStreamTxns = 0;
+	slot->failedStreamBytes = 0;
+	slot->failedTotalTxns = 0;
+	slot->failedTotalBytes = 0;
 	ctx->slot = slot;
 
 	ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx);
@@ -1794,11 +1800,17 @@ UpdateDecodingStats(LogicalDecodingContext *ctx)
 	repSlotStat.spill_txns = rb->spillTxns;
 	repSlotStat.spill_count = rb->spillCount;
 	repSlotStat.spill_bytes = rb->spillBytes;
+	repSlotStat.failed_spill_txns = 0;
+	repSlotStat.failed_spill_bytes = 0;
 	repSlotStat.stream_txns = rb->streamTxns;
 	repSlotStat.stream_count = rb->streamCount;
 	repSlotStat.stream_bytes = rb->streamBytes;
+	repSlotStat.failed_stream_txns = 0;
+	repSlotStat.failed_stream_bytes = 0;
 	repSlotStat.total_txns = rb->totalTxns;
 	repSlotStat.total_bytes = rb->totalBytes;
+	repSlotStat.failed_total_txns = 0;
+	repSlotStat.failed_total_bytes = 0;
 
 	pgstat_report_replslot(&repSlotStat);
 
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index b8c5e2a..ca35c42 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2051,6 +2051,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	PG_TRY();
 	{
 		ReorderBufferChange *change;
+		ReplicationSlot *slot = MyReplicationSlot;
 
 		if (using_subtxn)
 			BeginInternalSubTransaction(streaming ? "stream" : "replay");
@@ -2409,9 +2410,13 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		 * which we have already accounted in ReorderBufferIterTXNNext.
 		 */
 		if (!rbtxn_is_streamed(txn))
+		{
 			rb->totalTxns++;
+			slot->failedTotalTxns++;
+		}
 
 		rb->totalBytes += txn->total_size;
+		slot->failedTotalBytes += txn->total_size;
 
 		/*
 		 * Done with current changes, send the last message for this set of
@@ -3582,11 +3587,17 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	/* update the statistics iff we have spilled anything */
 	if (spilled)
 	{
+		ReplicationSlot *slot = MyReplicationSlot;
+		int spill_txn;
+
 		rb->spillCount += 1;
 		rb->spillBytes += size;
+		slot->failedSpillBytes += size;
 
 		/* don't consider already serialized transactions */
-		rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
+		spill_txn = (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
+		rb->spillTxns += spill_txn;
+		slot->failedSpillTxns += spill_txn;
 
 		/* update the decoding stats */
 		UpdateDecodingStats((LogicalDecodingContext *) rb->private_data);
@@ -3861,6 +3872,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	CommandId	command_id;
 	Size		stream_bytes;
 	bool		txn_is_streamed;
+	ReplicationSlot *slot = MyReplicationSlot;
 
 	/* We can never reach here for a subtransaction. */
 	Assert(txn->toptxn == NULL);
@@ -3956,9 +3968,11 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 
 	rb->streamCount += 1;
 	rb->streamBytes += stream_bytes;
+	slot->failedStreamBytes += stream_bytes;
 
 	/* Don't consider already streamed transaction. */
 	rb->streamTxns += (txn_is_streamed) ? 0 : 1;
+	slot->failedStreamTxns += (txn_is_streamed) ? 0 : 1;
 
 	/* update the decoding stats */
 	UpdateDecodingStats((LogicalDecodingContext *) rb->private_data);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 8c18b4e..ac5e953 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -44,6 +44,7 @@
 #include "common/string.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/logical.h"
 #include "replication/slot.h"
 #include "storage/fd.h"
 #include "storage/proc.h"
@@ -471,6 +472,11 @@ ReplicationSlotRelease(void)
 
 	Assert(slot != NULL && slot->active_pid != 0);
 
+	if (SlotIsLogical(slot))
+	{
+		UpdateDecodingFailedStats();
+	}
+
 	if (slot->data.persistency == RS_EPHEMERAL)
 	{
 		/*
@@ -1839,3 +1845,40 @@ RestoreSlotFromDisk(const char *name)
 				(errmsg("too many replication slots active before shutdown"),
 				 errhint("Increase max_replication_slots and try again.")));
 }
+
+/*
+ * Report failed stats for a slot.
+ */
+void
+UpdateDecodingFailedStats()
+{
+	PgStat_StatReplSlotEntry repSlotStat;
+	ReplicationSlot *slot = MyReplicationSlot;
+
+	Assert(MyReplicationSlot != NULL);
+
+	namestrcpy(&repSlotStat.slotname, NameStr(slot->data.name));
+	repSlotStat.spill_txns = 0;
+	repSlotStat.spill_count = 0;
+	repSlotStat.spill_bytes = 0;
+	repSlotStat.failed_spill_txns = slot->failedSpillTxns;
+	repSlotStat.failed_spill_bytes = slot->failedSpillBytes;
+	repSlotStat.stream_txns = 0;
+	repSlotStat.stream_count = 0;
+	repSlotStat.stream_bytes = 0;
+	repSlotStat.failed_stream_txns = slot->failedStreamTxns;
+	repSlotStat.failed_stream_bytes = slot->failedStreamBytes;
+	repSlotStat.total_txns = 0;
+	repSlotStat.total_bytes = 0;
+	repSlotStat.failed_total_txns = slot->failedTotalTxns;
+	repSlotStat.failed_total_bytes = slot->failedTotalBytes;
+
+	pgstat_report_replslot(&repSlotStat);
+
+	slot->failedSpillTxns = 0;
+	slot->failedSpillBytes = 0;
+	slot->failedStreamTxns = 0;
+	slot->failedStreamBytes = 0;
+	slot->failedTotalTxns = 0;
+	slot->failedTotalBytes = 0;
+}
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index f0e09ea..f518964 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -2313,7 +2313,7 @@ pg_stat_get_archiver(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_REPLICATION_SLOT_COLS 10
+#define PG_STAT_GET_REPLICATION_SLOT_COLS 16
 	text	   *slotname_text = PG_GETARG_TEXT_P(0);
 	NameData	slotname;
 	TupleDesc	tupdesc;
@@ -2336,17 +2336,29 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 					   INT8OID, -1, 0);
 	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "spill_bytes",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "stream_txns",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "failed_spill_txns",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "stream_count",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "failed_spill_bytes",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_bytes",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_txns",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "total_txns",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "stream_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "total_bytes",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "stream_bytes",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "stats_reset",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "failed_stream_txns",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "failed_stream_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 12, "total_txns",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 13, "total_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 14, "failed_total_txns",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 15, "failed_total_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 16, "stats_reset",
 					   TIMESTAMPTZOID, -1, 0);
 	BlessTupleDesc(tupdesc);
 
@@ -2366,16 +2378,22 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 	values[1] = Int64GetDatum(slotent->spill_txns);
 	values[2] = Int64GetDatum(slotent->spill_count);
 	values[3] = Int64GetDatum(slotent->spill_bytes);
-	values[4] = Int64GetDatum(slotent->stream_txns);
-	values[5] = Int64GetDatum(slotent->stream_count);
-	values[6] = Int64GetDatum(slotent->stream_bytes);
-	values[7] = Int64GetDatum(slotent->total_txns);
-	values[8] = Int64GetDatum(slotent->total_bytes);
+	values[4] = Int64GetDatum(slotent->failed_spill_txns);
+	values[5] = Int64GetDatum(slotent->failed_spill_bytes);
+	values[6] = Int64GetDatum(slotent->stream_txns);
+	values[7] = Int64GetDatum(slotent->stream_count);
+	values[8] = Int64GetDatum(slotent->stream_bytes);
+	values[9] = Int64GetDatum(slotent->failed_stream_txns);
+	values[10] = Int64GetDatum(slotent->failed_stream_bytes);
+	values[11] = Int64GetDatum(slotent->total_txns);
+	values[12] = Int64GetDatum(slotent->total_bytes);
+	values[13] = Int64GetDatum(slotent->failed_total_txns);
+	values[14] = Int64GetDatum(slotent->failed_total_bytes);
 
 	if (slotent->stat_reset_timestamp == 0)
-		nulls[9] = true;
+		nulls[15] = true;
 	else
-		values[9] = TimestampTzGetDatum(slotent->stat_reset_timestamp);
+		values[15] = TimestampTzGetDatum(slotent->stat_reset_timestamp);
 
 	/* Returns the record as Datum */
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fde251f..11cf01b 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5313,9 +5313,9 @@
   proname => 'pg_stat_get_replication_slot', prorows => '1', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'text',
-  proallargtypes => '{text,text,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{slot_name,slot_name,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes,total_txns,total_bytes,stats_reset}',
+  proallargtypes => '{text,text,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{slot_name,slot_name,spill_txns,spill_count,spill_bytes,failed_spill_txns,failed_spill_bytes,stream_txns,stream_count,stream_bytes,failed_stream_txns,failed_stream_bytes,total_txns,total_bytes,failed_total_txns,failed_total_bytes,stats_reset}',
   prosrc => 'pg_stat_get_replication_slot' },
 { oid => '6118', descr => 'statistics: information about subscription',
   proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 9612c0a..5064408 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -532,11 +532,17 @@ typedef struct PgStat_MsgReplSlot
 	PgStat_Counter m_spill_txns;
 	PgStat_Counter m_spill_count;
 	PgStat_Counter m_spill_bytes;
+	PgStat_Counter m_failed_spill_txns;
+	PgStat_Counter m_failed_spill_bytes;
 	PgStat_Counter m_stream_txns;
 	PgStat_Counter m_stream_count;
 	PgStat_Counter m_stream_bytes;
+	PgStat_Counter m_failed_stream_txns;
+	PgStat_Counter m_failed_stream_bytes;
 	PgStat_Counter m_total_txns;
 	PgStat_Counter m_total_bytes;
+	PgStat_Counter m_failed_total_txns;
+	PgStat_Counter m_failed_total_bytes;
 } PgStat_MsgReplSlot;
 
 
@@ -900,11 +906,17 @@ typedef struct PgStat_StatReplSlotEntry
 	PgStat_Counter spill_txns;
 	PgStat_Counter spill_count;
 	PgStat_Counter spill_bytes;
+	PgStat_Counter failed_spill_txns;
+	PgStat_Counter failed_spill_bytes;
 	PgStat_Counter stream_txns;
 	PgStat_Counter stream_count;
 	PgStat_Counter stream_bytes;
+	PgStat_Counter failed_stream_txns;
+	PgStat_Counter failed_stream_bytes;
 	PgStat_Counter total_txns;
 	PgStat_Counter total_bytes;
+	PgStat_Counter failed_total_txns;
+	PgStat_Counter failed_total_bytes;
 	TimestampTz stat_reset_timestamp;
 } PgStat_StatReplSlotEntry;
 
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 2eb7e3a..8d0bef6 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -164,6 +164,20 @@ typedef struct ReplicationSlot
 	XLogRecPtr	candidate_xmin_lsn;
 	XLogRecPtr	candidate_restart_valid;
 	XLogRecPtr	candidate_restart_lsn;
+
+	/*
+	 * Statistics about failed transactions.
+	 *
+	 * These failed statistics works to make a distinction between
+	 * successful and unsuccessful transactions, by utilizing other
+	 * corresponding stats of transactions respectively.
+	 */
+	int64       failedSpillTxns;
+	int64       failedSpillBytes;
+	int64       failedStreamTxns;
+	int64       failedStreamBytes;
+	int64       failedTotalTxns;
+	int64       failedTotalBytes;
 } ReplicationSlot;
 
 #define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid)
@@ -223,5 +237,6 @@ extern void StartupReplicationSlots(void);
 extern void CheckPointReplicationSlots(void);
 
 extern void CheckSlotRequirements(void);
+extern void UpdateDecodingFailedStats(void);
 
 #endif							/* SLOT_H */
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index e5ab112..41724bb 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2054,14 +2054,20 @@ pg_stat_replication_slots| SELECT s.slot_name,
     s.spill_txns,
     s.spill_count,
     s.spill_bytes,
+    s.failed_spill_txns,
+    s.failed_spill_bytes,
     s.stream_txns,
     s.stream_count,
     s.stream_bytes,
+    s.failed_stream_txns,
+    s.failed_stream_bytes,
     s.total_txns,
     s.total_bytes,
+    s.failed_total_txns,
+    s.failed_total_bytes,
     s.stats_reset
    FROM pg_replication_slots r,
-    LATERAL pg_stat_get_replication_slot((r.slot_name)::text) s(slot_name, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes, total_txns, total_bytes, stats_reset)
+    LATERAL pg_stat_get_replication_slot((r.slot_name)::text) s(slot_name, spill_txns, spill_count, spill_bytes, failed_spill_txns, failed_spill_bytes, stream_txns, stream_count, stream_bytes, failed_stream_txns, failed_stream_bytes, total_txns, total_bytes, failed_total_txns, failed_total_bytes, stats_reset)
   WHERE (r.datoid IS NOT NULL);
 pg_stat_slru| SELECT s.name,
     s.blks_zeroed,


view thread (113+ messages)  latest in thread

reply

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Reply to all the recipients using the --to and --cc options:
  reply via email

  To: [email protected]
  Cc: [email protected], [email protected]
  Subject: Re: Failed transaction statistics to measure the logical replication progress
  In-Reply-To: <OSBPR01MB48887CA8F40C8D984A6DC00CED199@OSBPR01MB4888.jpnprd01.prod.outlook.com>

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

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