Skip to content

Commit 9290ad1

Browse files
author
Amit Kapila
committed
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 view to check these stats. Author: Tomas Vondra, with bug-fixes and minor changes by Dilip Kumar Reviewed-by: Amit Kapila Discussion: https://postgr.es/m/688b0b7f-2f6c-d827-c27b-216a8e3ea700@2ndquadrant.com
1 parent 168d206 commit 9290ad1

File tree

9 files changed

+101
-9
lines changed

9 files changed

+101
-9
lines changed

‎doc/src/sgml/monitoring.sgml

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1972,6 +1972,26 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
19721972
<entry><type>timestamp with time zone</type></entry>
19731973
<entry>Send time of last reply message received from standby server</entry>
19741974
</row>
1975+
<row>
1976+
<entry><structfield>spill_bytes</structfield></entry>
1977+
<entry><type>bigint</type></entry>
1978+
<entry>Amount of decoded transaction data spilled to disk.</entry>
1979+
</row>
1980+
<row>
1981+
<entry><structfield>spill_txns</structfield></entry>
1982+
<entry><type>bigint</type></entry>
1983+
<entry>Number of transactions spilled to disk after the memory used by
1984+
logical decoding exceeds <literal>logical_decoding_work_mem</literal>. The
1985+
counter gets incremented both for toplevel transactions and
1986+
subtransactions.</entry>
1987+
</row>
1988+
<row>
1989+
<entry><structfield>spill_count</structfield></entry>
1990+
<entry><type>bigint</type></entry>
1991+
<entry>Number of times transactions were spilled to disk. Transactions
1992+
may get spilled repeatedly, and this counter gets incremented on every
1993+
such invocation.</entry>
1994+
</row>
19751995
</tbody>
19761996
</tgroup>
19771997
</table>

‎src/backend/catalog/system_views.sql

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -776,7 +776,10 @@ CREATE VIEW pg_stat_replication AS
776776
W.replay_lag,
777777
W.sync_priority,
778778
W.sync_state,
779-
W.reply_time
779+
W.reply_time,
780+
W.spill_txns,
781+
W.spill_count,
782+
W.spill_bytes
780783
FROM pg_stat_get_activity(NULL) AS S
781784
JOIN pg_stat_get_wal_senders() AS W ON (S.pid = W.pid)
782785
LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid);

‎src/backend/replication/logical/reorderbuffer.c

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -308,6 +308,10 @@ ReorderBufferAllocate(void)
308308
buffer->outbufsize = 0;
309309
buffer->size = 0;
310310

311+
buffer->spillCount = 0;
312+
buffer->spillTxns = 0;
313+
buffer->spillBytes = 0;
314+
311315
buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;
312316

313317
dlist_init(&buffer->toplevel_by_lsn);
@@ -2415,6 +2419,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
24152419
int fd = -1;
24162420
XLogSegNo curOpenSegNo = 0;
24172421
Size spilled = 0;
2422+
Size size = txn->size;
24182423

24192424
elog(DEBUG2, "spill %u changes in XID %u to disk",
24202425
(uint32) txn->nentries_mem, txn->xid);
@@ -2473,6 +2478,13 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
24732478
spilled++;
24742479
}
24752480

2481+
/* update the statistics */
2482+
rb->spillCount += 1;
2483+
rb->spillBytes += size;
2484+
2485+
/* Don't consider already serialized transaction. */
2486+
rb->spillTxns += txn->serialized ? 0 : 1;
2487+
24762488
Assert(spilled == txn->nentries_mem);
24772489
Assert(dlist_is_empty(&txn->changes));
24782490
txn->nentries_mem = 0;

‎src/backend/replication/walsender.c

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,7 @@ static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time);
248248
static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now);
249249
static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch);
250250

251+
static void UpdateSpillStats(LogicalDecodingContext *ctx);
251252
static void XLogRead(WALSegmentContext *segcxt, char *buf, XLogRecPtr startptr, Size count);
252253

253254

@@ -1261,7 +1262,8 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
12611262
/*
12621263
* LogicalDecodingContext 'update_progress' callback.
12631264
*
1264-
* Write the current position to the lag tracker (see XLogSendPhysical).
1265+
* Write the current position to the lag tracker (see XLogSendPhysical),
1266+
* and update the spill statistics.
12651267
*/
12661268
static void
12671269
WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid)
@@ -1280,6 +1282,11 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
12801282

12811283
LagTrackerWrite(lsn, now);
12821284
sendTime = now;
1285+
1286+
/*
1287+
* Update statistics about transactions that spilled to disk.
1288+
*/
1289+
UpdateSpillStats(ctx);
12831290
}
12841291

12851292
/*
@@ -2318,6 +2325,9 @@ InitWalSenderSlot(void)
23182325
walsnd->state = WALSNDSTATE_STARTUP;
23192326
walsnd->latch = &MyProc->procLatch;
23202327
walsnd->replyTime = 0;
2328+
walsnd->spillTxns = 0;
2329+
walsnd->spillCount = 0;
2330+
walsnd->spillBytes = 0;
23212331
SpinLockRelease(&walsnd->mutex);
23222332
/* don't need the lock anymore */
23232333
MyWalSnd = (WalSnd *) walsnd;
@@ -3219,7 +3229,7 @@ offset_to_interval(TimeOffset offset)
32193229
Datum
32203230
pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
32213231
{
3222-
#define PG_STAT_GET_WAL_SENDERS_COLS 12
3232+
#define PG_STAT_GET_WAL_SENDERS_COLS 15
32233233
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
32243234
TupleDesc tupdesc;
32253235
Tuplestorestate *tupstore;
@@ -3274,6 +3284,9 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
32743284
int pid;
32753285
WalSndState state;
32763286
TimestampTz replyTime;
3287+
int64 spillTxns;
3288+
int64 spillCount;
3289+
int64 spillBytes;
32773290
Datum values[PG_STAT_GET_WAL_SENDERS_COLS];
32783291
bool nulls[PG_STAT_GET_WAL_SENDERS_COLS];
32793292

@@ -3294,6 +3307,9 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
32943307
applyLag = walsnd->applyLag;
32953308
priority = walsnd->sync_standby_priority;
32963309
replyTime = walsnd->replyTime;
3310+
spillTxns = walsnd->spillTxns;
3311+
spillCount = walsnd->spillCount;
3312+
spillBytes = walsnd->spillBytes;
32973313
SpinLockRelease(&walsnd->mutex);
32983314

32993315
memset(nulls, 0, sizeof(nulls));
@@ -3375,6 +3391,11 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
33753391
nulls[11] = true;
33763392
else
33773393
values[11] = TimestampTzGetDatum(replyTime);
3394+
3395+
/* spill to disk */
3396+
values[12] = Int64GetDatum(spillTxns);
3397+
values[13] = Int64GetDatum(spillCount);
3398+
values[14] = Int64GetDatum(spillBytes);
33783399
}
33793400

33803401
tuplestore_putvalues(tupstore, tupdesc, values, nulls);
@@ -3611,3 +3632,20 @@ LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now)
36113632
Assert(time != 0);
36123633
return now - time;
36133634
}
3635+
3636+
static void
3637+
UpdateSpillStats(LogicalDecodingContext *ctx)
3638+
{
3639+
ReorderBuffer *rb = ctx->reorder;
3640+
3641+
SpinLockAcquire(&MyWalSnd->mutex);
3642+
3643+
MyWalSnd->spillTxns = rb->spillTxns;
3644+
MyWalSnd->spillCount = rb->spillCount;
3645+
MyWalSnd->spillBytes = rb->spillBytes;
3646+
3647+
elog(DEBUG2, "UpdateSpillStats: updating stats %p %ld %ld %ld",
3648+
rb, rb->spillTxns, rb->spillCount, rb->spillBytes);
3649+
3650+
SpinLockRelease(&MyWalSnd->mutex);
3651+
}

‎src/include/catalog/catversion.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,6 @@
5353
*/
5454

5555
/* yyyymmddN */
56-
#define CATALOG_VERSION_NO 201911121
56+
#define CATALOG_VERSION_NO 201911211
5757

5858
#endif

‎src/include/catalog/pg_proc.dat

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5166,9 +5166,9 @@
51665166
proname => 'pg_stat_get_wal_senders', prorows => '10', proisstrict => 'f',
51675167
proretset => 't', provolatile => 's', proparallel => 'r',
51685168
prorettype => 'record', proargtypes => '',
5169-
proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz}',
5170-
proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o}',
5171-
proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time}',
5169+
proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,int8,int8}',
5170+
proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
5171+
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}',
51725172
prosrc => 'pg_stat_get_wal_senders' },
51735173
{ oid => '3317', descr => 'statistics: information about WAL receiver',
51745174
proname => 'pg_stat_get_wal_receiver', proisstrict => 'f', provolatile => 's',

‎src/include/replication/reorderbuffer.h

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -402,6 +402,17 @@ struct ReorderBuffer
402402

403403
/* memory accounting */
404404
Size size;
405+
406+
/*
407+
* Statistics about transactions spilled to disk.
408+
*
409+
* A single transaction may be spilled repeatedly, which is why we keep
410+
* two different counters. For spilling, the transaction counter includes
411+
* both toplevel transactions and subtransactions.
412+
*/
413+
int64 spillCount; /* spill-to-disk invocation counter */
414+
int64 spillTxns; /* number of transactions spilled to disk */
415+
int64 spillBytes; /* amount of data spilled to disk */
405416
};
406417

407418

‎src/include/replication/walsender_private.h

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,11 @@ typedef struct WalSnd
8080
* Timestamp of the last message received from standby.
8181
*/
8282
TimestampTz replyTime;
83+
84+
/* Statistics for transactions spilled to disk. */
85+
int64 spillTxns;
86+
int64 spillCount;
87+
int64 spillBytes;
8388
} WalSnd;
8489

8590
extern WalSnd *MyWalSnd;

‎src/test/regress/expected/rules.out

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1952,9 +1952,12 @@ pg_stat_replication| SELECT s.pid,
19521952
w.replay_lag,
19531953
w.sync_priority,
19541954
w.sync_state,
1955-
w.reply_time
1955+
w.reply_time,
1956+
w.spill_txns,
1957+
w.spill_count,
1958+
w.spill_bytes
19561959
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)
1957-
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) ON ((s.pid = w.pid)))
1960+
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)))
19581961
LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
19591962
pg_stat_ssl| SELECT s.pid,
19601963
s.ssl,

0 commit comments

Comments
 (0)