From 569385da0bb7c3d40aec99842c7ea0504f8ba674 Mon Sep 17 00:00:00 2001 From: cagrib Date: Tue, 7 Jul 2026 14:17:19 +0200 Subject: [PATCH v1] Add ALTER SUBSCRIPTION ... REFRESH TABLE to resync one table 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 a single 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 [ WITH (copy_data = true, truncate = true) ] It resets the state of just the named relation in pg_subscription_rel back to init (optionally truncating the local copy first) so that a tablesync worker re-copies that one table, reusing the existing table synchronization machinery. Publication membership and other tables are left untouched. This mirrors REFRESH SEQUENCES, which already re-synchronizes already-subscribed sequences. 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 | 180 +++++++++++++++++ src/backend/parser/gram.y | 11 ++ src/include/nodes/parsenodes.h | 2 + src/test/subscription/t/039_refresh_table.pl | 198 +++++++++++++++++++ 4 files changed, 391 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..87029a5c034 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -82,6 +82,7 @@ #define SUBOPT_LSN 0x00020000 #define SUBOPT_ORIGIN 0x00040000 #define SUBOPT_CONFLICT_LOG_DEST 0x00080000 +#define SUBOPT_TRUNCATE 0x00100000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -99,6 +100,7 @@ typedef struct SubOpts bool enabled; bool create_slot; bool copy_data; + bool truncate; bool refresh; bool binary; char streaming; @@ -183,6 +185,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->create_slot = true; if (IsSet(supported_opts, SUBOPT_COPY_DATA)) opts->copy_data = true; + if (IsSet(supported_opts, SUBOPT_TRUNCATE)) + opts->truncate = true; if (IsSet(supported_opts, SUBOPT_REFRESH)) opts->refresh = true; if (IsSet(supported_opts, SUBOPT_BINARY)) @@ -264,6 +268,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->specified_opts |= SUBOPT_COPY_DATA; opts->copy_data = defGetBoolean(defel); } + else if (IsSet(supported_opts, SUBOPT_TRUNCATE) && + strcmp(defel->defname, "truncate") == 0) + { + if (IsSet(opts->specified_opts, SUBOPT_TRUNCATE)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_TRUNCATE; + opts->truncate = defGetBoolean(defel); + } else if (IsSet(supported_opts, SUBOPT_SYNCHRONOUS_COMMIT) && strcmp(defel->defname, "synchronous_commit") == 0) { @@ -1405,6 +1418,133 @@ AlterSubscription_refresh_seq(Subscription *sub) PG_END_TRY(); } +/* + * Resynchronize a single already-subscribed table. + * + * Resets the relation's pg_subscription_rel state back to init (optionally + * truncating the local copy first) so that, once the subscription is enabled, + * a tablesync worker re-copies just this one table. This is subscriber-local + * and does not touch publication membership, so sibling subscribers are + * unaffected. + * + * 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, bool copy_data, + bool truncate, RangeVar *rv) +{ + Relation rel; + Oid relid; + char relstate; + XLogRecPtr relstatelsn; + + /* + * Lock pg_subscription_rel with AccessExclusiveLock, matching + * AlterSubscription_refresh(), so the relation's state cannot change under + * us for the duration of the command. + */ + rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock); + + 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)); + + /* 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(); + } + + /* Optionally clear the local copy so the re-copy starts from empty. */ + if (copy_data && truncate) + { + TruncateStmt *tstmt = makeNode(TruncateStmt); + + tstmt->relations = list_make1(rv); + tstmt->restart_seqs = false; + tstmt->behavior = DROP_RESTRICT; + + ExecuteTruncate(tstmt); + } + + /* + * Reset the relation state. With copy_data it goes back to init so a + * tablesync worker re-copies it on enable; otherwise straight to ready. + */ + UpdateSubscriptionRelState(sub->oid, relid, + copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY, + 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. @@ -1644,6 +1784,10 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, supported_opts = SUBOPT_COPY_DATA; break; + case ALTER_SUBSCRIPTION_REFRESH_TABLE: + supported_opts = SUBOPT_COPY_DATA | SUBOPT_TRUNCATE; + break; + case ALTER_SUBSCRIPTION_SKIP: supported_opts = SUBOPT_LSN; break; @@ -2296,6 +2440,42 @@ 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.")); + + /* TRUNCATE only makes sense when we are going to re-copy. */ + if (!opts.copy_data) + { + if (IsSet(opts.specified_opts, SUBOPT_TRUNCATE) && + opts.truncate) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("truncate = true requires copy_data = true")); + + opts.truncate = false; + } + + PreventInTransactionBlock(isTopLevel, + "ALTER SUBSCRIPTION ... REFRESH TABLE"); + + AlterSubscription_refresh_table(sub, opts.copy_data, + opts.truncate, stmt->relation); + + 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..b71dd58ecf4 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -11617,6 +11617,17 @@ AlterSubscriptionStmt: n->subname = $3; $$ = (Node *) n; } + | ALTER SUBSCRIPTION name REFRESH TABLE qualified_name opt_definition + { + AlterSubscriptionStmt *n = + makeNode(AlterSubscriptionStmt); + + n->kind = ALTER_SUBSCRIPTION_REFRESH_TABLE; + n->subname = $3; + n->relation = $6; + n->options = $7; + $$ = (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..9c5208b7bc1 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 */ + RangeVar *relation; /* Table 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..b5d16e96bbe --- /dev/null +++ b/src/test/subscription/t/039_refresh_table.pl @@ -0,0 +1,198 @@ + +# Copyright (c) 2021-2026, PostgreSQL Global Development Group + +# Tests for ALTER SUBSCRIPTION ... REFRESH TABLE, which re-copies a single +# already-subscribed table 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'); + +# truncate = true requires copy_data = true. +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res WITH (copy_data = false, truncate = true)", + qr/truncate = true requires copy_data = true/, + 'REFRESH TABLE rejected with truncate but no copy_data'); + +# 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'); + +# 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 (defaults: copy_data = true, truncate = true). +$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'); + +# copy_data = false variant: reset to ready without re-copying or truncating. +$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', "DELETE FROM tab_res WHERE a <= 10"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res WITH (copy_data = false)" +); +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', + 'copy_data = false sets state directly to ready'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"), + '91', 'copy_data = false does not truncate or re-copy'); + +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub ENABLE"); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); -- 2.54.0