From b63ba0a637cd10ec341d807d9eaf4279142ced12 Mon Sep 17 00:00:00 2001 From: Peter Smith Date: Wed, 16 Aug 2023 19:02:36 +1000 Subject: [PATCH v1] Add worker_type to pg_stat_subscription --- doc/src/sgml/monitoring.sgml | 26 ++++++++++++++ src/backend/catalog/system_views.sql | 1 + src/backend/replication/logical/launcher.c | 55 +++++++++++++++++++++--------- src/include/catalog/pg_proc.dat | 6 ++-- src/test/regress/expected/rules.out | 3 +- 5 files changed, 70 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index f4fc5d8..ad25b5c 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -1993,6 +1993,32 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i + worker_type text + + + Type of the subscription worker process. Possible values are: + + + + a: apply worker + + + + + p: parallel apply worker + + + + + t: tablesync worker + + + + + + + + leader_pid integer diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index af65af6..166ad5f 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -950,6 +950,7 @@ CREATE VIEW pg_stat_subscription AS su.oid AS subid, su.subname, st.pid, + st.worker_type, st.leader_pid, st.relid, st.received_lsn, diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 7cc0a16..95f87d5 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -1272,7 +1272,7 @@ GetLeaderApplyWorkerPid(pid_t pid) Datum pg_stat_get_subscription(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_SUBSCRIPTION_COLS 9 +#define PG_STAT_GET_SUBSCRIPTION_COLS 10 Oid subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0); int i; ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; @@ -1298,40 +1298,61 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS) if (OidIsValid(subid) && worker.subid != subid) continue; + values[0] = ObjectIdGetDatum(worker.subid); + worker_pid = worker.proc->pid; + values[1] = Int32GetDatum(worker_pid); + + switch (worker.type) + { + case WORKERTYPE_APPLY: + values[2] = CStringGetTextDatum("a"); + break; + case WORKERTYPE_PARALLEL_APPLY: + values[2] = CStringGetTextDatum("p"); + break; + case WORKERTYPE_TABLESYNC: + values[2] = CStringGetTextDatum("t"); + break; + case WORKERTYPE_UNKNOWN: /* should not be possible */ + nulls[2] = true; + } - values[0] = ObjectIdGetDatum(worker.subid); - if (isTablesyncWorker(&worker)) - values[1] = ObjectIdGetDatum(worker.relid); - else - nulls[1] = true; - values[2] = Int32GetDatum(worker_pid); if (isParallelApplyWorker(&worker)) values[3] = Int32GetDatum(worker.leader_pid); else nulls[3] = true; - if (XLogRecPtrIsInvalid(worker.last_lsn)) + if (isTablesyncWorker(&worker)) + values[4] = ObjectIdGetDatum(worker.relid); + else nulls[4] = true; + + if (XLogRecPtrIsInvalid(worker.last_lsn)) + nulls[5] = true; else - values[4] = LSNGetDatum(worker.last_lsn); + values[5] = LSNGetDatum(worker.last_lsn); + if (worker.last_send_time == 0) - nulls[5] = true; + nulls[6] = true; else - values[5] = TimestampTzGetDatum(worker.last_send_time); + values[6] = TimestampTzGetDatum(worker.last_send_time); + if (worker.last_recv_time == 0) - nulls[6] = true; + nulls[7] = true; else - values[6] = TimestampTzGetDatum(worker.last_recv_time); + values[7] = TimestampTzGetDatum(worker.last_recv_time); + if (XLogRecPtrIsInvalid(worker.reply_lsn)) - nulls[7] = true; + nulls[8] = true; else - values[7] = LSNGetDatum(worker.reply_lsn); + values[8] = LSNGetDatum(worker.reply_lsn); + if (worker.reply_time == 0) - nulls[8] = true; + nulls[9] = true; else - values[8] = TimestampTzGetDatum(worker.reply_time); + values[9] = TimestampTzGetDatum(worker.reply_time); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 6996073..a3455f2 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5466,9 +5466,9 @@ proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', proparallel => 'r', prorettype => 'record', proargtypes => 'oid', - proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz}', - proargmodes => '{i,o,o,o,o,o,o,o,o,o}', - proargnames => '{subid,subid,relid,pid,leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time}', + proallargtypes => '{oid,oid,int4,text,int4,oid,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz}', + proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{subid,subid,pid,worker_type,leader_pid,relid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time}', prosrc => 'pg_stat_get_subscription' }, { oid => '2026', descr => 'statistics: current backend PID', proname => 'pg_backend_pid', provolatile => 's', proparallel => 'r', diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index e07afcd..82935e8 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2119,6 +2119,7 @@ pg_stat_ssl| SELECT pid, pg_stat_subscription| SELECT su.oid AS subid, su.subname, st.pid, + st.worker_type, st.leader_pid, st.relid, st.received_lsn, @@ -2127,7 +2128,7 @@ pg_stat_subscription| SELECT su.oid AS subid, st.latest_end_lsn, st.latest_end_time FROM (pg_subscription su - LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, leader_pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid = su.oid))); + LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, pid, worker_type, leader_pid, relid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid = su.oid))); pg_stat_subscription_stats| SELECT ss.subid, s.subname, ss.apply_error_count, -- 1.8.3.1