From 5fed4ef615c690258f68da3f13ea3c9e0d7900fa Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Mon, 25 Aug 2025 10:03:28 +0800 Subject: [PATCH v66 3/3] Add a dead_tuple_retention_active column in pg_stat_subscription To monitor worker's conflict retention status, this patch also introduces a new column 'dead_tuple_retention_active' in the pg_stat_subscription view. This column indicates whether the apply worker is effectively retaining conflict information. The value is set to true only if retain_dead_tuples is enabled for the associated subscription, and the retention duration for conflict detection by the apply worker has not exceeded max_retention_duration. --- doc/src/sgml/monitoring.sgml | 12 ++++++++++++ src/backend/catalog/system_views.sql | 3 ++- src/backend/replication/logical/launcher.c | 18 +++++++++++++++++- src/include/catalog/pg_proc.dat | 6 +++--- src/test/regress/expected/rules.out | 5 +++-- src/test/subscription/t/035_conflicts.pl | 10 +++++++++- 6 files changed, 46 insertions(+), 8 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 3f4a27a736e..96270f03bf2 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -2114,6 +2114,18 @@ description | Waiting for a newly initialized WAL file to reach durable storage sender; NULL for parallel apply workers + + + + dead_tuple_retention_active boolean + + + True if retain_dead_tuples + is enabled and the retention duration for information used in conflict detection is + within max_retention_duration; NULL for + parallel apply workers and table synchronization workers. + + diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index fbd3d1900f4..4679cb70213 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -996,7 +996,8 @@ CREATE VIEW pg_stat_subscription AS st.last_msg_send_time, st.last_msg_receipt_time, st.latest_end_lsn, - st.latest_end_time + st.latest_end_time, + st.dead_tuple_retention_active FROM pg_subscription su LEFT JOIN pg_stat_get_subscription(NULL) st ON (st.subid = su.oid); diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 73d2c22ae8b..6d3559038bd 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -1575,7 +1575,7 @@ GetLeaderApplyWorkerPid(pid_t pid) Datum pg_stat_get_subscription(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_SUBSCRIPTION_COLS 10 +#define PG_STAT_GET_SUBSCRIPTION_COLS 11 Oid subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0); int i; ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; @@ -1652,6 +1652,22 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS) elog(ERROR, "unknown worker type"); } + /* + * Use the worker's oldest_nonremovable_xid instead of + * pg_subscription.subretentionactive to determine whether retention + * is active, as retention resumption might not be complete even when + * subretentionactive is set to true; this is because the launcher + * assigns the initial oldest_nonremovable_xid after the apply worker + * updates the catalog (see resume_conflict_info_retention). + * + * Only the leader apply worker manages conflict retention (see + * maybe_advance_nonremovable_xid() for details). + */ + if (!isParallelApplyWorker(&worker) && !isTablesyncWorker(&worker)) + values[10] = TransactionIdIsValid(worker.oldest_nonremovable_xid); + else + nulls[10] = true; + 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 118d6da1ace..3810f3883b7 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5696,9 +5696,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,text}', - proargmodes => '{i,o,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,worker_type}', + proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz,text,bool}', + proargmodes => '{i,o,o,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,worker_type,dead_tuple_retention_active}', 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 35e8aad7701..183fc193ad3 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2174,9 +2174,10 @@ pg_stat_subscription| SELECT su.oid AS subid, st.last_msg_send_time, st.last_msg_receipt_time, st.latest_end_lsn, - st.latest_end_time + st.latest_end_time, + st.dead_tuple_retention_active 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, worker_type) ON ((st.subid = su.oid))); + 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, worker_type, dead_tuple_retention_active) ON ((st.subid = su.oid))); pg_stat_subscription_stats| SELECT ss.subid, s.subname, ss.apply_error_count, diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index 36aeb14c563..c1b8ede81cb 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -214,6 +214,10 @@ ok( $node_B->poll_query_until( ), "the xmin value of slot 'pg_conflict_detection' is valid on Node B"); +my $result = $node_B->safe_psql('postgres', + "SELECT dead_tuple_retention_active FROM pg_stat_subscription WHERE subname='$subname_BA';"); +is($result, qq(t), 'worker on node B retains conflict information'); + ################################################## # Check that the retain_dead_tuples option can be enabled only for disabled # subscriptions. Validate the NOTICE message during the subscription DDL, and @@ -254,6 +258,10 @@ ok( $node_A->poll_query_until( ), "the xmin value of slot 'pg_conflict_detection' is valid on Node A"); +$result = $node_A->safe_psql('postgres', + "SELECT dead_tuple_retention_active FROM pg_stat_subscription WHERE subname='$subname_AB';"); +is($result, qq(t), 'worker on node A retains conflict information'); + ################################################## # Check the WARNING when changing the origin to ANY, if retain_dead_tuples is # enabled. This warns of the possibility of receiving changes from origins @@ -281,7 +289,7 @@ $node_A->psql('postgres', $node_A->safe_psql('postgres', "INSERT INTO tab VALUES (1, 1), (2, 2);"); $node_A->wait_for_catchup($subname_BA); -my $result = $node_B->safe_psql('postgres', "SELECT * FROM tab;"); +$result = $node_B->safe_psql('postgres', "SELECT * FROM tab;"); is($result, qq(1|1 2|2), 'check replicated insert on node B'); -- 2.31.1