v3-0002-Switch-on-worker-type.patch

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