From fd360cb1c504a4011335d2da616efc2dbfc23e7d Mon Sep 17 00:00:00 2001 From: cagrib Date: Fri, 23 Oct 2026 10:55:05 +0200 Subject: [PATCH v13 1/2] Add ALTER SUBSCRIPTION ... REFRESH TABLE to resync individual tables A subscribed table whose contents have diverged from the publisher can currently only be repaired by dropping and recreating the subscription, which re-copies every table and loses the progress of the others. Add ALTER SUBSCRIPTION ... REFRESH TABLE, which truncates the named tables on the subscriber and resets their pg_subscription_rel state to init, so that a tablesync worker copies each of them afresh once the subscription is enabled. Publication membership is not touched, so other subscribers of the same publication are unaffected. All of the named tables are validated before any of them is reset, so a bad table name aborts the command without touching the others, and they are truncated in one command, which lets a set connected by foreign keys be re-seeded as a unit. The caller needs the TRUNCATE privilege on each named table, checked again once the relation is locked. The subscription must be disabled, so there is no apply worker holding cached relation state and no race against a concurrently launched tablesync worker. A table that another subscription also populates is refused, since this command cannot reset that subscription's state and its rows would never be copied back. The truncate is attributed to the subscription's own replication origin, the one it would have carried had it arrived from the publisher, so that it travels under the same origin filter as the rows that replace it. Without this, a node subscribed to this one with origin = none would receive the discard but not the re-copy, which carries an origin of its own, and would be left empty -- including when it is the node the re-copy reads from, as in a two-way arrangement. Partitioned tables and partitions are not accepted yet. --- doc/src/sgml/ref/alter_subscription.sgml | 99 +++ src/backend/commands/subscriptioncmds.c | 491 +++++++++++ src/backend/commands/tablecmds.c | 9 +- src/backend/parser/gram.y | 10 + src/bin/psql/tab-complete.in.c | 11 +- src/include/commands/tablecmds.h | 4 + src/include/nodes/parsenodes.h | 2 + src/test/subscription/meson.build | 1 + src/test/subscription/t/039_refresh_table.pl | 843 +++++++++++++++++++ 9 files changed, 1462 insertions(+), 8 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..d0f96e727fa 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,103 @@ 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. Partitioned tables and partitions cannot be refreshed. + + + See for details of how + re-copying a table interacts with the + origin + parameter. Because this command copies a table the subscription + already had, it reports the same warning that + CREATE SUBSCRIPTION does, for the tables it re-seeds. + + + The truncate is attributed to the subscription's own replication origin, + the one a TRUNCATE arriving from the publisher would + have carried. A node subscribed to this one therefore receives the + truncate only if it accepts changes from that origin, which is also the + condition under which it will receive the rows that the re-copy brings + in. A subscriber using + origin = none + receives neither, so it is not emptied by a repair whose re-copy it + would have discarded. This is what keeps the rows when the node being + repaired is one half of a two-way arrangement and copies from the node + its truncate would otherwise reach. + + + The command repairs one node. A node that replicates from this one + keeps whatever it had, and may need a REFRESH TABLE + of its own to agree with it again. + + + 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 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. + + + + + If the command fails, run it again. It removes the table + synchronization slot and origin of a table that was still + synchronizing, which cannot be undone, so a table can be left unable to + resume, with its table synchronization worker reporting that its + replication slot is missing. Completing the command clears that. + + + + + ENABLE diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 36bb20a1f6e..39238d2fc9f 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" @@ -53,11 +54,13 @@ #include "storage/lock.h" #include "utils/acl.h" #include "utils/builtins.h" +#include "utils/fmgroids.h" #include "utils/guc.h" #include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/pg_lsn.h" +#include "utils/rel.h" #include "utils/syscache.h" /* @@ -1494,6 +1497,436 @@ 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. + */ +static void +CheckRefreshTableNotInOtherSubscriptions(Relation pgsubrel, Subscription *sub, + List *relids) +{ + foreach_oid(relid, relids) + { + ScanKeyData skey; + SysScanDesc scan; + HeapTuple tup; + + ScanKeyInit(&skey, Anum_pg_subscription_rel_srrelid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(relid)); + + 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); + + 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)); + } + + systable_endscan(scan); + } +} + +/* + * Resynchronize one or more already-subscribed tables. + * + * Truncates the local copy of each named table and resets its + * pg_subscription_rel state back to init so that, once the subscription is + * enabled, a tablesync worker re-copies just those tables. This is + * subscriber-local and does not touch publication membership, so sibling + * subscribers are unaffected. + * + * Every named relation is validated before any of them is reset, so the + * command is all-or-nothing: a bad table name aborts the whole command + * without touching the others. The tables are truncated together, which lets + * a set connected by foreign keys be re-seeded as a unit. + * + * The caller must ensure the subscription is disabled: with no apply worker + * running there is no cached relation state to invalidate and no race against + * a concurrently launched tablesync worker. + */ +static void +AlterSubscription_refresh_table(Subscription *sub, List *relations) +{ + Relation pgsubrel; + ListCell *lc; + List *subrelids = NIL; + List *rels = NIL; + List *relids = NIL; + List *relids_logged = NIL; + List *originrelids = NIL; + List *slotrelids = NIL; + char originname[NAMEDATALEN]; + ReplOriginId saved_origin; + bool origin_none; + + /* + * Lock user relations before pg_subscription_rel, never the other way + * round. AlterSubscription_refresh() takes AccessShareLock on each + * relation of the publication and only then opens pg_subscription_rel, at + * AccessExclusiveLock, to remove the rows of the ones that are gone. A + * command that held pg_subscription_rel while waiting for a relation lock + * would close a cycle with it, on any lock level: this one needs + * RowExclusiveLock there, which conflicts with that AccessExclusiveLock + * just as surely. So the catalogue is opened further down, once every + * relation lock this command needs is held, and the one lookup that has + * to happen before that releases it again right away. + * + * First pass: validate every named relation before changing anything, so + * the command is all-or-nothing. Duplicates in the list are ignored. + * + * Only AccessShareLock is taken here, and the level the truncate needs is + * taken further down, once the relation set is known to be refreshable. + * Taking it here would mean waiting for any tablesync worker copying one + * of these relations for another subscription to finish, only to reject + * the command because of that very subscription. + */ + foreach(lc, relations) + { + RangeVar *rv = lfirst_node(RangeVar, lc); + Oid relid; + char relstate; + XLogRecPtr relstatelsn; + Relation userrel; + + relid = RangeVarGetRelid(rv, AccessShareLock, false); + + if (get_rel_relkind(relid) == RELKIND_SEQUENCE) + ereport(ERROR, + errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("cannot refresh sequence \"%s\" as a table", + rv->relname), + errhint("Use ALTER SUBSCRIPTION ... REFRESH SEQUENCES instead.")); + + /* + * 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) + 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. + * + * This is not the check the command relies on, because REVOKE can + * still take the privilege away afterwards; it runs here so that a + * user who does not have it is turned away before the command starts + * escalating locks and waiting. It is checked again once the + * relation is locked. + */ + userrel = table_open(relid, NoLock); + truncate_check_perms(relid, userrel->rd_rel); + table_close(userrel, NoLock); + + 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); + } + + relids = subrelids; + + /* + * 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. + * + * pg_subscription_rel is closed again as soon as the scan is done, and + * with it the lock, because the relation locks below are taken next and + * holding a lock on it across them is what would deadlock. Releasing it + * costs nothing: this check is advisory anyway, and the one that decides + * runs below with those locks held. + */ + pgsubrel = table_open(SubscriptionRelRelationId, AccessShareLock); + CheckRefreshTableNotInOtherSubscriptions(pgsubrel, sub, 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. + */ + foreach_oid(relid, relids) + LockRelationOid(relid, 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 + * RowExclusiveLock, the level the updates below need. + */ + 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. + */ + CheckRefreshTableNotInOtherSubscriptions(pgsubrel, sub, relids); + + /* + * Open what is about to be truncated. The locks are already held. + */ + foreach_oid(relid, relids) + { + Relation userrel = table_open(relid, NoLock); + + rels = lappend(rels, userrel); + if (RelationIsLogicallyLogged(userrel)) + relids_logged = lappend_oid(relids_logged, relid); + + /* + * ExecuteTruncateGuts() applies these only to the relations it adds + * itself through CASCADE, leaving the ones passed in to the caller, + * as ExecuteTruncate() does. truncate_check_rel() is also what + * invokes the truncate object access hook, which an audit or label + * provider needs to see for a command that discards a table's + * contents. + */ + truncate_check_rel(relid, userrel->rd_rel); + truncate_check_activity(userrel); + + /* + * This is the privilege check that counts. The one in the first pass + * ran under AccessShareLock, which REVOKE does not conflict with: + * objectNamesToOids() deliberately takes no more than that, so the + * privilege could have been revoked since. AccessExclusiveLock is + * held now, so the answer cannot change before the truncate. + * + * ExecuteTruncate() gets this for free, because its permission check + * lives in a RangeVarGetRelidExtended() callback that is re-run with + * the lock held whenever an invalidation arrives while it is being + * taken. This command resolves its relations without a callback, so + * it has to repeat the check itself. + */ + truncate_check_perms(relid, userrel->rd_rel); + } + + /* + * 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. + * + * Attribute the truncate to the subscription's own origin. The rows that + * replace what it discards arrive under an origin as well, so without + * this the two halves of the resync would travel under different origin + * filters: a node subscribed to this one with origin = NONE would receive + * the discard but not the re-copy, and would be left empty. Tagged this + * way, a subscriber either takes both halves or neither. + */ + ReplicationOriginNameForLogicalRep(sub->oid, InvalidOid, originname, + sizeof(originname)); + saved_origin = replorigin_xact_state.origin; + replorigin_xact_state.origin = replorigin_by_name(originname, false); + + PG_TRY(); + { + ExecuteTruncateGuts(rels, relids, relids_logged, DROP_RESTRICT, false, + false); + } + PG_FINALLY(); + { + /* + * Restore it before anything else runs: left set, it would reach + * RecordTransactionCommit(), which expects a configured origin + * session to advance. + */ + replorigin_xact_state.origin = saved_origin; + } + PG_END_TRY(); + + 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)); + } + + /* + * Everything that can fail for an ordinary reason is done, so the two + * steps that cannot be rolled back come last. + * + * First the tablesync origin of every relation that was caught mid-sync. + * Deleting the catalog row is transactional, but + * replorigin_drop_by_name() also clears the in-memory progress of the + * origin, and that part is not, so a later failure would leave the origin + * behind without it. Pass missing_ok = true as the origin may not exist + * yet. + */ + foreach_oid(relid, originrelids) + { + ReplicationOriginNameForLogicalRep(sub->oid, relid, originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + } + + /* + * A subscription with origin = none is asking not to take in rows that + * originated elsewhere, and the copy this command arranges cannot honour + * that, because a row's origin is only known from WAL. CREATE + * SUBSCRIPTION and ALTER SUBSCRIPTION ... REFRESH PUBLICATION warn about + * exactly this, and skip the relations the subscription already has on + * the grounds that those are kept current through WAL, where origins are + * known. That reasoning does not cover this command, which is what puts + * such a relation back through a copy, so the relations being re-seeded + * are the ones to check here and the rest of the subscription's are the + * ones to skip. + */ + origin_none = sub->origin && + pg_strcasecmp(sub->origin, LOGICALREP_ORIGIN_NONE) == 0; + + /* + * Then the tablesync slots left on the publisher by those same relations. + * This is last of all, as it is the only step whose effect reaches beyond + * this node. Should it fail partway, re-running the command completes + * the job, because the slots already gone are dropped with missing_ok. + * One connection serves that and the origin check above, and re-seeding + * relations that are all in ready state under origin = any needs no + * publisher connection at all. + */ + if (slotrelids != NIL || origin_none) + { + 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(); + { + if (origin_none) + { + List *subrel_states; + Oid *skip_oids; + int nskip = 0; + + subrel_states = GetSubscriptionRelations(sub->oid, true, false, + false); + skip_oids = palloc_array(Oid, list_length(subrel_states)); + + foreach(lc, subrel_states) + { + SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc); + + if (!list_member_oid(subrelids, relstate->relid)) + skip_oids[nskip++] = relstate->relid; + } + + qsort(skip_oids, nskip, sizeof(Oid), oid_cmp); + check_publications_origin_tables(wrconn, sub->publications, + true, sub->retaindeadtuples, + sub->origin, skip_oids, + nskip, sub->name); + } + + 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(pgsubrel, NoLock); +} + /* * Common checks for altering failover, two_phase, and retain_dead_tuples * options. @@ -2401,6 +2834,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/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 0274d892f2e..79adc9dde65 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -368,9 +368,6 @@ typedef enum addFkConstraintSides #define child_dependency_type(child_is_partition) \ ((child_is_partition) ? DEPENDENCY_AUTO : DEPENDENCY_NORMAL) -static void truncate_check_rel(Oid relid, Form_pg_class reltuple); -static void truncate_check_perms(Oid relid, Form_pg_class reltuple); -static void truncate_check_activity(Relation rel); static void RangeVarCallbackForTruncate(const RangeVar *relation, Oid relId, Oid oldRelId, void *arg); static List *MergeAttributes(List *columns, const List *supers, char relpersistence, @@ -2398,7 +2395,7 @@ ExecuteTruncateGuts(List *explicit_rels, * Check that a given relation is safe to truncate. Subroutine for * ExecuteTruncate() and RangeVarCallbackForTruncate(). */ -static void +void truncate_check_rel(Oid relid, Form_pg_class reltuple) { char *relname = NameStr(reltuple->relname); @@ -2451,7 +2448,7 @@ truncate_check_rel(Oid relid, Form_pg_class reltuple) /* * Check that current user has the permission to truncate given relation. */ -static void +void truncate_check_perms(Oid relid, Form_pg_class reltuple) { char *relname = NameStr(reltuple->relname); @@ -2469,7 +2466,7 @@ truncate_check_perms(Oid relid, Form_pg_class reltuple) * truncate. This is split with truncate_check_rel() as * RangeVarCallbackForTruncate() cannot open a Relation yet. */ -static void +void truncate_check_activity(Relation rel) { /* diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 0563453fe24..9e748965677 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -11121,6 +11121,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 b3bfe050b18..a81e8499ca1 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2336,12 +2336,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/commands/tablecmds.h b/src/include/commands/tablecmds.h index c3d8518cb62..fa14542bd75 100644 --- a/src/include/commands/tablecmds.h +++ b/src/include/commands/tablecmds.h @@ -17,6 +17,7 @@ #include "access/htup.h" #include "catalog/dependency.h" #include "catalog/objectaddress.h" +#include "catalog/pg_class.h" #include "nodes/parsenodes.h" #include "storage/lockdefs.h" #include "utils/relcache.h" @@ -65,6 +66,9 @@ extern void ExecuteTruncateGuts(List *explicit_rels, DropBehavior behavior, bool restart_seqs, bool run_as_table_owner); +extern void truncate_check_rel(Oid relid, Form_pg_class reltuple); +extern void truncate_check_perms(Oid relid, Form_pg_class reltuple); +extern void truncate_check_activity(Relation rel); extern void SetRelationHasSubclass(Oid relationId, bool relhassubclass); diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 0debcd193ab..edd0e973ad6 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -4421,6 +4421,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; @@ -4433,6 +4434,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/meson.build b/src/test/subscription/meson.build index e71e95c6297..37f480bcac2 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -48,6 +48,7 @@ tests += { 't/036_sequences.pl', 't/037_except.pl', 't/038_walsnd_shutdown_timeout.pl', + 't/039_refresh_table.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 new file mode 100644 index 00000000000..2a5077659bb --- /dev/null +++ b/src/test/subscription/t/039_refresh_table.pl @@ -0,0 +1,843 @@ + +# 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, and does not accept partitioned tables or partitions. +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'); +# Logical decoding on the subscriber too, so that the two-way case further +# down can publish back to the publisher. +$node_subscriber->init(allows_streaming => 'logical'); +$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; + 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', + "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'); + +# 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;", + 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'); +# A table that was not reset is READY, or SYNCDONE if the apply worker had +# not yet promoted it when the subscription was disabled; both are what +# wait_for_subscription_sync() waits for, and only 'i' means reset. +like( $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'" + ), + qr/^[rs]$/, + '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'" +); +like($other_state_before, qr/^[rs]$/, 'tab_other is synced 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'" + ), + $other_state_before, + 'tab_other left untouched (state unchanged)'); +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'); + +# 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'); +like( $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'" + ), + qr/^tab_inh_c=[rs],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_inh"); + +# A subscription with origin = none does not want rows that originated +# elsewhere, but a copy cannot tell where a row came from, so CREATE +# SUBSCRIPTION and REFRESH PUBLICATION warn when they copy under that setting. +# They skip the tables the subscription already has, because those are kept +# current through WAL, where origins are known. This command is the exception +# to that, since it puts an already-subscribed table back through a copy, so +# it has to warn about the tables it re-seeds. +# +# The warning fires when the publisher is itself subscribed to the table, so +# the publisher subscribes back to the subscriber here, giving the two-way +# arrangement origin = none exists for. tab_org2 is in the same publication +# but is not subscribed back, and is there to check that the warning follows +# the tables actually re-copied. +my $subscriber_connstr = $node_subscriber->connstr . ' dbname=postgres'; + +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_org (id int primary key); + CREATE TABLE tab_org2 (id int primary key); + INSERT INTO tab_org SELECT generate_series(1, 10); + INSERT INTO tab_org2 SELECT generate_series(1, 10); + CREATE PUBLICATION tap_pub_org FOR TABLE tab_org, tab_org2; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_org (id int primary key); + CREATE TABLE tab_org2 (id int primary key); + CREATE PUBLICATION tap_pub_org_back FOR TABLE tab_org; +)); + +# The publisher's own subscription to tab_org is what makes the rows on it +# carry an origin as far as the subscriber is concerned. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE SUBSCRIPTION tap_sub_org_back + CONNECTION '$subscriber_connstr' + PUBLICATION tap_pub_org_back + WITH (origin = none, copy_data = off, enabled = false, + create_slot = false, slot_name = NONE); +)); + +my ($ret, $stdout, $stderr) = $node_subscriber->psql( + 'postgres', qq( + CREATE SUBSCRIPTION tap_sub_org + CONNECTION '$publisher_connstr' + PUBLICATION tap_pub_org + WITH (origin = none); +)); +like( + $stderr, + qr/WARNING: ( [A-Z0-9]+:)? subscription "tap_sub_org" requested copy_data with origin = NONE but might copy data that had a different origin/, + 'CREATE SUBSCRIPTION warns when copying under origin = none'); + +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_org'); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_org DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_org' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION tap_sub_org REFRESH TABLE tab_org"); +is($ret, 0, 'REFRESH TABLE succeeds under origin = none'); +like( + $stderr, + qr/WARNING: ( [A-Z0-9]+:)? subscription "tap_sub_org" requested copy_data with origin = NONE but might copy data that had a different origin/, + 'REFRESH TABLE warns about re-copying a table that has another origin'); + +# tab_org2 is not written by any other subscription, so re-seeding it says +# nothing about origins. +($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION tap_sub_org REFRESH TABLE tab_org2"); +is($ret, 0, 'REFRESH TABLE succeeds for a table with no other origin'); +unlike( + $stderr, + qr/requested copy_data with origin = NONE/, + 'REFRESH TABLE does not warn for a table no other subscription writes'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_org ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_org'); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM tab_org"), + '10', + 'the re-seeded table is copied back under origin = none'); + +$node_publisher->safe_psql('postgres', + "DROP SUBSCRIPTION tap_sub_org_back"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_org"); +$node_subscriber->safe_psql('postgres', "DROP PUBLICATION tap_pub_org_back"); +$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_org"); + +# The truncate this command performs is attributed to the subscription's own +# origin, the same one the apply worker would have used had the truncate +# arrived from the publisher. A node subscribed to this one with +# origin = none therefore receives neither half of the resync, where before it +# received the discard but not the re-copy, which carries an origin of its own. +# +# tab_bidi is replicated both ways, the arrangement where this matters most: +# there, the node the truncate would reach is the very node the re-copy reads +# from, so letting it through empties both. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_bidi (id int primary key); + INSERT INTO tab_bidi SELECT generate_series(1, 10); + CREATE PUBLICATION tap_pub_bidi FOR TABLE tab_bidi; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_bidi (id int primary key); + CREATE PUBLICATION tap_pub_bidi_back FOR TABLE tab_bidi; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE SUBSCRIPTION tap_sub_bidi CONNECTION '$publisher_connstr' + PUBLICATION tap_pub_bidi WITH (origin = none); +)); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_bidi'); +$node_publisher->safe_psql( + 'postgres', qq( + CREATE SUBSCRIPTION tap_sub_bidi_back CONNECTION '$subscriber_connstr' + PUBLICATION tap_pub_bidi_back WITH (origin = none, copy_data = false); +)); +$node_publisher->wait_for_subscription_sync($node_subscriber, + 'tap_sub_bidi_back'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_bidi DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_bidi' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +# The command and the checks that follow it share one session, because the +# origin is per-session state that must not outlive the statement that set it: +# every later write in the session would be attributed to the subscription, +# and silently dropped by any subscriber using origin = none. +my $bidi_psql = $node_subscriber->background_psql('postgres', + on_error_stop => 0); +is($bidi_psql->query('SELECT pg_replication_origin_session_is_setup()'), + 'f', 'no origin is set before the command'); +my ($bidi_out, $bidi_ret) = + $bidi_psql->query("ALTER SUBSCRIPTION tap_sub_bidi REFRESH TABLE tab_bidi"); +unlike($bidi_psql->{stderr}, qr/ERROR/, + 'REFRESH TABLE succeeds on a table replicated both ways'); +$bidi_psql->{stderr} = ''; +is($bidi_psql->query('SELECT pg_replication_origin_session_is_setup()'), + 'f', 'the origin set for the truncate does not outlive the command'); + +# Catching the other node up means a replicated truncate would have arrived. +$node_subscriber->wait_for_catchup('tap_sub_bidi_back'); +is($node_publisher->safe_psql('postgres', "SELECT count(*) FROM tab_bidi"), + '10', + 'the truncate is not replicated to a subscriber using origin = none'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_bidi ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_bidi'); +is($node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_bidi"), + '10', 'the re-copy finds the other node still holding the rows'); + +# The same has to hold when the command fails part way through, which a +# BEFORE TRUNCATE trigger that raises arranges from inside the truncate +# itself. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_bidi DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_bidi' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE FUNCTION refuse_truncate() RETURNS trigger LANGUAGE plpgsql + AS \$\$ BEGIN RAISE EXCEPTION 'truncate refused'; END \$\$; + CREATE TRIGGER tab_bidi_refuse BEFORE TRUNCATE ON tab_bidi + FOR EACH STATEMENT EXECUTE FUNCTION refuse_truncate(); +)); + +($bidi_out, $bidi_ret) = + $bidi_psql->query("ALTER SUBSCRIPTION tap_sub_bidi REFRESH TABLE tab_bidi"); +like($bidi_psql->{stderr}, qr/truncate refused/, + 'the command fails inside the truncate'); +$bidi_psql->{stderr} = ''; +is($bidi_psql->query('SELECT pg_replication_origin_session_is_setup()'), + 'f', 'the origin is restored when the command fails'); +is( $node_subscriber->safe_psql( + 'postgres', qq( + SELECT srsubstate FROM pg_subscription_rel + WHERE srrelid = 'tab_bidi'::regclass +)), + 'r', + 'the failed command leaves the relation as it found it'); +$bidi_psql->quit; + +$node_publisher->safe_psql('postgres', + "DROP SUBSCRIPTION tap_sub_bidi_back"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_bidi"); +$node_subscriber->safe_psql( + 'postgres', qq( + DROP PUBLICATION tap_pub_bidi_back; + DROP TRIGGER tab_bidi_refuse ON tab_bidi; + DROP FUNCTION refuse_truncate(); +)); +$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_bidi"); + +# 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 = 'i') 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"); + +# 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"); + +# The command must not hold pg_subscription_rel while it waits for a lock on +# a user relation. ALTER SUBSCRIPTION ... REFRESH PUBLICATION takes the two +# the other way round, AccessShareLock on each relation of the publication +# first and then AccessExclusiveLock on pg_subscription_rel to remove the rows +# of the ones that are gone, so holding the catalogue across such a wait would +# close a deadlock cycle with it. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_lk1 (a int primary key); + CREATE TABLE tab_lk2 (a int primary key); + CREATE TABLE tab_lk_gone (a int primary key); + INSERT INTO tab_lk1 SELECT generate_series(1, 10); + INSERT INTO tab_lk2 SELECT generate_series(1, 10); + CREATE PUBLICATION tap_pub_lk FOR TABLE tab_lk1, tab_lk2; + CREATE PUBLICATION tap_pub_lk2 FOR TABLE tab_lk_gone; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_lk1 (a int primary key); + CREATE TABLE tab_lk2 (a int primary key); + CREATE TABLE tab_lk_gone (a int primary key); +)); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_lk CONNECTION '$publisher_connstr' PUBLICATION tap_pub_lk" +); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_lk2 CONNECTION '$publisher_connstr' PUBLICATION tap_pub_lk2" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_lk'); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_lk2'); + +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub_lk DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_lk' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +# Point the other subscription's publication at tab_lk1, so that its refresh +# takes AccessShareLock on tab_lk1 and then opens pg_subscription_rel at +# AccessExclusiveLock to remove tab_lk_gone. +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_lk2 SET TABLE tab_lk1"); + +# Park tab_lk2, so the command below stops in its first pass: tab_lk1 resolved, +# none of the locks it truncates under taken yet. +$node_subscriber->safe_psql('postgres', + "BEGIN; LOCK TABLE tab_lk2 IN ACCESS EXCLUSIVE MODE; PREPARE TRANSACTION 'refresh_lk';" +); + +my $bg_lk = $node_subscriber->background_psql('postgres', on_error_stop => 0); +my $bg_lk_pid = $bg_lk->query('SELECT pg_backend_pid()'); +$bg_lk->query_until(qr/refresh_fired/, + "\\echo refresh_fired\nALTER SUBSCRIPTION tap_sub_lk REFRESH TABLE tab_lk1, tab_lk2;\n" +); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 1 FROM pg_locks WHERE pid = $bg_lk_pid AND NOT granted") + or die "Timed out waiting for REFRESH TABLE to block on a relation lock"; + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_locks WHERE pid = $bg_lk_pid AND relation = 'pg_subscription_rel'::regclass" + ), + '0', + 'REFRESH TABLE holds no lock on pg_subscription_rel while waiting for a relation lock' +); + +# The other subscription's refresh now takes AccessShareLock on tab_lk1 and +# asks for pg_subscription_rel. It runs in the background as well: were the +# catalogue held above, it would stop here, and the cycle would close only +# once the parked lock is released below. +my $bg_pub = $node_subscriber->background_psql('postgres', on_error_stop => 0); +$bg_pub->query_until(qr/pubrefresh_fired/, + "\\echo pubrefresh_fired\nALTER SUBSCRIPTION tap_sub_lk2 REFRESH PUBLICATION WITH (copy_data = false);\n" +); + +$node_subscriber->safe_psql('postgres', "COMMIT PREPARED 'refresh_lk'"); +$bg_pub->query('SELECT 1'); +$bg_lk->query('SELECT 1'); + +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_subscription_rel sr JOIN pg_subscription s ON s.oid = sr.srsubid WHERE s.subname = 'tap_sub_lk2' AND sr.srrelid = 'tab_lk1'::regclass" + ), + '1', + 'REFRESH PUBLICATION completes alongside a waiting REFRESH TABLE'); +unlike($bg_pub->{stderr}, qr/deadlock detected/, + 'REFRESH PUBLICATION is not deadlocked by REFRESH TABLE'); +$bg_pub->{stderr} = ''; +$bg_pub->quit; + +# That refresh has just made tab_lk1 part of two subscriptions, so the command +# that was waiting is refused once it resumes. The point is that it is given +# an answer at all: before the catalogue was moved out of the way, the two +# commands deadlocked here. +like( + $bg_lk->{stderr}, + qr/table "tab_lk1" is also part of the subscription "tap_sub_lk2"/, + 'the waiting REFRESH TABLE is answered rather than deadlocked'); +unlike($bg_lk->{stderr}, qr/deadlock detected/, + 'no deadlock between REFRESH TABLE and REFRESH PUBLICATION'); +$bg_lk->{stderr} = ''; +$bg_lk->quit; + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_lk"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_lk2"); + +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); -- 2.54.0