v20240819-0003-Reorganize-tablesync-Code-and-Introduce-sy.patch

From 61c07d967a7b04be73d2bad1379899695d4b4415 Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Mon, 12 Aug 2024 14:43:11 +0530
Subject: [PATCH v20240819 3/5] Reorganize tablesync Code and Introduce
 syncutils

Reorganized the tablesync code by creating a new syncutils file.
This refactoring will facilitate the development of sequence
synchronization worker code.

This commit separates code reorganization from functional changes,
making it clearer to reviewers that only existing code has been moved.
The changes in this patch can be merged with subsequent patches during
the commit process.
---
 src/backend/catalog/pg_subscription.c         |   4 +-
 src/backend/replication/logical/Makefile      |   1 +
 .../replication/logical/applyparallelworker.c |   4 +-
 src/backend/replication/logical/meson.build   |   1 +
 src/backend/replication/logical/syncutils.c   | 589 ++++++++++++++++++
 src/backend/replication/logical/tablesync.c   | 570 +----------------
 src/backend/replication/logical/worker.c      |  18 +-
 src/include/catalog/pg_subscription_rel.h     |   2 +-
 src/include/replication/worker_internal.h     |  10 +-
 9 files changed, 614 insertions(+), 585 deletions(-)
 create mode 100644 src/backend/replication/logical/syncutils.c

diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..077903f059 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -457,13 +457,13 @@ RemoveSubscriptionRel(Oid subid, Oid relid)
 }
 
 /*
- * Does the subscription have any relations?
+ * Does the subscription have any tables?
  *
  * Use this function only to know true/false, and when you have no need for the
  * List returned by GetSubscriptionRelations.
  */
 bool
-HasSubscriptionRelations(Oid subid)
+HasSubscriptionTables(Oid subid)
 {
 	Relation	rel;
 	ScanKeyData skey[1];
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index ba03eeff1c..3964a30109 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -27,6 +27,7 @@ OBJS = \
 	reorderbuffer.o \
 	slotsync.o \
 	snapbuild.o \
+	syncutils.o \
 	tablesync.o \
 	worker.o
 
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index e7f7d4c5e4..983ead5929 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -309,7 +309,7 @@ pa_can_start(void)
 	 * should_apply_changes_for_rel) as we won't know remote_final_lsn by that
 	 * time. So, we don't start the new parallel apply worker in this case.
 	 */
-	if (!AllTablesyncsReady())
+	if (!SyncAllTablesReady())
 		return false;
 
 	return true;
@@ -962,7 +962,7 @@ ParallelApplyWorkerMain(Datum main_arg)
 	 * the subscription relation state.
 	 */
 	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
-								  invalidate_syncing_table_states,
+								  SyncInvalidateRelationStates,
 								  (Datum) 0);
 
 	set_apply_error_context_origin(originname);
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index 3dec36a6de..27a0e30ab7 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -13,6 +13,7 @@ backend_sources += files(
   'reorderbuffer.c',
   'slotsync.c',
   'snapbuild.c',
+  'syncutils.c',
   'tablesync.c',
   'worker.c',
 )
diff --git a/src/backend/replication/logical/syncutils.c b/src/backend/replication/logical/syncutils.c
new file mode 100644
index 0000000000..378b8eba04
--- /dev/null
+++ b/src/backend/replication/logical/syncutils.c
@@ -0,0 +1,589 @@
+/*-------------------------------------------------------------------------
+ * syncutils.c
+ *	  PostgreSQL logical replication: common synchronization code
+ *
+ * Copyright (c) 2024, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/syncutils.c
+ *
+ * NOTES
+ *	  This file contains code common to table synchronization workers, and
+ *	  the sequence synchronization worker.
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "catalog/pg_subscription_rel.h"
+#include "pgstat.h"
+#include "replication/logicallauncher.h"
+#include "replication/origin.h"
+#include "replication/slot.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+
+typedef enum
+{
+	SYNC_RELATION_STATE_NEEDS_REBUILD,
+	SYNC_RELATION_STATE_REBUILD_STARTED,
+	SYNC_RELATION_STATE_VALID,
+} SyncingRelationsState;
+
+static SyncingRelationsState relation_states_validity = SYNC_RELATION_STATE_NEEDS_REBUILD;
+static List *table_states_not_ready = NIL;
+static bool FetchRelationStates(bool *started_tx);
+
+/*
+ * Exit routine for synchronization worker.
+ */
+void
+pg_attribute_noreturn()
+SyncFinishWorker(void)
+{
+	/*
+	 * Commit any outstanding transaction. This is the usual case, unless
+	 * there was nothing to do for the table.
+	 */
+	if (IsTransactionState())
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(true);
+	}
+
+	/* And flush all writes. */
+	XLogFlush(GetXLogWriteRecPtr());
+
+	StartTransactionCommand();
+	ereport(LOG,
+			(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
+					MySubscription->name,
+					get_rel_name(MyLogicalRepWorker->relid))));
+	CommitTransactionCommand();
+
+	/* Find the leader apply worker and signal it. */
+	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
+
+	/* Stop gracefully */
+	proc_exit(0);
+}
+
+/*
+ * Callback from syscache invalidation.
+ */
+void
+SyncInvalidateRelationStates(Datum arg, int cacheid, uint32 hashvalue)
+{
+	relation_states_validity = SYNC_RELATION_STATE_NEEDS_REBUILD;
+}
+
+/*
+ * Handle table synchronization cooperation from the synchronization
+ * worker.
+ *
+ * If the sync worker is in CATCHUP state and reached (or passed) the
+ * predetermined synchronization point in the WAL stream, mark the table as
+ * SYNCDONE and finish.
+ */
+static void
+process_syncing_tables_for_sync(XLogRecPtr current_lsn)
+{
+	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+
+	if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
+		current_lsn >= MyLogicalRepWorker->relstate_lsn)
+	{
+		TimeLineID	tli;
+		char		syncslotname[NAMEDATALEN] = {0};
+		char		originname[NAMEDATALEN] = {0};
+
+		MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
+		MyLogicalRepWorker->relstate_lsn = current_lsn;
+
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+		/*
+		 * UpdateSubscriptionRelState must be called within a transaction.
+		 */
+		if (!IsTransactionState())
+			StartTransactionCommand();
+
+		UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+								   MyLogicalRepWorker->relid,
+								   MyLogicalRepWorker->relstate,
+								   MyLogicalRepWorker->relstate_lsn);
+
+		/*
+		 * End streaming so that LogRepWorkerWalRcvConn can be used to drop
+		 * the slot.
+		 */
+		walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
+
+		/*
+		 * Cleanup the tablesync slot.
+		 *
+		 * This has to be done after updating the state because otherwise if
+		 * there is an error while doing the database operations we won't be
+		 * able to rollback dropped slot.
+		 */
+		ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
+										MyLogicalRepWorker->relid,
+										syncslotname,
+										sizeof(syncslotname));
+
+		/*
+		 * It is important to give an error if we are unable to drop the slot,
+		 * otherwise, it won't be dropped till the corresponding subscription
+		 * is dropped. So passing missing_ok = false.
+		 */
+		ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
+
+		CommitTransactionCommand();
+		pgstat_report_stat(false);
+
+		/*
+		 * Start a new transaction to clean up the tablesync origin tracking.
+		 * This transaction will be ended within the SyncFinishWorker(). Now,
+		 * even, if we fail to remove this here, the apply worker will ensure
+		 * to clean it up afterward.
+		 *
+		 * We need to do this after the table state is set to SYNCDONE.
+		 * Otherwise, if an error occurs while performing the database
+		 * operation, the worker will be restarted and the in-memory state of
+		 * replication progress (remote_lsn) won't be rolled-back which would
+		 * have been cleared before restart. So, the restarted worker will use
+		 * invalid replication progress state resulting in replay of
+		 * transactions that have already been applied.
+		 */
+		StartTransactionCommand();
+
+		ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
+										   MyLogicalRepWorker->relid,
+										   originname,
+										   sizeof(originname));
+
+		/*
+		 * Resetting the origin session removes the ownership of the slot.
+		 * This is needed to allow the origin to be dropped.
+		 */
+		replorigin_session_reset();
+		replorigin_session_origin = InvalidRepOriginId;
+		replorigin_session_origin_lsn = InvalidXLogRecPtr;
+		replorigin_session_origin_timestamp = 0;
+
+		/*
+		 * Drop the tablesync's origin tracking if exists.
+		 *
+		 * There is a chance that the user is concurrently performing refresh
+		 * for the subscription where we remove the table state and its origin
+		 * or the apply worker would have removed this origin. So passing
+		 * missing_ok = true.
+		 */
+		replorigin_drop_by_name(originname, true, false);
+
+		SyncFinishWorker();
+	}
+	else
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+}
+
+/*
+ * Handle table synchronization cooperation from the apply worker.
+ *
+ * Walk over all subscription tables that are individually tracked by the
+ * apply process (currently, all that have state other than
+ * SUBREL_STATE_READY) and manage synchronization for them.
+ *
+ * If there are tables that need synchronizing and are not being synchronized
+ * yet, start sync workers for them (if there are free slots for sync
+ * workers).  To prevent starting the sync worker for the same relation at a
+ * high frequency after a failure, we store its last start time with each sync
+ * state info.  We start the sync worker for the same relation after waiting
+ * at least wal_retrieve_retry_interval.
+ *
+ * For tables that are being synchronized already, check if sync workers
+ * either need action from the apply worker or have finished.  This is the
+ * SYNCWAIT to CATCHUP transition.
+ *
+ * If the synchronization position is reached (SYNCDONE), then the table can
+ * be marked as READY and is no longer tracked.
+ */
+static void
+process_syncing_tables_for_apply(XLogRecPtr current_lsn)
+{
+	struct tablesync_start_time_mapping
+	{
+		Oid			relid;
+		TimestampTz last_start_time;
+	};
+	static HTAB *last_start_times = NULL;
+	ListCell   *lc;
+	bool		started_tx = false;
+	bool		should_exit = false;
+
+	Assert(!IsTransactionState());
+
+	/* We need up-to-date sync state info for subscription tables here. */
+	FetchRelationStates(&started_tx);
+
+	/*
+	 * Prepare a hash table for tracking last start times of workers, to avoid
+	 * immediate restarts.  We don't need it if there are no tables that need
+	 * syncing.
+	 */
+	if (table_states_not_ready != NIL && !last_start_times)
+	{
+		HASHCTL		ctl;
+
+		ctl.keysize = sizeof(Oid);
+		ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
+		last_start_times = hash_create("Logical replication table sync worker start times",
+									   256, &ctl, HASH_ELEM | HASH_BLOBS);
+	}
+
+	/*
+	 * Clean up the hash table when we're done with all tables (just to
+	 * release the bit of memory).
+	 */
+	else if (table_states_not_ready == NIL && last_start_times)
+	{
+		hash_destroy(last_start_times);
+		last_start_times = NULL;
+	}
+
+	/*
+	 * Process all tables that are being synchronized.
+	 */
+	foreach(lc, table_states_not_ready)
+	{
+		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
+
+		if (rstate->state == SUBREL_STATE_SYNCDONE)
+		{
+			/*
+			 * Apply has caught up to the position where the table sync has
+			 * finished.  Mark the table as ready so that the apply will just
+			 * continue to replicate it normally.
+			 */
+			if (current_lsn >= rstate->lsn)
+			{
+				char		originname[NAMEDATALEN];
+
+				rstate->state = SUBREL_STATE_READY;
+				rstate->lsn = current_lsn;
+				if (!started_tx)
+				{
+					StartTransactionCommand();
+					started_tx = true;
+				}
+
+				/*
+				 * Remove the tablesync origin tracking if exists.
+				 *
+				 * There is a chance that the user is concurrently performing
+				 * refresh for the subscription where we remove the table
+				 * state and its origin or the tablesync worker would have
+				 * already removed this origin. We can't rely on tablesync
+				 * worker to remove the origin tracking as if there is any
+				 * error while dropping we won't restart it to drop the
+				 * origin. So passing missing_ok = true.
+				 */
+				ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
+												   rstate->relid,
+												   originname,
+												   sizeof(originname));
+				replorigin_drop_by_name(originname, true, false);
+
+				/*
+				 * Update the state to READY only after the origin cleanup.
+				 */
+				UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+										   rstate->relid, rstate->state,
+										   rstate->lsn);
+			}
+		}
+		else
+		{
+			LogicalRepWorker *syncworker;
+
+			/*
+			 * Look for a sync worker for this relation.
+			 */
+			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+												rstate->relid, false);
+
+			if (syncworker)
+			{
+				/* Found one, update our copy of its state */
+				SpinLockAcquire(&syncworker->relmutex);
+				rstate->state = syncworker->relstate;
+				rstate->lsn = syncworker->relstate_lsn;
+				if (rstate->state == SUBREL_STATE_SYNCWAIT)
+				{
+					/*
+					 * Sync worker is waiting for apply.  Tell sync worker it
+					 * can catchup now.
+					 */
+					syncworker->relstate = SUBREL_STATE_CATCHUP;
+					syncworker->relstate_lsn =
+						Max(syncworker->relstate_lsn, current_lsn);
+				}
+				SpinLockRelease(&syncworker->relmutex);
+
+				/* If we told worker to catch up, wait for it. */
+				if (rstate->state == SUBREL_STATE_SYNCWAIT)
+				{
+					/* Signal the sync worker, as it may be waiting for us. */
+					if (syncworker->proc)
+						logicalrep_worker_wakeup_ptr(syncworker);
+
+					/* Now safe to release the LWLock */
+					LWLockRelease(LogicalRepWorkerLock);
+
+					if (started_tx)
+					{
+						/*
+						 * We must commit the existing transaction to release
+						 * the existing locks before entering a busy loop.
+						 * This is required to avoid any undetected deadlocks
+						 * due to any existing lock as deadlock detector won't
+						 * be able to detect the waits on the latch.
+						 */
+						CommitTransactionCommand();
+						pgstat_report_stat(false);
+					}
+
+					/*
+					 * Enter busy loop and wait for synchronization worker to
+					 * reach expected state (or die trying).
+					 */
+					StartTransactionCommand();
+					started_tx = true;
+
+					wait_for_relation_state_change(rstate->relid,
+												   SUBREL_STATE_SYNCDONE);
+				}
+				else
+					LWLockRelease(LogicalRepWorkerLock);
+			}
+			else
+			{
+				/*
+				 * If there is no sync worker for this table yet, count
+				 * running sync workers for this subscription, while we have
+				 * the lock.
+				 */
+				int			nsyncworkers =
+					logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
+
+				/* Now safe to release the LWLock */
+				LWLockRelease(LogicalRepWorkerLock);
+
+				/*
+				 * If there are free sync worker slot(s), start a new sync
+				 * worker for the table.
+				 */
+				if (nsyncworkers < max_sync_workers_per_subscription)
+				{
+					TimestampTz now = GetCurrentTimestamp();
+					struct tablesync_start_time_mapping *hentry;
+					bool		found;
+
+					hentry = hash_search(last_start_times, &rstate->relid,
+										 HASH_ENTER, &found);
+
+					if (!found ||
+						TimestampDifferenceExceeds(hentry->last_start_time, now,
+												   wal_retrieve_retry_interval))
+					{
+						logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
+												 MyLogicalRepWorker->dbid,
+												 MySubscription->oid,
+												 MySubscription->name,
+												 MyLogicalRepWorker->userid,
+												 rstate->relid,
+												 DSM_HANDLE_INVALID);
+						hentry->last_start_time = now;
+					}
+				}
+			}
+		}
+	}
+
+	if (started_tx)
+	{
+		/*
+		 * Even when the two_phase mode is requested by the user, it remains
+		 * as 'pending' until all tablesyncs have reached READY state.
+		 *
+		 * When this happens, we restart the apply worker and (if the
+		 * conditions are still ok) then the two_phase tri-state will become
+		 * 'enabled' at that time.
+		 *
+		 * Note: If the subscription has no tables then leave the state as
+		 * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
+		 * work.
+		 */
+		if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
+		{
+			CommandCounterIncrement();	/* make updates visible */
+			if (SyncAllTablesReady())
+			{
+				ereport(LOG,
+						(errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
+								MySubscription->name)));
+				should_exit = true;
+			}
+		}
+
+		CommitTransactionCommand();
+		pgstat_report_stat(true);
+	}
+
+	if (should_exit)
+	{
+		/*
+		 * Reset the last-start time for this worker so that the launcher will
+		 * restart it without waiting for wal_retrieve_retry_interval.
+		 */
+		ApplyLauncherForgetWorkerStartTime(MySubscription->oid);
+
+		proc_exit(0);
+	}
+}
+
+/*
+ * Process possible state change(s) of tables that are being synchronized.
+ */
+void
+SyncProcessRelations(XLogRecPtr current_lsn)
+{
+	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;
+
+		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");
+	}
+}
+
+/*
+ * Common code to fetch the up-to-date sync state info into the static lists.
+ *
+ * Returns true if subscription has 1 or more tables, else false.
+ *
+ * Note: If this function started the transaction (indicated by the parameter)
+ * then it is the caller's responsibility to commit it.
+ */
+static bool
+FetchRelationStates(bool *started_tx)
+{
+	static bool has_subtables = false;
+
+	*started_tx = false;
+
+	if (relation_states_validity != SYNC_RELATION_STATE_VALID)
+	{
+		MemoryContext oldctx;
+		List	   *rstates;
+		ListCell   *lc;
+		SubscriptionRelState *rstate;
+
+		relation_states_validity = SYNC_RELATION_STATE_REBUILD_STARTED;
+
+		/* Clean the old lists. */
+		list_free_deep(table_states_not_ready);
+		table_states_not_ready = NIL;
+
+		if (!IsTransactionState())
+		{
+			StartTransactionCommand();
+			*started_tx = true;
+		}
+
+		/* Fetch all non-ready tables. */
+		rstates = GetSubscriptionRelations(MySubscription->oid, true);
+
+		/* Allocate the tracking info in a permanent memory context. */
+		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+		foreach(lc, rstates)
+		{
+			rstate = palloc(sizeof(SubscriptionRelState));
+			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
+			table_states_not_ready = lappend(table_states_not_ready, rstate);
+		}
+		MemoryContextSwitchTo(oldctx);
+
+		/*
+		 * Does the subscription have tables?
+		 *
+		 * If there were not-READY tables found then we know it does. But
+		 * if table_states_not_ready was empty we still need to check again to
+		 * see if there are 0 tables.
+		 */
+		has_subtables = (table_states_not_ready != NIL) ||
+			HasSubscriptionTables(MySubscription->oid);
+
+		/*
+		 * If the subscription relation cache has been invalidated since we
+		 * entered this routine, we still use and return the relations we just
+		 * finished constructing, to avoid infinite loops, but we leave the
+		 * table states marked as stale so that we'll rebuild it again on next
+		 * access. Otherwise, we mark the table states as valid.
+		 */
+		if (relation_states_validity == SYNC_RELATION_STATE_REBUILD_STARTED)
+			relation_states_validity = SYNC_RELATION_STATE_VALID;
+	}
+
+	return has_subtables;
+}
+
+/*
+ * If the subscription has no tables then return false.
+ *
+ * Otherwise, are all tablesyncs READY?
+ *
+ * Note: This function is not suitable to be called from outside of apply or
+ * tablesync workers because MySubscription needs to be already initialized.
+ */
+bool
+SyncAllTablesReady(void)
+{
+	bool		started_tx = false;
+	bool		has_subrels = false;
+
+	/* We need up-to-date sync state info for subscription tables here. */
+	has_subrels = FetchRelationStates(&started_tx);
+
+	if (started_tx)
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(true);
+	}
+
+	/*
+	 * Return false when there are no tables in subscription or not all tables
+	 * are in ready state; true otherwise.
+	 */
+	return has_subrels && (table_states_not_ready == NIL);
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e03e761392..cea5e09a8c 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -117,59 +117,13 @@
 #include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
-#include "utils/memutils.h"
 #include "utils/rls.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 #include "utils/usercontext.h"
 
-typedef enum
-{
-	SYNC_TABLE_STATE_NEEDS_REBUILD,
-	SYNC_TABLE_STATE_REBUILD_STARTED,
-	SYNC_TABLE_STATE_VALID,
-} SyncingTablesState;
-
-static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
-static List *table_states_not_ready = NIL;
-static bool FetchTableStates(bool *started_tx);
-
 static StringInfo copybuf = NULL;
 
-/*
- * Exit routine for synchronization worker.
- */
-static void
-pg_attribute_noreturn()
-finish_sync_worker(void)
-{
-	/*
-	 * Commit any outstanding transaction. This is the usual case, unless
-	 * there was nothing to do for the table.
-	 */
-	if (IsTransactionState())
-	{
-		CommitTransactionCommand();
-		pgstat_report_stat(true);
-	}
-
-	/* And flush all writes. */
-	XLogFlush(GetXLogWriteRecPtr());
-
-	StartTransactionCommand();
-	ereport(LOG,
-			(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
-					MySubscription->name,
-					get_rel_name(MyLogicalRepWorker->relid))));
-	CommitTransactionCommand();
-
-	/* Find the leader apply worker and signal it. */
-	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
-
-	/* Stop gracefully */
-	proc_exit(0);
-}
-
 /*
  * Wait until the relation sync state is set in the catalog to the expected
  * one; return true when it happens.
@@ -180,7 +134,7 @@ finish_sync_worker(void)
  * Currently, this is used in the apply worker when transitioning from
  * CATCHUP state to SYNCDONE.
  */
-static bool
+bool
 wait_for_relation_state_change(Oid relid, char expected_state)
 {
 	char		state;
@@ -274,423 +228,6 @@ wait_for_worker_state_change(char expected_state)
 	return false;
 }
 
-/*
- * Callback from syscache invalidation.
- */
-void
-invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
-{
-	table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
-}
-
-/*
- * Handle table synchronization cooperation from the synchronization
- * worker.
- *
- * If the sync worker is in CATCHUP state and reached (or passed) the
- * predetermined synchronization point in the WAL stream, mark the table as
- * SYNCDONE and finish.
- */
-static void
-process_syncing_tables_for_sync(XLogRecPtr current_lsn)
-{
-	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
-
-	if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
-		current_lsn >= MyLogicalRepWorker->relstate_lsn)
-	{
-		TimeLineID	tli;
-		char		syncslotname[NAMEDATALEN] = {0};
-		char		originname[NAMEDATALEN] = {0};
-
-		MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
-		MyLogicalRepWorker->relstate_lsn = current_lsn;
-
-		SpinLockRelease(&MyLogicalRepWorker->relmutex);
-
-		/*
-		 * UpdateSubscriptionRelState must be called within a transaction.
-		 */
-		if (!IsTransactionState())
-			StartTransactionCommand();
-
-		UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
-								   MyLogicalRepWorker->relid,
-								   MyLogicalRepWorker->relstate,
-								   MyLogicalRepWorker->relstate_lsn);
-
-		/*
-		 * End streaming so that LogRepWorkerWalRcvConn can be used to drop
-		 * the slot.
-		 */
-		walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
-
-		/*
-		 * Cleanup the tablesync slot.
-		 *
-		 * This has to be done after updating the state because otherwise if
-		 * there is an error while doing the database operations we won't be
-		 * able to rollback dropped slot.
-		 */
-		ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
-										MyLogicalRepWorker->relid,
-										syncslotname,
-										sizeof(syncslotname));
-
-		/*
-		 * It is important to give an error if we are unable to drop the slot,
-		 * otherwise, it won't be dropped till the corresponding subscription
-		 * is dropped. So passing missing_ok = false.
-		 */
-		ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
-
-		CommitTransactionCommand();
-		pgstat_report_stat(false);
-
-		/*
-		 * Start a new transaction to clean up the tablesync origin tracking.
-		 * This transaction will be ended within the finish_sync_worker().
-		 * Now, even, if we fail to remove this here, the apply worker will
-		 * ensure to clean it up afterward.
-		 *
-		 * We need to do this after the table state is set to SYNCDONE.
-		 * Otherwise, if an error occurs while performing the database
-		 * operation, the worker will be restarted and the in-memory state of
-		 * replication progress (remote_lsn) won't be rolled-back which would
-		 * have been cleared before restart. So, the restarted worker will use
-		 * invalid replication progress state resulting in replay of
-		 * transactions that have already been applied.
-		 */
-		StartTransactionCommand();
-
-		ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
-										   MyLogicalRepWorker->relid,
-										   originname,
-										   sizeof(originname));
-
-		/*
-		 * Resetting the origin session removes the ownership of the slot.
-		 * This is needed to allow the origin to be dropped.
-		 */
-		replorigin_session_reset();
-		replorigin_session_origin = InvalidRepOriginId;
-		replorigin_session_origin_lsn = InvalidXLogRecPtr;
-		replorigin_session_origin_timestamp = 0;
-
-		/*
-		 * Drop the tablesync's origin tracking if exists.
-		 *
-		 * There is a chance that the user is concurrently performing refresh
-		 * for the subscription where we remove the table state and its origin
-		 * or the apply worker would have removed this origin. So passing
-		 * missing_ok = true.
-		 */
-		replorigin_drop_by_name(originname, true, false);
-
-		finish_sync_worker();
-	}
-	else
-		SpinLockRelease(&MyLogicalRepWorker->relmutex);
-}
-
-/*
- * Handle table synchronization cooperation from the apply worker.
- *
- * Walk over all subscription tables that are individually tracked by the
- * apply process (currently, all that have state other than
- * SUBREL_STATE_READY) and manage synchronization for them.
- *
- * If there are tables that need synchronizing and are not being synchronized
- * yet, start sync workers for them (if there are free slots for sync
- * workers).  To prevent starting the sync worker for the same relation at a
- * high frequency after a failure, we store its last start time with each sync
- * state info.  We start the sync worker for the same relation after waiting
- * at least wal_retrieve_retry_interval.
- *
- * For tables that are being synchronized already, check if sync workers
- * either need action from the apply worker or have finished.  This is the
- * SYNCWAIT to CATCHUP transition.
- *
- * If the synchronization position is reached (SYNCDONE), then the table can
- * be marked as READY and is no longer tracked.
- */
-static void
-process_syncing_tables_for_apply(XLogRecPtr current_lsn)
-{
-	struct tablesync_start_time_mapping
-	{
-		Oid			relid;
-		TimestampTz last_start_time;
-	};
-	static HTAB *last_start_times = NULL;
-	ListCell   *lc;
-	bool		started_tx = false;
-	bool		should_exit = false;
-
-	Assert(!IsTransactionState());
-
-	/* We need up-to-date sync state info for subscription tables here. */
-	FetchTableStates(&started_tx);
-
-	/*
-	 * Prepare a hash table for tracking last start times of workers, to avoid
-	 * immediate restarts.  We don't need it if there are no tables that need
-	 * syncing.
-	 */
-	if (table_states_not_ready != NIL && !last_start_times)
-	{
-		HASHCTL		ctl;
-
-		ctl.keysize = sizeof(Oid);
-		ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
-		last_start_times = hash_create("Logical replication table sync worker start times",
-									   256, &ctl, HASH_ELEM | HASH_BLOBS);
-	}
-
-	/*
-	 * Clean up the hash table when we're done with all tables (just to
-	 * release the bit of memory).
-	 */
-	else if (table_states_not_ready == NIL && last_start_times)
-	{
-		hash_destroy(last_start_times);
-		last_start_times = NULL;
-	}
-
-	/*
-	 * Process all tables that are being synchronized.
-	 */
-	foreach(lc, table_states_not_ready)
-	{
-		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
-
-		if (rstate->state == SUBREL_STATE_SYNCDONE)
-		{
-			/*
-			 * Apply has caught up to the position where the table sync has
-			 * finished.  Mark the table as ready so that the apply will just
-			 * continue to replicate it normally.
-			 */
-			if (current_lsn >= rstate->lsn)
-			{
-				char		originname[NAMEDATALEN];
-
-				rstate->state = SUBREL_STATE_READY;
-				rstate->lsn = current_lsn;
-				if (!started_tx)
-				{
-					StartTransactionCommand();
-					started_tx = true;
-				}
-
-				/*
-				 * Remove the tablesync origin tracking if exists.
-				 *
-				 * There is a chance that the user is concurrently performing
-				 * refresh for the subscription where we remove the table
-				 * state and its origin or the tablesync worker would have
-				 * already removed this origin. We can't rely on tablesync
-				 * worker to remove the origin tracking as if there is any
-				 * error while dropping we won't restart it to drop the
-				 * origin. So passing missing_ok = true.
-				 */
-				ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
-												   rstate->relid,
-												   originname,
-												   sizeof(originname));
-				replorigin_drop_by_name(originname, true, false);
-
-				/*
-				 * Update the state to READY only after the origin cleanup.
-				 */
-				UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
-										   rstate->relid, rstate->state,
-										   rstate->lsn);
-			}
-		}
-		else
-		{
-			LogicalRepWorker *syncworker;
-
-			/*
-			 * Look for a sync worker for this relation.
-			 */
-			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-
-			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-												rstate->relid, false);
-
-			if (syncworker)
-			{
-				/* Found one, update our copy of its state */
-				SpinLockAcquire(&syncworker->relmutex);
-				rstate->state = syncworker->relstate;
-				rstate->lsn = syncworker->relstate_lsn;
-				if (rstate->state == SUBREL_STATE_SYNCWAIT)
-				{
-					/*
-					 * Sync worker is waiting for apply.  Tell sync worker it
-					 * can catchup now.
-					 */
-					syncworker->relstate = SUBREL_STATE_CATCHUP;
-					syncworker->relstate_lsn =
-						Max(syncworker->relstate_lsn, current_lsn);
-				}
-				SpinLockRelease(&syncworker->relmutex);
-
-				/* If we told worker to catch up, wait for it. */
-				if (rstate->state == SUBREL_STATE_SYNCWAIT)
-				{
-					/* Signal the sync worker, as it may be waiting for us. */
-					if (syncworker->proc)
-						logicalrep_worker_wakeup_ptr(syncworker);
-
-					/* Now safe to release the LWLock */
-					LWLockRelease(LogicalRepWorkerLock);
-
-					if (started_tx)
-					{
-						/*
-						 * We must commit the existing transaction to release
-						 * the existing locks before entering a busy loop.
-						 * This is required to avoid any undetected deadlocks
-						 * due to any existing lock as deadlock detector won't
-						 * be able to detect the waits on the latch.
-						 */
-						CommitTransactionCommand();
-						pgstat_report_stat(false);
-					}
-
-					/*
-					 * Enter busy loop and wait for synchronization worker to
-					 * reach expected state (or die trying).
-					 */
-					StartTransactionCommand();
-					started_tx = true;
-
-					wait_for_relation_state_change(rstate->relid,
-												   SUBREL_STATE_SYNCDONE);
-				}
-				else
-					LWLockRelease(LogicalRepWorkerLock);
-			}
-			else
-			{
-				/*
-				 * If there is no sync worker for this table yet, count
-				 * running sync workers for this subscription, while we have
-				 * the lock.
-				 */
-				int			nsyncworkers =
-					logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
-
-				/* Now safe to release the LWLock */
-				LWLockRelease(LogicalRepWorkerLock);
-
-				/*
-				 * If there are free sync worker slot(s), start a new sync
-				 * worker for the table.
-				 */
-				if (nsyncworkers < max_sync_workers_per_subscription)
-				{
-					TimestampTz now = GetCurrentTimestamp();
-					struct tablesync_start_time_mapping *hentry;
-					bool		found;
-
-					hentry = hash_search(last_start_times, &rstate->relid,
-										 HASH_ENTER, &found);
-
-					if (!found ||
-						TimestampDifferenceExceeds(hentry->last_start_time, now,
-												   wal_retrieve_retry_interval))
-					{
-						logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
-												 MyLogicalRepWorker->dbid,
-												 MySubscription->oid,
-												 MySubscription->name,
-												 MyLogicalRepWorker->userid,
-												 rstate->relid,
-												 DSM_HANDLE_INVALID);
-						hentry->last_start_time = now;
-					}
-				}
-			}
-		}
-	}
-
-	if (started_tx)
-	{
-		/*
-		 * Even when the two_phase mode is requested by the user, it remains
-		 * as 'pending' until all tablesyncs have reached READY state.
-		 *
-		 * When this happens, we restart the apply worker and (if the
-		 * conditions are still ok) then the two_phase tri-state will become
-		 * 'enabled' at that time.
-		 *
-		 * Note: If the subscription has no tables then leave the state as
-		 * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
-		 * work.
-		 */
-		if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
-		{
-			CommandCounterIncrement();	/* make updates visible */
-			if (AllTablesyncsReady())
-			{
-				ereport(LOG,
-						(errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
-								MySubscription->name)));
-				should_exit = true;
-			}
-		}
-
-		CommitTransactionCommand();
-		pgstat_report_stat(true);
-	}
-
-	if (should_exit)
-	{
-		/*
-		 * Reset the last-start time for this worker so that the launcher will
-		 * restart it without waiting for wal_retrieve_retry_interval.
-		 */
-		ApplyLauncherForgetWorkerStartTime(MySubscription->oid);
-
-		proc_exit(0);
-	}
-}
-
-/*
- * Process possible state change(s) of tables that are being synchronized.
- */
-void
-process_syncing_tables(XLogRecPtr current_lsn)
-{
-	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;
-
-		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");
-	}
-}
-
 /*
  * Create list of columns for COPY based on logical relation mapping.
  */
@@ -1320,7 +857,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 		case SUBREL_STATE_SYNCDONE:
 		case SUBREL_STATE_READY:
 		case SUBREL_STATE_UNKNOWN:
-			finish_sync_worker();	/* doesn't return */
+			SyncFinishWorker(); /* doesn't return */
 	}
 
 	/* Calculate the name of the tablesync slot. */
@@ -1561,77 +1098,6 @@ copy_table_done:
 	return slotname;
 }
 
-/*
- * Common code to fetch the up-to-date sync state info into the static lists.
- *
- * Returns true if subscription has 1 or more tables, else false.
- *
- * Note: If this function started the transaction (indicated by the parameter)
- * then it is the caller's responsibility to commit it.
- */
-static bool
-FetchTableStates(bool *started_tx)
-{
-	static bool has_subrels = false;
-
-	*started_tx = false;
-
-	if (table_states_validity != SYNC_TABLE_STATE_VALID)
-	{
-		MemoryContext oldctx;
-		List	   *rstates;
-		ListCell   *lc;
-		SubscriptionRelState *rstate;
-
-		table_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED;
-
-		/* Clean the old lists. */
-		list_free_deep(table_states_not_ready);
-		table_states_not_ready = NIL;
-
-		if (!IsTransactionState())
-		{
-			StartTransactionCommand();
-			*started_tx = true;
-		}
-
-		/* Fetch all non-ready tables. */
-		rstates = GetSubscriptionRelations(MySubscription->oid, true);
-
-		/* Allocate the tracking info in a permanent memory context. */
-		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
-		foreach(lc, rstates)
-		{
-			rstate = palloc(sizeof(SubscriptionRelState));
-			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
-			table_states_not_ready = lappend(table_states_not_ready, rstate);
-		}
-		MemoryContextSwitchTo(oldctx);
-
-		/*
-		 * Does the subscription have tables?
-		 *
-		 * If there were not-READY relations found then we know it does. But
-		 * if table_states_not_ready was empty we still need to check again to
-		 * see if there are 0 tables.
-		 */
-		has_subrels = (table_states_not_ready != NIL) ||
-			HasSubscriptionRelations(MySubscription->oid);
-
-		/*
-		 * If the subscription relation cache has been invalidated since we
-		 * entered this routine, we still use and return the relations we just
-		 * finished constructing, to avoid infinite loops, but we leave the
-		 * table states marked as stale so that we'll rebuild it again on next
-		 * access. Otherwise, we mark the table states as valid.
-		 */
-		if (table_states_validity == SYNC_TABLE_STATE_REBUILD_STARTED)
-			table_states_validity = SYNC_TABLE_STATE_VALID;
-	}
-
-	return has_subrels;
-}
-
 /*
  * Execute the initial sync with error handling. Disable the subscription,
  * if it's required.
@@ -1717,37 +1183,7 @@ TablesyncWorkerMain(Datum main_arg)
 
 	run_tablesync_worker();
 
-	finish_sync_worker();
-}
-
-/*
- * If the subscription has no tables then return false.
- *
- * Otherwise, are all tablesyncs READY?
- *
- * Note: This function is not suitable to be called from outside of apply or
- * tablesync workers because MySubscription needs to be already initialized.
- */
-bool
-AllTablesyncsReady(void)
-{
-	bool		started_tx = false;
-	bool		has_subrels = false;
-
-	/* We need up-to-date sync state info for subscription tables here. */
-	has_subrels = FetchTableStates(&started_tx);
-
-	if (started_tx)
-	{
-		CommitTransactionCommand();
-		pgstat_report_stat(true);
-	}
-
-	/*
-	 * Return false when there are no tables in subscription or not all tables
-	 * are in ready state; true otherwise.
-	 */
-	return has_subrels && (table_states_not_ready == NIL);
+	SyncFinishWorker();
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 245e9be6f2..6ef21c4273 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1026,7 +1026,7 @@ apply_handle_commit(StringInfo s)
 	apply_handle_commit_internal(&commit_data);
 
 	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(commit_data.end_lsn);
+	SyncProcessRelations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
@@ -1148,7 +1148,7 @@ apply_handle_prepare(StringInfo s)
 	in_remote_transaction = false;
 
 	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(prepare_data.end_lsn);
+	SyncProcessRelations(prepare_data.end_lsn);
 
 	/*
 	 * Since we have already prepared the transaction, in a case where the
@@ -1204,7 +1204,7 @@ apply_handle_commit_prepared(StringInfo s)
 	in_remote_transaction = false;
 
 	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(prepare_data.end_lsn);
+	SyncProcessRelations(prepare_data.end_lsn);
 
 	clear_subscription_skip_lsn(prepare_data.end_lsn);
 
@@ -1270,7 +1270,7 @@ apply_handle_rollback_prepared(StringInfo s)
 	in_remote_transaction = false;
 
 	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(rollback_data.rollback_end_lsn);
+	SyncProcessRelations(rollback_data.rollback_end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
@@ -1405,7 +1405,7 @@ apply_handle_stream_prepare(StringInfo s)
 	pgstat_report_stat(false);
 
 	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(prepare_data.end_lsn);
+	SyncProcessRelations(prepare_data.end_lsn);
 
 	/*
 	 * Similar to prepare case, the subskiplsn could be left in a case of
@@ -2247,7 +2247,7 @@ apply_handle_stream_commit(StringInfo s)
 	}
 
 	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(commit_data.end_lsn);
+	SyncProcessRelations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 
@@ -3645,7 +3645,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 			maybe_reread_subscription();
 
 			/* Process any table synchronization changes. */
-			process_syncing_tables(last_received);
+			SyncProcessRelations(last_received);
 		}
 
 		/* Cleanup the memory. */
@@ -4525,7 +4525,7 @@ run_apply_worker()
 	 * work.
 	 */
 	if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
-		AllTablesyncsReady())
+		SyncAllTablesReady())
 	{
 		/* Start streaming with two_phase enabled */
 		options.proto.logical.twophase = true;
@@ -4679,7 +4679,7 @@ SetupApplyOrSyncWorker(int worker_slot)
 	 * the subscription relation state.
 	 */
 	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
-								  invalidate_syncing_table_states,
+								  SyncInvalidateRelationStates,
 								  (Datum) 0);
 }
 
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 8244ad537a..f11eea6824 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -89,7 +89,7 @@ extern void UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
 extern char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn);
 extern void RemoveSubscriptionRel(Oid subid, Oid relid);
 
-extern bool HasSubscriptionRelations(Oid subid);
+extern bool HasSubscriptionTables(Oid subid);
 extern List *GetSubscriptionRelations(Oid subid, bool not_ready);
 
 #endif							/* PG_SUBSCRIPTION_REL_H */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 9646261d7e..ff25874d5e 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -250,18 +250,20 @@ extern void logicalrep_worker_stop(Oid subid, Oid relid);
 extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
+extern void pg_attribute_noreturn() SyncFinishWorker(void);
 
 extern int	logicalrep_sync_worker_count(Oid subid);
 
 extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 											   char *originname, Size szoriginname);
 
-extern bool AllTablesyncsReady(void);
+extern bool SyncAllTablesReady(void);
 extern void UpdateTwoPhaseState(Oid suboid, char new_state);
 
-extern void process_syncing_tables(XLogRecPtr current_lsn);
-extern void invalidate_syncing_table_states(Datum arg, int cacheid,
-											uint32 hashvalue);
+extern bool wait_for_relation_state_change(Oid relid, char expected_state);
+extern void SyncProcessRelations(XLogRecPtr current_lsn);
+extern void SyncInvalidateRelationStates(Datum arg, int cacheid,
+										 uint32 hashvalue);
 
 extern void stream_start_internal(TransactionId xid, bool first_segment);
 extern void stream_stop_internal(TransactionId xid);
-- 
2.34.1