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);