v20251024-0002-Add-worker-type-argument-to-logicalrep_wor.patch

From 85e350f069379c4064d727efed26ea6e713b197d Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Fri, 24 Oct 2025 15:03:50 +0530
Subject: [PATCH v20251024 2/4] Add worker type argument to
 logicalrep_worker_stop and logicalrep_worker_find

Extend the logicalrep_worker_stop and logicalrep_worker_find functions to
accept a worker type argument. This change makes it possible to distinguish
between different logical replication worker types, such as table sync and
sequence sync workers. The enhancement does not alter existing behavior but
prepares the code for future patches that will introduce sequence
synchronization workers.
---
 src/backend/commands/subscriptioncmds.c     |  4 ++--
 src/backend/replication/logical/launcher.c  | 26 ++++++++++++++-------
 src/backend/replication/logical/tablesync.c |  7 +++---
 src/backend/replication/logical/worker.c    |  2 +-
 src/include/replication/worker_internal.h   |  4 +++-
 5 files changed, 28 insertions(+), 15 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index a0974d71de1..0b18f2feae4 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1082,7 +1082,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 
 				sub_remove_rels = lappend(sub_remove_rels, remove_rel);
 
-				logicalrep_worker_stop(sub->oid, relid);
+				logicalrep_worker_stop(sub->oid, relid, WORKERTYPE_TABLESYNC);
 
 				/*
 				 * For READY state, we would have already dropped the
@@ -2134,7 +2134,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	{
 		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-		logicalrep_worker_stop(w->subid, w->relid);
+		logicalrep_worker_stop(w->subid, w->relid, w->type);
 	}
 	list_free(subworkers);
 
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 218cefe86e2..d42e7815e8c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -246,19 +246,23 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 
 /*
  * Walks the workers array and searches for one that matches given
- * subscription id and relid.
+ * subscription id, relid and type.
  *
  * We are only interested in the leader apply worker or table sync worker.
+ * For apply workers, the relid should be set to InvalidOid, as they manage
+ * changes across all tables and sequences. For table sync workers, the relid
+ * should be set to the OID of the relation being synchronized.
  */
 LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid subid, Oid relid, LogicalRepWorkerType wtype,
+					   bool only_running)
 {
 	int			i;
 	LogicalRepWorker *res = NULL;
 
 	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
-	/* Search for attached worker for a given subscription id. */
+	/* Search for the attached worker matching the specified criteria. */
 	for (i = 0; i < max_logical_replication_workers; i++)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
@@ -268,7 +272,7 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 			continue;
 
 		if (w->in_use && w->subid == subid && w->relid == relid &&
-			(!only_running || w->proc))
+			w->type == wtype && (!only_running || w->proc))
 		{
 			res = w;
 			break;
@@ -628,15 +632,18 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
 
 /*
  * Stop the logical replication worker for subid/relid, if any.
+ *
+ * Similar to logicalrep_worker_find, relid should be set to a valid OID only
+ * for table sync workers.
  */
 void
-logicalrep_worker_stop(Oid subid, Oid relid)
+logicalrep_worker_stop(Oid subid, Oid relid, LogicalRepWorkerType wtype)
 {
 	LogicalRepWorker *worker;
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-	worker = logicalrep_worker_find(subid, relid, false);
+	worker = logicalrep_worker_find(subid, relid, wtype, false);
 
 	if (worker)
 	{
@@ -700,10 +707,12 @@ void
 logicalrep_worker_wakeup(Oid subid, Oid relid)
 {
 	LogicalRepWorker *worker;
+	LogicalRepWorkerType wtype =
+		OidIsValid(relid) ? WORKERTYPE_TABLESYNC : WORKERTYPE_APPLY;
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-	worker = logicalrep_worker_find(subid, relid, true);
+	worker = logicalrep_worker_find(subid, relid, wtype, true);
 
 	if (worker)
 		logicalrep_worker_wakeup_ptr(worker);
@@ -1260,7 +1269,8 @@ ApplyLauncherMain(Datum main_arg)
 				continue;
 
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-			w = logicalrep_worker_find(sub->oid, InvalidOid, false);
+			w = logicalrep_worker_find(sub->oid, InvalidOid, WORKERTYPE_APPLY,
+									   false);
 
 			if (w != NULL)
 			{
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 40e1ed3c20e..8f606c8a59c 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -161,7 +161,7 @@ wait_for_table_state_change(Oid relid, char expected_state)
 		/* Check if the sync worker is still running and bail if not. */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 		worker = logicalrep_worker_find(MyLogicalRepWorker->subid, relid,
-										false);
+										WORKERTYPE_TABLESYNC, false);
 		LWLockRelease(LogicalRepWorkerLock);
 		if (!worker)
 			break;
@@ -208,7 +208,7 @@ wait_for_worker_state_change(char expected_state)
 		 */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 		worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-										InvalidOid, false);
+										InvalidOid, WORKERTYPE_APPLY, false);
 		if (worker && worker->proc)
 			logicalrep_worker_wakeup_ptr(worker);
 		LWLockRelease(LogicalRepWorkerLock);
@@ -477,7 +477,8 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-												rstate->relid, false);
+												rstate->relid,
+												WORKERTYPE_TABLESYNC, true);
 
 			if (syncworker)
 			{
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5df5a4612b6..7aaedfb1efa 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3285,7 +3285,7 @@ FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid,
 		 */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 		leader = logicalrep_worker_find(MyLogicalRepWorker->subid,
-										InvalidOid, false);
+										InvalidOid, WORKERTYPE_APPLY, false);
 		if (!leader)
 		{
 			ereport(ERROR,
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index ae352f6e691..e97daa348d2 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -255,6 +255,7 @@ extern PGDLLIMPORT List *table_states_not_ready;
 
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
+												LogicalRepWorkerType wtype,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running,
 									 bool acquire_lock);
@@ -263,7 +264,8 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm,
 									 bool retain_dead_tuples);
-extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_worker_stop(Oid subid, Oid relid,
+								   LogicalRepWorkerType wtype);
 extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
-- 
2.43.0