v1-0001-refresh-table.patch

From 569385da0bb7c3d40aec99842c7ea0504f8ba674 Mon Sep 17 00:00:00 2001
From: cagrib <cagri.biroglu@adyen.com>
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