0010-Track-statistics-for-streaming.patch

From bf0e4ba0c0441cbbf52989f35b49fdbf7da01fdc Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilip.kumar@enterprisedb.com>
Date: Mon, 18 Nov 2019 12:19:49 +0530
Subject: [PATCH 10/13] Track statistics for streaming

---
 doc/src/sgml/monitoring.sgml                    | 25 +++++++++++++++++++++
 src/backend/catalog/system_views.sql            |  5 ++++-
 src/backend/replication/logical/reorderbuffer.c | 13 +++++++++++
 src/backend/replication/walsender.c             | 30 ++++++++++++++++++++-----
 src/include/catalog/pg_proc.dat                 |  6 ++---
 src/include/replication/reorderbuffer.h         | 13 +++++++----
 src/include/replication/walsender_private.h     |  5 +++++
 src/test/regress/expected/rules.out             |  7 ++++--
 8 files changed, 89 insertions(+), 15 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index a3c5f86..3de62c0 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -1992,6 +1992,31 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
       may get spilled repeatedly, and this counter gets incremented on every
       such invocation.</entry>
     </row>
+    <row>
+     <entry><structfield>stream_txns</structfield></entry>
+     <entry><type>integer</type></entry>
+     <entry>Number of in-progress transactions streamed to subscriber after
+      memory used by logical decoding exceeds <literal>logical_work_mem</literal>.
+      Streaming only works with toplevel transactions (subtransactions can't
+      be streamed independently), so the counter does not get incremented for
+      subtransactions.
+      </entry>
+    </row>
+    <row>
+     <entry><structfield>stream_count</structfield></entry>
+     <entry><type>integer</type></entry>
+     <entry>Number of times in-progress transactions were streamed to subscriber.
+      Transactions may get streamed repeatedly, and this counter gets incremented
+      on every such invocation.
+      </entry>
+    </row>
+    <row>
+     <entry><structfield>stream_bytes</structfield></entry>
+     <entry><type>integer</type></entry>
+     <entry>Amount of decoded in-progress transaction data streamed to subscriber.
+      </entry>
+    </row>
+
    </tbody>
    </tgroup>
   </table>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index f7800f0..5897611 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -779,7 +779,10 @@ CREATE VIEW pg_stat_replication AS
             W.reply_time,
             W.spill_txns,
             W.spill_count,
-            W.spill_bytes
+            W.spill_bytes,
+            W.stream_txns,
+            W.stream_count,
+            W.stream_bytes
     FROM pg_stat_get_activity(NULL) AS S
         JOIN pg_stat_get_wal_senders() AS W ON (S.pid = W.pid)
         LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid);
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 0ab3191..83eb4df 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -358,6 +358,10 @@ ReorderBufferAllocate(void)
 	buffer->spillTxns = 0;
 	buffer->spillBytes = 0;
 
+	buffer->streamCount = 0;
+	buffer->streamTxns = 0;
+	buffer->streamBytes = 0;
+
 	buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;
 
 	dlist_init(&buffer->toplevel_by_lsn);
@@ -3732,6 +3736,15 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	PG_END_TRY();
 
 	/*
+	 * Update the stream statistics.
+	 */
+	rb->streamCount += 1;
+	rb->streamBytes += txn->size;
+
+	/* Don't consider already streamed transaction. */
+	rb->streamTxns += (rbtxn_is_streamed(txn)) ? 0 : 1;
+
+	/*
 	 * Discard the changes that we just streamed, and mark the transactions
 	 * as streamed (if they contained changes).
 	 */
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index a2ae283..8619a01 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1269,7 +1269,7 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
  * LogicalDecodingContext 'update_progress' callback.
  *
  * Write the current position to the lag tracker (see XLogSendPhysical),
- * and update the spill statistics.
+ * and update the spill/stream statistics.
  */
 static void
 WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid)
@@ -1290,7 +1290,8 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
 	sendTime = now;
 
 	/*
-	 * Update statistics about transactions that spilled to disk.
+	 * Update statistics about transactions that spilled to disk or streamed to
+	 * subscriber (before being committed).
 	 */
 	UpdateSpillStats(ctx);
 }
@@ -2334,6 +2335,9 @@ InitWalSenderSlot(void)
 			walsnd->spillTxns = 0;
 			walsnd->spillCount = 0;
 			walsnd->spillBytes = 0;
+			walsnd->streamTxns = 0;
+			walsnd->streamCount = 0;
+			walsnd->streamBytes = 0;
 			SpinLockRelease(&walsnd->mutex);
 			/* don't need the lock anymore */
 			MyWalSnd = (WalSnd *) walsnd;
@@ -3235,7 +3239,7 @@ offset_to_interval(TimeOffset offset)
 Datum
 pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_WAL_SENDERS_COLS	15
+#define PG_STAT_GET_WAL_SENDERS_COLS	18
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
 	Tuplestorestate *tupstore;
@@ -3293,6 +3297,9 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		int64		spillTxns;
 		int64		spillCount;
 		int64		spillBytes;
+		int64		streamTxns;
+		int64		streamCount;
+		int64		streamBytes;
 		Datum		values[PG_STAT_GET_WAL_SENDERS_COLS];
 		bool		nulls[PG_STAT_GET_WAL_SENDERS_COLS];
 
@@ -3316,6 +3323,9 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		spillTxns = walsnd->spillTxns;
 		spillCount = walsnd->spillCount;
 		spillBytes = walsnd->spillBytes;
+		streamTxns = walsnd->streamTxns;
+		streamCount = walsnd->streamCount;
+		streamBytes = walsnd->streamBytes;
 		SpinLockRelease(&walsnd->mutex);
 
 		memset(nulls, 0, sizeof(nulls));
@@ -3402,6 +3412,11 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 			values[12] = Int64GetDatum(spillTxns);
 			values[13] = Int64GetDatum(spillCount);
 			values[14] = Int64GetDatum(spillBytes);
+
+			/* stream over-sized transactions */
+			values[15] = Int64GetDatum(streamTxns);
+			values[16] = Int64GetDatum(streamCount);
+			values[17] = Int64GetDatum(streamBytes);
 		}
 
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
@@ -3650,8 +3665,13 @@ UpdateSpillStats(LogicalDecodingContext *ctx)
 	MyWalSnd->spillCount = rb->spillCount;
 	MyWalSnd->spillBytes = rb->spillBytes;
 
-	elog(DEBUG2, "UpdateSpillStats: updating stats %p %ld %ld %ld",
-		 rb, rb->spillTxns, rb->spillCount, rb->spillBytes);
+	MyWalSnd->streamTxns = rb->streamTxns;
+	MyWalSnd->streamCount = rb->streamCount;
+	MyWalSnd->streamBytes = rb->streamBytes;
+
+	elog(DEBUG2, "UpdateSpillStats: updating stats %p %ld %ld %ld %ld %ld %ld",
+		 rb, rb->spillTxns, rb->spillCount, rb->spillBytes,
+		 rb->streamTxns, rb->streamCount, rb->streamBytes);
 
 	SpinLockRelease(&MyWalSnd->mutex);
 }
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fa0a2a1..9a508bf 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5166,9 +5166,9 @@
   proname => 'pg_stat_get_wal_senders', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => '',
-  proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,int8,int8}',
-  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time,spill_txns,spill_count,spill_bytes}',
+  proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,int8,int8,int8,int8,int8}',
+  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes}',
   prosrc => 'pg_stat_get_wal_senders' },
 { oid => '3317', descr => 'statistics: information about WAL receiver',
   proname => 'pg_stat_get_wal_receiver', proisstrict => 'f', provolatile => 's',
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 7d08e2f..c183fce 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -511,15 +511,20 @@ struct ReorderBuffer
 	Size		size;
 
 	/*
-	 * Statistics about transactions spilled to disk.
+	 * Statistics about transactions streamed or spilled to disk.
 	 *
-	 * A single transaction may be spilled repeatedly, which is why we keep
-	 * two different counters. For spilling, the transaction counter includes
-	 * both toplevel transactions and subtransactions.
+	 * A single transaction may be streamed/spilled repeatedly, which is
+	 * why we keep two different counters. For spilling, the transaction
+	 * counter includes both toplevel transactions and subtransactions.
+	 * For streaming, it only includes toplevel transactions (we never
+	 * stream individual subtransactions).
 	 */
 	int64		spillCount;		/* spill-to-disk invocation counter */
 	int64		spillTxns;		/* number of transactions spilled to disk  */
 	int64		spillBytes;		/* amount of data spilled to disk */
+	int64		streamCount;	/* streaming invocation counter */
+	int64		streamTxns;		/* number of transactions spilled to disk */
+	int64		streamBytes;	/* amount of data streamed to subscriber */
 };
 
 
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index a6b3205..7efc332 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -85,6 +85,11 @@ typedef struct WalSnd
 	int64		spillTxns;
 	int64		spillCount;
 	int64		spillBytes;
+
+	/* Statistics for in-progress transactions streamed to subscriber. */
+	int64           streamTxns;
+	int64           streamCount;
+	int64           streamBytes;
 } WalSnd;
 
 extern WalSnd *MyWalSnd;
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index c9cc569..9fe3dd5 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1955,9 +1955,12 @@ pg_stat_replication| SELECT s.pid,
     w.reply_time,
     w.spill_txns,
     w.spill_count,
-    w.spill_bytes
+    w.spill_bytes,
+    w.stream_txns,
+    w.stream_count,
+    w.stream_bytes
    FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc)
-     JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time, spill_txns, spill_count, spill_bytes) ON ((s.pid = w.pid)))
+     JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_ssl| SELECT s.pid,
     s.ssl,
-- 
1.8.3.1