From b3f9ab416059f62af03d0c812bb3ed2616205f43 Mon Sep 17 00:00:00 2001 From: cagrib Date: Fri, 23 Oct 2026 10:55:05 +0200 Subject: [PATCH v14 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 tablesync slots and origins of relations caught mid-sync are dropped last, once the publisher has been reached and everything else that can fail is done, as neither drop can be rolled back. The slots go first, the order the tablesync worker uses for its own: dropping an origin also discards its in-memory progress, which a rollback does not restore, so a relation must have no slot left to resume from by the time its origin is dropped. 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. Only the WAL record of the truncate carries that origin, through a new parameter of ExecuteTruncateGuts(), and only while it is inserted. Held any longer, it would also attribute what the truncate's triggers write, and an abort reached meanwhile, such as of a subtransaction a trigger rolls back, would take it for an origin set up in the session and try to advance that. The triggers fire as they would for TRUNCATE in the same session. Partitioned tables and partitions are not accepted yet. --- doc/src/sgml/ref/alter_subscription.sgml | 114 ++ src/backend/commands/subscriptioncmds.c | 487 ++++++++ src/backend/commands/tablecmds.c | 43 +- src/backend/parser/gram.y | 10 + src/backend/replication/logical/worker.c | 3 +- src/bin/psql/tab-complete.in.c | 11 +- src/include/commands/tablecmds.h | 8 +- src/include/nodes/parsenodes.h | 2 + src/test/subscription/meson.build | 1 + src/test/subscription/t/039_refresh_table.pl | 1112 ++++++++++++++++++ 10 files changed, 1778 insertions(+), 13 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..3551275dbd3 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,118 @@ 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 truncate fires the tables' TRUNCATE triggers the + way a TRUNCATE run in the same session would, not the + way a replicated one would: which of them fire depends on that session's + . The re-copy is made by + a table synchronization worker, which like the rest of the apply process + runs with session_replication_role set to + replica. From a session using the default setting, + only a trigger configured with + ENABLE ALWAYS + therefore fires for both the truncate and the rows the re-copy brings + in. Changes made by the triggers are not attributed to the + subscription's origin; they replicate as any other change made in that + session would. + + + 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..1b35aa9074d 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,432 @@ 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]; + 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. + * + * Only the truncate itself is tagged. Triggers on these tables fire as + * they would for TRUNCATE, and what they write is an ordinary local + * change. + */ + ReplicationOriginNameForLogicalRep(sub->oid, InvalidOid, originname, + sizeof(originname)); + ExecuteTruncateGuts(rels, relids, relids_logged, DROP_RESTRICT, false, + false, replorigin_by_name(originname, false)); + + foreach(lc, rels) + table_close((Relation) lfirst(lc), NoLock); + + /* + * Reset each relation to init so that a tablesync worker re-copies it + * once the subscription is enabled. + */ + foreach_oid(relid, subrelids) + { + UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_INIT, + InvalidXLogRecPtr, false); + + ereport(DEBUG1, + errmsg_internal("table \"%s.%s\" of subscription \"%s\" reset for resync", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), sub->name)); + } + + /* + * Everything local that can fail for an ordinary reason is done. What is + * left is to drop the tablesync slots and origins of the relations caught + * mid-sync, and neither can be rolled back, so whatever else needs the + * publisher comes first: a publisher that cannot be reached then fails + * the command before anything irreversible has happened. + * + * 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. + * + * One connection serves that check and the slot drops, and re-seeding + * relations that are all in ready state under origin = any needs no + * publisher connection at all. + */ + origin_none = sub->origin && + pg_strcasecmp(sub->origin, LOGICALREP_ORIGIN_NONE) == 0; + + if (slotrelids != NIL || origin_none) + { + WalReceiverConn *wrconn; + char *err = NULL; + 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); + } + + /* + * Then the tablesync slots left on the publisher by the relations + * caught mid-sync. They go before the origins, for the reason + * given below. Should a drop fail partway, re-running the + * command completes the job, because the slots already gone are + * dropped with missing_ok. + */ + 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(); + } + + /* + * The tablesync origins go last, after the slots, which is also the order + * the tablesync worker drops its own in. 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: should the command fail + * after this, the origin would come back without its progress. A + * relation in finishedcopy state resumes from its origin, so it could + * then replay changes it had already applied. Dropped after the slots, + * an origin can only lose its progress once the relation has no slot left + * to resume from, which its tablesync worker reports until the command is + * run again. 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); + } + + table_close(pgsubrel, NoLock); +} + /* * Common checks for altering failover, two_phase, and retain_dead_tuples * options. @@ -2401,6 +2830,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 47f55121574..ae779d06cdc 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -87,6 +87,7 @@ #include "partitioning/partbounds.h" #include "partitioning/partdesc.h" #include "pgstat.h" +#include "replication/origin.h" #include "rewrite/rewriteDefine.h" #include "rewrite/rewriteHandler.h" #include "rewrite/rewriteManip.h" @@ -368,9 +369,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, @@ -1987,7 +1985,8 @@ ExecuteTruncate(TruncateStmt *stmt) } ExecuteTruncateGuts(rels, relids, relids_logged, - stmt->behavior, stmt->restart_seqs, false); + stmt->behavior, stmt->restart_seqs, false, + InvalidReplOriginId); /* And close the rels */ foreach(cell, rels) @@ -2010,13 +2009,18 @@ ExecuteTruncate(TruncateStmt *stmt) * relids_logged is the list of Oids (a subset of relids) that require * WAL-logging. This is all a bit redundant, but the existing callers have * this information handy in this form. + * + * If origin is valid, the WAL record that lets the truncation be logically + * decoded is attributed to that replication origin rather than to the + * session's. Nothing else the truncation writes is, including whatever its + * triggers do. */ void ExecuteTruncateGuts(List *explicit_rels, List *relids, List *relids_logged, DropBehavior behavior, bool restart_seqs, - bool run_as_table_owner) + bool run_as_table_owner, ReplOriginId origin) { List *rels; List *seq_relids = NIL; @@ -2354,7 +2358,28 @@ ExecuteTruncateGuts(List *explicit_rels, XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN); - (void) XLogInsert(RM_HEAP_ID, XLOG_HEAP_TRUNCATE); + if (origin == InvalidReplOriginId) + (void) XLogInsert(RM_HEAP_ID, XLOG_HEAP_TRUNCATE); + else + { + ReplOriginId saved_origin = replorigin_xact_state.origin; + + /* + * Install the origin for the insertion alone. The commit and + * abort paths take a set origin to be the session's configured + * one, and advance that, so they must never see this one. + */ + replorigin_xact_state.origin = origin; + PG_TRY(); + { + (void) XLogInsert(RM_HEAP_ID, XLOG_HEAP_TRUNCATE); + } + PG_FINALLY(); + { + replorigin_xact_state.origin = saved_origin; + } + PG_END_TRY(); + } } /* @@ -2397,7 +2422,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); @@ -2450,7 +2475,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); @@ -2468,7 +2493,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/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 44d735cdc55..7d91167f8f3 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -3819,7 +3819,8 @@ apply_handle_truncate(StringInfo s) relids_logged, DROP_RESTRICT, restart_seqs, - !MySubscription->runasowner); + !MySubscription->runasowner, + InvalidReplOriginId); foreach(lc, remote_rels) { LogicalRepRelMapEntry *rel = lfirst(lc); diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 96de78cd087..6a9c39cdd1e 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2342,12 +2342,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..a5ad1dc8653 100644 --- a/src/include/commands/tablecmds.h +++ b/src/include/commands/tablecmds.h @@ -15,8 +15,10 @@ #define TABLECMDS_H #include "access/htup.h" +#include "access/xlogdefs.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" @@ -64,7 +66,11 @@ extern void ExecuteTruncateGuts(List *explicit_rels, List *relids_logged, DropBehavior behavior, bool restart_seqs, - bool run_as_table_owner); + bool run_as_table_owner, + ReplOriginId origin); +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..e2ca8421428 --- /dev/null +++ b/src/test/subscription/t/039_refresh_table.pl @@ -0,0 +1,1112 @@ + +# 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'); + +# Triggers fire on the truncate as they would on TRUNCATE, in the session +# that issued the command, and that session may have a replication origin of +# its own. The command has to leave that origin alone, and must not be upset +# by whatever the triggers do with it, or by subtransactions they roll back. +$node_subscriber->safe_psql( + 'postgres', qq( + DROP TRIGGER tab_bidi_refuse ON tab_bidi; + SELECT pg_replication_origin_create('regress_refresh_origin'); +)); +my $bidi_pid = $bidi_psql->query('SELECT pg_backend_pid()'); +$bidi_psql->query( + "SELECT pg_replication_origin_session_setup('regress_refresh_origin')"); +$bidi_psql->query("ALTER SUBSCRIPTION tap_sub_bidi REFRESH TABLE tab_bidi"); +unlike($bidi_psql->{stderr}, qr/ERROR/, + 'REFRESH TABLE succeeds in a session with its own origin'); +$bidi_psql->{stderr} = ''; +is($bidi_psql->query('SELECT pg_replication_origin_session_is_setup()'), + 't', 'the session keeps its own origin'); + +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE FUNCTION reset_origin() RETURNS trigger LANGUAGE plpgsql + AS \$\$ BEGIN PERFORM pg_replication_origin_session_reset(); RETURN NULL; END \$\$; + CREATE TRIGGER tab_bidi_reset BEFORE TRUNCATE ON tab_bidi + FOR EACH STATEMENT EXECUTE FUNCTION reset_origin(); +)); +$bidi_psql->query("ALTER SUBSCRIPTION tap_sub_bidi REFRESH TABLE tab_bidi"); +unlike($bidi_psql->{stderr}, qr/ERROR/, + 'REFRESH TABLE succeeds when a trigger resets the session origin'); +$bidi_psql->{stderr} = ''; +is($bidi_psql->query('SELECT pg_backend_pid()'), + $bidi_pid, 'the session survives a trigger resetting its origin'); + +$node_subscriber->safe_psql( + 'postgres', qq( + DROP TRIGGER tab_bidi_reset ON tab_bidi; + CREATE TABLE refresh_audit (tab text primary key); + INSERT INTO refresh_audit VALUES ('tab_bidi'); + CREATE FUNCTION audit_truncate() RETURNS trigger LANGUAGE plpgsql + AS \$\$ BEGIN + INSERT INTO refresh_audit VALUES (TG_TABLE_NAME); + RETURN NULL; + EXCEPTION WHEN unique_violation THEN + RETURN NULL; + END \$\$; + CREATE TRIGGER tab_bidi_audit AFTER TRUNCATE ON tab_bidi + FOR EACH STATEMENT EXECUTE FUNCTION audit_truncate(); +)); +$bidi_psql->query("ALTER SUBSCRIPTION tap_sub_bidi REFRESH TABLE tab_bidi"); +unlike($bidi_psql->{stderr}, qr/ERROR/, + 'REFRESH TABLE succeeds when a trigger rolls back a subtransaction'); +$bidi_psql->{stderr} = ''; +is($bidi_psql->query('SELECT pg_backend_pid()'), + $bidi_pid, + 'the session survives a trigger rolling back a subtransaction'); + +# None of that changed what the truncate is attributed to. +$node_subscriber->wait_for_catchup('tap_sub_bidi_back'); +is($node_publisher->safe_psql('postgres', "SELECT count(*) FROM tab_bidi"), + '10', 'the truncate is still not replicated to origin = none'); + +# A session terminated while a trigger runs exits without its abort tripping +# over the origin. +$node_subscriber->safe_psql( + 'postgres', qq( + DROP TRIGGER tab_bidi_audit ON tab_bidi; + CREATE FUNCTION slow_truncate() RETURNS trigger LANGUAGE plpgsql + AS \$\$ BEGIN PERFORM pg_sleep($PostgreSQL::Test::Utils::timeout_default); RETURN NULL; END \$\$; + CREATE TRIGGER tab_bidi_slow BEFORE TRUNCATE ON tab_bidi + FOR EACH STATEMENT EXECUTE FUNCTION slow_truncate(); +)); +my $bidi_log_offset = -s $node_subscriber->logfile; +$bidi_psql->query_until(qr/slow_fired/, + "\\echo slow_fired\nALTER SUBSCRIPTION tap_sub_bidi REFRESH TABLE tab_bidi;\n" +); +$node_subscriber->poll_query_until('postgres', + "SELECT wait_event = 'PgSleep' FROM pg_stat_activity WHERE pid = $bidi_pid" +) or die "Timed out waiting for the truncate trigger to start"; +$node_subscriber->safe_psql('postgres', + "SELECT pg_terminate_backend($bidi_pid)"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE pid = $bidi_pid") + or die "Timed out waiting for the terminated session to exit"; +ok( !$node_subscriber->log_contains( + qr/was terminated by signal/, + $bidi_log_offset), + 'a session terminated inside the truncate exits cleanly'); +$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_slow ON tab_bidi; + DROP FUNCTION refuse_truncate(), reset_origin(), audit_truncate(), + slow_truncate(); + DROP TABLE refresh_audit; + SELECT pg_replication_origin_drop('regress_refresh_origin'); +)); +$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); +my $bg_pub_pid = $bg_pub->query('SELECT pg_backend_pid()'); +$bg_pub->query_until(qr/pubrefresh_fired/, + "\\echo pubrefresh_fired\nALTER SUBSCRIPTION tap_sub_lk2 REFRESH PUBLICATION WITH (copy_data = false);\n" +); + +# It has to be over before the parked lock is released, or the command +# waiting there could get ahead of it and find tab_lk1 still free. Stop +# waiting as well if that command is what blocks it, which is the failure +# tested for here, so that it is reported rather than waited out. Ask +# pg_stat_activity, not pg_subscription_rel: in that failure, a query on the +# catalogue would queue behind the lock the blocked refresh is waiting for. +my $pub_refresh_done = + "SELECT state = 'idle' AND query ~ 'REFRESH PUBLICATION' FROM pg_stat_activity WHERE pid = $bg_pub_pid"; +$node_subscriber->poll_query_until('postgres', + "SELECT ($pub_refresh_done) OR $bg_lk_pid = ANY (pg_blocking_pids($bg_pub_pid))" +) or die "Timed out waiting for REFRESH PUBLICATION"; +is($node_subscriber->safe_psql('postgres', $pub_refresh_done), + 't', 'REFRESH PUBLICATION completes while REFRESH TABLE waits'); + +$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 adds tab_lk1 to its subscription'); +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"); + +# The tablesync slots and origins of relations caught mid-sync are dropped +# last, and neither can be rolled back, so a publisher that cannot be reached +# has to fail the command before them: dropping a tablesync origin discards +# its in-memory progress, which a rollback does not bring back. +# +# Park a relation at FINISHEDCOPY, where it has both. Its copy is held back +# by a trigger until a row has been inserted on the publisher, so that the +# row comes in the catch-up that follows the copy, where it collides with +# one planted on the subscriber. Catch-up keeps failing on it from then on. +$node_publisher->safe_psql( + 'postgres', qq( + CREATE TABLE tab_down (a int primary key); + INSERT INTO tab_down SELECT generate_series(1, 10); + CREATE PUBLICATION tap_pub_down FOR TABLE tab_down; +)); +$node_subscriber->safe_psql( + 'postgres', qq( + CREATE TABLE tab_down (a int primary key); + INSERT INTO tab_down VALUES (11); + CREATE FUNCTION hold_copy() RETURNS trigger LANGUAGE plpgsql + AS \$\$ BEGIN PERFORM pg_advisory_xact_lock_shared(4711); RETURN NEW; END \$\$; + CREATE TRIGGER tab_down_hold BEFORE INSERT ON tab_down + FOR EACH ROW EXECUTE FUNCTION hold_copy(); + ALTER TABLE tab_down ENABLE ALWAYS TRIGGER tab_down_hold; +)); +my $hold_psql = $node_subscriber->background_psql('postgres'); +$hold_psql->query('SELECT pg_advisory_lock(4711)'); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_down CONNECTION '$publisher_connstr' PUBLICATION tap_pub_down" +); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 1 FROM pg_locks WHERE locktype = 'advisory' AND objid = 4711 AND NOT granted" +) or die "Timed out waiting for the copy of tab_down to start"; +$node_publisher->safe_psql('postgres', "INSERT INTO tab_down VALUES (11)"); +$hold_psql->query('SELECT pg_advisory_unlock(4711)'); +$hold_psql->quit; +$node_subscriber->poll_query_until('postgres', + "SELECT srsubstate = 'f' FROM pg_subscription_rel WHERE srrelid = 'tab_down'::regclass" +) or die "Timed out waiting for tab_down to stop at FINISHEDCOPY"; + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_down DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub_down' AND pid IS NOT NULL" +) or die "Timed out waiting for subscription workers to stop"; + +my $down_origin = $node_subscriber->safe_psql('postgres', + "SELECT 'pg_' || srsubid || '_' || srrelid FROM pg_subscription_rel WHERE srrelid = 'tab_down'::regclass" +); +my $down_progress_sql = + "SELECT remote_lsn FROM pg_replication_origin_status WHERE external_id = '$down_origin'"; +my $down_progress = + $node_subscriber->safe_psql('postgres', $down_progress_sql); +isnt($down_progress, '', 'the tablesync origin of tab_down has progress'); + +my $unreachable_connstr = + $node_publisher->connstr . ' dbname=regress_doesnotexist'; +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_down CONNECTION '$unreachable_connstr'"); +refresh_should_fail( + "ALTER SUBSCRIPTION tap_sub_down REFRESH TABLE tab_down", + qr/could not connect to the publisher/, + 'REFRESH TABLE fails when the publisher cannot be reached'); +is( $node_subscriber->safe_psql('postgres', $down_progress_sql), + $down_progress, + 'the tablesync origin keeps its progress when the publisher cannot be reached' +); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_down CONNECTION '$publisher_connstr'"); + +# The publisher can still fail the command once it has been reached, in the +# middle of dropping the slots, so the origins are dropped after them. Keep +# the tablesync slot of tab_down busy, which makes its drop wait, and cancel +# the drop while it waits. +my $down_slot = $node_subscriber->safe_psql('postgres', + "SELECT 'pg_' || srsubid || '_sync_' || srrelid || '_' || system_identifier FROM pg_subscription_rel, pg_control_system() WHERE srrelid = 'tab_down'::regclass" +); +my $down_slot_free = + "SELECT NOT active FROM pg_replication_slots WHERE slot_name = '$down_slot'"; +$node_publisher->poll_query_until('postgres', $down_slot_free) + or die "Timed out waiting for the tablesync slot to be released"; + +my ($recv_stdout, $recv_stderr) = ('', ''); +my $recv_timeout = + IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default); +my $recv = IPC::Run::start( + [ + 'pg_recvlogical', + '--dbname' => $node_publisher->connstr('postgres'), + '--slot' => $down_slot, + '--option' => 'proto_version=1', + '--option' => 'publication_names=tap_pub_down', + '--file' => '-', + '--verbose', + '--no-loop', + '--start', + ], + '>' => \$recv_stdout, + '2>' => \$recv_stderr, + $recv_timeout); +pump_until($recv, $recv_timeout, \$recv_stderr, qr/streaming initiated/) + or die "pg_recvlogical did not start streaming from the tablesync slot"; + +my $slot_drop_waiting = + "FROM pg_stat_activity WHERE application_name = 'tap_sub_down' AND wait_event = 'ReplicationSlotDrop'"; +my $bg_down = + $node_subscriber->background_psql('postgres', on_error_stop => 0); +$bg_down->query_until(qr/drop_fired/, + "\\echo drop_fired\nALTER SUBSCRIPTION tap_sub_down REFRESH TABLE tab_down;\n" +); +$node_publisher->poll_query_until('postgres', + "SELECT count(*) = 1 $slot_drop_waiting") + or die "Timed out waiting for REFRESH TABLE to wait for the tablesync slot"; +$node_publisher->safe_psql('postgres', + "SELECT pg_cancel_backend(pid) $slot_drop_waiting"); +$bg_down->query('SELECT 1'); +like( + $bg_down->{stderr}, + qr/could not drop replication slot "\Q$down_slot\E" on publisher/, + 'REFRESH TABLE fails when a tablesync slot cannot be dropped'); +$bg_down->{stderr} = ''; +$bg_down->quit; +is( $node_subscriber->safe_psql('postgres', $down_progress_sql), + $down_progress, + 'the tablesync origin keeps its progress when its slot cannot be dropped' +); + +$recv->kill_kill; +$node_publisher->poll_query_until('postgres', $down_slot_free) + or die "Timed out waiting for pg_recvlogical to release the tablesync slot"; + +# Running the command again completes it, and the table is copied afresh. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_down REFRESH TABLE tab_down"); +is( $node_publisher->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_replication_slots WHERE slot_name = '$down_slot'" + ), + '0', + 'the tablesync slot is dropped once the command completes'); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_replication_origin WHERE roname = '$down_origin'" + ), + '0', + 'the tablesync origin is dropped once the command completes'); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub_down ENABLE"); +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_down'); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_down"), + '11', + 'tab_down is copied afresh once the command completes'); + +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_down"); + +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); -- 2.54.0