Adding a LogicalRepWorker type field

From c277577e33927e4eb62c334d87be3f16804257a6 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Wed, 2 Aug 2023 12:18:01 +1000
Subject: [PATCH v2] Switch on worker type

---
 src/backend/replication/logical/tablesync.c | 32 ++++++++++++++--------
 src/backend/replication/logical/worker.c    | 42 +++++++++++++++++------------
 2 files changed, 46 insertions(+), 28 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e359286..98ebbc9 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -647,18 +647,28 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
-	/*
-	 * Skip for parallel apply workers because they only operate on tables
-	 * that are in a READY state. See pa_can_start() and
-	 * should_apply_changes_for_rel().
-	 */
-	if (IsParallelApplyWorker())
-		return;
+	switch (MyLogicalRepWorker->type)
+	{
+		case TYPE_TABLESYNC_WORKER:
+			process_syncing_tables_for_sync(current_lsn);
+			break;
 
-	if (IsTablesyncWorker())
-		process_syncing_tables_for_sync(current_lsn);
-	else
-		process_syncing_tables_for_apply(current_lsn);
+		case TYPE_APPLY_WORKER:
+			process_syncing_tables_for_apply(current_lsn);
+			break;
+
+		case TYPE_PARALLEL_APPLY_WORKER:
+			/*
+			 * Skip for parallel apply workers because they only operate on tables
+			 * that are in a READY state. See pa_can_start() and
+			 * should_apply_changes_for_rel().
+			 */
+			break;
+
+		case TYPE_UNKNOWN:
+			Assert(false);
+			break;
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index d165db6..dabfa48 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -486,25 +486,33 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (IsTablesyncWorker())
-		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (IsParallelApplyWorker())
+	switch (MyLogicalRepWorker->type)
 	{
-		/* We don't synchronize rel's that are in unknown state. */
-		if (rel->state != SUBREL_STATE_READY &&
-			rel->state != SUBREL_STATE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
-							MySubscription->name),
-					 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
-
-		return rel->state == SUBREL_STATE_READY;
+		case TYPE_TABLESYNC_WORKER:
+			return MyLogicalRepWorker->relid == rel->localreloid;
+
+		case TYPE_APPLY_WORKER:
+			return (rel->state == SUBREL_STATE_READY ||
+					(rel->state == SUBREL_STATE_SYNCDONE &&
+					 rel->statelsn <= remote_final_lsn));
+
+		case TYPE_PARALLEL_APPLY_WORKER:
+			/* We don't synchronize rel's that are in unknown state. */
+			if (rel->state != SUBREL_STATE_READY &&
+				rel->state != SUBREL_STATE_UNKNOWN)
+					ereport(ERROR,
+							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
+									MySubscription->name),
+							 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+			return rel->state == SUBREL_STATE_READY;
+
+		case TYPE_UNKNOWN:
+			Assert(false);
+			return false;
 	}
-	else
-		return (rel->state == SUBREL_STATE_READY ||
-				(rel->state == SUBREL_STATE_SYNCDONE &&
-				 rel->statelsn <= remote_final_lsn));
+
+	return false; /* dummy for compiler */
 }
 
 /*
-- 
1.8.3.1

From 19d80078daf5d85dda49ff43a6afb51f49cdef26 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Wed, 2 Aug 2023 10:15:13 +1000
Subject: [PATCH v2] Add LogicalRepWorkerType enum

---
 .../replication/logical/applyparallelworker.c      | 12 +++---
 src/backend/replication/logical/launcher.c         | 24 ++++++++----
 src/backend/replication/logical/tablesync.c        |  4 +-
 src/backend/replication/logical/worker.c           | 44 +++++++++++-----------
 src/include/replication/worker_internal.h          | 42 ++++++++++-----------
 src/tools/pgindent/typedefs.list                   |  1 +
 6 files changed, 68 insertions(+), 59 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 6fb9614..6d25fde 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -265,7 +265,7 @@ static bool
 pa_can_start(void)
 {
 	/* Only leader apply workers can start parallel apply workers. */
-	if (!am_leader_apply_worker())
+	if (!IsLeaderApplyWorker())
 		return false;
 
 	/*
@@ -555,7 +555,7 @@ pa_find_worker(TransactionId xid)
 static void
 pa_free_worker(ParallelApplyWorkerInfo *winfo)
 {
-	Assert(!am_parallel_apply_worker());
+	Assert(!IsParallelApplyWorker());
 	Assert(winfo->in_use);
 	Assert(pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
 
@@ -1506,7 +1506,7 @@ pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
 
 	if (fileset_state == FS_SERIALIZE_DONE)
 	{
-		Assert(am_leader_apply_worker());
+		Assert(IsLeaderApplyWorker());
 		Assert(MyLogicalRepWorker->stream_fileset);
 		wshared->fileset = *MyLogicalRepWorker->stream_fileset;
 	}
@@ -1522,7 +1522,7 @@ pa_get_fileset_state(void)
 {
 	PartialFileSetState fileset_state;
 
-	Assert(am_parallel_apply_worker());
+	Assert(IsParallelApplyWorker());
 
 	SpinLockAcquire(&MyParallelShared->mutex);
 	fileset_state = MyParallelShared->fileset_state;
@@ -1593,7 +1593,7 @@ pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
 void
 pa_decr_and_wait_stream_block(void)
 {
-	Assert(am_parallel_apply_worker());
+	Assert(IsParallelApplyWorker());
 
 	/*
 	 * It is only possible to not have any pending stream chunks when we are
@@ -1620,7 +1620,7 @@ pa_decr_and_wait_stream_block(void)
 void
 pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
 {
-	Assert(am_leader_apply_worker());
+	Assert(IsLeaderApplyWorker());
 
 	/*
 	 * Unlock the shared object lock so that parallel apply worker can
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 542af7d..a4afdc6 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -259,7 +259,7 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
 		/* Skip parallel apply workers. */
-		if (isParallelApplyWorker(w))
+		if (IsParallelApplyWorker1(w))
 			continue;
 
 		if (w->in_use && w->subid == subid && w->relid == relid &&
@@ -427,6 +427,13 @@ retry:
 	}
 
 	/* Prepare the worker slot. */
+	if (OidIsValid(relid))
+		worker->type = TYPE_TABLESYNC_WORKER;
+	else if (is_parallel_apply_worker)
+		worker->type = TYPE_PARALLEL_APPLY_WORKER;
+	else
+		worker->type = TYPE_APPLY_WORKER;
+
 	worker->launch_time = now;
 	worker->in_use = true;
 	worker->generation++;
@@ -601,7 +608,7 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 
 	if (worker)
 	{
-		Assert(!isParallelApplyWorker(worker));
+		Assert(!IsParallelApplyWorker1(worker));
 		logicalrep_worker_stop_internal(worker, SIGTERM);
 	}
 
@@ -643,7 +650,7 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 	worker = &LogicalRepCtx->workers[slot_no];
-	Assert(isParallelApplyWorker(worker));
+	Assert(IsParallelApplyWorker1(worker));
 
 	/*
 	 * Only stop the worker if the generation matches and the worker is alive.
@@ -729,7 +736,7 @@ static void
 logicalrep_worker_detach(void)
 {
 	/* Stop the parallel apply workers. */
-	if (am_leader_apply_worker())
+	if (IsLeaderApplyWorker())
 	{
 		List	   *workers;
 		ListCell   *lc;
@@ -749,7 +756,7 @@ logicalrep_worker_detach(void)
 		{
 			LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-			if (isParallelApplyWorker(w))
+			if (IsParallelApplyWorker1(w))
 				logicalrep_worker_stop_internal(w, SIGTERM);
 		}
 
@@ -772,6 +779,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
 {
 	Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));
 
+	worker->type = TYPE_UNKNOWN;
 	worker->in_use = false;
 	worker->proc = NULL;
 	worker->dbid = InvalidOid;
@@ -868,7 +876,7 @@ logicalrep_pa_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->subid == subid && isParallelApplyWorker(w))
+		if (w->subid == subid && IsParallelApplyWorker1(w))
 			res++;
 	}
 
@@ -1237,7 +1245,7 @@ GetLeaderApplyWorkerPid(pid_t pid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
+		if (IsParallelApplyWorker1(w) && w->proc && pid == w->proc->pid)
 		{
 			leader_pid = w->leader_pid;
 			break;
@@ -1290,7 +1298,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 			nulls[1] = true;
 		values[2] = Int32GetDatum(worker_pid);
 
-		if (isParallelApplyWorker(&worker))
+		if (IsParallelApplyWorker1(&worker))
 			values[3] = Int32GetDatum(worker.leader_pid);
 		else
 			nulls[3] = true;
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6d46165..e359286 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -652,10 +652,10 @@ process_syncing_tables(XLogRecPtr current_lsn)
 	 * that are in a READY state. See pa_can_start() and
 	 * should_apply_changes_for_rel().
 	 */
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 		return;
 
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		process_syncing_tables_for_sync(current_lsn);
 	else
 		process_syncing_tables_for_apply(current_lsn);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index dd353fd..d165db6 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -486,9 +486,9 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (am_parallel_apply_worker())
+	else if (IsParallelApplyWorker())
 	{
 		/* We don't synchronize rel's that are in unknown state. */
 		if (rel->state != SUBREL_STATE_READY &&
@@ -1054,7 +1054,7 @@ apply_handle_begin_prepare(StringInfo s)
 	LogicalRepPreparedTxnData begin_data;
 
 	/* Tablesync should never receive prepare. */
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("tablesync worker received a BEGIN PREPARE message")));
@@ -1293,7 +1293,7 @@ apply_handle_stream_prepare(StringInfo s)
 				 errmsg_internal("STREAM PREPARE message without STREAM STOP")));
 
 	/* Tablesync should never receive prepare. */
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("tablesync worker received a STREAM PREPARE message")));
@@ -1423,7 +1423,7 @@ apply_handle_origin(StringInfo s)
 	 */
 	if (!in_streamed_transaction &&
 		(!in_remote_transaction ||
-		 (IsTransactionState() && !am_tablesync_worker())))
+		 (IsTransactionState() && !IsTablesyncWorker())))
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("ORIGIN message sent out of order")));
@@ -2020,7 +2020,7 @@ apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
 	int			fileno;
 	off_t		offset;
 
-	if (!am_parallel_apply_worker())
+	if (!IsParallelApplyWorker())
 		maybe_start_skipping_changes(lsn);
 
 	/* Make sure we have an open transaction */
@@ -3440,7 +3440,7 @@ store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 	 * Skip for parallel apply workers, because the lsn_mapping is maintained
 	 * by the leader apply worker.
 	 */
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 		return;
 
 	/* Need to do this in permanent context */
@@ -3832,7 +3832,7 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 static void
 apply_worker_exit(void)
 {
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 	{
 		/*
 		 * Don't stop the parallel apply worker as the leader will detect the
@@ -3851,7 +3851,7 @@ apply_worker_exit(void)
 	 * subscription is still active, and so that we won't leak that hash table
 	 * entry if it isn't.
 	 */
-	if (!am_tablesync_worker())
+	if (!IsTablesyncWorker())
 		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 	proc_exit(0);
@@ -3894,7 +3894,7 @@ maybe_reread_subscription(void)
 						MySubscription->name)));
 
 		/* Ensure we remove no-longer-useful entry for worker's start time */
-		if (!am_tablesync_worker() && !am_parallel_apply_worker())
+		if (!IsTablesyncWorker() && !IsParallelApplyWorker())
 			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 		proc_exit(0);
 	}
@@ -3932,7 +3932,7 @@ maybe_reread_subscription(void)
 		newsub->owner != MySubscription->owner ||
 		!equal(newsub->publications, MySubscription->publications))
 	{
-		if (am_parallel_apply_worker())
+		if (IsParallelApplyWorker())
 			ereport(LOG,
 					(errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
 							MySubscription->name)));
@@ -4359,7 +4359,7 @@ start_table_sync(XLogRecPtr *origin_startpos, char **myslotname)
 {
 	char	   *syncslotname = NULL;
 
-	Assert(am_tablesync_worker());
+	Assert(IsTablesyncWorker());
 
 	PG_TRY();
 	{
@@ -4416,7 +4416,7 @@ start_apply(XLogRecPtr origin_startpos)
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
-			pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+			pgstat_report_subscription_error(MySubscription->oid, !IsTablesyncWorker());
 
 			PG_RE_THROW();
 		}
@@ -4465,7 +4465,7 @@ InitializeApplyWorker(void)
 						MyLogicalRepWorker->subid)));
 
 		/* Ensure we remove no-longer-useful entry for worker's start time */
-		if (!am_tablesync_worker() && !am_parallel_apply_worker())
+		if (!IsTablesyncWorker() && !IsParallelApplyWorker())
 			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 		proc_exit(0);
 	}
@@ -4491,7 +4491,7 @@ InitializeApplyWorker(void)
 								  subscription_change_cb,
 								  (Datum) 0);
 
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		ereport(LOG,
 				(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
 						MySubscription->name,
@@ -4545,7 +4545,7 @@ ApplyWorkerMain(Datum main_arg)
 	elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
 		 MySubscription->conninfo);
 
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 	{
 		start_table_sync(&origin_startpos, &myslotname);
 
@@ -4658,7 +4658,7 @@ ApplyWorkerMain(Datum main_arg)
 	options.proto.logical.twophase = false;
 	options.proto.logical.origin = pstrdup(MySubscription->origin);
 
-	if (!am_tablesync_worker())
+	if (!IsTablesyncWorker())
 	{
 		/*
 		 * Even when the two_phase mode is requested by the user, it remains
@@ -4727,7 +4727,7 @@ DisableSubscriptionAndExit(void)
 
 	/* Report the worker failed during either table synchronization or apply */
 	pgstat_report_subscription_error(MyLogicalRepWorker->subid,
-									 !am_tablesync_worker());
+									 !IsTablesyncWorker());
 
 	/* Disable the subscription */
 	StartTransactionCommand();
@@ -4735,7 +4735,7 @@ DisableSubscriptionAndExit(void)
 	CommitTransactionCommand();
 
 	/* Ensure we remove no-longer-useful entry for worker's start time */
-	if (!am_tablesync_worker() && !am_parallel_apply_worker())
+	if (!IsTablesyncWorker() && !IsParallelApplyWorker())
 		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 	/* Notify the subscription has been disabled and exit */
@@ -4761,7 +4761,7 @@ IsLogicalWorker(void)
 bool
 IsLogicalParallelApplyWorker(void)
 {
-	return IsLogicalWorker() && am_parallel_apply_worker();
+	return IsLogicalWorker() && IsParallelApplyWorker();
 }
 
 /*
@@ -4826,7 +4826,7 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 	XLogRecPtr	myskiplsn = MySubscription->skiplsn;
 	bool		started_tx = false;
 
-	if (likely(XLogRecPtrIsInvalid(myskiplsn)) || am_parallel_apply_worker())
+	if (likely(XLogRecPtrIsInvalid(myskiplsn)) || IsParallelApplyWorker())
 		return;
 
 	if (!IsTransactionState())
@@ -5060,7 +5060,7 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
 {
 	*winfo = NULL;
 
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 	{
 		return TRANS_PARALLEL_APPLY;
 	}
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 343e781..e40c255 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -27,8 +27,29 @@
 #include "storage/spin.h"
 
 
+/* Different types of worker */
+typedef enum LogicalRepWorkerType
+{
+	TYPE_UNKNOWN = 0,
+	TYPE_TABLESYNC_WORKER,
+	TYPE_APPLY_WORKER,
+	TYPE_PARALLEL_APPLY_WORKER
+} LogicalRepWorkerType;
+
+
+#define IsLeaderApplyWorker1(worker) ((worker)->type == TYPE_APPLY_WORKER)
+#define IsParallelApplyWorker1(worker) ((worker)->type == TYPE_PARALLEL_APPLY_WORKER)
+#define IsTablesyncWorker1(worker) ((worker)->type == TYPE_TABLESYNC_WORKER)
+
+#define IsLeaderApplyWorker() IsLeaderApplyWorker1(MyLogicalRepWorker)
+#define IsParallelApplyWorker() IsParallelApplyWorker1(MyLogicalRepWorker)
+#define IsTablesyncWorker() IsTablesyncWorker1(MyLogicalRepWorker)
+
 typedef struct LogicalRepWorker
 {
+	/* What type of worker is this? */
+	LogicalRepWorkerType type;
+
 	/* Time at which this worker was launched. */
 	TimestampTz launch_time;
 
@@ -305,25 +326,4 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)
-
-static inline bool
-am_tablesync_worker(void)
-{
-	return OidIsValid(MyLogicalRepWorker->relid);
-}
-
-static inline bool
-am_leader_apply_worker(void)
-{
-	return (!am_tablesync_worker() &&
-			!isParallelApplyWorker(MyLogicalRepWorker));
-}
-
-static inline bool
-am_parallel_apply_worker(void)
-{
-	return isParallelApplyWorker(MyLogicalRepWorker);
-}
-
 #endif							/* WORKER_INTERNAL_H */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index e941fb6..0b39394 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1498,6 +1498,7 @@ LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
+LogicalRepWorkerType
 LogicalRewriteMappingData
 LogicalTape
 LogicalTapeSet
-- 
1.8.3.1

From 35bd37aff2be272aa1ce3a0edd70f0e3f1373563 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Wed, 2 Aug 2023 16:34:24 +1000
Subject: [PATCH v3] Add LogicalRepWorkerType enum

---
 .../replication/logical/applyparallelworker.c      | 12 +++---
 src/backend/replication/logical/launcher.c         | 24 ++++++++----
 src/backend/replication/logical/tablesync.c        |  4 +-
 src/backend/replication/logical/worker.c           | 44 +++++++++++-----------
 src/include/replication/worker_internal.h          | 33 ++++++++--------
 src/tools/pgindent/typedefs.list                   |  1 +
 6 files changed, 64 insertions(+), 54 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 6fb9614..6d25fde 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -265,7 +265,7 @@ static bool
 pa_can_start(void)
 {
 	/* Only leader apply workers can start parallel apply workers. */
-	if (!am_leader_apply_worker())
+	if (!IsLeaderApplyWorker())
 		return false;
 
 	/*
@@ -555,7 +555,7 @@ pa_find_worker(TransactionId xid)
 static void
 pa_free_worker(ParallelApplyWorkerInfo *winfo)
 {
-	Assert(!am_parallel_apply_worker());
+	Assert(!IsParallelApplyWorker());
 	Assert(winfo->in_use);
 	Assert(pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
 
@@ -1506,7 +1506,7 @@ pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
 
 	if (fileset_state == FS_SERIALIZE_DONE)
 	{
-		Assert(am_leader_apply_worker());
+		Assert(IsLeaderApplyWorker());
 		Assert(MyLogicalRepWorker->stream_fileset);
 		wshared->fileset = *MyLogicalRepWorker->stream_fileset;
 	}
@@ -1522,7 +1522,7 @@ pa_get_fileset_state(void)
 {
 	PartialFileSetState fileset_state;
 
-	Assert(am_parallel_apply_worker());
+	Assert(IsParallelApplyWorker());
 
 	SpinLockAcquire(&MyParallelShared->mutex);
 	fileset_state = MyParallelShared->fileset_state;
@@ -1593,7 +1593,7 @@ pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
 void
 pa_decr_and_wait_stream_block(void)
 {
-	Assert(am_parallel_apply_worker());
+	Assert(IsParallelApplyWorker());
 
 	/*
 	 * It is only possible to not have any pending stream chunks when we are
@@ -1620,7 +1620,7 @@ pa_decr_and_wait_stream_block(void)
 void
 pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
 {
-	Assert(am_leader_apply_worker());
+	Assert(IsLeaderApplyWorker());
 
 	/*
 	 * Unlock the shared object lock so that parallel apply worker can
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 542af7d..9022b64 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -259,7 +259,7 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
 		/* Skip parallel apply workers. */
-		if (isParallelApplyWorker(w))
+		if (is_worker_type(w, TYPE_PARALLEL_APPLY_WORKER))
 			continue;
 
 		if (w->in_use && w->subid == subid && w->relid == relid &&
@@ -427,6 +427,13 @@ retry:
 	}
 
 	/* Prepare the worker slot. */
+	if (OidIsValid(relid))
+		worker->type = TYPE_TABLESYNC_WORKER;
+	else if (is_parallel_apply_worker)
+		worker->type = TYPE_PARALLEL_APPLY_WORKER;
+	else
+		worker->type = TYPE_APPLY_WORKER;
+
 	worker->launch_time = now;
 	worker->in_use = true;
 	worker->generation++;
@@ -601,7 +608,7 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 
 	if (worker)
 	{
-		Assert(!isParallelApplyWorker(worker));
+		Assert(!is_worker_type(worker, TYPE_PARALLEL_APPLY_WORKER));
 		logicalrep_worker_stop_internal(worker, SIGTERM);
 	}
 
@@ -643,7 +650,7 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 	worker = &LogicalRepCtx->workers[slot_no];
-	Assert(isParallelApplyWorker(worker));
+	Assert(is_worker_type(worker, TYPE_PARALLEL_APPLY_WORKER));
 
 	/*
 	 * Only stop the worker if the generation matches and the worker is alive.
@@ -729,7 +736,7 @@ static void
 logicalrep_worker_detach(void)
 {
 	/* Stop the parallel apply workers. */
-	if (am_leader_apply_worker())
+	if (IsLeaderApplyWorker())
 	{
 		List	   *workers;
 		ListCell   *lc;
@@ -749,7 +756,7 @@ logicalrep_worker_detach(void)
 		{
 			LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-			if (isParallelApplyWorker(w))
+			if (is_worker_type(w, TYPE_PARALLEL_APPLY_WORKER))
 				logicalrep_worker_stop_internal(w, SIGTERM);
 		}
 
@@ -772,6 +779,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
 {
 	Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));
 
+	worker->type = TYPE_UNKNOWN;
 	worker->in_use = false;
 	worker->proc = NULL;
 	worker->dbid = InvalidOid;
@@ -868,7 +876,7 @@ logicalrep_pa_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->subid == subid && isParallelApplyWorker(w))
+		if (w->subid == subid && is_worker_type(w, TYPE_PARALLEL_APPLY_WORKER))
 			res++;
 	}
 
@@ -1237,7 +1245,7 @@ GetLeaderApplyWorkerPid(pid_t pid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
+		if (is_worker_type(w, TYPE_PARALLEL_APPLY_WORKER) && w->proc && pid == w->proc->pid)
 		{
 			leader_pid = w->leader_pid;
 			break;
@@ -1290,7 +1298,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 			nulls[1] = true;
 		values[2] = Int32GetDatum(worker_pid);
 
-		if (isParallelApplyWorker(&worker))
+		if (is_worker_type(&worker, TYPE_PARALLEL_APPLY_WORKER))
 			values[3] = Int32GetDatum(worker.leader_pid);
 		else
 			nulls[3] = true;
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6d46165..e359286 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -652,10 +652,10 @@ process_syncing_tables(XLogRecPtr current_lsn)
 	 * that are in a READY state. See pa_can_start() and
 	 * should_apply_changes_for_rel().
 	 */
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 		return;
 
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		process_syncing_tables_for_sync(current_lsn);
 	else
 		process_syncing_tables_for_apply(current_lsn);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index dd353fd..d165db6 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -486,9 +486,9 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (am_parallel_apply_worker())
+	else if (IsParallelApplyWorker())
 	{
 		/* We don't synchronize rel's that are in unknown state. */
 		if (rel->state != SUBREL_STATE_READY &&
@@ -1054,7 +1054,7 @@ apply_handle_begin_prepare(StringInfo s)
 	LogicalRepPreparedTxnData begin_data;
 
 	/* Tablesync should never receive prepare. */
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("tablesync worker received a BEGIN PREPARE message")));
@@ -1293,7 +1293,7 @@ apply_handle_stream_prepare(StringInfo s)
 				 errmsg_internal("STREAM PREPARE message without STREAM STOP")));
 
 	/* Tablesync should never receive prepare. */
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("tablesync worker received a STREAM PREPARE message")));
@@ -1423,7 +1423,7 @@ apply_handle_origin(StringInfo s)
 	 */
 	if (!in_streamed_transaction &&
 		(!in_remote_transaction ||
-		 (IsTransactionState() && !am_tablesync_worker())))
+		 (IsTransactionState() && !IsTablesyncWorker())))
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("ORIGIN message sent out of order")));
@@ -2020,7 +2020,7 @@ apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
 	int			fileno;
 	off_t		offset;
 
-	if (!am_parallel_apply_worker())
+	if (!IsParallelApplyWorker())
 		maybe_start_skipping_changes(lsn);
 
 	/* Make sure we have an open transaction */
@@ -3440,7 +3440,7 @@ store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 	 * Skip for parallel apply workers, because the lsn_mapping is maintained
 	 * by the leader apply worker.
 	 */
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 		return;
 
 	/* Need to do this in permanent context */
@@ -3832,7 +3832,7 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 static void
 apply_worker_exit(void)
 {
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 	{
 		/*
 		 * Don't stop the parallel apply worker as the leader will detect the
@@ -3851,7 +3851,7 @@ apply_worker_exit(void)
 	 * subscription is still active, and so that we won't leak that hash table
 	 * entry if it isn't.
 	 */
-	if (!am_tablesync_worker())
+	if (!IsTablesyncWorker())
 		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 	proc_exit(0);
@@ -3894,7 +3894,7 @@ maybe_reread_subscription(void)
 						MySubscription->name)));
 
 		/* Ensure we remove no-longer-useful entry for worker's start time */
-		if (!am_tablesync_worker() && !am_parallel_apply_worker())
+		if (!IsTablesyncWorker() && !IsParallelApplyWorker())
 			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 		proc_exit(0);
 	}
@@ -3932,7 +3932,7 @@ maybe_reread_subscription(void)
 		newsub->owner != MySubscription->owner ||
 		!equal(newsub->publications, MySubscription->publications))
 	{
-		if (am_parallel_apply_worker())
+		if (IsParallelApplyWorker())
 			ereport(LOG,
 					(errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
 							MySubscription->name)));
@@ -4359,7 +4359,7 @@ start_table_sync(XLogRecPtr *origin_startpos, char **myslotname)
 {
 	char	   *syncslotname = NULL;
 
-	Assert(am_tablesync_worker());
+	Assert(IsTablesyncWorker());
 
 	PG_TRY();
 	{
@@ -4416,7 +4416,7 @@ start_apply(XLogRecPtr origin_startpos)
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
-			pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+			pgstat_report_subscription_error(MySubscription->oid, !IsTablesyncWorker());
 
 			PG_RE_THROW();
 		}
@@ -4465,7 +4465,7 @@ InitializeApplyWorker(void)
 						MyLogicalRepWorker->subid)));
 
 		/* Ensure we remove no-longer-useful entry for worker's start time */
-		if (!am_tablesync_worker() && !am_parallel_apply_worker())
+		if (!IsTablesyncWorker() && !IsParallelApplyWorker())
 			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 		proc_exit(0);
 	}
@@ -4491,7 +4491,7 @@ InitializeApplyWorker(void)
 								  subscription_change_cb,
 								  (Datum) 0);
 
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 		ereport(LOG,
 				(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
 						MySubscription->name,
@@ -4545,7 +4545,7 @@ ApplyWorkerMain(Datum main_arg)
 	elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
 		 MySubscription->conninfo);
 
-	if (am_tablesync_worker())
+	if (IsTablesyncWorker())
 	{
 		start_table_sync(&origin_startpos, &myslotname);
 
@@ -4658,7 +4658,7 @@ ApplyWorkerMain(Datum main_arg)
 	options.proto.logical.twophase = false;
 	options.proto.logical.origin = pstrdup(MySubscription->origin);
 
-	if (!am_tablesync_worker())
+	if (!IsTablesyncWorker())
 	{
 		/*
 		 * Even when the two_phase mode is requested by the user, it remains
@@ -4727,7 +4727,7 @@ DisableSubscriptionAndExit(void)
 
 	/* Report the worker failed during either table synchronization or apply */
 	pgstat_report_subscription_error(MyLogicalRepWorker->subid,
-									 !am_tablesync_worker());
+									 !IsTablesyncWorker());
 
 	/* Disable the subscription */
 	StartTransactionCommand();
@@ -4735,7 +4735,7 @@ DisableSubscriptionAndExit(void)
 	CommitTransactionCommand();
 
 	/* Ensure we remove no-longer-useful entry for worker's start time */
-	if (!am_tablesync_worker() && !am_parallel_apply_worker())
+	if (!IsTablesyncWorker() && !IsParallelApplyWorker())
 		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 	/* Notify the subscription has been disabled and exit */
@@ -4761,7 +4761,7 @@ IsLogicalWorker(void)
 bool
 IsLogicalParallelApplyWorker(void)
 {
-	return IsLogicalWorker() && am_parallel_apply_worker();
+	return IsLogicalWorker() && IsParallelApplyWorker();
 }
 
 /*
@@ -4826,7 +4826,7 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 	XLogRecPtr	myskiplsn = MySubscription->skiplsn;
 	bool		started_tx = false;
 
-	if (likely(XLogRecPtrIsInvalid(myskiplsn)) || am_parallel_apply_worker())
+	if (likely(XLogRecPtrIsInvalid(myskiplsn)) || IsParallelApplyWorker())
 		return;
 
 	if (!IsTransactionState())
@@ -5060,7 +5060,7 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
 {
 	*winfo = NULL;
 
-	if (am_parallel_apply_worker())
+	if (IsParallelApplyWorker())
 	{
 		return TRANS_PARALLEL_APPLY;
 	}
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 343e781..2c9a069 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -27,8 +27,20 @@
 #include "storage/spin.h"
 
 
+/* Different types of worker */
+typedef enum LogicalRepWorkerType
+{
+	TYPE_UNKNOWN = 0,
+	TYPE_TABLESYNC_WORKER,
+	TYPE_APPLY_WORKER,
+	TYPE_PARALLEL_APPLY_WORKER
+} LogicalRepWorkerType;
+
 typedef struct LogicalRepWorker
 {
+	/* What type of worker is this? */
+	LogicalRepWorkerType type;
+
 	/* Time at which this worker was launched. */
 	TimestampTz launch_time;
 
@@ -305,25 +317,14 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)
-
 static inline bool
-am_tablesync_worker(void)
+is_worker_type(LogicalRepWorker *worker, LogicalRepWorkerType type)
 {
-	return OidIsValid(MyLogicalRepWorker->relid);
+	return worker->type == type;
 }
 
-static inline bool
-am_leader_apply_worker(void)
-{
-	return (!am_tablesync_worker() &&
-			!isParallelApplyWorker(MyLogicalRepWorker));
-}
-
-static inline bool
-am_parallel_apply_worker(void)
-{
-	return isParallelApplyWorker(MyLogicalRepWorker);
-}
+#define IsLeaderApplyWorker() is_worker_type(MyLogicalRepWorker, TYPE_APPLY_WORKER)
+#define IsParallelApplyWorker() is_worker_type(MyLogicalRepWorker, TYPE_PARALLEL_APPLY_WORKER)
+#define IsTablesyncWorker() is_worker_type(MyLogicalRepWorker, TYPE_TABLESYNC_WORKER)
 
 #endif							/* WORKER_INTERNAL_H */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index e941fb6..0b39394 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1498,6 +1498,7 @@ LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
+LogicalRepWorkerType
 LogicalRewriteMappingData
 LogicalTape
 LogicalTapeSet
-- 
1.8.3.1

From 99c7b33014f009473db33a21e133c946d22fb511 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Wed, 2 Aug 2023 16:37:49 +1000
Subject: [PATCH v3] Switch on worker type

---
 src/backend/replication/logical/tablesync.c | 32 ++++++++++++++--------
 src/backend/replication/logical/worker.c    | 42 +++++++++++++++++------------
 2 files changed, 46 insertions(+), 28 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e359286..98ebbc9 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -647,18 +647,28 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
-	/*
-	 * Skip for parallel apply workers because they only operate on tables
-	 * that are in a READY state. See pa_can_start() and
-	 * should_apply_changes_for_rel().
-	 */
-	if (IsParallelApplyWorker())
-		return;
+	switch (MyLogicalRepWorker->type)
+	{
+		case TYPE_TABLESYNC_WORKER:
+			process_syncing_tables_for_sync(current_lsn);
+			break;
 
-	if (IsTablesyncWorker())
-		process_syncing_tables_for_sync(current_lsn);
-	else
-		process_syncing_tables_for_apply(current_lsn);
+		case TYPE_APPLY_WORKER:
+			process_syncing_tables_for_apply(current_lsn);
+			break;
+
+		case TYPE_PARALLEL_APPLY_WORKER:
+			/*
+			 * Skip for parallel apply workers because they only operate on tables
+			 * that are in a READY state. See pa_can_start() and
+			 * should_apply_changes_for_rel().
+			 */
+			break;
+
+		case TYPE_UNKNOWN:
+			Assert(false);
+			break;
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index d165db6..dabfa48 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -486,25 +486,33 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (IsTablesyncWorker())
-		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (IsParallelApplyWorker())
+	switch (MyLogicalRepWorker->type)
 	{
-		/* We don't synchronize rel's that are in unknown state. */
-		if (rel->state != SUBREL_STATE_READY &&
-			rel->state != SUBREL_STATE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
-							MySubscription->name),
-					 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
-
-		return rel->state == SUBREL_STATE_READY;
+		case TYPE_TABLESYNC_WORKER:
+			return MyLogicalRepWorker->relid == rel->localreloid;
+
+		case TYPE_APPLY_WORKER:
+			return (rel->state == SUBREL_STATE_READY ||
+					(rel->state == SUBREL_STATE_SYNCDONE &&
+					 rel->statelsn <= remote_final_lsn));
+
+		case TYPE_PARALLEL_APPLY_WORKER:
+			/* We don't synchronize rel's that are in unknown state. */
+			if (rel->state != SUBREL_STATE_READY &&
+				rel->state != SUBREL_STATE_UNKNOWN)
+					ereport(ERROR,
+							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
+									MySubscription->name),
+							 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+			return rel->state == SUBREL_STATE_READY;
+
+		case TYPE_UNKNOWN:
+			Assert(false);
+			return false;
 	}
-	else
-		return (rel->state == SUBREL_STATE_READY ||
-				(rel->state == SUBREL_STATE_SYNCDONE &&
-				 rel->statelsn <= remote_final_lsn));
+
+	return false; /* dummy for compiler */
 }
 
 /*
-- 
1.8.3.1

From a8996e809ee8d4f9a381f3589169b1a5a1077b1b Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Tue, 8 Aug 2023 17:17:26 +1000
Subject: [PATCH v4] Add LogicalRepWorkerType enum

Current HEAD, deduces a LogicalRepWorker's type from different fields ('relid'
and 'leader_pid'). But, the logical replication worker type is already known at
the time of launching the LogicalRepWorker and it never changes for the lifetime
of that process. So, instead of deducing the type, it is simpler to just store
it one time, and access it directly thereafter.

This patch adds a new enum LogicalRepWorkerType. A "type" field is added to
LogicalRepWorker. This type field is assigned when the worker is launched. Inline
functions is_leader_apply_worker(), is_parallel_apply_worker(), is_tablesync worker()
now just directly return the specified worker type.
---
 .../replication/logical/applyparallelworker.c      | 15 ++++---
 src/backend/replication/logical/launcher.c         | 51 +++++++++++++---------
 src/backend/replication/logical/tablesync.c        |  9 ++--
 src/backend/replication/logical/worker.c           | 40 ++++++++---------
 src/include/replication/worker_internal.h          | 29 +++++++-----
 src/tools/pgindent/typedefs.list                   |  1 +
 6 files changed, 84 insertions(+), 61 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 1d4e83c..779aec5 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -265,7 +265,7 @@ static bool
 pa_can_start(void)
 {
 	/* Only leader apply workers can start parallel apply workers. */
-	if (!am_leader_apply_worker())
+	if (!is_leader_apply_worker(MyLogicalRepWorker))
 		return false;
 
 	/*
@@ -435,7 +435,8 @@ pa_launch_parallel_worker(void)
 		return NULL;
 	}
 
-	launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+	launched = logicalrep_worker_launch(WORKERTYPE_PARALLEL_APPLY,
+										MyLogicalRepWorker->dbid,
 										MySubscription->oid,
 										MySubscription->name,
 										MyLogicalRepWorker->userid,
@@ -555,7 +556,7 @@ pa_find_worker(TransactionId xid)
 static void
 pa_free_worker(ParallelApplyWorkerInfo *winfo)
 {
-	Assert(!am_parallel_apply_worker());
+	Assert(!is_parallel_apply_worker(MyLogicalRepWorker));
 	Assert(winfo->in_use);
 	Assert(pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
 
@@ -1506,7 +1507,7 @@ pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
 
 	if (fileset_state == FS_SERIALIZE_DONE)
 	{
-		Assert(am_leader_apply_worker());
+		Assert(is_leader_apply_worker(MyLogicalRepWorker));
 		Assert(MyLogicalRepWorker->stream_fileset);
 		wshared->fileset = *MyLogicalRepWorker->stream_fileset;
 	}
@@ -1522,7 +1523,7 @@ pa_get_fileset_state(void)
 {
 	PartialFileSetState fileset_state;
 
-	Assert(am_parallel_apply_worker());
+	Assert(is_parallel_apply_worker(MyLogicalRepWorker));
 
 	SpinLockAcquire(&MyParallelShared->mutex);
 	fileset_state = MyParallelShared->fileset_state;
@@ -1593,7 +1594,7 @@ pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
 void
 pa_decr_and_wait_stream_block(void)
 {
-	Assert(am_parallel_apply_worker());
+	Assert(is_parallel_apply_worker(MyLogicalRepWorker));
 
 	/*
 	 * It is only possible to not have any pending stream chunks when we are
@@ -1620,7 +1621,7 @@ pa_decr_and_wait_stream_block(void)
 void
 pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
 {
-	Assert(am_leader_apply_worker());
+	Assert(is_leader_apply_worker(MyLogicalRepWorker));
 
 	/*
 	 * Unlock the shared object lock so that parallel apply worker can
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index e231fa7..7dc078a 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -259,7 +259,7 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
 		/* Skip parallel apply workers. */
-		if (isParallelApplyWorker(w))
+		if (is_parallel_apply_worker(w))
 			continue;
 
 		if (w->in_use && w->subid == subid && w->relid == relid &&
@@ -303,7 +303,8 @@ logicalrep_workers_find(Oid subid, bool only_running)
  * Returns true on success, false on failure.
  */
 bool
-logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
+logicalrep_worker_launch(LogicalRepWorkerType wtype,
+						 Oid dbid, Oid subid, const char *subname, Oid userid,
 						 Oid relid, dsm_handle subworker_dsm)
 {
 	BackgroundWorker bgw;
@@ -315,10 +316,18 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	int			nsyncworkers;
 	int			nparallelapplyworkers;
 	TimestampTz now;
-	bool		is_parallel_apply_worker = (subworker_dsm != DSM_HANDLE_INVALID);
+	bool		is_tablesync_wkr = (wtype == WORKERTYPE_TABLESYNC);
+	bool		is_parallel_apply_wkr = (wtype == WORKERTYPE_PARALLEL_APPLY);
 
-	/* Sanity check - tablesync worker cannot be a subworker */
-	Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
+	/*
+	 * Sanity checks:
+	 * - must be valid worker type
+	 * - tablesync workers are only ones to have relid
+	 * - parallel apply worker is the only kind of subworker
+	 */
+	Assert(wtype != WORKERTYPE_UNKNOWN);
+	Assert(is_tablesync_wkr == OidIsValid(relid));
+	Assert(is_parallel_apply_wkr == (subworker_dsm != DSM_HANDLE_INVALID));
 
 	ereport(DEBUG1,
 			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -393,7 +402,7 @@ retry:
 	 * sync worker limit per subscription. So, just return silently as we
 	 * might get here because of an otherwise harmless race condition.
 	 */
-	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
+	if (is_tablesync_wkr && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
@@ -405,7 +414,7 @@ retry:
 	 * Return false if the number of parallel apply workers reached the limit
 	 * per subscription.
 	 */
-	if (is_parallel_apply_worker &&
+	if (is_parallel_apply_wkr &&
 		nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
@@ -427,6 +436,7 @@ retry:
 	}
 
 	/* Prepare the worker slot. */
+	worker->type = wtype;
 	worker->launch_time = now;
 	worker->in_use = true;
 	worker->generation++;
@@ -438,8 +448,8 @@ retry:
 	worker->relstate = SUBREL_STATE_UNKNOWN;
 	worker->relstate_lsn = InvalidXLogRecPtr;
 	worker->stream_fileset = NULL;
-	worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
-	worker->parallel_apply = is_parallel_apply_worker;
+	worker->leader_pid = is_parallel_apply_wkr ? MyProcPid : InvalidPid;
+	worker->parallel_apply = is_parallel_apply_wkr;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -458,7 +468,7 @@ retry:
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
 
-	if (is_parallel_apply_worker)
+	if (is_parallel_apply_wkr)
 	{
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
@@ -466,7 +476,7 @@ retry:
 				 subid);
 		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
 	}
-	else if (OidIsValid(relid))
+	else if (is_tablesync_wkr)
 	{
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
@@ -488,7 +498,7 @@ retry:
 	bgw.bgw_notify_pid = MyProcPid;
 	bgw.bgw_main_arg = Int32GetDatum(slot);
 
-	if (is_parallel_apply_worker)
+	if (is_parallel_apply_wkr)
 		memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
 
 	if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
@@ -607,7 +617,7 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 
 	if (worker)
 	{
-		Assert(!isParallelApplyWorker(worker));
+		Assert(!is_parallel_apply_worker(worker));
 		logicalrep_worker_stop_internal(worker, SIGTERM);
 	}
 
@@ -649,7 +659,7 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 	worker = &LogicalRepCtx->workers[slot_no];
-	Assert(isParallelApplyWorker(worker));
+	Assert(is_parallel_apply_worker(worker));
 
 	/*
 	 * Only stop the worker if the generation matches and the worker is alive.
@@ -735,7 +745,7 @@ static void
 logicalrep_worker_detach(void)
 {
 	/* Stop the parallel apply workers. */
-	if (am_leader_apply_worker())
+	if (is_leader_apply_worker(MyLogicalRepWorker))
 	{
 		List	   *workers;
 		ListCell   *lc;
@@ -755,7 +765,7 @@ logicalrep_worker_detach(void)
 		{
 			LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-			if (isParallelApplyWorker(w))
+			if (is_parallel_apply_worker(w))
 				logicalrep_worker_stop_internal(w, SIGTERM);
 		}
 
@@ -874,7 +884,7 @@ logicalrep_pa_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->subid == subid && isParallelApplyWorker(w))
+		if (w->subid == subid && is_parallel_apply_worker(w))
 			res++;
 	}
 
@@ -1180,7 +1190,8 @@ ApplyLauncherMain(Datum main_arg)
 				(elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
 			{
 				ApplyLauncherSetWorkerStartTime(sub->oid, now);
-				logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
+				logicalrep_worker_launch(WORKERTYPE_APPLY,
+										 sub->dbid, sub->oid, sub->name,
 										 sub->owner, InvalidOid,
 										 DSM_HANDLE_INVALID);
 			}
@@ -1243,7 +1254,7 @@ GetLeaderApplyWorkerPid(pid_t pid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
+		if (is_parallel_apply_worker(w) && w->proc && pid == w->proc->pid)
 		{
 			leader_pid = w->leader_pid;
 			break;
@@ -1296,7 +1307,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 			nulls[1] = true;
 		values[2] = Int32GetDatum(worker_pid);
 
-		if (isParallelApplyWorker(&worker))
+		if (is_parallel_apply_worker(&worker))
 			values[3] = Int32GetDatum(worker.leader_pid);
 		else
 			nulls[3] = true;
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 651a775..25343be 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -587,7 +587,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 						TimestampDifferenceExceeds(hentry->last_start_time, now,
 												   wal_retrieve_retry_interval))
 					{
-						logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+						logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
+												 MyLogicalRepWorker->dbid,
 												 MySubscription->oid,
 												 MySubscription->name,
 												 MyLogicalRepWorker->userid,
@@ -653,10 +654,10 @@ process_syncing_tables(XLogRecPtr current_lsn)
 	 * that are in a READY state. See pa_can_start() and
 	 * should_apply_changes_for_rel().
 	 */
-	if (am_parallel_apply_worker())
+	if (is_parallel_apply_worker(MyLogicalRepWorker))
 		return;
 
-	if (am_tablesync_worker())
+	if (is_tablesync_worker(MyLogicalRepWorker))
 		process_syncing_tables_for_sync(current_lsn);
 	else
 		process_syncing_tables_for_apply(current_lsn);
@@ -1598,7 +1599,7 @@ start_table_sync(XLogRecPtr *origin_startpos, char **slotname)
 {
 	char	   *sync_slotname = NULL;
 
-	Assert(am_tablesync_worker());
+	Assert(is_tablesync_worker(MyLogicalRepWorker));
 
 	PG_TRY();
 	{
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a20d4c1..82b09ff 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -485,9 +485,9 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (am_tablesync_worker())
+	if (is_tablesync_worker(MyLogicalRepWorker))
 		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (am_parallel_apply_worker())
+	else if (is_parallel_apply_worker(MyLogicalRepWorker))
 	{
 		/* We don't synchronize rel's that are in unknown state. */
 		if (rel->state != SUBREL_STATE_READY &&
@@ -1053,7 +1053,7 @@ apply_handle_begin_prepare(StringInfo s)
 	LogicalRepPreparedTxnData begin_data;
 
 	/* Tablesync should never receive prepare. */
-	if (am_tablesync_worker())
+	if (is_tablesync_worker(MyLogicalRepWorker))
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("tablesync worker received a BEGIN PREPARE message")));
@@ -1292,7 +1292,7 @@ apply_handle_stream_prepare(StringInfo s)
 				 errmsg_internal("STREAM PREPARE message without STREAM STOP")));
 
 	/* Tablesync should never receive prepare. */
-	if (am_tablesync_worker())
+	if (is_tablesync_worker(MyLogicalRepWorker))
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("tablesync worker received a STREAM PREPARE message")));
@@ -1422,7 +1422,7 @@ apply_handle_origin(StringInfo s)
 	 */
 	if (!in_streamed_transaction &&
 		(!in_remote_transaction ||
-		 (IsTransactionState() && !am_tablesync_worker())))
+		 (IsTransactionState() && !is_tablesync_worker(MyLogicalRepWorker))))
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("ORIGIN message sent out of order")));
@@ -2019,7 +2019,7 @@ apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
 	int			fileno;
 	off_t		offset;
 
-	if (!am_parallel_apply_worker())
+	if (!is_parallel_apply_worker(MyLogicalRepWorker))
 		maybe_start_skipping_changes(lsn);
 
 	/* Make sure we have an open transaction */
@@ -3452,7 +3452,7 @@ store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 	 * Skip for parallel apply workers, because the lsn_mapping is maintained
 	 * by the leader apply worker.
 	 */
-	if (am_parallel_apply_worker())
+	if (is_parallel_apply_worker(MyLogicalRepWorker))
 		return;
 
 	/* Need to do this in permanent context */
@@ -3844,7 +3844,7 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 static void
 apply_worker_exit(void)
 {
-	if (am_parallel_apply_worker())
+	if (is_parallel_apply_worker(MyLogicalRepWorker))
 	{
 		/*
 		 * Don't stop the parallel apply worker as the leader will detect the
@@ -3863,7 +3863,7 @@ apply_worker_exit(void)
 	 * subscription is still active, and so that we won't leak that hash table
 	 * entry if it isn't.
 	 */
-	if (am_leader_apply_worker())
+	if (is_leader_apply_worker(MyLogicalRepWorker))
 		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 	proc_exit(0);
@@ -3906,7 +3906,7 @@ maybe_reread_subscription(void)
 						MySubscription->name)));
 
 		/* Ensure we remove no-longer-useful entry for worker's start time */
-		if (am_leader_apply_worker())
+		if (is_leader_apply_worker(MyLogicalRepWorker))
 			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 		proc_exit(0);
@@ -3945,7 +3945,7 @@ maybe_reread_subscription(void)
 		newsub->owner != MySubscription->owner ||
 		!equal(newsub->publications, MySubscription->publications))
 	{
-		if (am_parallel_apply_worker())
+		if (is_parallel_apply_worker(MyLogicalRepWorker))
 			ereport(LOG,
 					(errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
 							MySubscription->name)));
@@ -4436,7 +4436,7 @@ start_apply(XLogRecPtr origin_startpos)
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
-			pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+			pgstat_report_subscription_error(MySubscription->oid, !is_tablesync_worker(MyLogicalRepWorker));
 
 			PG_RE_THROW();
 		}
@@ -4590,7 +4590,7 @@ InitializeLogRepWorker(void)
 						MyLogicalRepWorker->subid)));
 
 		/* Ensure we remove no-longer-useful entry for worker's start time */
-		if (am_leader_apply_worker())
+		if (is_leader_apply_worker(MyLogicalRepWorker))
 			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 		proc_exit(0);
@@ -4617,7 +4617,7 @@ InitializeLogRepWorker(void)
 								  subscription_change_cb,
 								  (Datum) 0);
 
-	if (am_tablesync_worker())
+	if (is_tablesync_worker(MyLogicalRepWorker))
 		ereport(LOG,
 				(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
 						MySubscription->name,
@@ -4637,7 +4637,7 @@ SetupApplyOrSyncWorker(int worker_slot)
 	/* Attach to slot */
 	logicalrep_worker_attach(worker_slot);
 
-	Assert(am_tablesync_worker() || am_leader_apply_worker());
+	Assert(is_tablesync_worker(MyLogicalRepWorker) || is_leader_apply_worker(MyLogicalRepWorker));
 
 	/* Setup signal handling */
 	pqsignal(SIGHUP, SignalHandlerForConfigReload);
@@ -4709,7 +4709,7 @@ DisableSubscriptionAndExit(void)
 
 	/* Report the worker failed during either table synchronization or apply */
 	pgstat_report_subscription_error(MyLogicalRepWorker->subid,
-									 !am_tablesync_worker());
+									 !is_tablesync_worker(MyLogicalRepWorker));
 
 	/* Disable the subscription */
 	StartTransactionCommand();
@@ -4717,7 +4717,7 @@ DisableSubscriptionAndExit(void)
 	CommitTransactionCommand();
 
 	/* Ensure we remove no-longer-useful entry for worker's start time */
-	if (am_leader_apply_worker())
+	if (is_leader_apply_worker(MyLogicalRepWorker))
 		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
 	/* Notify the subscription has been disabled and exit */
@@ -4743,7 +4743,7 @@ IsLogicalWorker(void)
 bool
 IsLogicalParallelApplyWorker(void)
 {
-	return IsLogicalWorker() && am_parallel_apply_worker();
+	return IsLogicalWorker() && is_parallel_apply_worker(MyLogicalRepWorker);
 }
 
 /*
@@ -4808,7 +4808,7 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 	XLogRecPtr	myskiplsn = MySubscription->skiplsn;
 	bool		started_tx = false;
 
-	if (likely(XLogRecPtrIsInvalid(myskiplsn)) || am_parallel_apply_worker())
+	if (likely(XLogRecPtrIsInvalid(myskiplsn)) || is_parallel_apply_worker(MyLogicalRepWorker))
 		return;
 
 	if (!IsTransactionState())
@@ -5042,7 +5042,7 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
 {
 	*winfo = NULL;
 
-	if (am_parallel_apply_worker())
+	if (is_parallel_apply_worker(MyLogicalRepWorker))
 	{
 		return TRANS_PARALLEL_APPLY;
 	}
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 672a711..0ae0949 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -27,9 +27,20 @@
 #include "storage/shm_toc.h"
 #include "storage/spin.h"
 
+/* Different types of worker */
+typedef enum LogicalRepWorkerType
+{
+	WORKERTYPE_UNKNOWN = 0,
+	WORKERTYPE_TABLESYNC,
+	WORKERTYPE_APPLY,
+	WORKERTYPE_PARALLEL_APPLY
+} LogicalRepWorkerType;
 
 typedef struct LogicalRepWorker
 {
+	/* What type of worker is this? */
+	LogicalRepWorkerType type;
+
 	/* Time at which this worker was launched. */
 	TimestampTz launch_time;
 
@@ -232,7 +243,8 @@ extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
+									 Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
@@ -315,25 +327,22 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)
-
 static inline bool
-am_tablesync_worker(void)
+is_tablesync_worker(LogicalRepWorker *w)
 {
-	return OidIsValid(MyLogicalRepWorker->relid);
+	return w->type == WORKERTYPE_TABLESYNC;
 }
 
 static inline bool
-am_leader_apply_worker(void)
+is_leader_apply_worker(LogicalRepWorker *w)
 {
-	return (!am_tablesync_worker() &&
-			!isParallelApplyWorker(MyLogicalRepWorker));
+	return w->type == WORKERTYPE_APPLY;
 }
 
 static inline bool
-am_parallel_apply_worker(void)
+is_parallel_apply_worker(LogicalRepWorker *w)
 {
-	return isParallelApplyWorker(MyLogicalRepWorker);
+	return w->type == WORKERTYPE_PARALLEL_APPLY;
 }
 
 #endif							/* WORKER_INTERNAL_H */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 66823bc..52a8789 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1500,6 +1500,7 @@ LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
+LogicalRepWorkerType
 LogicalRewriteMappingData
 LogicalTape
 LogicalTapeSet
-- 
1.8.3.1

From 1900bb2c1305b59713d66fa6194515cfdbf16c93 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Tue, 8 Aug 2023 17:53:42 +1000
Subject: [PATCH v4] Switch on worker type.

Now that the LogicalRepWorker has an enum type we can code to switch on that type,
which should be more efficient than cascading if/else.
---
 src/backend/replication/logical/launcher.c  | 51 ++++++++++++++++-------------
 src/backend/replication/logical/tablesync.c | 31 +++++++++++-------
 src/backend/replication/logical/worker.c    | 42 ++++++++++++++----------
 3 files changed, 73 insertions(+), 51 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 7dc078a..45a7739 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -468,30 +468,35 @@ retry:
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
 
-	if (is_parallel_apply_wkr)
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication parallel apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
-	}
-	else if (is_tablesync_wkr)
+	switch (wtype)
 	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication tablesync worker for subscription %u sync %u",
-				 subid,
-				 relid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
-	}
-	else
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+		case WORKERTYPE_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+			break;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication parallel apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
+			break;
+
+		case WORKERTYPE_TABLESYNC:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication tablesync worker for subscription %u sync %u",
+					 subid,
+					 relid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("logicalrep_worker_launch: unknown worker type"));
 	}
 
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 25343be..0b12461 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -649,18 +649,27 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
-	/*
-	 * Skip for parallel apply workers because they only operate on tables
-	 * that are in a READY state. See pa_can_start() and
-	 * should_apply_changes_for_rel().
-	 */
-	if (is_parallel_apply_worker(MyLogicalRepWorker))
-		return;
+	switch (MyLogicalRepWorker->type)
+	{
+		case WORKERTYPE_TABLESYNC:
+			process_syncing_tables_for_sync(current_lsn);
+			break;
 
-	if (is_tablesync_worker(MyLogicalRepWorker))
-		process_syncing_tables_for_sync(current_lsn);
-	else
-		process_syncing_tables_for_apply(current_lsn);
+		case WORKERTYPE_APPLY:
+			process_syncing_tables_for_apply(current_lsn);
+			break;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			/*
+			 * Skip for parallel apply workers because they only operate on tables
+			 * that are in a READY state. See pa_can_start() and
+			 * should_apply_changes_for_rel().
+			 */
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("process_syncing_tables: Unknown worker type"));
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 82b09ff..93188f5 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -485,25 +485,33 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (is_tablesync_worker(MyLogicalRepWorker))
-		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (is_parallel_apply_worker(MyLogicalRepWorker))
+	switch (MyLogicalRepWorker->type)
 	{
-		/* We don't synchronize rel's that are in unknown state. */
-		if (rel->state != SUBREL_STATE_READY &&
-			rel->state != SUBREL_STATE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
-							MySubscription->name),
-					 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
-
-		return rel->state == SUBREL_STATE_READY;
+		case WORKERTYPE_TABLESYNC:
+			return MyLogicalRepWorker->relid == rel->localreloid;
+
+		case WORKERTYPE_APPLY:
+			return (rel->state == SUBREL_STATE_READY ||
+					(rel->state == SUBREL_STATE_SYNCDONE &&
+					 rel->statelsn <= remote_final_lsn));
+
+		case WORKERTYPE_PARALLEL_APPLY:
+				/* We don't synchronize rel's that are in unknown state. */
+				if (rel->state != SUBREL_STATE_READY &&
+					rel->state != SUBREL_STATE_UNKNOWN)
+						ereport(ERROR,
+								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+								 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
+										MySubscription->name),
+								 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+
+				return rel->state == SUBREL_STATE_READY;
+
+		case WORKERTYPE_UNKNOWN:
+				ereport(ERROR, errmsg_internal("should_apply_changes_for_rel: Unknown worker type"));
 	}
-	else
-		return (rel->state == SUBREL_STATE_READY ||
-				(rel->state == SUBREL_STATE_SYNCDONE &&
-				 rel->statelsn <= remote_final_lsn));
+
+	return false; /* dummy for compiler */
 }
 
 /*
-- 
1.8.3.1

From 9e35a9d521023ed4d54f5738b3c99407a5fe3e02 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 10 Aug 2023 12:11:27 +1000
Subject: [PATCH v5] Switch on worker type.

---
 src/backend/replication/logical/launcher.c  | 51 ++++++++++++++++-------------
 src/backend/replication/logical/tablesync.c | 31 +++++++++++-------
 src/backend/replication/logical/worker.c    | 42 ++++++++++++++----------
 3 files changed, 73 insertions(+), 51 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 2a1602c..3ad315c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -468,30 +468,35 @@ retry:
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
 
-	if (is_parallel_apply_worker)
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication parallel apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
-	}
-	else if (is_tablesync_worker)
+	switch (worker->type)
 	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication tablesync worker for subscription %u sync %u",
-				 subid,
-				 relid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
-	}
-	else
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+		case WORKERTYPE_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication parallel apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
+			break;
+
+		case WORKERTYPE_TABLESYNC:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication tablesync worker for subscription %u sync %u",
+					 subid,
+					 relid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
+			break;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("logicalrep_worker_launch: unknown worker type"));
 	}
 
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 67bdd14..89d9cec 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -649,18 +649,27 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
-	/*
-	 * Skip for parallel apply workers because they only operate on tables
-	 * that are in a READY state. See pa_can_start() and
-	 * should_apply_changes_for_rel().
-	 */
-	if (am_parallel_apply_worker())
-		return;
+	switch (MyLogicalRepWorker->type)
+	{
+		case WORKERTYPE_PARALLEL_APPLY:
+			/*
+			 * Skip for parallel apply workers because they only operate on tables
+			 * that are in a READY state. See pa_can_start() and
+			 * should_apply_changes_for_rel().
+			 */
+			break;
 
-	if (am_tablesync_worker())
-		process_syncing_tables_for_sync(current_lsn);
-	else
-		process_syncing_tables_for_apply(current_lsn);
+		case WORKERTYPE_TABLESYNC:
+			process_syncing_tables_for_sync(current_lsn);
+			break;
+
+		case WORKERTYPE_APPLY:
+			process_syncing_tables_for_apply(current_lsn);
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("process_syncing_tables: Unknown worker type"));
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a20d4c1..f596366 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -485,25 +485,33 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (am_tablesync_worker())
-		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (am_parallel_apply_worker())
+	switch (MyLogicalRepWorker->type)
 	{
-		/* We don't synchronize rel's that are in unknown state. */
-		if (rel->state != SUBREL_STATE_READY &&
-			rel->state != SUBREL_STATE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
-							MySubscription->name),
-					 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+		case WORKERTYPE_TABLESYNC:
+			return MyLogicalRepWorker->relid == rel->localreloid;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			/* We don't synchronize rel's that are in unknown state. */
+			if (rel->state != SUBREL_STATE_READY &&
+				rel->state != SUBREL_STATE_UNKNOWN)
+				ereport(ERROR,
+						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
+								MySubscription->name),
+						 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+
+			return rel->state == SUBREL_STATE_READY;
+
+		case WORKERTYPE_APPLY:
+			return (rel->state == SUBREL_STATE_READY ||
+					(rel->state == SUBREL_STATE_SYNCDONE &&
+					 rel->statelsn <= remote_final_lsn));
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("should_apply_changes_for_rel: Unknown worker type"));
+		}
 
-		return rel->state == SUBREL_STATE_READY;
-	}
-	else
-		return (rel->state == SUBREL_STATE_READY ||
-				(rel->state == SUBREL_STATE_SYNCDONE &&
-				 rel->statelsn <= remote_final_lsn));
+	return false; /* dummy for compiler */
 }
 
 /*
-- 
1.8.3.1

From da405938e1f9040f241478fa03a9f9db4ebc615a Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 10 Aug 2023 11:39:12 +1000
Subject: [PATCH v5] Add LogicalRepWorkerType enum.

Current HEAD code deduces a LogicalRepWorker's type from the values of several
different fields ('relid' and 'leader_pid') whenever logic needs to know it.

In fact, the logical replication worker type is already known at the time of
launching the LogicalRepWorker and it never changes for the lifetime of that
process. Instead of deducing the type, it is simpler to just store it one time,
and access it directly thereafter.

This patch adds a new enum LogicalRepWorkerType. A 'type' field is added to
LogicalRepWorker. This field is assigned when the worker is launched.
---
 .../replication/logical/applyparallelworker.c      |  3 ++-
 src/backend/replication/logical/launcher.c         | 25 ++++++++++++++++------
 src/backend/replication/logical/tablesync.c        |  3 ++-
 src/include/replication/worker_internal.h          | 23 +++++++++++++++-----
 src/tools/pgindent/typedefs.list                   |  1 +
 5 files changed, 41 insertions(+), 14 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 1d4e83c..4e8ee29 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -435,7 +435,8 @@ pa_launch_parallel_worker(void)
 		return NULL;
 	}
 
-	launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+	launched = logicalrep_worker_launch(WORKERTYPE_PARALLEL_APPLY,
+										MyLogicalRepWorker->dbid,
 										MySubscription->oid,
 										MySubscription->name,
 										MyLogicalRepWorker->userid,
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index e231fa7..2a1602c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -303,7 +303,8 @@ logicalrep_workers_find(Oid subid, bool only_running)
  * Returns true on success, false on failure.
  */
 bool
-logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
+logicalrep_worker_launch(LogicalRepWorkerType wtype,
+						 Oid dbid, Oid subid, const char *subname, Oid userid,
 						 Oid relid, dsm_handle subworker_dsm)
 {
 	BackgroundWorker bgw;
@@ -315,10 +316,18 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	int			nsyncworkers;
 	int			nparallelapplyworkers;
 	TimestampTz now;
-	bool		is_parallel_apply_worker = (subworker_dsm != DSM_HANDLE_INVALID);
+	bool		is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
+	bool		is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
 
-	/* Sanity check - tablesync worker cannot be a subworker */
-	Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
+	/*
+	 * Sanity checks:
+	 * - must be valid worker type
+	 * - tablesync workers are only ones to have relid
+	 * - parallel apply worker is the only kind of subworker
+	 */
+	Assert(wtype != WORKERTYPE_UNKNOWN);
+	Assert(is_tablesync_worker == OidIsValid(relid));
+	Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
 
 	ereport(DEBUG1,
 			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -393,7 +402,7 @@ retry:
 	 * sync worker limit per subscription. So, just return silently as we
 	 * might get here because of an otherwise harmless race condition.
 	 */
-	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
+	if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
@@ -427,6 +436,7 @@ retry:
 	}
 
 	/* Prepare the worker slot. */
+	worker->type = wtype;
 	worker->launch_time = now;
 	worker->in_use = true;
 	worker->generation++;
@@ -466,7 +476,7 @@ retry:
 				 subid);
 		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
 	}
-	else if (OidIsValid(relid))
+	else if (is_tablesync_worker)
 	{
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
@@ -1180,7 +1190,8 @@ ApplyLauncherMain(Datum main_arg)
 				(elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
 			{
 				ApplyLauncherSetWorkerStartTime(sub->oid, now);
-				logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
+				logicalrep_worker_launch(WORKERTYPE_APPLY,
+										 sub->dbid, sub->oid, sub->name,
 										 sub->owner, InvalidOid,
 										 DSM_HANDLE_INVALID);
 			}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 651a775..67bdd14 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -587,7 +587,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 						TimestampDifferenceExceeds(hentry->last_start_time, now,
 												   wal_retrieve_retry_interval))
 					{
-						logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+						logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
+												 MyLogicalRepWorker->dbid,
 												 MySubscription->oid,
 												 MySubscription->name,
 												 MyLogicalRepWorker->userid,
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 672a711..ce44776 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -27,9 +27,20 @@
 #include "storage/shm_toc.h"
 #include "storage/spin.h"
 
+/* Different types of worker */
+typedef enum LogicalRepWorkerType
+{
+	WORKERTYPE_UNKNOWN = 0,
+	WORKERTYPE_TABLESYNC,
+	WORKERTYPE_APPLY,
+	WORKERTYPE_PARALLEL_APPLY
+} LogicalRepWorkerType;
 
 typedef struct LogicalRepWorker
 {
+	/* What type of worker is this? */
+	LogicalRepWorkerType type;
+
 	/* Time at which this worker was launched. */
 	TimestampTz launch_time;
 
@@ -232,7 +243,8 @@ extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
+									 Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
@@ -315,19 +327,20 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)
+#define isTablesyncWorker(worker) ((worker)->type == WORKERTYPE_TABLESYNC)
+#define isLeaderApplyWorker(worker) ((worker)->type == WORKERTYPE_APPLY)
+#define isParallelApplyWorker(worker) ((worker)->type == WORKERTYPE_PARALLEL_APPLY)
 
 static inline bool
 am_tablesync_worker(void)
 {
-	return OidIsValid(MyLogicalRepWorker->relid);
+	return isTablesyncWorker(MyLogicalRepWorker);
 }
 
 static inline bool
 am_leader_apply_worker(void)
 {
-	return (!am_tablesync_worker() &&
-			!isParallelApplyWorker(MyLogicalRepWorker));
+	return isLeaderApplyWorker(MyLogicalRepWorker);
 }
 
 static inline bool
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 66823bc..52a8789 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1500,6 +1500,7 @@ LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
+LogicalRepWorkerType
 LogicalRewriteMappingData
 LogicalTape
 LogicalTapeSet
-- 
1.8.3.1

From 969055559fdfc93dddc51a9010f9741cd63b47c2 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 10 Aug 2023 11:39:12 +1000
Subject: [PATCH v7] Simplify determining logical replication worker types.

We deduce a LogicalRepWorker's type from the values of several different
fields ('relid' and 'leader_pid') whenever logic needs to know it.

In fact, the logical replication worker type is already known at the time
of launching the LogicalRepWorker and it never changes for the lifetime of
that process. Instead of deducing the type, it is simpler to just store it
one time, and access it directly thereafter.

Author: Peter Smith
Reviewed-by: Amit Kapila, Bharath Rupireddy
Discussion: http://postgr.es/m/CAHut+PttPSuP0yoZ=9zLDXKqTJ=d0bhxwKaEaNcaym1XqcvDEg@mail.gmail.com
---
 .../replication/logical/applyparallelworker.c |  3 ++-
 src/backend/replication/logical/launcher.c    | 27 +++++++++++++------
 src/backend/replication/logical/tablesync.c   |  3 ++-
 src/include/replication/worker_internal.h     | 21 +++++++++++----
 src/tools/pgindent/typedefs.list              |  1 +
 5 files changed, 40 insertions(+), 15 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 1d4e83c4c1..4e8ee2973e 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -435,7 +435,8 @@ pa_launch_parallel_worker(void)
 		return NULL;
 	}
 
-	launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+	launched = logicalrep_worker_launch(WORKERTYPE_PARALLEL_APPLY,
+										MyLogicalRepWorker->dbid,
 										MySubscription->oid,
 										MySubscription->name,
 										MyLogicalRepWorker->userid,
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index e231fa7f95..7454c8ca27 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -303,7 +303,8 @@ logicalrep_workers_find(Oid subid, bool only_running)
  * Returns true on success, false on failure.
  */
 bool
-logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
+logicalrep_worker_launch(LogicalRepWorkerType wtype,
+						 Oid dbid, Oid subid, const char *subname, Oid userid,
 						 Oid relid, dsm_handle subworker_dsm)
 {
 	BackgroundWorker bgw;
@@ -315,10 +316,18 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	int			nsyncworkers;
 	int			nparallelapplyworkers;
 	TimestampTz now;
-	bool		is_parallel_apply_worker = (subworker_dsm != DSM_HANDLE_INVALID);
-
-	/* Sanity check - tablesync worker cannot be a subworker */
-	Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
+	bool		is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
+	bool		is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
+
+	/*----------
+	 * Sanity checks:
+	 * - must be valid worker type
+	 * - tablesync workers are only ones to have relid
+	 * - parallel apply worker is the only kind of subworker
+	 */
+	Assert(wtype != WORKERTYPE_UNKNOWN);
+	Assert(is_tablesync_worker == OidIsValid(relid));
+	Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
 
 	ereport(DEBUG1,
 			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -393,7 +402,7 @@ retry:
 	 * sync worker limit per subscription. So, just return silently as we
 	 * might get here because of an otherwise harmless race condition.
 	 */
-	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
+	if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
@@ -427,6 +436,7 @@ retry:
 	}
 
 	/* Prepare the worker slot. */
+	worker->type = wtype;
 	worker->launch_time = now;
 	worker->in_use = true;
 	worker->generation++;
@@ -466,7 +476,7 @@ retry:
 				 subid);
 		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
 	}
-	else if (OidIsValid(relid))
+	else if (is_tablesync_worker)
 	{
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
@@ -1180,7 +1190,8 @@ ApplyLauncherMain(Datum main_arg)
 				(elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
 			{
 				ApplyLauncherSetWorkerStartTime(sub->oid, now);
-				logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
+				logicalrep_worker_launch(WORKERTYPE_APPLY,
+										 sub->dbid, sub->oid, sub->name,
 										 sub->owner, InvalidOid,
 										 DSM_HANDLE_INVALID);
 			}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 651a775065..67bdd14095 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -587,7 +587,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 						TimestampDifferenceExceeds(hentry->last_start_time, now,
 												   wal_retrieve_retry_interval))
 					{
-						logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+						logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
+												 MyLogicalRepWorker->dbid,
 												 MySubscription->oid,
 												 MySubscription->name,
 												 MyLogicalRepWorker->userid,
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 672a7117c0..d5c9de484e 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -27,9 +27,20 @@
 #include "storage/shm_toc.h"
 #include "storage/spin.h"
 
+/* Different types of worker */
+typedef enum LogicalRepWorkerType
+{
+	WORKERTYPE_UNKNOWN = 0,
+	WORKERTYPE_TABLESYNC,
+	WORKERTYPE_APPLY,
+	WORKERTYPE_PARALLEL_APPLY
+} LogicalRepWorkerType;
 
 typedef struct LogicalRepWorker
 {
+	/* What type of worker is this? */
+	LogicalRepWorkerType type;
+
 	/* Time at which this worker was launched. */
 	TimestampTz launch_time;
 
@@ -232,7 +243,8 @@ extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
+									 Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
@@ -315,19 +327,18 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)
+#define isParallelApplyWorker(worker) ((worker)->type == WORKERTYPE_PARALLEL_APPLY)
 
 static inline bool
 am_tablesync_worker(void)
 {
-	return OidIsValid(MyLogicalRepWorker->relid);
+	return (MyLogicalRepWorker->type == WORKERTYPE_TABLESYNC);
 }
 
 static inline bool
 am_leader_apply_worker(void)
 {
-	return (!am_tablesync_worker() &&
-			!isParallelApplyWorker(MyLogicalRepWorker));
+	return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
 }
 
 static inline bool
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 66823bc2a7..52a8789cc4 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1500,6 +1500,7 @@ LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
+LogicalRepWorkerType
 LogicalRewriteMappingData
 LogicalTape
 LogicalTapeSet
-- 
2.39.1

From 372da0f445a2bc33d9c7ee5f53a56ab1d771858e Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Fri, 11 Aug 2023 19:59:21 +1000
Subject: [PATCH v7] Add isTablesyncWorker

---
 src/backend/replication/logical/launcher.c | 4 ++--
 src/include/replication/worker_internal.h  | 3 ++-
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 7454c8c..7cc0a16 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -857,7 +857,7 @@ logicalrep_sync_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->subid == subid && OidIsValid(w->relid))
+		if (w->subid == subid && isTablesyncWorker(w))
 			res++;
 	}
 
@@ -1301,7 +1301,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 		worker_pid = worker.proc->pid;
 
 		values[0] = ObjectIdGetDatum(worker.subid);
-		if (OidIsValid(worker.relid))
+		if (isTablesyncWorker(&worker))
 			values[1] = ObjectIdGetDatum(worker.relid);
 		else
 			nulls[1] = true;
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index d5c9de4..a428663 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -328,11 +328,12 @@ extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
 #define isParallelApplyWorker(worker) ((worker)->type == WORKERTYPE_PARALLEL_APPLY)
+#define isTablesyncWorker(worker) ((worker)->type == WORKERTYPE_TABLESYNC)
 
 static inline bool
 am_tablesync_worker(void)
 {
-	return (MyLogicalRepWorker->type == WORKERTYPE_TABLESYNC);
+	return isTablesyncWorker(MyLogicalRepWorker);
 }
 
 static inline bool
-- 
1.8.3.1

From e6cabe717637cf4095cfae85806e199a122935d3 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Fri, 11 Aug 2023 19:41:39 +1000
Subject: [PATCH v7] Simplify determining logical replication worker types.

We deduce a LogicalRepWorker's type from the values of several different
fields ('relid' and 'leader_pid') whenever logic needs to know it.

In fact, the logical replication worker type is already known at the time
of launching the LogicalRepWorker and it never changes for the lifetime of
that process. Instead of deducing the type, it is simpler to just store it
one time, and access it directly thereafter.

Author: Peter Smith
Reviewed-by: Amit Kapila, Bharath Rupireddy
Discussion: http://postgr.es/m/CAHut+PttPSuP0yoZ=9zLDXKqTJ=d0bhxwKaEaNcaym1XqcvDEg@mail.gmail.com
---
 .../replication/logical/applyparallelworker.c      |  3 ++-
 src/backend/replication/logical/launcher.c         | 27 +++++++++++++++-------
 src/backend/replication/logical/tablesync.c        |  3 ++-
 src/include/replication/worker_internal.h          | 21 +++++++++++++----
 src/tools/pgindent/typedefs.list                   |  1 +
 5 files changed, 40 insertions(+), 15 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 1d4e83c..4e8ee29 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -435,7 +435,8 @@ pa_launch_parallel_worker(void)
 		return NULL;
 	}
 
-	launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+	launched = logicalrep_worker_launch(WORKERTYPE_PARALLEL_APPLY,
+										MyLogicalRepWorker->dbid,
 										MySubscription->oid,
 										MySubscription->name,
 										MyLogicalRepWorker->userid,
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index e231fa7..7454c8c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -303,7 +303,8 @@ logicalrep_workers_find(Oid subid, bool only_running)
  * Returns true on success, false on failure.
  */
 bool
-logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
+logicalrep_worker_launch(LogicalRepWorkerType wtype,
+						 Oid dbid, Oid subid, const char *subname, Oid userid,
 						 Oid relid, dsm_handle subworker_dsm)
 {
 	BackgroundWorker bgw;
@@ -315,10 +316,18 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	int			nsyncworkers;
 	int			nparallelapplyworkers;
 	TimestampTz now;
-	bool		is_parallel_apply_worker = (subworker_dsm != DSM_HANDLE_INVALID);
-
-	/* Sanity check - tablesync worker cannot be a subworker */
-	Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
+	bool		is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
+	bool		is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
+
+	/*----------
+	 * Sanity checks:
+	 * - must be valid worker type
+	 * - tablesync workers are only ones to have relid
+	 * - parallel apply worker is the only kind of subworker
+	 */
+	Assert(wtype != WORKERTYPE_UNKNOWN);
+	Assert(is_tablesync_worker == OidIsValid(relid));
+	Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
 
 	ereport(DEBUG1,
 			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -393,7 +402,7 @@ retry:
 	 * sync worker limit per subscription. So, just return silently as we
 	 * might get here because of an otherwise harmless race condition.
 	 */
-	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
+	if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
@@ -427,6 +436,7 @@ retry:
 	}
 
 	/* Prepare the worker slot. */
+	worker->type = wtype;
 	worker->launch_time = now;
 	worker->in_use = true;
 	worker->generation++;
@@ -466,7 +476,7 @@ retry:
 				 subid);
 		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
 	}
-	else if (OidIsValid(relid))
+	else if (is_tablesync_worker)
 	{
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
@@ -1180,7 +1190,8 @@ ApplyLauncherMain(Datum main_arg)
 				(elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
 			{
 				ApplyLauncherSetWorkerStartTime(sub->oid, now);
-				logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
+				logicalrep_worker_launch(WORKERTYPE_APPLY,
+										 sub->dbid, sub->oid, sub->name,
 										 sub->owner, InvalidOid,
 										 DSM_HANDLE_INVALID);
 			}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 651a775..67bdd14 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -587,7 +587,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 						TimestampDifferenceExceeds(hentry->last_start_time, now,
 												   wal_retrieve_retry_interval))
 					{
-						logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+						logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
+												 MyLogicalRepWorker->dbid,
 												 MySubscription->oid,
 												 MySubscription->name,
 												 MyLogicalRepWorker->userid,
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 672a711..d5c9de4 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -27,9 +27,20 @@
 #include "storage/shm_toc.h"
 #include "storage/spin.h"
 
+/* Different types of worker */
+typedef enum LogicalRepWorkerType
+{
+	WORKERTYPE_UNKNOWN = 0,
+	WORKERTYPE_TABLESYNC,
+	WORKERTYPE_APPLY,
+	WORKERTYPE_PARALLEL_APPLY
+} LogicalRepWorkerType;
 
 typedef struct LogicalRepWorker
 {
+	/* What type of worker is this? */
+	LogicalRepWorkerType type;
+
 	/* Time at which this worker was launched. */
 	TimestampTz launch_time;
 
@@ -232,7 +243,8 @@ extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
+									 Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
@@ -315,19 +327,18 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)
+#define isParallelApplyWorker(worker) ((worker)->type == WORKERTYPE_PARALLEL_APPLY)
 
 static inline bool
 am_tablesync_worker(void)
 {
-	return OidIsValid(MyLogicalRepWorker->relid);
+	return (MyLogicalRepWorker->type == WORKERTYPE_TABLESYNC);
 }
 
 static inline bool
 am_leader_apply_worker(void)
 {
-	return (!am_tablesync_worker() &&
-			!isParallelApplyWorker(MyLogicalRepWorker));
+	return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
 }
 
 static inline bool
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 66823bc..52a8789 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1500,6 +1500,7 @@ LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
+LogicalRepWorkerType
 LogicalRewriteMappingData
 LogicalTape
 LogicalTapeSet
-- 
1.8.3.1

From b0e09eec056e6ecaf9d7bea7b22af278a62ead61 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Mon, 14 Aug 2023 16:21:45 +1000
Subject: [PATCH v8] Switch on worker type.

---
 src/backend/replication/logical/launcher.c  | 56 +++++++++++++++--------------
 src/backend/replication/logical/tablesync.c | 31 ++++++++++------
 src/backend/replication/logical/worker.c    | 42 +++++++++++++---------
 3 files changed, 75 insertions(+), 54 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 7cc0a16..2119bf6 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -468,39 +468,43 @@ retry:
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
 
-	if (is_parallel_apply_worker)
+	switch (worker->type)
 	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication parallel apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
-	}
-	else if (is_tablesync_worker)
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication tablesync worker for subscription %u sync %u",
-				 subid,
-				 relid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
-	}
-	else
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+		case WORKERTYPE_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+			break;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication parallel apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
+
+			memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
+			break;
+
+		case WORKERTYPE_TABLESYNC:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication tablesync worker for subscription %u sync %u",
+					 subid,
+					 relid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("logicalrep_worker_launch: unknown worker type"));
 	}
 
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
 	bgw.bgw_notify_pid = MyProcPid;
 	bgw.bgw_main_arg = Int32GetDatum(slot);
 
-	if (is_parallel_apply_worker)
-		memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
-
 	if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
 	{
 		/* Failed to start worker, so clean up the worker slot. */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 67bdd14..89d9cec 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -649,18 +649,27 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
-	/*
-	 * Skip for parallel apply workers because they only operate on tables
-	 * that are in a READY state. See pa_can_start() and
-	 * should_apply_changes_for_rel().
-	 */
-	if (am_parallel_apply_worker())
-		return;
+	switch (MyLogicalRepWorker->type)
+	{
+		case WORKERTYPE_PARALLEL_APPLY:
+			/*
+			 * Skip for parallel apply workers because they only operate on tables
+			 * that are in a READY state. See pa_can_start() and
+			 * should_apply_changes_for_rel().
+			 */
+			break;
 
-	if (am_tablesync_worker())
-		process_syncing_tables_for_sync(current_lsn);
-	else
-		process_syncing_tables_for_apply(current_lsn);
+		case WORKERTYPE_TABLESYNC:
+			process_syncing_tables_for_sync(current_lsn);
+			break;
+
+		case WORKERTYPE_APPLY:
+			process_syncing_tables_for_apply(current_lsn);
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("process_syncing_tables: Unknown worker type"));
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a20d4c1..f596366 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -485,25 +485,33 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (am_tablesync_worker())
-		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (am_parallel_apply_worker())
+	switch (MyLogicalRepWorker->type)
 	{
-		/* We don't synchronize rel's that are in unknown state. */
-		if (rel->state != SUBREL_STATE_READY &&
-			rel->state != SUBREL_STATE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
-							MySubscription->name),
-					 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+		case WORKERTYPE_TABLESYNC:
+			return MyLogicalRepWorker->relid == rel->localreloid;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			/* We don't synchronize rel's that are in unknown state. */
+			if (rel->state != SUBREL_STATE_READY &&
+				rel->state != SUBREL_STATE_UNKNOWN)
+				ereport(ERROR,
+						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
+								MySubscription->name),
+						 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+
+			return rel->state == SUBREL_STATE_READY;
+
+		case WORKERTYPE_APPLY:
+			return (rel->state == SUBREL_STATE_READY ||
+					(rel->state == SUBREL_STATE_SYNCDONE &&
+					 rel->statelsn <= remote_final_lsn));
+
+		case WORKERTYPE_UNKNOWN:
+			ereport(ERROR, errmsg_internal("should_apply_changes_for_rel: Unknown worker type"));
+		}
 
-		return rel->state == SUBREL_STATE_READY;
-	}
-	else
-		return (rel->state == SUBREL_STATE_READY ||
-				(rel->state == SUBREL_STATE_SYNCDONE &&
-				 rel->statelsn <= remote_final_lsn));
+	return false; /* dummy for compiler */
 }
 
 /*
-- 
1.8.3.1

From 7f0470e576bf4f5fa157f959432a1b38f7c66ae2 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Mon, 21 Aug 2023 09:58:32 +1000
Subject: [PATCH v9] Switch on worker type.

---
 src/backend/replication/logical/launcher.c  | 57 ++++++++++++++++-------------
 src/backend/replication/logical/tablesync.c | 32 ++++++++++------
 src/backend/replication/logical/worker.c    | 43 +++++++++++++---------
 3 files changed, 78 insertions(+), 54 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 7cc0a16..72e44d5 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -468,39 +468,44 @@ retry:
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
 
-	if (is_parallel_apply_worker)
+	switch (worker->type)
 	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication parallel apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
-	}
-	else if (is_tablesync_worker)
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication tablesync worker for subscription %u sync %u",
-				 subid,
-				 relid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
-	}
-	else
-	{
-		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
-		snprintf(bgw.bgw_name, BGW_MAXLEN,
-				 "logical replication apply worker for subscription %u",
-				 subid);
-		snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+		case WORKERTYPE_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
+			break;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication parallel apply worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
+
+			memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
+			break;
+
+		case WORKERTYPE_TABLESYNC:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication tablesync worker for subscription %u sync %u",
+					 subid,
+					 relid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			/* Should never happen. */
+			elog(ERROR, "unknown worker type");
 	}
 
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
 	bgw.bgw_notify_pid = MyProcPid;
 	bgw.bgw_main_arg = Int32GetDatum(slot);
 
-	if (is_parallel_apply_worker)
-		memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
-
 	if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
 	{
 		/* Failed to start worker, so clean up the worker slot. */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 67bdd14..54cf957 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -649,18 +649,28 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
-	/*
-	 * Skip for parallel apply workers because they only operate on tables
-	 * that are in a READY state. See pa_can_start() and
-	 * should_apply_changes_for_rel().
-	 */
-	if (am_parallel_apply_worker())
-		return;
+	switch (MyLogicalRepWorker->type)
+	{
+		case WORKERTYPE_PARALLEL_APPLY:
+			/*
+			 * Skip for parallel apply workers because they only operate on tables
+			 * that are in a READY state. See pa_can_start() and
+			 * should_apply_changes_for_rel().
+			 */
+			break;
 
-	if (am_tablesync_worker())
-		process_syncing_tables_for_sync(current_lsn);
-	else
-		process_syncing_tables_for_apply(current_lsn);
+		case WORKERTYPE_TABLESYNC:
+			process_syncing_tables_for_sync(current_lsn);
+			break;
+
+		case WORKERTYPE_APPLY:
+			process_syncing_tables_for_apply(current_lsn);
+			break;
+
+		case WORKERTYPE_UNKNOWN:
+			/* Should never happen. */
+			elog(ERROR, "Unknown worker type");
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a20d4c1..d6b594d 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -485,25 +485,34 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
-	if (am_tablesync_worker())
-		return MyLogicalRepWorker->relid == rel->localreloid;
-	else if (am_parallel_apply_worker())
+	switch (MyLogicalRepWorker->type)
 	{
-		/* We don't synchronize rel's that are in unknown state. */
-		if (rel->state != SUBREL_STATE_READY &&
-			rel->state != SUBREL_STATE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
-							MySubscription->name),
-					 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+		case WORKERTYPE_TABLESYNC:
+			return MyLogicalRepWorker->relid == rel->localreloid;
+
+		case WORKERTYPE_PARALLEL_APPLY:
+			/* We don't synchronize rel's that are in unknown state. */
+			if (rel->state != SUBREL_STATE_READY &&
+				rel->state != SUBREL_STATE_UNKNOWN)
+				ereport(ERROR,
+						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
+								MySubscription->name),
+						 errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
+
+			return rel->state == SUBREL_STATE_READY;
+
+		case WORKERTYPE_APPLY:
+			return (rel->state == SUBREL_STATE_READY ||
+					(rel->state == SUBREL_STATE_SYNCDONE &&
+					 rel->statelsn <= remote_final_lsn));
+
+		case WORKERTYPE_UNKNOWN:
+			/* Should never happen. */
+			elog(ERROR, "Unknown worker type");
+		}
 
-		return rel->state == SUBREL_STATE_READY;
-	}
-	else
-		return (rel->state == SUBREL_STATE_READY ||
-				(rel->state == SUBREL_STATE_SYNCDONE &&
-				 rel->statelsn <= remote_final_lsn));
+	return false; /* dummy for compiler */
 }
 
 /*
-- 
1.8.3.1