Send time of last reply message received from standby server

</para></entry>

</row>

<row>

<entry role="catalog_table_entry"><para role="column_definition">

<structfield>spill_txns</structfield> <type>bigint</type>

</para>

<para>

Number of transactions spilled to disk after the memory used by

logical decoding exceeds <literal>logical_decoding_work_mem</literal>. The

counter gets incremented both for top-level transactions and

subtransactions.

</para></entry>

</row>

<row>

<entry role="catalog_table_entry"><para role="column_definition">

<structfield>spill_count</structfield> <type>bigint</type>

</para>

<para>

Number of times transactions were spilled to disk. Transactions

may get spilled repeatedly, and this counter gets incremented on every

such invocation.

</para></entry>

</row>

<row>

<entry role="catalog_table_entry"><para role="column_definition">

<structfield>spill_bytes</structfield> <type>bigint</type>

</para>

<para>

Amount of decoded transaction data spilled to disk.

</para></entry>

</row>

</tbody>

</tgroup>

</table>

mechanism will simply display NULL lag.

</para>

<para>

Tracking of spilled transactions works only for logical replication. In

physical replication, the tracking mechanism will display 0 for spilled

statistics.

</para>

<note>

<para>

The reported lag times are not predictions of how long it will take for

<listitem>

<!--

Author: Amit Kapila <akapila@postgresql.org>

2019-11-21 [9290ad198] Track statistics for spilling of changes from ReorderBuf

-->

<para>

Add columns to the <link

linkend="monitoring-stats-dynamic-views-table"><structname>pg_stat_replication</structname></link>

system view to report how much logical decoding information has been

spilled to disk (Tomas Vondra)

</para>

</listitem>

<listitem>

<!--

Author: Robert Haas <rhaas@postgresql.org>

2020-01-09 [ed10f32e3] Add pg_shmem_allocations view.

-->

W.replay_lag,

W.sync_priority,

W.sync_state,

W.reply_time,

W.spill_txns,

W.spill_count,

W.spill_bytes

W.reply_time

FROM pg_stat_get_activity(NULL)AS S

JOIN pg_stat_get_wal_senders()AS WON (S.pid=W.pid)

LEFT JOIN pg_authidAS UON (S.usesysid=U.oid);

buffer->outbufsize=0;

buffer->size=0;

buffer->spillCount=0;

buffer->spillTxns=0;

buffer->spillBytes=0;

buffer->current_restart_decoding_lsn=InvalidXLogRecPtr;

dlist_init(&buffer->toplevel_by_lsn);

intfd=-1;

XLogSegNocurOpenSegNo=0;

Sizespilled=0;

Sizesize=txn->size;

elog(DEBUG2,"spill %u changes in XID %u to disk",

(uint32)txn->nentries_mem,txn->xid);

spilled++;

}

rb->spillCount+=1;

rb->spillBytes+=size;

rb->spillTxns+=rbtxn_is_serialized(txn) ?0 :1;

Assert(spilled==txn->nentries_mem);

Assert(dlist_is_empty(&txn->changes));

txn->nentries_mem=0;

staticvoidWalSndSegmentOpen(XLogReaderState*state,XLogSegNonextSegNo,

TimeLineID*tli_p);

staticvoidUpdateSpillStats(LogicalDecodingContext*ctx);

WalSndUpdateProgress(LogicalDecodingContext*ctx,XLogRecPtrlsn,TransactionIdxid)

LagTrackerWrite(lsn,now);

sendTime=now;

UpdateSpillStats(ctx);

}

walsnd->sync_standby_priority=0;

walsnd->latch=&MyProc->procLatch;

walsnd->replyTime=0;

walsnd->spillTxns=0;

walsnd->spillCount=0;

walsnd->spillBytes=0;

SpinLockRelease(&walsnd->mutex);

MyWalSnd= (WalSnd*)walsnd;

pg_stat_get_wal_senders(PG_FUNCTION_ARGS)

{

ReturnSetInfo*rsinfo= (ReturnSetInfo*)fcinfo->resultinfo;

TupleDesctupdesc;

Tuplestorestate*tupstore;

intpid;

WalSndStatestate;

TimestampTzreplyTime;

int64spillTxns;

int64spillCount;

int64spillBytes;

boolis_sync_standby;

Datumvalues[PG_STAT_GET_WAL_SENDERS_COLS];

boolnulls[PG_STAT_GET_WAL_SENDERS_COLS];

applyLag=walsnd->applyLag;

priority=walsnd->sync_standby_priority;

replyTime=walsnd->replyTime;

spillTxns=walsnd->spillTxns;

spillCount=walsnd->spillCount;

spillBytes=walsnd->spillBytes;

SpinLockRelease(&walsnd->mutex);

nulls[11]= true;

values[11]=TimestampTzGetDatum(replyTime);

values[12]=Int64GetDatum(spillTxns);

values[13]=Int64GetDatum(spillCount);

values[14]=Int64GetDatum(spillBytes);

}

tuplestore_putvalues(tupstore,tupdesc,values,nulls);

Assert(time!=0);

returnnow-time;

}

UpdateSpillStats(LogicalDecodingContext*ctx)

{

ReorderBuffer*rb=ctx->reorder;

elog(DEBUG2,"UpdateSpillStats: updating stats %p %lld %lld %lld",

rb,

(long long)rb->spillTxns,

(long long)rb->spillCount,

(long long)rb->spillBytes);

SpinLockAcquire(&MyWalSnd->mutex);

MyWalSnd->spillTxns=rb->spillTxns;

MyWalSnd->spillCount=rb->spillCount;

MyWalSnd->spillBytes=rb->spillBytes;

SpinLockRelease(&MyWalSnd->mutex);

}

proname => 'pg_stat_get_wal_senders', prorows => '10', proisstrict => 'f',

proretset => 't', provolatile => 's', proparallel => 'r',

prorettype => 'record', proargtypes => '',

proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,int8,int8}',

proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',

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}',

proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz}',

proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o}',

proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time}',

prosrc => 'pg_stat_get_wal_senders' },

{ oid => '3317', descr => 'statistics: information about WAL receiver',

proname => 'pg_stat_get_wal_receiver', proisstrict => 'f', provolatile => 's',

Sizesize;

int64spillCount;/* spill-to-disk invocation counter */

int64spillTxns;/* number of transactions spilled to disk */

int64spillBytes;/* amount of data spilled to disk */

};

TimestampTzreplyTime;

int64spillTxns;

int64spillCount;

int64spillBytes;

}WalSnd;

externWalSnd*MyWalSnd;

w.replay_lag,

w.sync_priority,

w.sync_state,

w.reply_time,

w.spill_txns,

w.spill_count,

w.spill_bytes

w.reply_time

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, leader_pid)

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

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

LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));

pg_stat_slru| SELECT s.name,

s.blks_zeroed,