fix_stats_1.patch

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 7899060f02f..a692030d6ba 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2054,17 +2054,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				rb->begin(rb, txn);
 		}
 
-		/*
-		 * Update total transaction count and total transaction bytes, if
-		 * transaction is streamed or spilled it will be updated while the
-		 * transaction gets spilled or streamed.
-		 */
-		if (!rb->streamBytes && !rb->spillBytes)
-		{
-			rb->totalTxns++;
-			rb->totalBytes += rb->size;
-		}
-
 		ReorderBufferIterTXNInit(rb, txn, &iterstate);
 		while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL)
 		{
@@ -2373,6 +2362,15 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 			specinsert = NULL;
 		}
 
+		/*
+		 * Update total transaction count and total transaction bytes
+		 * processed. Ensure to not count the streamed transaction multiple
+		 * times.
+		 */
+		rb->totalBytes += rb->size;
+		if (!rbtxn_is_streamed(txn))
+			rb->totalTxns++;
+
 		/* clean up the iterator */
 		ReorderBufferIterTXNFinish(rb, iterstate);
 		iterstate = NULL;
@@ -3538,11 +3536,9 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	{
 		rb->spillCount += 1;
 		rb->spillBytes += size;
-		rb->totalBytes += size;
 
 		/* don't consider already serialized transactions */
 		rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
-		rb->totalTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
 	}
 
 	Assert(spilled == txn->nentries_mem);
@@ -3908,11 +3904,9 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 
 	rb->streamCount += 1;
 	rb->streamBytes += stream_bytes;
-	rb->totalBytes += stream_bytes;
 
 	/* Don't consider already streamed transaction. */
 	rb->streamTxns += (txn_is_streamed) ? 0 : 1;
-	rb->totalTxns += (txn_is_streamed) ? 0 : 1;
 
 	Assert(dlist_is_empty(&txn->changes));
 	Assert(txn->nentries == 0);