From 0029daa2b6999c363a40bf8aed14d44ef812c8df Mon Sep 17 00:00:00 2001 From: Huseyin Demir Date: Mon, 7 Sep 2026 19:19:31 +0200 Subject: [PATCH] Report changes discarded for relations not in the subscription When an apply worker receives a change for a relation that is not in SUBREL_STATE_READY, should_apply_changes_for_rel() returns false and the change is discarded with no log message at any elevation, no conflict and no statistics counter. That is correct while a tablesync worker is still copying the table, but it is indistinguishable from a relation that has no pg_subscription_rel row at all, where the discard is permanent and both the replication origin and the publisher slot advance past the discarded rows. Report the discard at DEBUG1 when the relation is not part of the subscription, separating the case where the subscription tracks no tables at all, and at DEBUG2 for a relation that is merely not synchronized yet. Reporting is throttled per relation to one message per sync state change or per wal_retrieve_retry_interval, whichever comes first, since the gate is reached once per change. --- src/backend/replication/logical/worker.c | 87 ++++++++++++++++++++- src/include/replication/logicalrelation.h | 11 ++- src/test/subscription/t/015_stream.pl | 39 +++++++++ src/test/subscription/t/024_add_drop_pub.pl | 80 +++++++++++++++++++ 4 files changed, 212 insertions(+), 5 deletions(-) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 7781bb1c168..f4dd5bf5a7a 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -671,6 +671,77 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, } } +/* + * Report that changes for this relation are not being applied. + * + * A relation that is not ready is skipped silently on every change. That is + * correct while a tablesync worker is still copying it, but it is + * indistinguishable from a relation missing from pg_subscription_rel + * altogether -- one whose changes are discarded, and acknowledged to the + * publisher as applied, for as long as the subscription runs. Report the + * latter at DEBUG1 and the former, which is transient and expected, at DEBUG2. + * + * The gate this serves is reached once per change, so reporting is throttled + * per relation: a change of sync state is reported immediately, and otherwise + * at most one report is made per wal_retrieve_retry_interval. + * + * begin_replication_step() has already stamped the statement start time for + * this change, so use that rather than reading the clock again. + */ +static void +report_unapplied_change(LogicalRepRelMapEntry *rel) +{ + TimestampTz now = GetCurrentStatementStartTimestamp(); + + if (rel->state == rel->reportedstate && + !TimestampDifferenceExceeds(rel->lastreported, now, + wal_retrieve_retry_interval)) + return; + + if (rel->state == SUBREL_STATE_UNKNOWN) + { + /* + * Distinguish a subscription that tracks no tables at all -- what a + * pre-17 pg_upgrade or a restore that loses pg_subscription_rel + * leaves behind, where every change for every relation is discarded + * -- from a single relation that was published without a refresh on + * this side. This is only reached on the throttled path, so the + * cached lookup is not made once per change. + * + * Only the apply worker asks. A parallel apply worker exists only + * because pa_can_start() found all tablesyncs ready, which requires + * the subscription to have tables, so the question is already + * answered for it. + */ + bool notables = (MyLogicalRepWorker->type == WORKERTYPE_APPLY && + !HasSubscriptionTablesCached()); + + ereport(DEBUG1, + errmsg_internal("logical replication apply worker for subscription \"%s\" is not applying changes for relation \"%s.%s\"", + MySubscription->name, + rel->remoterel.nspname, rel->remoterel.relname), + notables ? + errdetail_internal("The subscription has no tables.") : + errdetail_internal("The relation \"%s.%s\" is not part of the subscription.", + rel->remoterel.nspname, rel->remoterel.relname), + notables ? + errhint_internal("Use ALTER SUBSCRIPTION ... REFRESH PUBLICATION to add the published tables to the subscription.") : + errhint_internal("Use ALTER SUBSCRIPTION ... REFRESH PUBLICATION to add the relation to the subscription.")); + } + else + ereport(DEBUG2, + errmsg_internal("logical replication apply worker for subscription \"%s\" is not applying changes for relation \"%s.%s\"", + MySubscription->name, + rel->remoterel.nspname, rel->remoterel.relname), + errdetail_internal("Relation synchronization state is \"%c\", synchronization LSN %X/%08X, transaction finish LSN %X/%08X.", + rel->state, + LSN_FORMAT_ARGS(rel->statelsn), + LSN_FORMAT_ARGS(remote_ctx.finish_lsn))); + + rel->reportedstate = rel->state; + rel->lastreported = now; +} + /* * Should this worker apply changes for given relation. * @@ -714,12 +785,20 @@ should_apply_changes_for_rel(LogicalRepRelMapEntry *rel) MySubscription->name), errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized."))); - return rel->state == SUBREL_STATE_READY; + if (rel->state == SUBREL_STATE_READY) + return true; + + report_unapplied_change(rel); + return false; case WORKERTYPE_APPLY: - return (rel->state == SUBREL_STATE_READY || - (rel->state == SUBREL_STATE_SYNCDONE && - rel->statelsn <= remote_ctx.finish_lsn)); + if (rel->state == SUBREL_STATE_READY || + (rel->state == SUBREL_STATE_SYNCDONE && + rel->statelsn <= remote_ctx.finish_lsn)) + return true; + + report_unapplied_change(rel); + return false; case WORKERTYPE_SEQUENCESYNC: /* Should never happen. */ diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index efe0f9d6031..4e730d1a3fe 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -34,9 +34,18 @@ typedef struct LogicalRepRelMapEntry bool updatable; /* Can apply updates/deletes? */ Oid localindexoid; /* which index to use, or InvalidOid if none */ - /* Sync state. */ + /* + * Sync state. + * + * reportedstate and lastreported throttle the reports made by + * report_unapplied_change(): the sync state that was last reported for + * this relation, and when. reportedstate is kept next to state so that + * it occupies padding that follows it anyway. + */ char state; + char reportedstate; XLogRecPtr statelsn; + TimestampTz lastreported; } LogicalRepRelMapEntry; extern void logicalrep_relmap_update(LogicalRepRelation *remoterel); diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl index ac96bc3f009..1bc90f055de 100644 --- a/src/test/subscription/t/015_stream.pl +++ b/src/test/subscription/t/015_stream.pl @@ -201,6 +201,45 @@ $node_subscriber->safe_psql('postgres', q{SELECT 1}); test_streaming($node_publisher, $node_subscriber, $appname, 1); +# Verify that a parallel apply worker reports changes it does not apply. +# +# A table that is published but never refreshed on the subscriber has no +# pg_subscription_rel row at all. GetSubscriptionRelations() only returns rows +# that exist, so such a table is invisible to AllTablesyncsReady(), +# pa_can_start() still hands the streamed transaction to a parallel apply +# worker, and that worker discards every change for the table. +$node_publisher->safe_psql('postgres', "CREATE TABLE test_tab_3 (a int)"); +$node_subscriber->safe_psql('postgres', "CREATE TABLE test_tab_3 (a int)"); +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION tap_pub ADD TABLE test_tab_3"); + +my $pa_offset = -s $node_subscriber->logfile; + +# Large enough to be streamed, given logical_decoding_work_mem = 64kB. +$node_publisher->safe_psql('postgres', + "INSERT INTO test_tab_3 SELECT i FROM generate_series(1, 5000) s(i)"); + +# The bgw_type in log_line_prefix's %b distinguishes the parallel apply worker +# ("logical replication parallel worker") from the leader ("logical replication +# apply worker"), so this pins the report to the parallel path specifically. +$node_subscriber->wait_for_log( + qr/logical replication parallel worker\[\d+\] DEBUG: ( [A-Z0-9]+:)? logical replication apply worker for subscription "tap_sub" is not applying changes for relation "public\.test_tab_3"/, + $pa_offset); + +ok( $node_subscriber->log_contains( + qr/DETAIL: ( [A-Z0-9]+:)? The relation "public\.test_tab_3" is not part of the subscription\./, + $pa_offset), + 'parallel apply worker names the relation it is not applying changes for' +); + +$node_publisher->wait_for_catchup($appname); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_3"); +is($result, qq(0), + 'parallel apply worker discards changes for a relation not in the subscription' +); + # Test that the deadlock is detected among the leader and parallel apply # workers. diff --git a/src/test/subscription/t/024_add_drop_pub.pl b/src/test/subscription/t/024_add_drop_pub.pl index 63adb1b464c..f399107476a 100644 --- a/src/test/subscription/t/024_add_drop_pub.pl +++ b/src/test/subscription/t/024_add_drop_pub.pl @@ -135,6 +135,86 @@ is( $result, qq(1 'check that the incremental data is replicated after the publication is created' ); +# Verify that the apply worker reports changes it does not apply because the +# relation is not part of the subscription. Such changes are discarded and +# still acknowledged to the publisher as applied, so the report is the only +# indication that they were received at all. +# +# wal_retrieve_retry_interval throttles the report; set it high so that the +# message is emitted exactly once no matter how many changes arrive, which is +# what the count below checks. +$node_subscriber->append_conf( + 'postgresql.conf', qq( +log_min_messages = debug1 +wal_retrieve_retry_interval = 10min +)); +$node_subscriber->reload; + +$node_publisher->safe_psql('postgres', "CREATE TABLE tab_4 (a int)"); +$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_4 (a int)"); + +# Add the table to the publication, but do not refresh the subscription, so +# the subscriber has no pg_subscription_rel entry for it. +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_3 ADD TABLE tab_4"); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_4 SELECT generate_series(1, 10)"); + +$node_subscriber->wait_for_log( + qr/DEBUG: ( [A-Z0-9]+:)? logical replication apply worker for subscription "tap_sub" is not applying changes for relation "public\.tab_4"/, + $offset); + +# The subscription does have other tables, so this must be reported as a +# single missing relation rather than as a subscription with no tables. +ok( $node_subscriber->log_contains( + qr/DETAIL: ( [A-Z0-9]+:)? The relation "public\.tab_4" is not part of the subscription\./, + $offset), + 'relation not in the subscription is reported with the relation name'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_4"); +is($result, qq(0), + 'changes for a relation not in the subscription are discarded'); + +# Send more changes and confirm the report is throttled rather than emitted +# once per change. +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_4 SELECT generate_series(11, 100)"); +$node_publisher->wait_for_catchup('tap_sub'); + +my $log = + PostgreSQL::Test::Utils::slurp_file($node_subscriber->logfile, $offset); +my $count = () = + $log =~ /is not applying changes for relation "public\.tab_4"/g; +is($count, 1, 'the report is emitted once rather than once per change'); + +# Restore the retry interval before refreshing: tablesync worker startup is +# gated on the same GUC, so leaving it at 10min would stall the sync below. +$node_subscriber->append_conf('postgresql.conf', + "wal_retrieve_retry_interval = 5s\n"); +$node_subscriber->reload; + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub'); + +$node_publisher->safe_psql('postgres', "INSERT INTO tab_4 VALUES (101)"); +$node_publisher->wait_for_catchup('tap_sub'); + +# Once the relation is part of the subscription its changes are applied, and +# the initial copy brings across the rows that were discarded earlier. +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*), max(a) FROM tab_4"); +is($result, qq(101|101), + 'changes are applied once the relation is part of the subscription'); + +$node_subscriber->append_conf('postgresql.conf', + "log_min_messages = warning\n"); +$node_subscriber->reload; + # shutdown $node_subscriber->stop('fast'); $node_publisher->stop('fast'); -- 2.50.1 (Apple Git-155)