Track statistics for spilling of changes from ReorderBuffer.

This adds the statistics about transactions spilled to disk from
ReorderBuffer. Users can query the pg_stat_replication_slots view to check
these stats and call pg_stat_reset_replication_slot to reset the stats of
a particular slot. Users can pass NULL in pg_stat_reset_replication_slot
to reset stats of all the slots.

This commit extends the statistics collector to track this information
about slots.

Author: Sawada Masahiko and Amit Kapila
Reviewed-by: Amit Kapila and Dilip Kumar
Discussion: https://postgr.es/m/CA+fd4k5_pPAYRTDrO2PbtTOe0eHQpBvuqmCr8ic39uTNmR49Eg@mail.gmail.com

src/backend/replication/logical/decode.c

@@ -650,6 +650,12 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	/* replay actions of all transaction + subtransactions in order */
 	ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
 						commit_time, origin_id, origin_lsn);
+
+	/*
+	 * Update the decoding stats at transaction commit/abort. It is not clear
+	 * that sending more or less frequently than this would be better.
+	 */
+	UpdateDecodingStats(ctx);
 }
 
 /*
@@ -669,6 +675,9 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	}
 
 	ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
+
+	/* update the decoding stats */
+	UpdateDecodingStats(ctx);
 }
 
 /*

src/backend/replication/logical/logical.c

@@ -32,6 +32,7 @@
 #include "access/xlog_internal.h"
 #include "fmgr.h"
 #include "miscadmin.h"
+#include "pgstat.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
 #include "replication/origin.h"
@@ -1460,3 +1461,31 @@ ResetLogicalStreamingState(void)
 	CheckXidAlive = InvalidTransactionId;
 	bsysscan = false;
 }
+
+/*
+ * Report stats for a slot.
+ */
+void
+UpdateDecodingStats(LogicalDecodingContext *ctx)
+{
+	ReorderBuffer *rb = ctx->reorder;
+
+	/*
+	 * Nothing to do if we haven't spilled anything since the last time the
+	 * stats has been sent.
+	 */
+	if (rb->spillBytes <= 0)
+		return;
+
+	elog(DEBUG2, "UpdateSpillStats: updating stats %p %lld %lld %lld",
+		 rb,
+		 (long long) rb->spillTxns,
+		 (long long) rb->spillCount,
+		 (long long) rb->spillBytes);
+
+	pgstat_report_replslot(NameStr(ctx->slot->data.name),
+						   rb->spillTxns, rb->spillCount, rb->spillBytes);
+	rb->spillTxns = 0;
+	rb->spillCount = 0;
+	rb->spillBytes = 0;
+}

src/backend/replication/logical/reorderbuffer.c

@@ -343,6 +343,10 @@ ReorderBufferAllocate(void)
 	buffer->outbufsize = 0;
 	buffer->size = 0;
 
+	buffer->spillTxns = 0;
+	buffer->spillCount = 0;
+	buffer->spillBytes = 0;
+
 	buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;
 
 	dlist_init(&buffer->toplevel_by_lsn);
@@ -1579,6 +1583,13 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	{
 		ReorderBufferRestoreCleanup(rb, txn);
 		txn->txn_flags &= ~RBTXN_IS_SERIALIZED;
+
+		/*
+		 * We set this flag to indicate if the transaction is ever serialized.
+		 * We need this to accurately update the stats as otherwise the same
+		 * transaction can be counted as serialized multiple times.
+		 */
+		txn->txn_flags |= RBTXN_IS_SERIALIZED_CLEAR;
 	}
 
 	/* also reset the number of entries in the transaction */
@@ -3112,6 +3123,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	int			fd = -1;
 	XLogSegNo	curOpenSegNo = 0;
 	Size		spilled = 0;
+	Size		size = txn->size;
 
 	elog(DEBUG2, "spill %u changes in XID %u to disk",
 		 (uint32) txn->nentries_mem, txn->xid);
@@ -3170,6 +3182,16 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 		spilled++;
 	}
 
+	/* update the statistics iff we have spilled anything */
+	if (spilled)
+	{
+		rb->spillCount += 1;
+		rb->spillBytes += size;
+
+		/* don't consider already serialized transactions */
+		rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
+	}
+
 	Assert(spilled == txn->nentries_mem);
 	Assert(dlist_is_empty(&txn->changes));
 	txn->nentries_mem = 0;

src/backend/replication/slot.c

@@ -99,7 +99,6 @@ ReplicationSlot *MyReplicationSlot = NULL;
 int			max_replication_slots = 0;	/* the maximum number of replication
 										 * slots */
 
-static ReplicationSlot *SearchNamedReplicationSlot(const char *name);
 static int ReplicationSlotAcquireInternal(ReplicationSlot *slot,
 										  const char *name, SlotAcquireBehavior behavior);
 static void ReplicationSlotDropAcquired(void);
@@ -314,6 +313,15 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 
 	LWLockRelease(ReplicationSlotControlLock);
 
+	/*
+	 * Create statistics entry for the new logical slot. We don't collect any
+	 * stats for physical slots, so no need to create an entry for the same.
+	 * See ReplicationSlotDropPtr for why we need to do this before releasing
+	 * ReplicationSlotAllocationLock.
+	 */
+	if (SlotIsLogical(slot))
+		pgstat_report_replslot(NameStr(slot->data.name), 0, 0, 0);
+
 	/*
 	 * Now that the slot has been marked as in_use and active, it's safe to
 	 * let somebody else try to allocate a slot.
@@ -331,7 +339,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
  *
  * The caller must hold ReplicationSlotControlLock in shared mode.
  */
-static ReplicationSlot *
+ReplicationSlot *
 SearchNamedReplicationSlot(const char *name)
 {
 	int			i;
@@ -683,6 +691,19 @@ ReplicationSlotDropPtr(ReplicationSlot *slot)
 		ereport(WARNING,
 				(errmsg("could not remove directory \"%s\"", tmppath)));
 
+	/*
+	 * Send a message to drop the replication slot to the stats collector.
+	 * Since there is no guarantee of the order of message transfer on a UDP
+	 * connection, it's possible that a message for creating a new slot
+	 * reaches before a message for removing the old slot. We send the drop
+	 * and create messages while holding ReplicationSlotAllocationLock to
+	 * reduce that possibility. If the messages reached in reverse, we would
+	 * lose one statistics update message. But the next update message will
+	 * create the statistics for the replication slot.
+	 */
+	if (SlotIsLogical(slot))
+		pgstat_report_replslot_drop(NameStr(slot->data.name));
+
 	/*
 	 * We release this at the very end, so that nobody starts trying to create
 	 * a slot while we're still cleaning up the detritus of the old one.