From fa82f9f0ece03430671273905073f3364e9ee6d4 Mon Sep 17 00:00:00 2001 From: cagrib Date: Tue, 7 Jul 2026 14:17:19 +0200 Subject: [PATCH v2] Add ALTER SUBSCRIPTION ... REFRESH TABLE to resync tables REFRESH PUBLICATION only starts syncing tables that were newly added to the publication; it never re-copies a table that is already part of the subscription. Today the only ways to re-seed an already-subscribed table are to churn publication membership (which is not local to one subscriber and can disturb sibling subscribers or cascade downstream) or to hand-edit pg_subscription_rel (unsupported and racy against the apply worker). Add a subscriber-local command: ALTER SUBSCRIPTION name REFRESH TABLE table_name [, ...] It truncates the local copy of each named table and resets its state in pg_subscription_rel back to init so that a tablesync worker re-copies just those tables, reusing the existing table synchronization machinery. All named relations are validated before any is reset, so the command is all-or-nothing, and they are truncated together so a set connected by foreign keys can be re-seeded as a unit. Publication membership and other tables are left untouched. This mirrors REFRESH SEQUENCES, which likewise re-synchronizes objects already known to the subscription rather than only discovering newly added ones. This first version requires the subscription to be disabled, which avoids racing the apply worker's cached relation state. A TAP test is included. --- src/backend/commands/subscriptioncmds.c | 187 +++++++++++++++ src/backend/parser/gram.y | 10 + src/include/nodes/parsenodes.h | 2 + src/test/subscription/t/039_refresh_table.pl | 228 +++++++++++++++++++ 4 files changed, 427 insertions(+) create mode 100644 src/test/subscription/t/039_refresh_table.pl diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 4292e7fb8f4..dd6096d5843 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1405,6 +1405,170 @@ AlterSubscription_refresh_seq(Subscription *sub) PG_END_TRY(); } +/* + * Resynchronize one or more already-subscribed tables. + * + * Truncates the local copy of each named table and resets its + * pg_subscription_rel state back to init so that, once the subscription is + * enabled, a tablesync worker re-copies just those tables. This is + * subscriber-local and does not touch publication membership, so sibling + * subscribers are unaffected. + * + * Every named relation is validated before any of them is reset, so the + * command is all-or-nothing: a bad table name aborts the whole command + * without touching the others. The tables are truncated together, which lets + * a set connected by foreign keys be re-seeded as a unit. + * + * The caller must ensure the subscription is disabled: with no apply worker + * running there is no cached relation state to invalidate and no race against + * a concurrently launched tablesync worker. + */ +static void +AlterSubscription_refresh_table(Subscription *sub, List *relations) +{ + Relation rel; + ListCell *lc; + List *relids = NIL; + List *truncrels = NIL; + TruncateStmt *tstmt; + + /* + * Lock pg_subscription_rel with AccessExclusiveLock, matching + * AlterSubscription_refresh(), so the relation states cannot change under + * us for the duration of the command. + */ + rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock); + + /* + * First pass: validate every named relation before changing anything, so + * the command is all-or-nothing. Duplicates in the list are ignored. + */ + foreach(lc, relations) + { + RangeVar *rv = lfirst_node(RangeVar, lc); + Oid relid; + char relstate; + XLogRecPtr relstatelsn; + + relid = RangeVarGetRelid(rv, AccessShareLock, false); + + if (get_rel_relkind(relid) == RELKIND_SEQUENCE) + ereport(ERROR, + errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("cannot refresh sequence \"%s\" as a table", + rv->relname), + errhint("Use ALTER SUBSCRIPTION ... REFRESH SEQUENCES instead.")); + + /* The relation must already be part of the subscription. */ + relstate = GetSubscriptionRelState(sub->oid, relid, &relstatelsn); + if (relstate == SUBREL_STATE_UNKNOWN) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" is not part of the subscription \"%s\"", + rv->relname, sub->name)); + + if (list_member_oid(relids, relid)) + continue; + + relids = lappend_oid(relids, relid); + truncrels = lappend(truncrels, rv); + } + + /* + * Second pass: stop any leftover tablesync worker for each relation and, + * if it was caught mid-sync, clean up its origin and publisher slot. + */ + foreach_oid(relid, relids) + { + char relstate; + XLogRecPtr relstatelsn; + + relstate = GetSubscriptionRelState(sub->oid, relid, &relstatelsn); + + /* Stop any leftover tablesync worker for this relation. */ + logicalrep_worker_stop(WORKERTYPE_TABLESYNC, sub->oid, relid); + + /* + * If the relation was caught mid-sync, drop its tablesync origin. For + * READY state the tablesync worker already dropped it, so nothing is + * needed there. Pass missing_ok = true as the origin may not exist + * yet. + */ + if (relstate != SUBREL_STATE_READY) + { + char originname[NAMEDATALEN]; + + ReplicationOriginNameForLogicalRep(sub->oid, relid, originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + } + + /* + * Likewise drop the tablesync slot on the publisher for a mid-sync + * relation. READY/SYNCDONE relations have no such slot, so the common + * case needs no publisher connection at all. + */ + if (relstate != SUBREL_STATE_READY && relstate != SUBREL_STATE_SYNCDONE) + { + char syncslotname[NAMEDATALEN] = {0}; + char *err = NULL; + WalReceiverConn *wrconn; + bool must_use_password; + + load_file("libpqwalreceiver", false); + + must_use_password = sub->passwordrequired && !sub->ownersuperuser; + wrconn = walrcv_connect(sub->conninfo, true, true, must_use_password, + sub->name, &err); + if (!wrconn) + ereport(ERROR, + errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("subscription \"%s\" could not connect to the publisher: %s", + sub->name, err)); + + PG_TRY(); + { + ReplicationSlotNameForTablesync(sub->oid, relid, syncslotname, + sizeof(syncslotname)); + ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); + } + PG_FINALLY(); + { + walrcv_disconnect(wrconn); + } + PG_END_TRY(); + } + } + + /* + * Clear the local copies so the re-copy starts from empty. Truncating the + * tables together lets a set connected by foreign keys be re-seeded as a + * unit. + */ + tstmt = makeNode(TruncateStmt); + tstmt->relations = truncrels; + tstmt->restart_seqs = false; + tstmt->behavior = DROP_RESTRICT; + ExecuteTruncate(tstmt); + + /* + * Reset each relation to init so that a tablesync worker re-copies it once + * the subscription is enabled. + */ + foreach_oid(relid, relids) + { + UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_INIT, + InvalidXLogRecPtr, false); + + ereport(DEBUG1, + errmsg_internal("table \"%s.%s\" of subscription \"%s\" reset for resync", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), sub->name)); + } + + table_close(rel, NoLock); +} + /* * Common checks for altering failover, two_phase, and retain_dead_tuples * options. @@ -2296,6 +2460,29 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, break; } + case ALTER_SUBSCRIPTION_REFRESH_TABLE: + { + /* + * The first version requires the subscription to be disabled. + * With no apply worker running there is no cached relation + * state to invalidate and no race against a concurrently + * launched tablesync worker while we reset the relation. + */ + if (sub->enabled) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("%s is not allowed for enabled subscriptions", + "ALTER SUBSCRIPTION ... REFRESH TABLE"), + errhint("Disable the subscription with ALTER SUBSCRIPTION ... DISABLE first.")); + + PreventInTransactionBlock(isTopLevel, + "ALTER SUBSCRIPTION ... REFRESH TABLE"); + + AlterSubscription_refresh_table(sub, stmt->relations); + + break; + } + case ALTER_SUBSCRIPTION_SKIP: { /* ALTER SUBSCRIPTION ... SKIP supports only LSN option */ diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index ff4e1388c55..82716c46947 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -11617,6 +11617,16 @@ AlterSubscriptionStmt: n->subname = $3; $$ = (Node *) n; } + | ALTER SUBSCRIPTION name REFRESH TABLE qualified_name_list + { + AlterSubscriptionStmt *n = + makeNode(AlterSubscriptionStmt); + + n->kind = ALTER_SUBSCRIPTION_REFRESH_TABLE; + n->subname = $3; + n->relations = $6; + $$ = (Node *) n; + } | ALTER SUBSCRIPTION name ADD_P PUBLICATION name_list opt_definition { AlterSubscriptionStmt *n = diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index e03556399ab..cfeb09ab417 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -4566,6 +4566,7 @@ typedef enum AlterSubscriptionType ALTER_SUBSCRIPTION_DROP_PUBLICATION, ALTER_SUBSCRIPTION_REFRESH_PUBLICATION, ALTER_SUBSCRIPTION_REFRESH_SEQUENCES, + ALTER_SUBSCRIPTION_REFRESH_TABLE, ALTER_SUBSCRIPTION_ENABLED, ALTER_SUBSCRIPTION_SKIP, } AlterSubscriptionType; @@ -4578,6 +4579,7 @@ typedef struct AlterSubscriptionStmt char *servername; /* Server name of publisher */ char *conninfo; /* Connection string to publisher */ List *publication; /* One or more publication to subscribe to */ + List *relations; /* Tables to resync (for REFRESH TABLE) */ List *options; /* List of DefElem nodes */ } AlterSubscriptionStmt; diff --git a/src/test/subscription/t/039_refresh_table.pl b/src/test/subscription/t/039_refresh_table.pl new file mode 100644 index 00000000000..b09f1e149bf --- /dev/null +++ b/src/test/subscription/t/039_refresh_table.pl @@ -0,0 +1,228 @@ + +# Copyright (c) 2021-2026, PostgreSQL Global Development Group + +# Tests for ALTER SUBSCRIPTION ... REFRESH TABLE, which re-copies one or more +# already-subscribed tables on the subscriber without touching publication +# membership or other tables. The first version requires the subscription +# to be disabled. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Initialize publisher and subscriber nodes +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init; +$node_subscriber->append_conf('postgresql.conf', + "wal_retrieve_retry_interval = 1ms"); +$node_subscriber->start; + +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; + +# Preexisting content on the publisher: two published tables. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_res (a int primary key, b text); + CREATE TABLE tab_other (a int primary key, b text); + INSERT INTO tab_res SELECT g, 'p' || g FROM generate_series(1, 100) g; + INSERT INTO tab_other SELECT g, 'q' || g FROM generate_series(1, 100) g; + CREATE PUBLICATION tap_pub FOR TABLE tab_res, tab_other; +)); + +# Matching structure on the subscriber, plus objects used for error cases. +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_res (a int primary key, b text); + CREATE TABLE tab_other (a int primary key, b text); + CREATE TABLE tab_local (a int primary key); + CREATE SEQUENCE seq_local; +)); + +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub" +); + +# Wait for initial sync of both tables. +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub'); + +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"), + '100', 'initial sync of tab_res'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_other"), + '100', 'initial sync of tab_other'); + +# A small helper: run SQL expected to fail, and check the error message. +sub refresh_should_fail +{ + my ($sql, $pattern, $desc) = @_; + my ($ret, $stdout, $stderr) = ('', '', ''); + $ret = $node_subscriber->psql('postgres', $sql, + stdout => \$stdout, stderr => \$stderr); + ok($ret != 0 && $stderr =~ /$pattern/, + $desc) + or diag("got ret=$ret stderr=$stderr"); +} + +# REFRESH TABLE is not allowed while the subscription is enabled. +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res", + qr/not allowed for enabled subscriptions/, + 'REFRESH TABLE rejected while enabled'); + +# Disable the subscription and wait for its workers to stop. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +# A table that is not part of the subscription is rejected. +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_local", + qr/is not part of the subscription/, + 'REFRESH TABLE rejected for table not in subscription'); + +# A sequence cannot be refreshed as a table. +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE seq_local", + qr/cannot refresh sequence/, + 'REFRESH TABLE rejected for a sequence'); + +# The command cannot run inside a transaction block. +refresh_should_fail( + "BEGIN; ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res;", + qr/cannot run inside a transaction block/, + 'REFRESH TABLE rejected inside a transaction block'); + +# All-or-nothing: a list containing one bad table aborts the whole command and +# leaves the valid table untouched. +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res, tab_local", + qr/is not part of the subscription/, + 'REFRESH TABLE with a bad table in the list is rejected'); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_res'" + ), + 'r', + 'valid table not reset when another table in the list is invalid'); + +# Introduce drift on the subscriber, only in tab_res. +$node_subscriber->safe_psql( + 'postgres', qq( + DELETE FROM tab_res WHERE a <= 40; + UPDATE tab_res SET b = 'CORRUPT' WHERE a = 60; +)); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"), + '60', 'drift introduced in tab_res'); + +# Record the pre-refresh state of the other table. +my $other_state_before = $node_subscriber->safe_psql('postgres', + "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_other'" +); +is($other_state_before, 'r', 'tab_other is ready before refresh'); + +# Resync just tab_res. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res"); + +# Only tab_res is reset to init; tab_other is untouched; local copy truncated. +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_res'" + ), + 'i', + 'tab_res reset to init state'); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_other'" + ), + 'r', + 'tab_other left untouched (still ready)'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"), + '0', 'tab_res truncated locally by REFRESH while disabled'); + +# Re-enable and wait for the single table to re-copy. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub'); + +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"), + '100', 'tab_res re-copied after enable'); +is( $node_subscriber->safe_psql('postgres', + "SELECT b FROM tab_res WHERE a = 60"), + 'p60', 'tab_res corruption repaired by resync'); + +# Full content matches the publisher. +my $pub_md5 = $node_publisher->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res"); +my $sub_md5 = $node_subscriber->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res"); +is($sub_md5, $pub_md5, 'tab_res matches publisher after resync'); + +# tab_other was never disturbed. +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_other"), + '100', 'tab_other intact throughout'); + +# Ongoing replication still works for the resynced table. +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_res VALUES (101, 'p101')"); +$node_publisher->wait_for_catchup('tap_sub'); +is( $node_subscriber->safe_psql('postgres', + "SELECT count(*) FROM tab_res WHERE a = 101"), + '1', 'streaming resumes on resynced table'); + +# Multiple tables can be resynced in a single command. Disable, drift both +# tables, and refresh them together. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +$node_subscriber->safe_psql( + 'postgres', qq( + DELETE FROM tab_res WHERE a <= 20; + DELETE FROM tab_other WHERE a <= 20; +)); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res, tab_other"); + +# Both listed tables are reset to init and truncated locally. +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT string_agg(srsubstate, ',' ORDER BY c.relname) FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname IN ('tab_other', 'tab_res')" + ), + 'i,i', + 'both listed tables reset to init state'); +is( $node_subscriber->safe_psql('postgres', + "SELECT (SELECT count(*) FROM tab_res) + (SELECT count(*) FROM tab_other)"), + '0', 'both listed tables truncated locally'); + +# Re-enable and wait for both tables to re-copy. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub'); + +my $pub_res = $node_publisher->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res"); +my $sub_res = $node_subscriber->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res"); +is($sub_res, $pub_res, 'tab_res matches publisher after multi-table resync'); + +my $pub_oth = $node_publisher->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_other"); +my $sub_oth = $node_subscriber->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_other"); +is($sub_oth, $pub_oth, 'tab_other matches publisher after multi-table resync'); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); -- 2.54.0