From cd16c8c4795d89eeb41312ddfdb7eea15642a72f Mon Sep 17 00:00:00 2001 From: cagrib Date: Sun, 1 Nov 2026 15:31:45 +0100 Subject: [PATCH v13 2/2] Allow ALTER SUBSCRIPTION ... REFRESH TABLE on partitioned tables Let the command accept a partitioned table and a partition. A partitioned table holds no data itself, so re-synchronizing one has to truncate its partitions as well, which mirrors what replicating a TRUNCATE of that table does. Only partitions are followed, never inheritance children: a tracked partitioned root is tracked because the publication uses publish_via_partition_root, so its partitions are refilled by the root's re-copy through tuple routing, whereas an inheritance child is an independent relation that may be subscribed in its own right or hold data that is not replicated at all. This widens the set of relations the command discards, so the check that no other subscription feeds them has to widen with it: tuples reach a partition through its ancestors, so a subscription that tracks any partitioned ancestor keeps the partition populated too. Those ancestors are not truncated and would otherwise not be locked, so lock them at the same level, which keeps the rule uniform that every relation the check consults is held at a level that conflicts with registering a subscription for it. Without that, CREATE SUBSCRIPTION could register one for an ancestor while the command runs, and WITH (copy_data = false) would record it as ready and never copy back what the truncate discards. The truncation set is collected under AccessShareLock, which ATTACH PARTITION and DETACH PARTITION CONCURRENTLY do not conflict with, so it is derived again once the locks are held: a partition attached in the meantime has to be truncated too, and one detached in the meantime has to be left alone. The named relations are also sorted so that an ancestor is locked before its descendants, the order the rest of the system uses for a partition hierarchy, because a root and its partition can both be members of the same subscription and the order they are named in is the user's to choose. The new test uses an injection point to hold the command between its final check and its commit, which is where a concurrently registered subscription would have to slip in. --- doc/src/sgml/ref/alter_subscription.sgml | 17 +- src/backend/commands/subscriptioncmds.c | 398 +++++++++++++++--- src/test/subscription/meson.build | 1 + src/test/subscription/t/039_refresh_table.pl | 300 ++++++++++++- .../subscription/t/040_refresh_table_race.pl | 139 ++++++ 5 files changed, 774 insertions(+), 81 deletions(-) create mode 100644 src/test/subscription/t/040_refresh_table_race.pl diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index d0f96e727fa..13c3ab4f213 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -303,7 +303,20 @@ ALTER SUBSCRIPTION name RENAME TO < Only tables that the subscription actually tracks, as listed in pg_subscription_rel, - can be named. Partitioned tables and partitions cannot be refreshed. + 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. + + + Re-copying a partitioned table truncates every one of its partitions, + just as replicating a TRUNCATE of that table does. + That includes a partition which exists only on the subscriber, or which + holds rows the publisher does not have: those rows are discarded and + nothing copies them back. See for details of how @@ -342,6 +355,8 @@ ALTER SUBSCRIPTION name RENAME TO < 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. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 39238d2fc9f..a76ad1b2312 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -29,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" @@ -1501,53 +1503,270 @@ AlterSubscription_refresh_seq(Subscription *sub, char *conninfo) * 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. If another subscription tracks the same - * relation, 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. + * 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 CheckRefreshTableNotInOtherSubscriptions(Relation pgsubrel, Subscription *sub, - List *relids) + List *subrelids, List *relids) { foreach_oid(relid, relids) { - ScanKeyData skey; - SysScanDesc scan; - HeapTuple tup; + 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. + * + * Only partitions have such feeders. get_partition_ancestors() scans + * pg_inherits without regard to relispartition, so for an inheritance + * child it would return the parent as well, and a parent does not + * route rows into its inheritance children. + */ + if (get_rel_relispartition(relid)) + feeders = lappend_oid(get_partition_ancestors(relid), relid); + else + feeders = list_make1_oid(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); - ScanKeyInit(&skey, Anum_pg_subscription_rel_srrelid, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(relid)); + /* + * 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); + } - scan = systable_beginscan(pgsubrel, - SubscriptionRelSrrelidSrsubidIndexId, - true, NULL, 1, &skey); + list_free(feeders); + } +} - while (HeapTupleIsValid(tup = systable_getnext(scan))) +/* + * Order the relations named in REFRESH TABLE so that a partitioned ancestor + * comes before its descendants. + * + * They are locked in this order further down, and the rest of the system + * takes a partition hierarchy from the root downwards: an INSERT into a + * partitioned table locks the root and then the partition it routes into. + * Locking a partition before its root deadlocks against that, and the command + * can be asked to, because both a root and its partition can be members of + * the same subscription when the publisher has them as independent tables, + * and the order they are named in is the user's to choose. + * + * Relations that are not related to each other keep the order they were named + * in. Only the named relations need this: the partitions collected for a + * named partitioned table come from find_all_inheritors(), which already + * returns each of them after its root. + */ +static List * +SortRefreshTableRelations(List *subrelids) +{ + List *result = NIL; + int nrels = list_length(subrelids); + int *depth; + int maxdepth = 0; + int i; + + if (nrels < 2) + return subrelids; + + depth = palloc(nrels * sizeof(int)); + + i = 0; + foreach_oid(relid, subrelids) + { + List *ancestors = get_partition_ancestors(relid); + + /* A relation is always deeper than any of its ancestors. */ + depth[i] = list_length(ancestors); + maxdepth = Max(maxdepth, depth[i]); + list_free(ancestors); + i++; + } + + for (int d = 0; d <= maxdepth; d++) + { + i = 0; + foreach_oid(relid, subrelids) { - Form_pg_subscription_rel subrel; - char *othername; + if (depth[i] == d) + result = lappend_oid(result, relid); + i++; + } + } - subrel = (Form_pg_subscription_rel) GETSTRUCT(tup); + pfree(depth); + list_free(subrelids); - if (subrel->srsubid == sub->oid) - continue; + return result; +} - othername = get_subscription_name(subrel->srsubid, false); +/* + * Lock the relations that CheckRefreshTableNotInOtherSubscriptions() consults + * but that are not themselves truncated. + * + * That check can only see the subscriptions that exist while it runs, so + * something has to keep another one from appearing for the same relations + * before this command commits. CREATE SUBSCRIPTION and ALTER SUBSCRIPTION + * ... REFRESH PUBLICATION both resolve a relation with AccessShareLock before + * registering it in pg_subscription_rel, so AccessExclusiveLock on that + * relation blocks them, and nothing weaker conflicts with AccessShareLock. + * + * The relations in the truncation set are locked that way in any case, because + * they are about to be truncated. A partitioned ancestor is not in the set + * and would otherwise not be locked at all, even though the check consults it: + * a subscription tracking an ancestor routes rows down into the partition, so + * it feeds a relation this command discards. Locking the ancestors here makes + * the rule uniform, namely that every relation the check looks at is held at a + * level that conflicts with registering a subscription for it. + * + * Ancestors are locked before their descendants, the order used elsewhere for + * a partition hierarchy. A relation already in the truncation set is left to + * the caller, which locks that set in an order that is itself top-down + * because SortRefreshTableRelations() put it in one. + */ +static void +LockRefreshTableFeeders(List *relids) +{ + foreach_oid(relid, relids) + { + List *ancestors; - 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)); + /* Only a partition has ancestors that route rows into it. */ + if (!get_rel_relispartition(relid)) + continue; + + ancestors = get_partition_ancestors(relid); + + /* get_partition_ancestors() returns the closest parent first. */ + for (int i = list_length(ancestors) - 1; i >= 0; i--) + { + Oid ancestorid = list_nth_oid(ancestors, i); + + if (!list_member_oid(relids, ancestorid)) + LockRelationOid(ancestorid, AccessExclusiveLock); } - systable_endscan(scan); + list_free(ancestors); + } +} + +/* + * Expand the relations named in REFRESH TABLE into the set that has to be + * truncated: each of them, plus the partitions of any that are partitioned. + * + * This mirrors apply_handle_truncate(). Only partitions are followed, never + * inheritance children; see the comment on the invariant in + * AlterSubscription_refresh_table(). + * + * The result is used both to build the truncation set and, once the relations + * are locked, to confirm that it has not changed in the meantime, so the two + * must be derived the same way. + */ +static List * +RefreshTableTruncationSet(List *subrelids, LOCKMODE lockmode) +{ + List *result = NIL; + + foreach_oid(relid, subrelids) + { + /* It may already be in the set as another named table's partition. */ + if (list_member_oid(result, relid)) + continue; + + result = lappend_oid(result, relid); + + if (get_rel_relkind(relid) != RELKIND_PARTITIONED_TABLE) + continue; + + foreach_oid(childrelid, find_all_inheritors(relid, lockmode, NULL)) + { + Relation childrel; + + if (list_member_oid(result, 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; + } + + /* Keep the lock, the caller opens the relation again later. */ + table_close(childrel, NoLock); + result = lappend_oid(result, childrelid); + } } + + return result; } /* @@ -1621,31 +1840,28 @@ AlterSubscription_refresh_table(Subscription *sub, List *relations) rv->relname), errhint("Use ALTER SUBSCRIPTION ... REFRESH SEQUENCES instead.")); - /* - * Partitioned tables are not supported yet. Re-synchronizing one has - * to truncate its partitions, and the relations that feed them are - * then not the named relation alone, which the checks below are not - * prepared for. - */ - if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot refresh partitioned table \"%s\"", - rv->relname)); - - if (get_rel_relispartition(relid)) - ereport(ERROR, - errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot refresh table \"%s\" because it is a partition", - rv->relname)); - /* 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)) @@ -1653,7 +1869,9 @@ AlterSubscription_refresh_table(Subscription *sub, List *relations) /* * The local copy is discarded below, so require the same privilege - * TRUNCATE itself would. + * TRUNCATE itself would. As in ExecuteTruncate(), the privilege is + * required of the relations named in the command, not of the + * partitions that come with them. * * This is not the check the command relies on, because REVOKE can * still take the privilege away afterwards; it runs here so that a @@ -1683,7 +1901,28 @@ AlterSubscription_refresh_table(Subscription *sub, List *relations) slotrelids = lappend_oid(slotrelids, relid); } - relids = subrelids; + /* + * Lock the hierarchy from the root down, whatever order the relations + * were named in. + */ + subrelids = SortRefreshTableRelations(subrelids); + + /* + * A partitioned table holds no data itself, so its partitions have to be + * truncated as well. + * + * Only partitions are followed, 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. + */ + relids = RefreshTableTruncationSet(subrelids, AccessShareLock); /* * Everything that will be truncated has to be fed by this subscription @@ -1698,18 +1937,39 @@ AlterSubscription_refresh_table(Subscription *sub, List *relations) * runs below with those locks held. */ pgsubrel = table_open(SubscriptionRelRelationId, AccessShareLock); - CheckRefreshTableNotInOtherSubscriptions(pgsubrel, sub, relids); + CheckRefreshTableNotInOtherSubscriptions(pgsubrel, sub, subrelids, relids); table_close(pgsubrel, AccessShareLock); /* * The relation set is now known to be fed by this subscription alone, and * this subscription has no running workers, so nothing that replicates * into these relations is still holding them. Take the level the - * truncate needs, which is also what makes the re-check below conclusive. + * truncate needs, and the same level on the ancestors the check above + * relies on, so that the re-check below is conclusive. */ + LockRefreshTableFeeders(relids); + foreach_oid(relid, relids) LockRelationOid(relid, AccessExclusiveLock); + /* + * The set used to take those locks was collected under AccessShareLock, + * which ATTACH PARTITION and DETACH PARTITION CONCURRENTLY do not + * conflict with, so a partition may have come or gone since. Derive it + * again now that the locks are held and use that instead: a partition + * attached in the meantime has to be truncated as well, or the re-copy + * would add rows on top of the ones it kept, and one detached in the + * meantime has to be left alone, as this subscription no longer feeds it + * and nothing would copy its rows back. + * + * A relation that has dropped out of the set stays locked until commit, + * which is harmless. Nothing can change the set from here on: ATTACH and + * DETACH both need a lock on the root that AccessExclusiveLock conflicts + * with, and find_all_inheritors() has just locked any newly attached + * partition at that same level. + */ + relids = RefreshTableTruncationSet(subrelids, AccessExclusiveLock); + /* * Every relation lock is held, so pg_subscription_rel can now be opened * without any risk of waiting for one while holding it. Take @@ -1718,22 +1978,27 @@ AlterSubscription_refresh_table(Subscription *sub, List *relations) pgsubrel = table_open(SubscriptionRelRelationId, RowExclusiveLock); /* - * Check membership again with the locks held. The check above runs first - * only so that the common rejection does not have to wait for a tablesync - * worker to finish copying, and it is not conclusive on its own: it held - * pg_subscription_rel at AccessShareLock and let it go again, and even - * held at RowExclusiveLock it would prove nothing, as that is also what - * CREATE SUBSCRIPTION takes. What makes this one conclusive is that - * every relation it consults is now held at AccessExclusiveLock, so no - * further subscription can be registered for any of them until this - * command commits. It matters because CREATE SUBSCRIPTION ... WITH - * (copy_data = false) records the relation as ready, and would therefore - * never copy back what the truncate discards. + * With the set settled, check its membership again. The check above runs + * first only so that the common rejection does not have to wait for a + * tablesync worker to finish copying, and it is not conclusive on its + * own: it held pg_subscription_rel at AccessShareLock and let it go + * again, and even held at RowExclusiveLock it would prove nothing, as + * that is also what CREATE SUBSCRIPTION takes. What makes this one + * conclusive is that every relation it consults is now held at + * AccessExclusiveLock, the truncation set and the partitioned ancestors + * that route rows into it alike, so no further subscription can be + * registered for any of them until this command commits. It matters + * because CREATE SUBSCRIPTION ... WITH (copy_data = false) records the + * relation as ready, and would therefore never copy back what the + * truncate discards. */ - CheckRefreshTableNotInOtherSubscriptions(pgsubrel, sub, relids); + CheckRefreshTableNotInOtherSubscriptions(pgsubrel, sub, subrelids, relids); + + INJECTION_POINT("subscription-refresh-table-after-subscription-check", NULL); /* - * Open what is about to be truncated. The locks are already held. + * The set is settled, so open what is about to be truncated. The locks + * are already held. */ foreach_oid(relid, relids) { @@ -1767,7 +2032,8 @@ AlterSubscription_refresh_table(Subscription *sub, List *relations) * taken. This command resolves its relations without a callback, so * it has to repeat the check itself. */ - truncate_check_perms(relid, userrel->rd_rel); + if (list_member_oid(subrelids, relid)) + truncate_check_perms(relid, userrel->rd_rel); } /* diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index 37f480bcac2..49bfbcbfd40 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -49,6 +49,7 @@ tests += { 't/037_except.pl', 't/038_walsnd_shutdown_timeout.pl', 't/039_refresh_table.pl', + 't/040_refresh_table_race.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_refresh_table.pl b/src/test/subscription/t/039_refresh_table.pl index 2a5077659bb..e7591415245 100644 --- a/src/test/subscription/t/039_refresh_table.pl +++ b/src/test/subscription/t/039_refresh_table.pl @@ -4,7 +4,7 @@ # 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, and does not accept partitioned tables or partitions. +# to be disabled. use strict; use warnings FATAL => 'all'; use PostgreSQL::Test::Cluster; @@ -47,8 +47,6 @@ $node_subscriber->safe_psql( CREATE TABLE tab_other (a int primary key, b text); CREATE TABLE tab_local (a int primary key); CREATE SEQUENCE seq_local; - CREATE TABLE tab_pt (a int primary key) PARTITION BY RANGE (a); - CREATE TABLE tab_pt_1 PARTITION OF tab_pt FOR VALUES FROM (1) TO (11); )); $node_subscriber->safe_psql('postgres', @@ -101,17 +99,6 @@ refresh_should_fail( qr/cannot refresh sequence/, 'REFRESH TABLE rejected for a sequence'); -# Neither a partitioned table nor a partition can be refreshed yet. -refresh_should_fail( - "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_pt", - qr/cannot refresh partitioned table "tab_pt"/, - 'REFRESH TABLE rejects a partitioned table'); - -refresh_should_fail( - "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_pt_1", - qr/cannot refresh table "tab_pt_1" because it is a partition/, - 'REFRESH TABLE rejects a partition'); - # The command cannot run inside a transaction block. refresh_should_fail( "BEGIN; ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res;", @@ -248,6 +235,134 @@ my $sub_oth = $node_subscriber->safe_psql('postgres', 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. @@ -336,8 +451,40 @@ is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_inh_c"), '2', 'child re-copied after refreshing both'); +# An inheritance parent does not route rows into its children, so another +# subscription tracking only the parent must not block refreshing the child. +# The ancestor test has to be restricted to partitions for this: pg_inherits +# records inheritance and partitioning alike, so walking it unconditionally +# would report the inheritance parent as a feeder of the child. +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION tap_pub_inh_ponly FOR TABLE ONLY tab_inh_p"); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_inh_ponly CONNECTION '$publisher_connstr' PUBLICATION tap_pub_inh_ponly WITH (copy_data = false, enabled = false, create_slot = false, slot_name = NONE)" +); +$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_c"); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT r.srsubstate 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' AND c.relname = 'tab_inh_c'" + ), + 'i', + 'inheritance child can be refreshed when another subscription tracks only the parent' +); + +$node_subscriber->safe_psql('postgres', + "DROP SUBSCRIPTION tap_sub_inh_ponly"); +$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_inh_ponly"); + # 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 subscription with origin = none does not want rows that originated @@ -544,6 +691,68 @@ $node_subscriber->safe_psql( )); $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_bidi"); +# A partitioned table and one of its partitions can both be members of the +# same subscription, when the publisher has them as independent tables and +# only the subscriber puts them in a hierarchy. Naming the partition first +# then asks for the two to be locked in the opposite order to the one the rest +# of the system uses, which the command has to correct. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_mix_p (id int); + CREATE TABLE tab_mix_c (id int); + INSERT INTO tab_mix_p SELECT generate_series(-90, -81); + INSERT INTO tab_mix_c SELECT generate_series(-50, -41); + CREATE PUBLICATION tap_pub_mix_p FOR TABLE tab_mix_p; + CREATE PUBLICATION tap_pub_mix_c FOR TABLE tab_mix_c; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_mix_p (id int) PARTITION BY RANGE (id); + CREATE TABLE tab_mix_c PARTITION OF tab_mix_p FOR VALUES FROM (-100) TO (0); +)); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_mix CONNECTION '$publisher_connstr' PUBLICATION tap_pub_mix_p, tap_pub_mix_c" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_mix'); + +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_mix'" + ), + 'tab_mix_c,tab_mix_p', + 'a partition and its root can both be subscription members'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_mix DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_mix' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +# The partition is named before its root on purpose. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_mix REFRESH TABLE tab_mix_c, tab_mix_p"); + +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_mix'" + ), + 'tab_mix_c=i,tab_mix_p=i', + 'both the partition and its root are reset when named in either order'); +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM tab_mix_p"), + '0', + 'the whole hierarchy is truncated'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_mix ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_mix'); +is( $node_subscriber->safe_psql( + 'postgres', "SELECT count(*) FROM tab_mix_p"), + '20', + 'both members are re-copied'); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_mix"); + # 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 @@ -651,6 +860,69 @@ is( $node_subscriber->safe_psql( $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( diff --git a/src/test/subscription/t/040_refresh_table_race.pl b/src/test/subscription/t/040_refresh_table_race.pl new file mode 100644 index 00000000000..95e910ddb27 --- /dev/null +++ b/src/test/subscription/t/040_refresh_table_race.pl @@ -0,0 +1,139 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# ALTER SUBSCRIPTION ... REFRESH TABLE refuses to run when another +# subscription also feeds the table, because the truncate would discard rows +# that the other subscription maintains but would never copy again. A +# subscription that feeds the table by routing rows through a partitioned +# ancestor is one of those, and the ancestor is not itself truncated, so it +# only counts if the command locks it. Check that a subscription cannot be +# registered on such an ancestor while the command runs. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +if ($ENV{enable_injection_points} ne 'yes') +{ + plan skip_all => 'Injection points not supported by this build'; +} + +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->start; + +# The leaf is published on its own, and the root is published with +# publish_via_partition_root, so a subscription to either one keeps the leaf +# populated. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_race (a int primary key) PARTITION BY RANGE (a); + CREATE TABLE tab_race_1 PARTITION OF tab_race FOR VALUES FROM (1) TO (12); + INSERT INTO tab_race SELECT generate_series(1, 10); + CREATE PUBLICATION tap_pub_race_leaf FOR TABLE tab_race_1; + CREATE PUBLICATION tap_pub_race_root FOR TABLE tab_race + WITH (publish_via_partition_root = true); +)); + +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; + +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE EXTENSION injection_points; + CREATE TABLE tab_race (a int primary key) PARTITION BY RANGE (a); + CREATE TABLE tab_race_1 PARTITION OF tab_race FOR VALUES FROM (1) TO (12); + CREATE SUBSCRIPTION tap_sub_race_leaf CONNECTION '$publisher_connstr' + PUBLICATION tap_pub_race_leaf; +)); +$node_subscriber->wait_for_subscription_sync($node_publisher, + 'tap_sub_race_leaf'); + +is($node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_race_1"), + '10', 'leaf subscription copied the baseline'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_race_leaf DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_race_leaf' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +# Pause the command once it has taken its locks and cleared its final check. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('subscription-refresh-table-after-subscription-check', 'wait')" +); + +my $refresh = $node_subscriber->background_psql('postgres'); +$refresh->query_until( + qr/starting_refresh/, q( + \echo starting_refresh + ALTER SUBSCRIPTION tap_sub_race_leaf REFRESH TABLE tab_race_1; +)); +$node_subscriber->wait_for_event('client backend', + 'subscription-refresh-table-after-subscription-check'); + +# Registering a subscription resolves the relation with AccessShareLock first, +# so this has to wait for the refresh to release the root. Run it in the +# background, since it is expected not to return yet. +my $create = $node_subscriber->background_psql('postgres'); +$create->query_until( + qr/starting_create/, qq( + \\echo starting_create + CREATE SUBSCRIPTION tap_sub_race_root CONNECTION '$publisher_connstr' + PUBLICATION tap_pub_race_root WITH (copy_data = false, enabled = false); +)); + +ok( $node_subscriber->poll_query_until( + 'postgres', qq( + SELECT count(*) > 0 FROM pg_locks + WHERE relation = 'tab_race'::regclass AND NOT granted +)), 'CREATE SUBSCRIPTION waits for the lock on the partitioned ancestor'); + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_subscription WHERE subname = 'tap_sub_race_root'" + ), + '0', + 'the ancestor subscription is not registered while the refresh holds the lock' +); + +# Let the refresh finish. The waiting CREATE SUBSCRIPTION can then proceed, +# and it sees a table that has already been reset. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_wakeup('subscription-refresh-table-after-subscription-check')" +); +ok($refresh->quit, 'refresh completed'); +ok($create->quit, 'CREATE SUBSCRIPTION completed once the lock was released'); + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT r.srsubstate 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_race_leaf'" + ), + 'i', + 'the refreshed leaf is reset for re-copy'); + +# The refresh went through because it was alone when it checked and stayed +# alone until it committed. Re-running it now is rejected, because the +# subscription on the ancestor is visible and would lose rows. +my ($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION tap_sub_race_leaf REFRESH TABLE tab_race_1"); +isnt($ret, 0, 'a later refresh is rejected'); +like( + $stderr, + qr/is a partition of "tab_race", which is part of the subscription "tap_sub_race_root"/, + 'rejected because the ancestor subscription now feeds the leaf'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_race_leaf ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, + 'tap_sub_race_leaf'); +is($node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_race_1"), + '10', 'the leaf is refilled by its own subscription'); + +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); -- 2.54.0