From 3d0d5398dee15233a0308b88b2b7df6f3578e950 Mon Sep 17 00:00:00 2001 From: Huseyin Demir Date: Fri, 30 Oct 2026 08:03:39 +0100 Subject: [PATCH] Log 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, conflict or 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 LOG when the relation is not part of the subscription, and at DEBUG2 for a relation that is merely not synchronized yet. Each relation is reported once per sync state for the lifetime of the apply worker, since the gate is reached once per change. Author: Huseyin Demir --- src/backend/replication/logical/worker.c | 53 ++++++++++++-- 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, 178 insertions(+), 5 deletions(-) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 44d735cdc55..83ae7b6ad5f 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -670,6 +670,43 @@ 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 LOG and the former, which is transient and expected, at DEBUG2. + * + * The gate this serves is reached once per change, so report each sync state + * of a relation only once. The record is kept in the relation map entry, so + * it lasts as long as the worker does. + */ +static void +report_unapplied_change(LogicalRepRelMapEntry *rel) +{ + if (rel->reported && rel->state == rel->reportedstate) + return; + + if (rel->state == SUBREL_STATE_UNKNOWN) + ereport(LOG, + errmsg("skipping changes for relation \"%s.%s\" in subscription \"%s\"", + rel->remoterel.nspname, rel->remoterel.relname, + MySubscription->name), + errdetail("The relation is not part of the subscription."), + errhint("Use ALTER SUBSCRIPTION ... REFRESH PUBLICATION to add it.")); + else + ereport(DEBUG2, + errmsg_internal("skipping changes for relation \"%s.%s\" in subscription \"%s\": synchronization state is \"%c\"", + rel->remoterel.nspname, rel->remoterel.relname, + MySubscription->name, rel->state)); + + rel->reported = true; + rel->reportedstate = rel->state; +} + /* * Should this worker apply changes for given relation. * @@ -713,12 +750,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 4063118d2b9..df1c984ec2a 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -38,8 +38,17 @@ typedef struct LogicalRepRelMapEntry * for a REPLICA IDENTITY FULL remote * relation? */ - /* Sync state. */ + /* + * Sync state. + * + * reported and reportedstate limit report_unapplied_change() to one + * report per sync state: whether a report was made, and for which state. + * They are kept next to state so that they occupy padding that follows it + * anyway. + */ char state; + bool reported; + char reportedstate; XLogRecPtr statelsn; } LogicalRepRelMapEntry; diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl index ac96bc3f009..880d0ed775d 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+\] LOG: ( [A-Z0-9]+:)? skipping changes for relation "public\.test_tab_3" in subscription "tap_sub"/, + $pa_offset); + +ok( $node_subscriber->log_contains( + qr/DETAIL: ( [A-Z0-9]+:)? The relation 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..240a8ded7da 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. +# +# Each relation is reported once per apply worker, however many changes arrive. +$node_publisher->safe_psql('postgres', "CREATE TABLE tab_4 (a int)"); +$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_4 (a int)"); +$node_publisher->safe_psql('postgres', "CREATE TABLE tab_5 (a int)"); +$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_5 (a int)"); + +# Add the tables to the publication, but do not refresh the subscription, so +# the subscriber has no pg_subscription_rel entry for them. +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_3 ADD TABLE tab_4, tab_5"); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_4 SELECT generate_series(1, 10)"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_5 SELECT generate_series(1, 10)"); + +foreach my $tab ('tab_4', 'tab_5') +{ + $node_subscriber->wait_for_log( + qr/LOG: ( [A-Z0-9]+:)? skipping changes for relation "public\.$tab" in subscription "tap_sub"/, + $offset); +} + +ok( $node_subscriber->log_contains( + qr/DETAIL: ( [A-Z0-9]+:)? The relation 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 each relation is reported once, rather than +# once per change. +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_4 SELECT generate_series(11, 100)"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_5 SELECT generate_series(11, 100)"); +$node_publisher->wait_for_catchup('tap_sub'); + +my $log = + PostgreSQL::Test::Utils::slurp_file($node_subscriber->logfile, $offset); +foreach my $tab ('tab_4', 'tab_5') +{ + my $count = () = $log =~ /skipping changes for relation "public\.$tab"/g; + is($count, 1, "$tab is reported once rather than once per change"); +} + +# A restarted apply worker starts with an empty relation map, so a relation +# that is still missing is reported again. +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub DISABLE"); +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub ENABLE"); +$offset = -s $node_subscriber->logfile; +$node_publisher->safe_psql('postgres', "INSERT INTO tab_4 VALUES (101)"); +$node_subscriber->wait_for_log( + qr/LOG: ( [A-Z0-9]+:)? skipping changes for relation "public\.tab_4" in subscription "tap_sub"/, + $offset); + +$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 (102)"); +$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(102|102), + 'changes are applied once the relation is part of the subscription'); + # shutdown $node_subscriber->stop('fast'); $node_publisher->stop('fast'); -- 2.54.0 (Apple Git-157)