From: Amit Kapila Date: Thu, 29 Oct 2020 03:41:51 +0000 (+0530) Subject: Track statistics for streaming of changes from ReorderBuffer. X-Git-Tag: REL_14_BETA1~1400 X-Git-Url: http://git.postgresql.org/gitweb/?a=commitdiff_plain;h=8e90ec5580d5345fef31005d7cc2215ba2125070;p=postgresql.git Track statistics for streaming of changes from ReorderBuffer. This adds the statistics about transactions streamed to the decoding output plugin 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. Commit 9868167500 has added the basic infrastructure to capture the stats of slot and this commit extends the statistics collector to track additional information about slots. Bump the catversion as we have added new columns in the catalog entry. Author: Ajin Cherian and Amit Kapila Reviewed-by: Sawada Masahiko and Dilip Kumar Discussion: https://postgr.es/m/CAA4eK1+chpEomLzgSoky-D31qev19AmECNiEAietPQUGEFhtVA@mail.gmail.com --- diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 313e44ed549..98e19954538 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -2632,6 +2632,44 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i + + + stream_txns bigint + + + Number of in-progress transactions streamed to the decoding output plugin + after the memory used by logical decoding of changes from WAL for this + slot exceeds logical_decoding_work_mem. Streaming only + works with toplevel transactions (subtransactions can't be streamed + independently), so the counter does not get incremented for subtransactions. + + + + + + stream_countbigint + + + Number of times in-progress transactions were streamed to the decoding + output plugin while decoding changes from WAL for this slot. Transactions + may get streamed repeatedly, and this counter gets incremented on every + such invocation. + + + + + + stream_bytesbigint + + + Amount of decoded in-progress transaction data streamed to the decoding + output plugin while decoding changes from WAL for this slot. This and other + streaming counters for this slot can be used to gauge the network I/O which + occurred during logical decoding and allow tuning logical_decoding_work_mem. + + + + stats_reset timestamp with time zone diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index c6dd084fbcc..5171ea05c7e 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -802,6 +802,9 @@ CREATE VIEW pg_stat_replication_slots AS s.spill_txns, s.spill_count, s.spill_bytes, + s.stream_txns, + s.stream_count, + s.stream_bytes, s.stats_reset FROM pg_stat_get_replication_slots() AS s; diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 822f0ebc628..f1dca2f25b7 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -1708,7 +1708,7 @@ pgstat_report_tempfile(size_t filesize) */ void pgstat_report_replslot(const char *slotname, int spilltxns, int spillcount, - int spillbytes) + int spillbytes, int streamtxns, int streamcount, int streambytes) { PgStat_MsgReplSlot msg; @@ -1721,6 +1721,9 @@ pgstat_report_replslot(const char *slotname, int spilltxns, int spillcount, msg.m_spill_txns = spilltxns; msg.m_spill_count = spillcount; msg.m_spill_bytes = spillbytes; + msg.m_stream_txns = streamtxns; + msg.m_stream_count = streamcount; + msg.m_stream_bytes = streambytes; pgstat_send(&msg, sizeof(PgStat_MsgReplSlot)); } @@ -6892,6 +6895,9 @@ pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len) replSlotStats[idx].spill_txns += msg->m_spill_txns; replSlotStats[idx].spill_count += msg->m_spill_count; replSlotStats[idx].spill_bytes += msg->m_spill_bytes; + replSlotStats[idx].stream_txns += msg->m_stream_txns; + replSlotStats[idx].stream_count += msg->m_stream_count; + replSlotStats[idx].stream_bytes += msg->m_stream_bytes; } } @@ -7125,6 +7131,9 @@ pgstat_reset_replslot(int i, TimestampTz ts) replSlotStats[i].spill_txns = 0; replSlotStats[i].spill_count = 0; replSlotStats[i].spill_bytes = 0; + replSlotStats[i].stream_txns = 0; + replSlotStats[i].stream_count = 0; + replSlotStats[i].stream_bytes = 0; replSlotStats[i].stat_reset_timestamp = ts; } diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 8675832f4d6..d5cfbeaa4af 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -1471,21 +1471,28 @@ 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. + * Nothing to do if we haven't spilled or streamed anything since the last + * time the stats has been sent. */ - if (rb->spillBytes <= 0) + if (rb->spillBytes <= 0 && rb->streamBytes <= 0) return; - elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld", + elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld", rb, (long long) rb->spillTxns, (long long) rb->spillCount, - (long long) rb->spillBytes); + (long long) rb->spillBytes, + (long long) rb->streamTxns, + (long long) rb->streamCount, + (long long) rb->streamBytes); pgstat_report_replslot(NameStr(ctx->slot->data.name), - rb->spillTxns, rb->spillCount, rb->spillBytes); + rb->spillTxns, rb->spillCount, rb->spillBytes, + rb->streamTxns, rb->streamCount, rb->streamBytes); rb->spillTxns = 0; rb->spillCount = 0; rb->spillBytes = 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 7a8bf760791..c1bd68011c5 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -346,6 +346,9 @@ ReorderBufferAllocate(void) buffer->spillTxns = 0; buffer->spillCount = 0; buffer->spillBytes = 0; + buffer->streamTxns = 0; + buffer->streamCount = 0; + buffer->streamBytes = 0; buffer->current_restart_decoding_lsn = InvalidXLogRecPtr; @@ -3482,6 +3485,8 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) { Snapshot snapshot_now; CommandId command_id; + Size stream_bytes; + bool txn_is_streamed; /* We can never reach here for a subtransaction. */ Assert(txn->toptxn == NULL); @@ -3562,10 +3567,25 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) txn->snapshot_now = NULL; } + /* + * Remember this information to be used later to update stats. We can't + * update the stats here as an error while processing the changes would + * lead to the accumulation of stats even though we haven't streamed all + * the changes. + */ + txn_is_streamed = rbtxn_is_streamed(txn); + stream_bytes = txn->total_size; + /* Process and send the changes to output plugin. */ ReorderBufferProcessTXN(rb, txn, InvalidXLogRecPtr, snapshot_now, command_id, true); + rb->streamCount += 1; + rb->streamBytes += stream_bytes; + + /* Don't consider already streamed transaction. */ + rb->streamTxns += (txn_is_streamed) ? 0 : 1; + Assert(dlist_is_empty(&txn->changes)); Assert(txn->nentries == 0); Assert(txn->nentries_mem == 0); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 220b4cd6e99..09be1d8c485 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -320,7 +320,7 @@ ReplicationSlotCreate(const char *name, bool db_specific, * ReplicationSlotAllocationLock. */ if (SlotIsLogical(slot)) - pgstat_report_replslot(NameStr(slot->data.name), 0, 0, 0); + pgstat_report_replslot(NameStr(slot->data.name), 0, 0, 0, 0, 0, 0); /* * Now that the slot has been marked as in_use and active, it's safe to diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index 472fa596e1f..a210fc93b41 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -2153,7 +2153,7 @@ pg_stat_get_archiver(PG_FUNCTION_ARGS) Datum pg_stat_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_REPLICATION_SLOT_COLS 5 +#define PG_STAT_GET_REPLICATION_SLOT_COLS 8 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; TupleDesc tupdesc; Tuplestorestate *tupstore; @@ -2201,11 +2201,14 @@ pg_stat_get_replication_slots(PG_FUNCTION_ARGS) values[1] = Int64GetDatum(s->spill_txns); values[2] = Int64GetDatum(s->spill_count); values[3] = Int64GetDatum(s->spill_bytes); + values[4] = Int64GetDatum(s->stream_txns); + values[5] = Int64GetDatum(s->stream_count); + values[6] = Int64GetDatum(s->stream_bytes); if (s->stat_reset_timestamp == 0) - nulls[4] = true; + nulls[7] = true; else - values[4] = TimestampTzGetDatum(s->stat_reset_timestamp); + values[7] = TimestampTzGetDatum(s->stat_reset_timestamp); tuplestore_putvalues(tupstore, tupdesc, values, nulls); } diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h index 8cf02fc0d85..73650f88e94 100644 --- a/src/include/catalog/catversion.h +++ b/src/include/catalog/catversion.h @@ -53,6 +53,6 @@ */ /* yyyymmddN */ -#define CATALOG_VERSION_NO 202010281 +#define CATALOG_VERSION_NO 202010291 #endif diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 24ec2cfed6a..d9770bbadd8 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5260,9 +5260,9 @@ proname => 'pg_stat_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', proparallel => 'r', prorettype => 'record', proargtypes => '', - proallargtypes => '{text,int8,int8,int8,timestamptz}', - proargmodes => '{o,o,o,o,o}', - proargnames => '{slot_name,spill_txns,spill_count,spill_bytes,stats_reset}', + proallargtypes => '{text,int8,int8,int8,int8,int8,int8,timestamptz}', + proargmodes => '{o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes,stats_reset}', prosrc => 'pg_stat_get_replication_slots' }, { oid => '6118', descr => 'statistics: information about subscription', proname => 'pg_stat_get_subscription', proisstrict => 'f', provolatile => 's', diff --git a/src/include/pgstat.h b/src/include/pgstat.h index a821ff4f158..257e515bfe7 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -492,6 +492,9 @@ typedef struct PgStat_MsgReplSlot PgStat_Counter m_spill_txns; PgStat_Counter m_spill_count; PgStat_Counter m_spill_bytes; + PgStat_Counter m_stream_txns; + PgStat_Counter m_stream_count; + PgStat_Counter m_stream_bytes; } PgStat_MsgReplSlot; @@ -823,6 +826,9 @@ typedef struct PgStat_ReplSlotStats PgStat_Counter spill_txns; PgStat_Counter spill_count; PgStat_Counter spill_bytes; + PgStat_Counter stream_txns; + PgStat_Counter stream_count; + PgStat_Counter stream_bytes; TimestampTz stat_reset_timestamp; } PgStat_ReplSlotStats; @@ -1387,7 +1393,7 @@ extern void pgstat_report_deadlock(void); extern void pgstat_report_checksum_failures_in_db(Oid dboid, int failurecount); extern void pgstat_report_checksum_failure(void); extern void pgstat_report_replslot(const char *slotname, int spilltxns, int spillcount, - int spillbytes); + int spillbytes, int streamtxns, int streamcount, int streambytes); extern void pgstat_report_replslot_drop(const char *slotname); extern void pgstat_initialize(void); diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 1c77819aad2..dfdda938b2a 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -551,6 +551,11 @@ 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 */ + + /* Statistics about transactions streamed to the decoding output plugin */ + int64 streamTxns; /* number of transactions streamed */ + int64 streamCount; /* streaming invocation counter */ + int64 streamBytes; /* amount of data streamed */ }; diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 492cdcf74c3..097ff5d111f 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2022,8 +2022,11 @@ pg_stat_replication_slots| SELECT s.slot_name, s.spill_txns, s.spill_count, s.spill_bytes, + s.stream_txns, + s.stream_count, + s.stream_bytes, s.stats_reset - FROM pg_stat_get_replication_slots() s(slot_name, spill_txns, spill_count, spill_bytes, stats_reset); + FROM pg_stat_get_replication_slots() s(slot_name, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes, stats_reset); pg_stat_slru| SELECT s.name, s.blks_zeroed, s.blks_hit,