public inbox for [email protected]
help / color / mirror / Atom feedFrom: Cagri Biroglu <[email protected]>
To: Masahiko Sawada <[email protected]>
To: [email protected] <[email protected]>
Cc: [email protected]
Subject: Re: Per-table resync for logical replication subscriptions
Date: Fri, 10 Jul 2026 12:38:10 +0200
Message-ID: <CAA36msq=JYBN+zm2MoKqXZCSjMhhdmhYwRWVZnhPKdE8zaUGFA@mail.gmail.com> (raw)
In-Reply-To: <CAD21AoD_RKYrzcR377Ptcr4doKnsF-+k80QTPsMwG9oe2O83hg@mail.gmail.com>
References: <CAA36mso9zgvMVr4UNHMMAQyU-dQy33EqxjZ=yEDG8SAB6yYDJw@mail.gmail.com>
<CAA36msq+KWyQPqJjZ-+fzefdC_9GLLLYX1ZvR49-CQ7yMeB=0Q@mail.gmail.com>
<CAD21AoD_RKYrzcR377Ptcr4doKnsF-+k80QTPsMwG9oe2O83hg@mail.gmail.com>
>
> Thanks a lot for looking at this.
>
> Before the point-by-point: one change after getting feedback on this
from the sketch in my first mail
is that I dropped the WITH (copy_data, truncate) options. For refreshing
> a table that is already subscribed, only one combination is useful, so
> REFRESH TABLE now always truncates the local copy and re-copies (what
> would have been copy_data = true, truncate = true). Without truncating,
> the re-copy hits duplicate-key errors or doubles the rows; skipping the
> copy leaves the drift in place. So neither other combination is
> meaningful, and there is nothing to configure.
>
> > REFRESH TABLE is better to me. It's more consistent with the existing
> > commands such as REFRESH SEQUENCES.
>
> Great, I've kept REFRESH TABLE.
>
> > I think it might be a good idea that the REFRESH TABLE command can
> > take multiple tables rather than one table.
>
> Agreed. v2 makes REFRESH TABLE take a comma-separated list:
>
> ALTER SUBSCRIPTION name REFRESH TABLE t1, t2, ...
>
> It's all-or-nothing: every named relation is validated (part of the
> subscription, not a sequence, exists) before any of them is reset, so a
> bad table name aborts the whole command without touching the others. The
> listed tables are also truncated together, which lets a set connected by
> foreign keys be re-seeded as a unit.
>
> One note on scope: this doesn't add resource risk over doing one table
> at a time, because the re-copy is still performed by tablesync workers
> that are capped by max_sync_workers_per_subscription; resetting N tables
> just queues them rather than starting N copies at once.
>
> > Why does the REFRESH TABLE command require for the subscription to be
> > disabled while REFRESH PUBLICATIONS doesn't?
>
> Good question, but I'd actually argue the disabled requirement is a
> reasonable design choice here rather than a limitation to remove.
>
> The reason the two commands differ is that REFRESH PUBLICATION only adds
> *new* relations to pg_subscription_rel (in init state), whereas REFRESH
> TABLE resets a relation that is already in ready state, and those hit
> different caches in the apply worker:
>
> - The set of not-ready relations is cached (table_states_not_ready in
> syncutils.c) and correctly invalidated by the SUBSCRIPTIONRELMAP
> syscache callback (InvalidateSyncingRelStates). So after the reset the
> apply worker does notice the table is non-ready and does launch a
> tablesync worker for it. That part works whether enabled or not.
>
> - However, whether the apply worker applies incoming changes for a
> relation is decided by should_apply_changes_for_rel(), which reads the
> per-relation LogicalRepRelMapEntry.state. That state is only refreshed
> from the catalog while it is not READY:
>
> /* relation.c, logicalrep_rel_open() */
> if (entry->state != SUBREL_STATE_READY)
> entry->state = GetSubscriptionRelState(...);
>
> For an already-ready table the entry stays READY, and the relcache
> invalidation only clears localrelvalid (rebuilds the attrmap); it does
> not reset entry->state. So the running apply worker keeps treating the
> table as READY and keeps applying live changes to it, at the same time
> as the freshly launched tablesync worker re-copies it, with no handoff
> on the sync LSN. In my testing that reliably diverges the table.
>
> For a newly added table (REFRESH PUBLICATION) there is no stale READY
> entry, so the normal init -> datasync -> syncdone -> ready handoff works
> without any disable.
>
> Requiring the subscription to be disabled sidesteps this cleanly. When
> you re-enable, a fresh apply worker starts and reads the reset table's
> state from the catalog as init, so the same init -> datasync -> syncdone
> -> ready handoff kicks in, exactly as for a newly added table. And the
> disable only needs to bracket the quick catalog reset, not the re-copy
> itself: REFRESH TABLE just truncates and flips the state, and after
> ENABLE the tablesync worker does the (possibly long) copy while the
> apply worker resumes the other tables as usual, so the pause is short
> and scoped to the metadata change, not the data copy.
>
> Supporting the enabled case would instead mean reaching into the running
> apply worker to force it to drop and re-read the cached state of the
> reset relation mid-flight. That is doable, but it adds moving parts and
> a new invalidation path to get right, for what is a deliberate,
> occasional maintenance action where briefly disabling the subscription
> does no harm. So my inclination is to keep the disabled requirement here
> rather than trade that simplicity away. Happy to reconsider if you think
> the enabled case matters enough to justify the extra machinery or i am
> missing sth.
Regards,
> Cagri Biroglu
>
Attachments:
[application/octet-stream] v2-0001-refresh-table.patch (19.3K, ../CAA36msq=JYBN+zm2MoKqXZCSjMhhdmhYwRWVZnhPKdE8zaUGFA@mail.gmail.com/3-v2-0001-refresh-table.patch)
download | inline diff:
From fa82f9f0ece03430671273905073f3364e9ee6d4 Mon Sep 17 00:00:00 2001
From: cagrib <[email protected]>
Date: Tue, 7 Jul 2026 14:17:19 +0200
Subject: [PATCH v2] Add ALTER SUBSCRIPTION ... REFRESH TABLE to resync tables
REFRESH PUBLICATION only starts syncing tables that were newly added to
the publication; it never re-copies a table that is already part of the
subscription. Today the only ways to re-seed an already-subscribed table
are to churn publication membership (which is not local to one subscriber
and can disturb sibling subscribers or cascade downstream) or to hand-edit
pg_subscription_rel (unsupported and racy against the apply worker).
Add a subscriber-local command:
ALTER SUBSCRIPTION name REFRESH TABLE table_name [, ...]
It truncates the local copy of each named table and resets its state in
pg_subscription_rel back to init so that a tablesync worker re-copies just
those tables, reusing the existing table synchronization machinery. All
named relations are validated before any is reset, so the command is
all-or-nothing, and they are truncated together so a set connected by
foreign keys can be re-seeded as a unit. Publication membership and other
tables are left untouched. This mirrors REFRESH SEQUENCES, which likewise
re-synchronizes objects already known to the subscription rather than only
discovering newly added ones.
This first version requires the subscription to be disabled, which avoids
racing the apply worker's cached relation state.
A TAP test is included.
---
src/backend/commands/subscriptioncmds.c | 187 +++++++++++++++
src/backend/parser/gram.y | 10 +
src/include/nodes/parsenodes.h | 2 +
src/test/subscription/t/039_refresh_table.pl | 228 +++++++++++++++++++
4 files changed, 427 insertions(+)
create mode 100644 src/test/subscription/t/039_refresh_table.pl
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 4292e7fb8f4..dd6096d5843 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1405,6 +1405,170 @@ AlterSubscription_refresh_seq(Subscription *sub)
PG_END_TRY();
}
+/*
+ * 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 rel;
+ ListCell *lc;
+ List *relids = NIL;
+ List *truncrels = NIL;
+ TruncateStmt *tstmt;
+
+ /*
+ * Lock pg_subscription_rel with AccessExclusiveLock, matching
+ * AlterSubscription_refresh(), so the relation states cannot change under
+ * us for the duration of the command.
+ */
+ rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock);
+
+ /*
+ * First pass: validate every named relation before changing anything, so
+ * the command is all-or-nothing. Duplicates in the list are ignored.
+ */
+ foreach(lc, relations)
+ {
+ RangeVar *rv = lfirst_node(RangeVar, lc);
+ Oid relid;
+ char relstate;
+ XLogRecPtr relstatelsn;
+
+ relid = RangeVarGetRelid(rv, AccessShareLock, false);
+
+ if (get_rel_relkind(relid) == RELKIND_SEQUENCE)
+ ereport(ERROR,
+ errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot refresh sequence \"%s\" as a table",
+ rv->relname),
+ errhint("Use ALTER SUBSCRIPTION ... REFRESH SEQUENCES instead."));
+
+ /* The relation must already be part of the subscription. */
+ relstate = GetSubscriptionRelState(sub->oid, relid, &relstatelsn);
+ if (relstate == SUBREL_STATE_UNKNOWN)
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("table \"%s\" is not part of the subscription \"%s\"",
+ rv->relname, sub->name));
+
+ if (list_member_oid(relids, relid))
+ continue;
+
+ relids = lappend_oid(relids, relid);
+ truncrels = lappend(truncrels, rv);
+ }
+
+ /*
+ * Second pass: stop any leftover tablesync worker for each relation and,
+ * if it was caught mid-sync, clean up its origin and publisher slot.
+ */
+ foreach_oid(relid, relids)
+ {
+ char relstate;
+ XLogRecPtr relstatelsn;
+
+ relstate = GetSubscriptionRelState(sub->oid, relid, &relstatelsn);
+
+ /* Stop any leftover tablesync worker for this relation. */
+ logicalrep_worker_stop(WORKERTYPE_TABLESYNC, sub->oid, relid);
+
+ /*
+ * If the relation was caught mid-sync, drop its tablesync origin. For
+ * READY state the tablesync worker already dropped it, so nothing is
+ * needed there. Pass missing_ok = true as the origin may not exist
+ * yet.
+ */
+ if (relstate != SUBREL_STATE_READY)
+ {
+ char originname[NAMEDATALEN];
+
+ ReplicationOriginNameForLogicalRep(sub->oid, relid, originname,
+ sizeof(originname));
+ replorigin_drop_by_name(originname, true, false);
+ }
+
+ /*
+ * Likewise drop the tablesync slot on the publisher for a mid-sync
+ * relation. READY/SYNCDONE relations have no such slot, so the common
+ * case needs no publisher connection at all.
+ */
+ if (relstate != SUBREL_STATE_READY && relstate != SUBREL_STATE_SYNCDONE)
+ {
+ char syncslotname[NAMEDATALEN] = {0};
+ char *err = NULL;
+ WalReceiverConn *wrconn;
+ bool must_use_password;
+
+ load_file("libpqwalreceiver", false);
+
+ must_use_password = sub->passwordrequired && !sub->ownersuperuser;
+ wrconn = walrcv_connect(sub->conninfo, true, true, must_use_password,
+ sub->name, &err);
+ if (!wrconn)
+ ereport(ERROR,
+ errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("subscription \"%s\" could not connect to the publisher: %s",
+ sub->name, err));
+
+ PG_TRY();
+ {
+ ReplicationSlotNameForTablesync(sub->oid, relid, syncslotname,
+ sizeof(syncslotname));
+ ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
+ }
+ PG_FINALLY();
+ {
+ walrcv_disconnect(wrconn);
+ }
+ PG_END_TRY();
+ }
+ }
+
+ /*
+ * 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.
+ */
+ tstmt = makeNode(TruncateStmt);
+ tstmt->relations = truncrels;
+ tstmt->restart_seqs = false;
+ tstmt->behavior = DROP_RESTRICT;
+ ExecuteTruncate(tstmt);
+
+ /*
+ * Reset each relation to init so that a tablesync worker re-copies it once
+ * the subscription is enabled.
+ */
+ foreach_oid(relid, relids)
+ {
+ 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));
+ }
+
+ table_close(rel, NoLock);
+}
+
/*
* Common checks for altering failover, two_phase, and retain_dead_tuples
* options.
@@ -2296,6 +2460,29 @@ 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."));
+
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... REFRESH TABLE");
+
+ AlterSubscription_refresh_table(sub, stmt->relations);
+
+ break;
+ }
+
case ALTER_SUBSCRIPTION_SKIP:
{
/* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index ff4e1388c55..82716c46947 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11617,6 +11617,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/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index e03556399ab..cfeb09ab417 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4566,6 +4566,7 @@ typedef enum AlterSubscriptionType
ALTER_SUBSCRIPTION_DROP_PUBLICATION,
ALTER_SUBSCRIPTION_REFRESH_PUBLICATION,
ALTER_SUBSCRIPTION_REFRESH_SEQUENCES,
+ ALTER_SUBSCRIPTION_REFRESH_TABLE,
ALTER_SUBSCRIPTION_ENABLED,
ALTER_SUBSCRIPTION_SKIP,
} AlterSubscriptionType;
@@ -4578,6 +4579,7 @@ typedef struct AlterSubscriptionStmt
char *servername; /* Server name of publisher */
char *conninfo; /* Connection string to publisher */
List *publication; /* One or more publication to subscribe to */
+ List *relations; /* Tables to resync (for REFRESH TABLE) */
List *options; /* List of DefElem nodes */
} AlterSubscriptionStmt;
diff --git a/src/test/subscription/t/039_refresh_table.pl b/src/test/subscription/t/039_refresh_table.pl
new file mode 100644
index 00000000000..b09f1e149bf
--- /dev/null
+++ b/src/test/subscription/t/039_refresh_table.pl
@@ -0,0 +1,228 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+# Tests for ALTER SUBSCRIPTION ... REFRESH TABLE, which re-copies one or more
+# already-subscribed tables on the subscriber without touching publication
+# membership or other tables. The first version requires the subscription
+# to be disabled.
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher and subscriber nodes
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ "wal_retrieve_retry_interval = 1ms");
+$node_subscriber->start;
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+
+# Preexisting content on the publisher: two published tables.
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ CREATE TABLE tab_res (a int primary key, b text);
+ CREATE TABLE tab_other (a int primary key, b text);
+ INSERT INTO tab_res SELECT g, 'p' || g FROM generate_series(1, 100) g;
+ INSERT INTO tab_other SELECT g, 'q' || g FROM generate_series(1, 100) g;
+ CREATE PUBLICATION tap_pub FOR TABLE tab_res, tab_other;
+));
+
+# Matching structure on the subscriber, plus objects used for error cases.
+$node_subscriber->safe_psql(
+ 'postgres', qq(
+ CREATE TABLE tab_res (a int primary key, b text);
+ CREATE TABLE tab_other (a int primary key, b text);
+ CREATE TABLE tab_local (a int primary key);
+ CREATE SEQUENCE seq_local;
+));
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub"
+);
+
+# Wait for initial sync of both tables.
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"),
+ '100', 'initial sync of tab_res');
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_other"),
+ '100', 'initial sync of tab_other');
+
+# A small helper: run SQL expected to fail, and check the error message.
+sub refresh_should_fail
+{
+ my ($sql, $pattern, $desc) = @_;
+ my ($ret, $stdout, $stderr) = ('', '', '');
+ $ret = $node_subscriber->psql('postgres', $sql,
+ stdout => \$stdout, stderr => \$stderr);
+ ok($ret != 0 && $stderr =~ /$pattern/,
+ $desc)
+ or diag("got ret=$ret stderr=$stderr");
+}
+
+# REFRESH TABLE is not allowed while the subscription is enabled.
+refresh_should_fail(
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res",
+ qr/not allowed for enabled subscriptions/,
+ 'REFRESH TABLE rejected while enabled');
+
+# Disable the subscription and wait for its workers to stop.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub DISABLE");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub' AND pid IS NOT NULL"
+) or die "Timed out waiting for subscription workers to stop";
+
+# A table that is not part of the subscription is rejected.
+refresh_should_fail(
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_local",
+ qr/is not part of the subscription/,
+ 'REFRESH TABLE rejected for table not in subscription');
+
+# A sequence cannot be refreshed as a table.
+refresh_should_fail(
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE seq_local",
+ qr/cannot refresh sequence/,
+ 'REFRESH TABLE rejected for a sequence');
+
+# The command cannot run inside a transaction block.
+refresh_should_fail(
+ "BEGIN; ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res;",
+ qr/cannot run inside a transaction block/,
+ 'REFRESH TABLE rejected inside a transaction block');
+
+# All-or-nothing: a list containing one bad table aborts the whole command and
+# leaves the valid table untouched.
+refresh_should_fail(
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res, tab_local",
+ qr/is not part of the subscription/,
+ 'REFRESH TABLE with a bad table in the list is rejected');
+is( $node_subscriber->safe_psql(
+ 'postgres',
+ "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_res'"
+ ),
+ 'r',
+ 'valid table not reset when another table in the list is invalid');
+
+# Introduce drift on the subscriber, only in tab_res.
+$node_subscriber->safe_psql(
+ 'postgres', qq(
+ DELETE FROM tab_res WHERE a <= 40;
+ UPDATE tab_res SET b = 'CORRUPT' WHERE a = 60;
+));
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"),
+ '60', 'drift introduced in tab_res');
+
+# Record the pre-refresh state of the other table.
+my $other_state_before = $node_subscriber->safe_psql('postgres',
+ "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_other'"
+);
+is($other_state_before, 'r', 'tab_other is ready before refresh');
+
+# Resync just tab_res.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res");
+
+# Only tab_res is reset to init; tab_other is untouched; local copy truncated.
+is( $node_subscriber->safe_psql(
+ 'postgres',
+ "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_res'"
+ ),
+ 'i',
+ 'tab_res reset to init state');
+is( $node_subscriber->safe_psql(
+ 'postgres',
+ "SELECT srsubstate FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname = 'tab_other'"
+ ),
+ 'r',
+ 'tab_other left untouched (still ready)');
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"),
+ '0', 'tab_res truncated locally by REFRESH while disabled');
+
+# Re-enable and wait for the single table to re-copy.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub ENABLE");
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"),
+ '100', 'tab_res re-copied after enable');
+is( $node_subscriber->safe_psql('postgres',
+ "SELECT b FROM tab_res WHERE a = 60"),
+ 'p60', 'tab_res corruption repaired by resync');
+
+# Full content matches the publisher.
+my $pub_md5 = $node_publisher->safe_psql('postgres',
+ "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res");
+my $sub_md5 = $node_subscriber->safe_psql('postgres',
+ "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res");
+is($sub_md5, $pub_md5, 'tab_res matches publisher after resync');
+
+# tab_other was never disturbed.
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_other"),
+ '100', 'tab_other intact throughout');
+
+# Ongoing replication still works for the resynced table.
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_res VALUES (101, 'p101')");
+$node_publisher->wait_for_catchup('tap_sub');
+is( $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM tab_res WHERE a = 101"),
+ '1', 'streaming resumes on resynced table');
+
+# Multiple tables can be resynced in a single command. Disable, drift both
+# tables, and refresh them together.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub DISABLE");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'tap_sub' AND pid IS NOT NULL"
+) or die "Timed out waiting for subscription workers to stop";
+
+$node_subscriber->safe_psql(
+ 'postgres', qq(
+ DELETE FROM tab_res WHERE a <= 20;
+ DELETE FROM tab_other WHERE a <= 20;
+));
+
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res, tab_other");
+
+# Both listed tables are reset to init and truncated locally.
+is( $node_subscriber->safe_psql(
+ 'postgres',
+ "SELECT string_agg(srsubstate, ',' ORDER BY c.relname) FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid WHERE c.relname IN ('tab_other', 'tab_res')"
+ ),
+ 'i,i',
+ 'both listed tables reset to init state');
+is( $node_subscriber->safe_psql('postgres',
+ "SELECT (SELECT count(*) FROM tab_res) + (SELECT count(*) FROM tab_other)"),
+ '0', 'both listed tables truncated locally');
+
+# Re-enable and wait for both tables to re-copy.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub ENABLE");
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+
+my $pub_res = $node_publisher->safe_psql('postgres',
+ "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res");
+my $sub_res = $node_subscriber->safe_psql('postgres',
+ "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_res");
+is($sub_res, $pub_res, 'tab_res matches publisher after multi-table resync');
+
+my $pub_oth = $node_publisher->safe_psql('postgres',
+ "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_other");
+my $sub_oth = $node_subscriber->safe_psql('postgres',
+ "SELECT md5(string_agg(a || ':' || b, ',' ORDER BY a)) FROM tab_other");
+is($sub_oth, $pub_oth, 'tab_other matches publisher after multi-table resync');
+
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
+
+done_testing();
--
2.54.0
view thread (9+ messages)
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected]
Subject: Re: Per-table resync for logical replication subscriptions
In-Reply-To: <CAA36msq=JYBN+zm2MoKqXZCSjMhhdmhYwRWVZnhPKdE8zaUGFA@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox