public inbox for [email protected]
help / color / mirror / Atom feedFrom: Cagri Biroglu <[email protected]>
To: [email protected]
Subject: Re: Per-table resync for logical replication subscriptions
Date: Wed, 8 Jul 2026 12:54:42 +0200
Message-ID: <CAA36msq+KWyQPqJjZ-+fzefdC_9GLLLYX1ZvR49-CQ7yMeB=0Q@mail.gmail.com> (raw)
In-Reply-To: <CAA36mso9zgvMVr4UNHMMAQyU-dQy33EqxjZ=yEDG8SAB6yYDJw@mail.gmail.com>
References: <CAA36mso9zgvMVr4UNHMMAQyU-dQy33EqxjZ=yEDG8SAB6yYDJw@mail.gmail.com>
here is the WIP patch for the disabled-subscription case.
It adds:
ALTER SUBSCRIPTION name REFRESH TABLE table_name
[ WITH (copy_data = true, truncate = true) ]
The command is subscriber-local: it resets just the named relation in
pg_subscription_rel back to the initial sync state (optionally
truncating the local copy first), so a tablesync worker re-copies that
one table on the next enable, reusing the existing tablesync machinery.
Publication membership and other tables are left untouched. As
discussed, this first version requires the subscription to be disabled.
The tests are included in the same patch: a TAP test at
src/test/subscription/t/039_refresh_table.pl that
- resyncs one drifted table on a two-node setup and confirms it
matches the publisher again after re-enabling;
- confirms only the targeted relation is reset (a sibling table is
left in 'r' state);
- checks the rejection cases: enabled subscription, table not in the
subscription, a sequence, truncate without copy_data, and running
inside a transaction block.
The full src/test/subscription suite and the core regression tests pass
with the patch applied.
Open questions if this direction makes sense ,I would still like feedback
on: the command spelling
(REFRESH TABLE vs RESYNC TABLE), and whether the enabled-subscription
case is worth doing as a follow-up (it needs the running apply worker to
notice and re-read the reset relation state).
Regards,
Cagri Biroglu
On Tue, Jul 7, 2026 at 11:41 AM Cagri Biroglu <[email protected]>
wrote:
> Hi hackers,
>
> I would like to propose a subscriber-side way to re-copy a single table
> that is already part of a logical replication subscription.
>
> At the moment, I do not think PostgreSQL has a supported way to do this
> for one table on one subscriber. ALTER SUBSCRIPTION ... REFRESH
> PUBLICATION only handles tables that were added to the subscribed
> publications since the subscription was created, or since the last refresh.
> It does not re-copy tables that are already known to the subscription.
>
> That is fine for the normal “new table was added to the publication” case,
> but it leaves a gap when only one subscriber needs one table to be
> re-seeded. Some examples are:
>
> -
>
> the subscriber table became inconsistent;
> -
>
> a row filter changed and the subscriber needs a fresh copy;
> -
>
> a forwarder/subscriber needs to repair one table without disturbing
> other subscribers.
>
> The usual workaround is to temporarily change publication membership,
> refresh the subscription, truncate the local table, add the table back, and
> refresh again. That works in simple topologies, but it is not really local
> to the subscriber. In a fan-out topology it can affect sibling subscribers,
> and in a forwarding topology the truncate can be replicated further
> downstream if truncates are published.
>
> There is also the unsupported option of updating pg_subscription_rel by
> hand to put the table back into the initial sync state. That is not safe
> either, because it races with the apply worker and does not handle the
> surrounding cleanup and invalidation work.
>
> There is already a similar mechanism for sequences. ALTER SUBSCRIPTION
> ... REFRESH SEQUENCES can re-synchronize sequence data for sequences
> already known to the subscription, while REFRESH PUBLICATION only
> discovers newly added ones. This proposal is essentially the table-side
> equivalent of that idea.
>
> What I am proposing is something like:
>
> ALTER SUBSCRIPTION name REFRESH TABLE table_name
> [ WITH (copy_data = true, truncate = true) ];
>
> or, if people prefer the wording:
>
> ALTER SUBSCRIPTION name RESYNC TABLE table_name
> [ WITH (copy_data = true, truncate = true) ];
>
> The command would be subscriber-local and would not change publication
> membership. Internally, it would reuse the existing table synchronization
> machinery by resetting the selected relation back to the initial sync
> state, so that a tablesync worker copies it again and moves it through the
> normal i -> d -> s -> r states.
>
> I tested the current behavior on current devel with a small three-node
> setup:
>
> -
>
> upstream publisher U;
> -
>
> forwarding subscriber/publisher F;
> -
>
> downstream subscriber D;
> -
>
> sibling subscriber S.
>
> The test script keeps a writer running and prints per-table checksums at
> each step.
>
> The important observations were:
>
> 1.
>
> ALTER SUBSCRIPTION ... REFRESH PUBLICATION on F is a no-op for an
> already subscribed table. The table remained in ready state, the tuple
> metadata did not change, no WAL was generated, and no sync worker was
> started.
> 2.
>
> The publication-membership workaround is not local. In the forwarding
> case, truncating the table on F also wiped the table on D. In the
> fan-out case, the sibling subscriber S was left behind because the
> publication changes were global but the repair was only intended for F.
> 3.
>
> A controlled subscriber-local reset of only the selected table on F
> produced matching checksums again without touching the publication or S.
> Doing the same thing as a bare catalog update while the subscription
> remained enabled was not reliable; the apply worker could continue with
> cached state and the table diverged.
>
> So I do not think the hard part is inventing a new synchronization
> mechanism. The existing tablesync path already does most of the work. The
> missing piece is a supported command that performs the reset safely:
> stopping or coordinating with workers, cleaning up per-table sync state and
> slots/origins where needed, optionally truncating the target table,
> invalidating caches, taking the right locks, and validating that the
> requested table is actually part of the subscription.
>
> For a first version, I think it would be reasonable to require the
> subscription to be disabled. That avoids the apply-worker race and keeps
> the patch smaller. A later version could support the enabled case, probably
> by making the running worker notice and re-read the changed relation state.
>
> I have a WIP patch for the disabled-subscription, executor changes, and a
> TAP test. Before polishing it further, I would like to get feedback on
> whether this direction makes sense ?
>
> Regards,
>
> Cagri Biroglu
>
Attachments:
[application/octet-stream] v1-0001-refresh-table.patch (18.9K, ../CAA36msq+KWyQPqJjZ-+fzefdC_9GLLLYX1ZvR49-CQ7yMeB=0Q@mail.gmail.com/3-v1-0001-refresh-table.patch)
download | inline diff:
From 569385da0bb7c3d40aec99842c7ea0504f8ba674 Mon Sep 17 00:00:00 2001
From: cagrib <[email protected]>
Date: Tue, 7 Jul 2026 14:17:19 +0200
Subject: [PATCH v1] Add ALTER SUBSCRIPTION ... REFRESH TABLE to resync one
table
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 a single 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
[ WITH (copy_data = true, truncate = true) ]
It resets the state of just the named relation in pg_subscription_rel
back to init (optionally truncating the local copy first) so that a
tablesync worker re-copies that one table, reusing the existing table
synchronization machinery. Publication membership and other tables are
left untouched. This mirrors REFRESH SEQUENCES, which already
re-synchronizes already-subscribed sequences.
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 | 180 +++++++++++++++++
src/backend/parser/gram.y | 11 ++
src/include/nodes/parsenodes.h | 2 +
src/test/subscription/t/039_refresh_table.pl | 198 +++++++++++++++++++
4 files changed, 391 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..87029a5c034 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -82,6 +82,7 @@
#define SUBOPT_LSN 0x00020000
#define SUBOPT_ORIGIN 0x00040000
#define SUBOPT_CONFLICT_LOG_DEST 0x00080000
+#define SUBOPT_TRUNCATE 0x00100000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -99,6 +100,7 @@ typedef struct SubOpts
bool enabled;
bool create_slot;
bool copy_data;
+ bool truncate;
bool refresh;
bool binary;
char streaming;
@@ -183,6 +185,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->create_slot = true;
if (IsSet(supported_opts, SUBOPT_COPY_DATA))
opts->copy_data = true;
+ if (IsSet(supported_opts, SUBOPT_TRUNCATE))
+ opts->truncate = true;
if (IsSet(supported_opts, SUBOPT_REFRESH))
opts->refresh = true;
if (IsSet(supported_opts, SUBOPT_BINARY))
@@ -264,6 +268,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_COPY_DATA;
opts->copy_data = defGetBoolean(defel);
}
+ else if (IsSet(supported_opts, SUBOPT_TRUNCATE) &&
+ strcmp(defel->defname, "truncate") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_TRUNCATE))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_TRUNCATE;
+ opts->truncate = defGetBoolean(defel);
+ }
else if (IsSet(supported_opts, SUBOPT_SYNCHRONOUS_COMMIT) &&
strcmp(defel->defname, "synchronous_commit") == 0)
{
@@ -1405,6 +1418,133 @@ AlterSubscription_refresh_seq(Subscription *sub)
PG_END_TRY();
}
+/*
+ * Resynchronize a single already-subscribed table.
+ *
+ * Resets the relation's pg_subscription_rel state back to init (optionally
+ * truncating the local copy first) so that, once the subscription is enabled,
+ * a tablesync worker re-copies just this one table. This is subscriber-local
+ * and does not touch publication membership, so sibling subscribers are
+ * unaffected.
+ *
+ * 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, bool copy_data,
+ bool truncate, RangeVar *rv)
+{
+ Relation rel;
+ Oid relid;
+ char relstate;
+ XLogRecPtr relstatelsn;
+
+ /*
+ * Lock pg_subscription_rel with AccessExclusiveLock, matching
+ * AlterSubscription_refresh(), so the relation's state cannot change under
+ * us for the duration of the command.
+ */
+ rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock);
+
+ 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));
+
+ /* 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();
+ }
+
+ /* Optionally clear the local copy so the re-copy starts from empty. */
+ if (copy_data && truncate)
+ {
+ TruncateStmt *tstmt = makeNode(TruncateStmt);
+
+ tstmt->relations = list_make1(rv);
+ tstmt->restart_seqs = false;
+ tstmt->behavior = DROP_RESTRICT;
+
+ ExecuteTruncate(tstmt);
+ }
+
+ /*
+ * Reset the relation state. With copy_data it goes back to init so a
+ * tablesync worker re-copies it on enable; otherwise straight to ready.
+ */
+ UpdateSubscriptionRelState(sub->oid, relid,
+ copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
+ 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.
@@ -1644,6 +1784,10 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
supported_opts = SUBOPT_COPY_DATA;
break;
+ case ALTER_SUBSCRIPTION_REFRESH_TABLE:
+ supported_opts = SUBOPT_COPY_DATA | SUBOPT_TRUNCATE;
+ break;
+
case ALTER_SUBSCRIPTION_SKIP:
supported_opts = SUBOPT_LSN;
break;
@@ -2296,6 +2440,42 @@ 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."));
+
+ /* TRUNCATE only makes sense when we are going to re-copy. */
+ if (!opts.copy_data)
+ {
+ if (IsSet(opts.specified_opts, SUBOPT_TRUNCATE) &&
+ opts.truncate)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("truncate = true requires copy_data = true"));
+
+ opts.truncate = false;
+ }
+
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... REFRESH TABLE");
+
+ AlterSubscription_refresh_table(sub, opts.copy_data,
+ opts.truncate, stmt->relation);
+
+ 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..b71dd58ecf4 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11617,6 +11617,17 @@ AlterSubscriptionStmt:
n->subname = $3;
$$ = (Node *) n;
}
+ | ALTER SUBSCRIPTION name REFRESH TABLE qualified_name opt_definition
+ {
+ AlterSubscriptionStmt *n =
+ makeNode(AlterSubscriptionStmt);
+
+ n->kind = ALTER_SUBSCRIPTION_REFRESH_TABLE;
+ n->subname = $3;
+ n->relation = $6;
+ n->options = $7;
+ $$ = (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..9c5208b7bc1 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 */
+ RangeVar *relation; /* Table 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..b5d16e96bbe
--- /dev/null
+++ b/src/test/subscription/t/039_refresh_table.pl
@@ -0,0 +1,198 @@
+
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+# Tests for ALTER SUBSCRIPTION ... REFRESH TABLE, which re-copies a single
+# already-subscribed table 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');
+
+# truncate = true requires copy_data = true.
+refresh_should_fail(
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res WITH (copy_data = false, truncate = true)",
+ qr/truncate = true requires copy_data = true/,
+ 'REFRESH TABLE rejected with truncate but no copy_data');
+
+# 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');
+
+# 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 (defaults: copy_data = true, truncate = true).
+$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');
+
+# copy_data = false variant: reset to ready without re-copying or truncating.
+$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', "DELETE FROM tab_res WHERE a <= 10");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub REFRESH TABLE tab_res WITH (copy_data = false)"
+);
+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',
+ 'copy_data = false sets state directly to ready');
+is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_res"),
+ '91', 'copy_data = false does not truncate or re-copy');
+
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub ENABLE");
+
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
+
+done_testing();
--
2.54.0
view thread (3+ messages) latest in thread
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]
Subject: Re: Per-table resync for logical replication subscriptions
In-Reply-To: <CAA36msq+KWyQPqJjZ-+fzefdC_9GLLLYX1ZvR49-CQ7yMeB=0Q@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