From b057ea87f1b485340ed25f3760625f6c8faf31ae Mon Sep 17 00:00:00 2001 From: cagrib Date: Tue, 7 Jul 2026 14:17:19 +0200 Subject: [PATCH v4] Add ALTER SUBSCRIPTION ... REFRESH TABLE to resync individual tables When a subscribed table diverges from the publisher, the only supported remedy is to drop and recreate the subscription, which re-copies every table and throws away the progress of the ones that were fine. Users have instead been editing pg_subscription_rel by hand, which is not supported and is easy to get wrong. Add a command that re-synchronizes named tables in place: ALTER SUBSCRIPTION name REFRESH TABLE tbl [, ...] Each named table is truncated on the subscriber and its state is reset so that a table synchronization worker copies it again once the subscription is enabled. The set of subscribed tables is unchanged, and other subscribers of the same publication are unaffected. Several tables may be named and they are truncated in a single ExecuteTruncateGuts() call, so a group connected by foreign keys can be re-seeded together. Validation is done up front for every table, so the command either applies to all of them or changes nothing. The tables are locked with AccessExclusiveLock when they are resolved, and TRUNCATE privilege is required on each, matching what discarding the data implies. A single connection to the publisher is used to drop the table synchronization slots, and that is done last, once the local catalog changes cannot fail. The guiding rule is that a relation must not be discarded unless the state of whatever will refill it is reset too. Three cases follow from it. A partitioned table that is itself subscribed, which happens when the publication uses publish_via_partition_root, has its partitions truncated as well, mirroring apply_handle_truncate(); its partitions are not tracked separately, and the root's re-copy refills them by tuple routing. Inheritance children are not followed, because a child is an independent relation that may be subscribed in its own right or hold data that is not replicated at all. And a relation that some other subscription also populates, either directly or by routing rows into it through a partitioned ancestor, is rejected outright, since we cannot reset another subscription's state on its behalf. The subscription must be disabled and all of its workers must have stopped, which keeps the change entirely local: no worker can be looking at the relation state while it is rewritten. Note that a table synchronization worker outlives its apply worker while it finishes a copy, so all worker types are checked. The command is disallowed when two_phase is enabled. --- doc/src/sgml/ref/alter_subscription.sgml | 72 ++ src/backend/commands/subscriptioncmds.c | 434 ++++++++++++ src/backend/parser/gram.y | 10 + src/bin/psql/tab-complete.in.c | 11 +- src/include/nodes/parsenodes.h | 2 + src/test/subscription/t/039_refresh_table.pl | 710 +++++++++++++++++++ 6 files changed, 1237 insertions(+), 2 deletions(-) create mode 100644 src/test/subscription/t/039_refresh_table.pl diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 545264e8a0a..2b08a343b0f 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -28,6 +28,7 @@ ALTER SUBSCRIPTION name ADD PUBLICA ALTER SUBSCRIPTION name DROP PUBLICATION publication_name [, ...] [ WITH ( publication_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name REFRESH PUBLICATION [ WITH ( refresh_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name REFRESH SEQUENCES +ALTER SUBSCRIPTION name REFRESH TABLE table_name [, ...] ALTER SUBSCRIPTION name ENABLE ALTER SUBSCRIPTION name DISABLE ALTER SUBSCRIPTION name SET ( subscription_parameter [= value] [, ... ] ) @@ -69,6 +70,7 @@ ALTER SUBSCRIPTION name RENAME TO < Commands ALTER SUBSCRIPTION ... REFRESH PUBLICATION, + ALTER SUBSCRIPTION ... REFRESH TABLE, ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ... with refresh option as true, ALTER SUBSCRIPTION ... SET (failover = true|false) and @@ -263,6 +265,76 @@ ALTER SUBSCRIPTION name RENAME TO < + + REFRESH TABLE + + + Re-synchronize the named tables with the publisher. Unlike + + ALTER SUBSCRIPTION ... REFRESH PUBLICATION, + which only starts copying tables that are new to the subscription, this + re-copies tables that are already subscribed and have finished their + initial synchronization. Each named table is truncated on the + subscriber and marked for initial synchronization again, so that a table + synchronization worker copies it afresh once the subscription is + enabled. Which tables are subscribed is not changed. + + + This is intended for repairing a table whose contents have diverged from + the publisher, and it replaces the alternative of dropping and + recreating the whole subscription, which would re-copy every table and + lose the progress of the others. Other subscribers of the same + publication are unaffected. + + + All of the named tables are truncated in a single command, so a set of + tables connected by foreign keys can be re-seeded together. The tables + must already be part of the subscription, and the command fails without + changing anything if any of them is not. The user must have the + TRUNCATE privilege on each named table. + + + The subscription must be disabled, and its apply worker must have + stopped, before this command can be used. The command is also not + allowed when the subscription has + two_phase + commit enabled, because the tables are copied again. + + + Only tables that the subscription actually tracks, as listed in + pg_subscription_rel, + can be named. For a partitioned table this depends on the publication: + with + publish_via_partition_root, + the partitioned table itself is tracked and refreshing it re-copies the + whole hierarchy; otherwise the individual partitions are tracked and + those are what can be refreshed. Refreshing one partition leaves the + others untouched. + + + Tables in an inheritance hierarchy are refreshed individually. + Refreshing a parent discards only the parent's own rows, never those of + its inheritance children, since a child is a separate relation which may + be subscribed in its own right or may hold data that is not replicated + at all. Name the children as well to refresh them too. + + + A table that another subscription also populates cannot be refreshed. + Discarding it would throw away the rows maintained by that subscription, + which would not copy them again, so the command reports an error instead. + This includes the case where the other subscription publishes a + partitioned ancestor of the table and routes rows into it. + + + + This command discards the local copy of the named tables. Any data + present only on the subscriber, and any changes not yet replicated back + to the publisher, are lost. + + + + + ENABLE diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 8e8db08bd93..82d555f2f0a 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -15,6 +15,7 @@ #include "postgres.h" #include "access/commit_ts.h" +#include "access/genam.h" #include "access/htup_details.h" #include "access/table.h" #include "access/twophase.h" @@ -28,6 +29,8 @@ #include "catalog/pg_authid_d.h" #include "catalog/pg_database_d.h" #include "catalog/pg_foreign_server.h" +#include "catalog/partition.h" +#include "catalog/pg_inherits.h" #include "catalog/pg_namespace.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" @@ -53,10 +56,12 @@ #include "storage/lock.h" #include "utils/acl.h" #include "utils/builtins.h" +#include "utils/fmgroids.h" #include "utils/guc.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/pg_lsn.h" +#include "utils/rel.h" #include "utils/syscache.h" /* @@ -1478,6 +1483,377 @@ AlterSubscription_refresh_seq(Subscription *sub, char *conninfo) } } +/* + * 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. + */ +/* + * Refuse to re-synchronize relations that another subscription also populates. + * + * REFRESH TABLE discards a relation and relies on this subscription's table + * synchronization to refill it. A relation is fed by any subscription that + * tracks it directly, and, when it is a partition, by any subscription that + * tracks a partitioned ancestor and routes tuples down into it. If such + * another subscription exists, its rows would be discarded here while its + * state stays untouched, so nothing would ever copy them back. We cannot + * reset another subscription's state on its behalf -- that would need its + * workers stopped and its own tablesync slots dropped over its own connection + * -- so this combination is rejected rather than silently losing data. + * + * subrelids is the set of relations named in the command, relids the full set + * that will be truncated, which additionally contains the partitions of any + * named partitioned table. + */ +static void +CheckRefreshTableNotShared(Relation pgsubrel, Subscription *sub, + List *subrelids, List *relids) +{ + foreach_oid(relid, relids) + { + List *feeders; + + /* + * Tuples reach a partition through its ancestors, so a subscription + * tracking any of them keeps this relation populated too. A named + * relation's own entry belongs to this subscription and is skipped + * below by the srsubid test. + */ + feeders = lappend_oid(get_partition_ancestors(relid), relid); + + foreach_oid(feederid, feeders) + { + ScanKeyData skey; + SysScanDesc scan; + HeapTuple tup; + + ScanKeyInit(&skey, Anum_pg_subscription_rel_srrelid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(feederid)); + + scan = systable_beginscan(pgsubrel, + SubscriptionRelSrrelidSrsubidIndexId, + true, NULL, 1, &skey); + + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_subscription_rel subrel; + char *othername; + + subrel = (Form_pg_subscription_rel) GETSTRUCT(tup); + + if (subrel->srsubid == sub->oid) + continue; + + othername = get_subscription_name(subrel->srsubid, false); + + /* + * Report the relation that is actually shared, which is not + * necessarily the one that was named, together with why it is + * in the truncation set. + */ + if (feederid != relid) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" is a partition of \"%s\", which is part of the subscription \"%s\"", + get_rel_name(relid), get_rel_name(feederid), + othername), + errdetail("Re-synchronizing it would discard rows that subscription \"%s\" routes into the partition but would not copy again.", + othername)); + else if (list_member_oid(subrelids, relid)) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" is also part of the subscription \"%s\"", + get_rel_name(relid), othername), + errdetail("Re-synchronizing it would discard rows that subscription \"%s\" maintains but would not copy again.", + othername)); + else + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" is also part of the subscription \"%s\"", + get_rel_name(relid), othername), + errdetail("Re-synchronizing a partitioned table truncates its partitions, discarding rows that subscription \"%s\" maintains but would not copy again.", + othername)); + } + + systable_endscan(scan); + } + + list_free(feeders); + } +} + +static void +AlterSubscription_refresh_table(Subscription *sub, List *relations) +{ + Relation rel; + ListCell *lc; + List *subrelids = NIL; + List *rels = NIL; + List *relids = NIL; + List *relids_logged = NIL; + List *originrelids = NIL; + List *slotrelids = NIL; + LOCKMODE lockmode = AccessExclusiveLock; + + /* + * 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. + * + * The relations are locked here at the level the truncate below needs, + * rather than being locked weakly now and upgraded later. + */ + foreach(lc, relations) + { + RangeVar *rv = lfirst_node(RangeVar, lc); + Oid relid; + char relstate; + XLogRecPtr relstatelsn; + Relation userrel; + AclResult aclresult; + + relid = RangeVarGetRelid(rv, lockmode, 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) + { + /* + * A partitioned table is tracked in pg_subscription_rel only when + * the publication uses publish_via_partition_root. Otherwise its + * partitions are tracked individually, and those are what can be + * refreshed, so point the user at them. + */ + if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" is not part of the subscription \"%s\"", + rv->relname, sub->name), + errhint("The publication may publish its partitions individually; refresh those instead.")); + + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" is not part of the subscription \"%s\"", + rv->relname, sub->name)); + } + + /* Don't complain about "REFRESH TABLE foo, foo". */ + if (list_member_oid(subrelids, relid)) + continue; + + /* + * The local copy is discarded below, so require the same privilege + * TRUNCATE itself would. + */ + aclresult = pg_class_aclcheck(relid, GetUserId(), ACL_TRUNCATE); + if (aclresult != ACLCHECK_OK) + aclcheck_error(aclresult, get_relkind_objtype(get_rel_relkind(relid)), + rv->relname); + + subrelids = lappend_oid(subrelids, relid); + + /* + * A relation that never reached READY may have left a tablesync + * origin behind; for READY the tablesync worker already dropped it. + */ + if (relstate != SUBREL_STATE_READY) + originrelids = lappend_oid(originrelids, relid); + + /* + * Only a relation caught mid-sync has a tablesync slot on the + * publisher; for READY and SYNCDONE the tablesync worker already + * dropped it. + */ + if (relstate != SUBREL_STATE_READY && relstate != SUBREL_STATE_SYNCDONE) + slotrelids = lappend_oid(slotrelids, relid); + + /* It may already be in the truncation set as another table's child. */ + if (list_member_oid(relids, relid)) + continue; + + /* We already hold the lock RangeVarGetRelid() took above. */ + userrel = table_open(relid, NoLock); + + rels = lappend(rels, userrel); + relids = lappend_oid(relids, relid); + if (RelationIsLogicallyLogged(userrel)) + relids_logged = lappend_oid(relids_logged, relid); + + /* + * A partitioned table holds no data itself, so its partitions have to + * be truncated as well. This mirrors apply_handle_truncate(). + * + * Only partitions are followed here, never inheritance children. The + * rule is that we must not discard a relation unless we also reset + * the state of whatever will refill it. A tracked partitioned root + * is tracked precisely because the publication uses + * publish_via_partition_root, in which case its partitions are not + * themselves in pg_subscription_rel and the root's re-copy refills + * them through tuple routing. An inheritance child, by contrast, is + * an independent relation that may be a subscription member in its + * own right, or may hold data that is not replicated at all; + * truncating it here would destroy rows that nothing is going to copy + * back. Naming it in the command refreshes it. + */ + if (userrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + List *children; + + children = find_all_inheritors(relid, lockmode, NULL); + + foreach_oid(childrelid, children) + { + Relation childrel; + + if (list_member_oid(relids, childrelid)) + continue; + + /* find_all_inheritors() already took the lock */ + childrel = table_open(childrelid, NoLock); + + /* + * Ignore temp tables of other backends, as ExecuteTruncate() + * does. + */ + if (RELATION_IS_OTHER_TEMP(childrel)) + { + table_close(childrel, lockmode); + continue; + } + + rels = lappend(rels, childrel); + relids = lappend_oid(relids, childrelid); + if (RelationIsLogicallyLogged(childrel)) + relids_logged = lappend_oid(relids_logged, childrelid); + } + } + } + + /* + * Everything that will be truncated has to be fed by this subscription + * alone, otherwise we would discard rows that only some other + * subscription would restore. Check before touching anything, so this + * stays all-or-nothing. + */ + CheckRefreshTableNotShared(rel, sub, subrelids, relids); + + /* + * Drop the tablesync origin of every relation that was caught mid-sync. + * Pass missing_ok = true as the origin may not exist yet. + */ + foreach_oid(relid, originrelids) + { + char originname[NAMEDATALEN]; + + ReplicationOriginNameForLogicalRep(sub->oid, relid, originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + } + + /* + * 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. + */ + ExecuteTruncateGuts(rels, relids, relids_logged, DROP_RESTRICT, false, + false); + + foreach(lc, rels) + table_close((Relation) lfirst(lc), NoLock); + + /* + * Reset each relation to init so that a tablesync worker re-copies it + * once the subscription is enabled. + */ + foreach_oid(relid, subrelids) + { + 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)); + } + + /* + * Finally, drop the tablesync slots left on the publisher by relations + * that were caught mid-sync. + * + * This is deliberately last. Dropping a remote slot is not + * transactional, so it cannot be undone if a later step fails, and + * everything above can still fail for ordinary reasons -- the truncate in + * particular, on a foreign key reference or a lock timeout. One + * connection serves all of the slots, and the common case of re-seeding + * relations that are all in ready state needs no publisher connection at + * all. + */ + if (slotrelids != NIL) + { + char *err = NULL; + WalReceiverConn *wrconn; + bool must_use_password; + + /* Load the library providing us libpq calls. */ + load_file("libpqwalreceiver", false); + + must_use_password = sub->passwordrequired && !sub->ownersuperuser; + wrconn = walrcv_connect(SubscriptionConninfo(sub), 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(); + { + foreach_oid(relid, slotrelids) + { + char syncslotname[NAMEDATALEN] = {0}; + + ReplicationSlotNameForTablesync(sub->oid, relid, syncslotname, + sizeof(syncslotname)); + ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); + } + } + PG_FINALLY(); + { + walrcv_disconnect(wrconn); + } + PG_END_TRY(); + } + + table_close(rel, NoLock); +} + /* * Common checks for altering failover, two_phase, and retain_dead_tuples * options. @@ -2384,6 +2760,64 @@ 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.")); + + /* + * The relations are re-initialized below, and that must not + * happen once two_phase is enabled. See + * ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for the details. + * There is no copy_data = false exception here, as this + * command exists precisely to copy the data again. + */ + if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("%s is not allowed when two_phase is enabled", + "ALTER SUBSCRIPTION ... REFRESH TABLE"), + errhint("Use ALTER SUBSCRIPTION ... SET (two_phase = false), or use DROP/CREATE SUBSCRIPTION.")); + + /* + * Being marked disabled is not the same as having stopped: + * ALTER SUBSCRIPTION ... DISABLE only wakes the workers so + * that they notice the change, and they exit asynchronously + * afterwards. An apply worker may still have the relation + * state cached, and a tablesync worker outlives its apply + * worker while it finishes a copy, so wait for all of them + * rather than stopping any of them ourselves. + * + * only_running is false, as DROP SUBSCRIPTION does for the + * same reason: a worker that is in_use but has not attached + * yet is the dangerous one, since it is about to read the + * relation state we are rewriting. + */ + if (logicalrep_workers_find(subid, false, true)) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot %s when logical replication worker is still running", + "ALTER SUBSCRIPTION ... REFRESH TABLE"), + errhint("Try again after some time.")); + + 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 17035fb4d15..8143598eaad 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -11627,6 +11627,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/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 190fff7ea0e..eeb6abd45d9 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2348,12 +2348,19 @@ match_previous_words(int pattern_id, else if (Matches("ALTER", "SUBSCRIPTION", MatchAny)) COMPLETE_WITH("CONNECTION", "ENABLE", "DISABLE", "OWNER TO", "RENAME TO", "REFRESH PUBLICATION", "REFRESH SEQUENCES", - "SERVER", "SET", "SKIP (", "ADD PUBLICATION", "DROP PUBLICATION"); + "REFRESH TABLE", "SERVER", "SET", "SKIP (", + "ADD PUBLICATION", "DROP PUBLICATION"); else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, "SERVER")) COMPLETE_WITH_QUERY(Query_for_list_of_servers); /* ALTER SUBSCRIPTION REFRESH */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH")) - COMPLETE_WITH("PUBLICATION", "SEQUENCES"); + COMPLETE_WITH("PUBLICATION", "SEQUENCES", "TABLE"); + /* ALTER SUBSCRIPTION REFRESH TABLE */ + else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, "REFRESH", "TABLE")) + COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables); + else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny, "REFRESH", "TABLE") && + ends_with(prev_wd, ',')) + COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables); /* ALTER SUBSCRIPTION REFRESH PUBLICATION */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION")) COMPLETE_WITH("WITH ("); diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 8a9df884276..6d40cba7b92 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -4608,6 +4608,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; @@ -4620,6 +4621,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..ee7868c232c --- /dev/null +++ b/src/test/subscription/t/039_refresh_table.pl @@ -0,0 +1,710 @@ + +# 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->append_conf('postgresql.conf', + "max_prepared_transactions = 10"); +# Several subscriptions coexist below, each needing an apply worker plus table +# synchronization workers. +$node_subscriber->append_conf('postgresql.conf', + "max_logical_replication_workers = 12"); +$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'); + +# A partitioned table holds no data itself, so refreshing it has to truncate +# its partitions as well. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_part (a int primary key, b text) PARTITION BY RANGE (a); + CREATE TABLE tab_part_1 PARTITION OF tab_part FOR VALUES FROM (1) TO (51); + CREATE TABLE tab_part_2 PARTITION OF tab_part FOR VALUES FROM (51) TO (101); + INSERT INTO tab_part SELECT g, 'r' || g FROM generate_series(1, 100) g; + CREATE PUBLICATION tap_pub_part FOR TABLE tab_part + WITH (publish_via_partition_root = true); +)); + +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_part (a int primary key, b text) PARTITION BY RANGE (a); + CREATE TABLE tab_part_1 PARTITION OF tab_part FOR VALUES FROM (1) TO (51); + CREATE TABLE tab_part_2 PARTITION OF tab_part FOR VALUES FROM (51) TO (101); +)); + +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_part CONNECTION '$publisher_connstr' PUBLICATION tap_pub_part" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_part'); + +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_part"), + '100', + 'initial sync of partitioned table'); + +# Publishing via the root means the subscription tracks the root, not the +# individual partitions. +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT string_agg(c.relname, ',' ORDER BY c.relname) FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid JOIN pg_subscription s ON s.oid = r.srsubid WHERE s.subname = 'tap_sub_part'" + ), + 'tab_part', + 'subscription tracks the partitioned root'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_part DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_part' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +$node_subscriber->safe_psql('postgres', "DELETE FROM tab_part WHERE a <= 30"); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_part REFRESH TABLE tab_part"); + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT (SELECT count(*) FROM tab_part_1) + (SELECT count(*) FROM tab_part_2)" + ), + '0', + 'partitions truncated by REFRESH TABLE on the partitioned root'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_part ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_part'); + +my $pub_part = $node_publisher->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_part"); +my $sub_part = $node_subscriber->safe_psql('postgres', + "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_part"); +is($sub_part, $pub_part, 'partitioned table matches publisher after resync'); + +# Without publish_via_partition_root it is the partitions that are tracked, not +# the root, so naming the root should point the user at them. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_leaf (a int primary key, b text) PARTITION BY RANGE (a); + CREATE TABLE tab_leaf_1 PARTITION OF tab_leaf FOR VALUES FROM (1) TO (51); + CREATE TABLE tab_leaf_2 PARTITION OF tab_leaf FOR VALUES FROM (51) TO (101); + INSERT INTO tab_leaf SELECT g, 's' || g FROM generate_series(1, 100) g; + CREATE PUBLICATION tap_pub_leaf FOR TABLE tab_leaf; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_leaf (a int primary key, b text) PARTITION BY RANGE (a); + CREATE TABLE tab_leaf_1 PARTITION OF tab_leaf FOR VALUES FROM (1) TO (51); + CREATE TABLE tab_leaf_2 PARTITION OF tab_leaf FOR VALUES FROM (51) TO (101); +)); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_leaf CONNECTION '$publisher_connstr' PUBLICATION tap_pub_leaf" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_leaf'); + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT string_agg(c.relname, ',' ORDER BY c.relname) FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid JOIN pg_subscription s ON s.oid = r.srsubid WHERE s.subname = 'tap_sub_leaf'" + ), + 'tab_leaf_1,tab_leaf_2', + 'without publish_via_partition_root the partitions are tracked'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_leaf DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_leaf' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +my ($lret, $lout, $lerr) = ('', '', ''); +$lret = $node_subscriber->psql( + 'postgres', + "ALTER SUBSCRIPTION tap_sub_leaf REFRESH TABLE tab_leaf", + stdout => \$lout, + stderr => \$lerr); +ok( $lret != 0 + && $lerr =~ /is not part of the subscription/ + && $lerr =~ /partitions individually/, + 'naming an untracked partitioned root points at its partitions' +) or diag("got ret=$lret stderr=$lerr"); + +# Refreshing one partition must leave the sibling partitions alone. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_leaf REFRESH TABLE tab_leaf_1"); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT (SELECT count(*) FROM tab_leaf_1) || '/' || (SELECT count(*) FROM tab_leaf_2)" + ), + '0/50', + 'refreshing one partition leaves its siblings alone'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_leaf ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_leaf'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_leaf"), + '100', + 'refreshed partition re-copied after enable'); + +# Refreshing an inheritance parent must not discard its children. A child is an +# independent relation: it may be a subscription member in its own right, in +# which case truncating it here would leave rows that nothing copies back. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_inh_p (a int, b text); + CREATE TABLE tab_inh_c (a int, b text) INHERITS (tab_inh_p); + INSERT INTO tab_inh_p VALUES (1, 'parent1'), (2, 'parent2'); + INSERT INTO tab_inh_c VALUES (101, 'child1'), (102, 'child2'); + CREATE PUBLICATION tap_pub_inh FOR TABLE tab_inh_p, tab_inh_c; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_inh_p (a int, b text); + CREATE TABLE tab_inh_c (a int, b text) INHERITS (tab_inh_p); +)); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_inh CONNECTION '$publisher_connstr' PUBLICATION tap_pub_inh" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_inh'); + +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM ONLY tab_inh_p"), + '2', + 'inheritance parent synced'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_inh_c"), + '2', + 'inheritance child synced'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_inh DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_inh' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_inh REFRESH TABLE tab_inh_p"); + +# Only the parent's own rows are discarded, and only the parent is reset. +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM ONLY tab_inh_p"), + '0', + 'inheritance parent truncated by REFRESH TABLE'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_inh_c"), + '2', + 'inheritance child left intact by REFRESH TABLE'); +is( $node_subscriber->safe_psql( + "postgres", + "SELECT string_agg(c.relname || '=' || r.srsubstate::text, ',' ORDER BY c.relname) FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid JOIN pg_subscription s ON s.oid = r.srsubid WHERE s.subname = 'tap_sub_inh'" + ), + 'tab_inh_c=r,tab_inh_p=i', + 'only the named inheritance parent is reset'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_inh ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_inh'); + +# The child must still hold its rows: nothing would have copied them back. +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM ONLY tab_inh_p"), + '2', + 'inheritance parent re-copied after enable'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_inh_c"), + '2', + 'inheritance child not lost by refreshing the parent'); + +# Naming both refreshes both. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_inh DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_inh' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_inh REFRESH TABLE tab_inh_p, tab_inh_c"); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_inh_p"), + '0', + 'naming parent and child discards both'); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_inh ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_inh'); +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM ONLY tab_inh_p"), + '2', + 'parent re-copied after refreshing both'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_inh_c"), + '2', + 'child re-copied after refreshing both'); + +# Done with the subscriptions above; free their workers before adding more. +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_part"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_leaf"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_inh"); + +# A table referenced by a foreign key can only be re-synchronized together with +# the referencing table, because the truncate is rejected on the existence of +# the constraint rather than on the referencing rows. This is what naming +# several tables in one command is for. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_fk_p (id int primary key); + CREATE TABLE tab_fk_c (id int primary key, pid int references tab_fk_p(id)); + INSERT INTO tab_fk_p SELECT generate_series(1, 5); + INSERT INTO tab_fk_c SELECT g, g FROM generate_series(1, 5) g; + CREATE PUBLICATION tap_pub_fk FOR TABLE tab_fk_p, tab_fk_c; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_fk_p (id int primary key); + CREATE TABLE tab_fk_c (id int primary key, pid int references tab_fk_p(id)); +)); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_fk CONNECTION '$publisher_connstr' PUBLICATION tap_pub_fk" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_fk'); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_fk DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_fk' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub_fk REFRESH TABLE tab_fk_p", + qr/cannot truncate a table referenced in a foreign key constraint/, + 'REFRESH TABLE of a foreign key target alone is rejected'); + +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_fk_p"), + '5', + 'the foreign key target keeps its rows after the rejected command'); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FILTER (WHERE srsubstate <> 'r') FROM pg_subscription_rel r + JOIN pg_subscription s ON s.oid = r.srsubid WHERE s.subname = 'tap_sub_fk'" + ), + '0', + 'no state was reset by the rejected command'); + +# Naming both succeeds, since they are truncated in one command. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_fk REFRESH TABLE tab_fk_p, tab_fk_c"); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_fk_p"), + '0', + 'foreign key target discarded when named with its referencing table'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_fk ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_fk'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_fk_p"), + '5', + 'foreign key target re-copied'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_fk_c"), + '5', + 'referencing table re-copied'); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_fk"); + +# A relation that a second subscription also populates cannot be +# re-synchronized: its rows would be discarded while that subscription's state +# stays untouched, so nothing would copy them back. Refuse instead. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_shared (a int primary key, b text); + INSERT INTO tab_shared SELECT g, 'v' || g FROM generate_series(1, 20) g; + CREATE PUBLICATION tap_pub_lo FOR TABLE tab_shared WHERE (a < 10); + CREATE PUBLICATION tap_pub_hi FOR TABLE tab_shared WHERE (a >= 10); +)); +$node_subscriber->safe_psql('postgres', + "CREATE TABLE tab_shared (a int primary key, b text)"); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_lo CONNECTION '$publisher_connstr' PUBLICATION tap_pub_lo" +); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_hi CONNECTION '$publisher_connstr' PUBLICATION tap_pub_hi" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_lo'); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_hi'); + +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM tab_shared"), + '20', + 'both subscriptions populated the shared table'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_hi DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_hi' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub_hi REFRESH TABLE tab_shared", + qr/table "tab_shared" is also part of the subscription "tap_sub_lo"/, + 'REFRESH TABLE rejects a table another subscription also populates'); + +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM tab_shared"), + '20', + 'the shared table is left untouched by the rejected command'); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_lo"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_hi"); + +# The same hazard reaches partitions through tuple routing: one subscription +# tracks the partitioned root, another tracks a partition directly. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_cross (a int primary key, b text) PARTITION BY RANGE (a); + CREATE TABLE tab_cross_1 PARTITION OF tab_cross FOR VALUES FROM (1) TO (51); + CREATE TABLE tab_cross_2 PARTITION OF tab_cross FOR VALUES FROM (51) TO (101); + INSERT INTO tab_cross SELECT g, 'w' || g FROM generate_series(1, 100) g; + CREATE PUBLICATION tap_pub_croot FOR TABLE tab_cross + WITH (publish_via_partition_root = true); + CREATE PUBLICATION tap_pub_cleaf FOR TABLE tab_cross_1; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_cross (a int primary key, b text) PARTITION BY RANGE (a); + CREATE TABLE tab_cross_1 PARTITION OF tab_cross FOR VALUES FROM (1) TO (51); + CREATE TABLE tab_cross_2 PARTITION OF tab_cross FOR VALUES FROM (51) TO (101); +)); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_croot CONNECTION '$publisher_connstr' PUBLICATION tap_pub_croot" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, + 'tap_sub_croot'); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_cleaf CONNECTION '$publisher_connstr' PUBLICATION tap_pub_cleaf WITH (copy_data = false)" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, + 'tap_sub_cleaf'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_croot DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_croot' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +# Naming the root would truncate the partition owned by the other subscription. +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub_croot REFRESH TABLE tab_cross", + qr/table "tab_cross_1" is also part of the subscription "tap_sub_cleaf"/, + 'REFRESH TABLE rejects a root whose partition another subscription populates' +); + +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_cross"), + '100', + 'the partitioned table is left untouched by the rejected command'); + +# And the reverse: naming the partition, whose ancestor the other subscription +# tracks and routes tuples into. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_cleaf DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_cleaf' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub_cleaf REFRESH TABLE tab_cross_1", + qr/table "tab_cross_1" is a partition of "tab_cross", which is part of the subscription "tap_sub_croot"/, + 'REFRESH TABLE rejects a partition whose ancestor another subscription tracks' +); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_croot"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_cleaf"); + +# Re-copying a table is not allowed once two_phase is enabled. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_2pc (a int primary key); + INSERT INTO tab_2pc SELECT generate_series(1, 10); + CREATE PUBLICATION tap_pub_2pc FOR TABLE tab_2pc; +)); +$node_subscriber->safe_psql('postgres', + "CREATE TABLE tab_2pc (a int primary key)"); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_2pc CONNECTION '$publisher_connstr' PUBLICATION tap_pub_2pc WITH (two_phase = on)" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_2pc'); + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_2pc'" + ), + 'e', + 'two_phase is enabled on tap_sub_2pc'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_2pc DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_2pc' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub_2pc REFRESH TABLE tab_2pc", + qr/not allowed when two_phase is enabled/, + 'REFRESH TABLE rejected when two_phase is enabled'); + +# The local copy is discarded, so the caller needs TRUNCATE on the table. +# +# Ownership is transferred while the role is still a superuser, because +# handing a subscription to a non-superuser requires a password in the +# connection string. The privilege being tested is only checked afterwards. +# +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE ROLE regress_refresh_user LOGIN SUPERUSER; + GRANT pg_create_subscription TO regress_refresh_user; + GRANT CREATE ON DATABASE postgres TO regress_refresh_user; + GRANT ALL ON TABLE tab_2pc TO regress_refresh_user; + REVOKE TRUNCATE ON TABLE tab_2pc FROM regress_refresh_user; +)); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_2pc SET (two_phase = false)"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_2pc OWNER TO regress_refresh_user"); +$node_subscriber->safe_psql('postgres', + "ALTER ROLE regress_refresh_user NOSUPERUSER"); + +my ($acl_ret, $acl_out, $acl_err) = ('', '', ''); +$acl_ret = $node_subscriber->psql( + 'postgres', + "ALTER SUBSCRIPTION tap_sub_2pc REFRESH TABLE tab_2pc", + extra_params => [ '-U', 'regress_refresh_user' ], + stdout => \$acl_out, + stderr => \$acl_err); +ok( $acl_ret != 0 && $acl_err =~ /permission denied/, + 'REFRESH TABLE requires TRUNCATE privilege on the table' +) or diag("got ret=$acl_ret stderr=$acl_err"); + +$node_subscriber->safe_psql('postgres', + "GRANT TRUNCATE ON TABLE tab_2pc TO regress_refresh_user"); +$node_subscriber->safe_psql( + 'postgres', + "ALTER SUBSCRIPTION tap_sub_2pc REFRESH TABLE tab_2pc", + extra_params => [ '-U', 'regress_refresh_user' ]); +is($node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_2pc"), + '0', 'REFRESH TABLE accepted once TRUNCATE privilege is granted'); + +# Dropping tap_sub_2pc has to reach the publisher to remove its slot, which +# its now passwordless non-superuser owner cannot do. +$node_subscriber->safe_psql('postgres', + "ALTER ROLE regress_refresh_user SUPERUSER"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_2pc"); +$node_subscriber->safe_psql('postgres', "DROP OWNED BY regress_refresh_user"); +$node_subscriber->safe_psql('postgres', "DROP ROLE regress_refresh_user"); +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); base-commit: 40af05bb6157670482ae31fd5b0a12ff2992c966 -- 2.54.0