From e6892c7ef32cd79780808517a4bd21e392d905b8 Mon Sep 17 00:00:00 2001
From: Julien Tachoires <julmon@gmail.com>
Date: Thu, 24 Oct 2024 12:55:09 +0200
Subject: [PATCH 5/6] Track data size written to disk during logical decoding

The pg_stat_replication_slots view now exposes a new counter named
"spill_write_bytes". This counter is in charge of tracking the amount
of data written to disk during logical decoding. This counter differs
from "spill_bytes" as the data can be compressed before being written
to disk.

When data compression is not enabled (default), "spill_bytes" and
"spill_write_bytes" are equal.
---
 contrib/test_decoding/expected/stats.out      | 12 +++----
 doc/src/sgml/monitoring.sgml                  | 22 ++++++++++---
 src/backend/catalog/system_views.sql          |  1 +
 src/backend/replication/logical/logical.c     |  5 ++-
 .../replication/logical/reorderbuffer.c       | 11 +++++--
 src/backend/utils/activity/pgstat_replslot.c  |  1 +
 src/backend/utils/adt/pgstatfuncs.c           | 31 ++++++++++---------
 src/include/catalog/pg_proc.dat               |  6 ++--
 src/include/pgstat.h                          |  1 +
 src/include/replication/reorderbuffer.h       |  4 ++-
 src/test/regress/expected/rules.out           |  3 +-
 11 files changed, 64 insertions(+), 33 deletions(-)

diff --git a/contrib/test_decoding/expected/stats.out b/contrib/test_decoding/expected/stats.out
index 78d36429c8..caad65b64c 100644
--- a/contrib/test_decoding/expected/stats.out
+++ b/contrib/test_decoding/expected/stats.out
@@ -78,17 +78,17 @@ SELECT slot_name, spill_txns = 0 AS spill_txns, spill_count = 0 AS spill_count,
 
 -- verify accessing/resetting stats for non-existent slot does something reasonable
 SELECT * FROM pg_stat_get_replication_slot('do-not-exist');
-  slot_name   | spill_txns | spill_count | spill_bytes | stream_txns | stream_count | stream_bytes | total_txns | total_bytes | stats_reset 
---------------+------------+-------------+-------------+-------------+--------------+--------------+------------+-------------+-------------
- do-not-exist |          0 |           0 |           0 |           0 |            0 |            0 |          0 |           0 | 
+  slot_name   | spill_txns | spill_count | spill_bytes | spill_write_bytes | stream_txns | stream_count | stream_bytes | total_txns | total_bytes | stats_reset 
+--------------+------------+-------------+-------------+-------------------+-------------+--------------+--------------+------------+-------------+-------------
+ do-not-exist |          0 |           0 |           0 |                 0 |           0 |            0 |            0 |          0 |           0 | 
 (1 row)
 
 SELECT pg_stat_reset_replication_slot('do-not-exist');
 ERROR:  replication slot "do-not-exist" does not exist
 SELECT * FROM pg_stat_get_replication_slot('do-not-exist');
-  slot_name   | spill_txns | spill_count | spill_bytes | stream_txns | stream_count | stream_bytes | total_txns | total_bytes | stats_reset 
---------------+------------+-------------+-------------+-------------+--------------+--------------+------------+-------------+-------------
- do-not-exist |          0 |           0 |           0 |           0 |            0 |            0 |          0 |           0 | 
+  slot_name   | spill_txns | spill_count | spill_bytes | spill_write_bytes | stream_txns | stream_count | stream_bytes | total_txns | total_bytes | stats_reset 
+--------------+------------+-------------+-------------+-------------------+-------------+--------------+--------------+------------+-------------+-------------
+ do-not-exist |          0 |           0 |           0 |                 0 |           0 |            0 |            0 |          0 |           0 | 
 (1 row)
 
 -- spilling the xact
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 331315f8d3..d251e4cfa8 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -1561,10 +1561,24 @@ description | Waiting for a newly initialized WAL file to reach durable storage
         <structfield>spill_bytes</structfield> <type>bigint</type>
        </para>
        <para>
-        Amount of decoded transaction data spilled to disk while performing
-        decoding of changes from WAL for this slot. This and other spill
-        counters can be used to gauge the I/O which occurred during logical
-        decoding and allow tuning <literal>logical_decoding_work_mem</literal>.
+        Amount of decoded transaction data requiring to be eventually compressed
+        and spilled to disk while performing decoding of changes from WAL for
+        this slot. This and other spill counters can be used to gauge the I/O
+        which occurred during logical decoding and allow tuning
+        <literal>logical_decoding_work_mem</literal>.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+        <structfield>spill_write_bytes</structfield> <type>bigint</type>
+       </para>
+       <para>
+        Amount of decoded transaction data eventually compressed and spilled
+        to disk while performing decoding of changes from WAL for this slot.
+        When using data compression, <literal>spill_write_bytes</literal>
+        should be significantly smaller than <literal>spill_bytes</literal>
+        which represents uncompressed data size.
       </para></entry>
      </row>
 
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index a087223677..7ab046cb70 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1030,6 +1030,7 @@ CREATE VIEW pg_stat_replication_slots AS
             s.spill_txns,
             s.spill_count,
             s.spill_bytes,
+            s.spill_write_bytes,
             s.stream_txns,
             s.stream_count,
             s.stream_bytes,
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 54fbbe6fea..951318a994 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -1942,11 +1942,12 @@ UpdateDecodingStats(LogicalDecodingContext *ctx)
 	if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && rb->totalBytes <= 0)
 		return;
 
-	elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld",
+	elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld %lld",
 		 rb,
 		 (long long) rb->spillTxns,
 		 (long long) rb->spillCount,
 		 (long long) rb->spillBytes,
+		 (long long) rb->spillWriteBytes,
 		 (long long) rb->streamTxns,
 		 (long long) rb->streamCount,
 		 (long long) rb->streamBytes,
@@ -1956,6 +1957,7 @@ UpdateDecodingStats(LogicalDecodingContext *ctx)
 	repSlotStat.spill_txns = rb->spillTxns;
 	repSlotStat.spill_count = rb->spillCount;
 	repSlotStat.spill_bytes = rb->spillBytes;
+	repSlotStat.spill_write_bytes = rb->spillWriteBytes;
 	repSlotStat.stream_txns = rb->streamTxns;
 	repSlotStat.stream_count = rb->streamCount;
 	repSlotStat.stream_bytes = rb->streamBytes;
@@ -1967,6 +1969,7 @@ UpdateDecodingStats(LogicalDecodingContext *ctx)
 	rb->spillTxns = 0;
 	rb->spillCount = 0;
 	rb->spillBytes = 0;
+	rb->spillWriteBytes = 0;
 	rb->streamTxns = 0;
 	rb->streamCount = 0;
 	rb->streamBytes = 0;
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 9407466393..2b8839a96d 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -257,7 +257,7 @@ static void ReorderBufferExecuteInvalidations(uint32 nmsgs, SharedInvalidationMe
  */
 static void ReorderBufferCheckMemoryLimit(ReorderBuffer *rb);
 static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
-static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
+static Size ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 										 int fd, ReorderBufferChange *change);
 static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 										TXNEntryFile *file, XLogSegNo *segno);
@@ -3756,6 +3756,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	XLogSegNo	curOpenSegNo = 0;
 	Size		spilled = 0;
 	Size		size = txn->size;
+	Size		written = 0;
 
 	elog(DEBUG2, "spill %u changes in XID %u to disk",
 		 (uint32) txn->nentries_mem, txn->xid);
@@ -3807,7 +3808,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 						 errmsg("could not open file \"%s\": %m", path)));
 		}
 
-		ReorderBufferSerializeChange(rb, txn, fd, change);
+		written += ReorderBufferSerializeChange(rb, txn, fd, change);
 		dlist_delete(&change->node);
 		ReorderBufferReturnChange(rb, change, false);
 
@@ -3822,6 +3823,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	{
 		rb->spillCount += 1;
 		rb->spillBytes += size;
+		rb->spillWriteBytes += written;
 
 		/* don't consider already serialized transactions */
 		rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
@@ -3842,7 +3844,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 /*
  * Serialize individual change to disk.
  */
-static void
+static Size
 ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 							 int fd, ReorderBufferChange *change)
 {
@@ -4054,6 +4056,9 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	 */
 	if (txn->final_lsn < change->lsn)
 		txn->final_lsn = change->lsn;
+
+	/* Return the size of data written on disk */
+	return ondisk->size - sizeof(ReorderBufferDiskChange);
 }
 
 /* Returns true, if the output plugin supports streaming, false, otherwise. */
diff --git a/src/backend/utils/activity/pgstat_replslot.c b/src/backend/utils/activity/pgstat_replslot.c
index ddf2ab9928..66d6e36ebf 100644
--- a/src/backend/utils/activity/pgstat_replslot.c
+++ b/src/backend/utils/activity/pgstat_replslot.c
@@ -91,6 +91,7 @@ pgstat_report_replslot(ReplicationSlot *slot, const PgStat_StatReplSlotEntry *re
 	REPLSLOT_ACC(spill_txns);
 	REPLSLOT_ACC(spill_count);
 	REPLSLOT_ACC(spill_bytes);
+	REPLSLOT_ACC(spill_write_bytes);
 	REPLSLOT_ACC(stream_txns);
 	REPLSLOT_ACC(stream_count);
 	REPLSLOT_ACC(stream_bytes);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index f7b50e0b5a..f7488b2a58 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1907,7 +1907,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 11
 	text	   *slotname_text = PG_GETARG_TEXT_P(0);
 	NameData	slotname;
 	TupleDesc	tupdesc;
@@ -1926,17 +1926,19 @@ 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, "spill_write_bytes",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "stream_count",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "stream_txns",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_bytes",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "total_txns",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "stream_bytes",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "total_bytes",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "total_txns",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "stats_reset",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "total_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "stats_reset",
 					   TIMESTAMPTZOID, -1, 0);
 	BlessTupleDesc(tupdesc);
 
@@ -1956,16 +1958,17 @@ 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->spill_write_bytes);
+	values[5] = Int64GetDatum(slotent->stream_txns);
+	values[6] = Int64GetDatum(slotent->stream_count);
+	values[7] = Int64GetDatum(slotent->stream_bytes);
+	values[8] = Int64GetDatum(slotent->total_txns);
+	values[9] = Int64GetDatum(slotent->total_bytes);
 
 	if (slotent->stat_reset_timestamp == 0)
-		nulls[9] = true;
+		nulls[10] = true;
 	else
-		values[9] = TimestampTzGetDatum(slotent->stat_reset_timestamp);
+		values[10] = 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 1ec0d6f6b5..6b95dbeebf 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5584,9 +5584,9 @@
 { oid => '6169', descr => 'statistics: information about replication slot',
   proname => 'pg_stat_get_replication_slot', 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,timestamptz}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{slot_name,slot_name,spill_txns,spill_count,spill_bytes,spill_write_bytes,stream_txns,stream_count,stream_bytes,total_txns,total_bytes,stats_reset}',
   prosrc => 'pg_stat_get_replication_slot' },
 
 { oid => '6230', descr => 'statistics: check if a stats object exists',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index df53fa2d4f..733cd74f86 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -403,6 +403,7 @@ typedef struct PgStat_StatReplSlotEntry
 	PgStat_Counter spill_txns;
 	PgStat_Counter spill_count;
 	PgStat_Counter spill_bytes;
+	PgStat_Counter spill_write_bytes;
 	PgStat_Counter stream_txns;
 	PgStat_Counter stream_count;
 	PgStat_Counter stream_bytes;
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 5f231b5f90..2b585db7c5 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -659,7 +659,9 @@ struct ReorderBuffer
 	 */
 	int64		spillTxns;		/* number of transactions spilled to disk */
 	int64		spillCount;		/* spill-to-disk invocation counter */
-	int64		spillBytes;		/* amount of data spilled to disk */
+	int64		spillBytes;		/* amount of data spilled to disk, before
+								 * compression */
+	int64		spillWriteBytes;	/* amount of data actually written to disk */
 
 	/* Statistics about transactions streamed to the decoding output plugin */
 	int64		streamTxns;		/* number of transactions streamed */
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2b47013f11..26ff02f89d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2095,6 +2095,7 @@ pg_stat_replication_slots| SELECT s.slot_name,
     s.spill_txns,
     s.spill_count,
     s.spill_bytes,
+    s.spill_write_bytes,
     s.stream_txns,
     s.stream_count,
     s.stream_bytes,
@@ -2102,7 +2103,7 @@ pg_stat_replication_slots| SELECT s.slot_name,
     s.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, spill_write_bytes, stream_txns, stream_count, stream_bytes, total_txns, total_bytes, stats_reset)
   WHERE (r.datoid IS NOT NULL);
 pg_stat_slru| SELECT name,
     blks_zeroed,
-- 
2.43.0

