From 6934332931e4fa11fa2b08e5ccdb5e6f3c6430ee Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Mon, 9 Jun 2025 17:45:24 +0530
Subject: [PATCH v20250610 5/6] New worker for sequence synchronization during
 subscription management

This patch introduces sequence synchronization:
Sequences have 2 states:
   - INIT (needs synchronizing)
   - READY (is already synchronized)

A new sequencesync worker is launched as needed to synchronize sequences.
It does the following:
    a) Retrieves remote values of sequences with pg_sequence_state() INIT.
    b) Logs a warning if the sequence parameters differ between the publisher and subscriber.
    c) Sets the local sequence values accordingly.
    d) Updates the local sequence state to READY.
    e) Repeats until all done; Commits synchronized sequences in batches of 100

Sequence synchronization occurs in 3 places:
1) CREATE SUBSCRIPTION
    - (PG18 command syntax is unchanged)
    - The subscriber retrieves sequences associated with publications.
    - Published sequences are added to pg_subscription_rel with INIT state.
    - Initiate the sequencesync worker (see above) to synchronize all
      sequences.

2) ALTER SUBSCRIPTION ... REFRESH PUBLICATION
    - (PG18 command syntax is unchanged)
    - Dropped published sequences are removed from pg_subscription_rel.
    - Newly published sequences are added to pg_subscription_rel with INIT state.
    - Initiate the sequencesync worker (see above) to synchronize only
      newly added sequences.

3) ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES
    - The patch introduces this new command to refresh all sequences
    - Dropped published sequences are removed from pg_subscription_rel.
    - Newly published sequences are added to pg_subscription_rel
    - All sequences in pg_subscription_rel are reset to INIT state.
    - Initiate the sequencesync worker (see above) to synchronize all
      sequences.
---
 src/backend/catalog/system_views.sql          |   1 +
 src/backend/commands/sequence.c               |  26 +-
 src/backend/commands/subscriptioncmds.c       |   4 +-
 src/backend/postmaster/bgworker.c             |   5 +-
 src/backend/replication/logical/Makefile      |   1 +
 src/backend/replication/logical/launcher.c    |  71 +-
 src/backend/replication/logical/meson.build   |   1 +
 .../replication/logical/sequencesync.c        | 628 ++++++++++++++++++
 src/backend/replication/logical/syncutils.c   |  70 +-
 src/backend/replication/logical/tablesync.c   |  48 +-
 src/backend/replication/logical/worker.c      |  73 +-
 .../utils/activity/pgstat_subscription.c      |  27 +-
 src/backend/utils/adt/pgstatfuncs.c           |  25 +-
 src/backend/utils/misc/guc_tables.c           |   2 +-
 src/include/catalog/pg_proc.dat               |   6 +-
 src/include/catalog/pg_subscription_rel.h     |   8 +
 src/include/commands/sequence.h               |   3 +
 src/include/pgstat.h                          |   6 +-
 src/include/replication/logicalworker.h       |   3 +-
 src/include/replication/worker_internal.h     |  30 +-
 src/test/subscription/meson.build             |   1 +
 src/test/subscription/t/036_sequences.pl      | 226 +++++++
 src/tools/pgindent/typedefs.list              |   1 +
 23 files changed, 1157 insertions(+), 109 deletions(-)
 create mode 100644 src/backend/replication/logical/sequencesync.c
 create mode 100644 src/test/subscription/t/036_sequences.pl

diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 9853fd50b35..dde8a71b84d 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1396,6 +1396,7 @@ CREATE VIEW pg_stat_subscription_stats AS
         ss.subid,
         s.subname,
         ss.apply_error_count,
+        ss.sequence_sync_error_count,
         ss.sync_error_count,
         ss.confl_insert_exists,
         ss.confl_update_origin_differs,
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index cf357650a24..60035e1979b 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -110,7 +110,6 @@ static void init_params(ParseState *pstate, List *options, bool for_identity,
 						Form_pg_sequence_data seqdataform,
 						bool *need_seq_rewrite,
 						List **owned_by);
-static void do_setval(Oid relid, int64 next, bool iscalled);
 static void process_owned_by(Relation seqrel, List *owned_by, bool for_identity);
 
 
@@ -941,9 +940,12 @@ lastval(PG_FUNCTION_ARGS)
  * restore the state of a sequence exactly during data-only restores -
  * it is the only way to clear the is_called flag in an existing
  * sequence.
+ *
+ * log_cnt is currently used only by the sequence syncworker to set the
+ * log_cnt for sequences while synchronizing values from the publisher.
  */
-static void
-do_setval(Oid relid, int64 next, bool iscalled)
+void
+SetSequence(Oid relid, int64 next, int64 log_cnt, bool is_called)
 {
 	SeqTable	elm;
 	Relation	seqrel;
@@ -994,7 +996,7 @@ do_setval(Oid relid, int64 next, bool iscalled)
 						minv, maxv)));
 
 	/* Set the currval() state only if iscalled = true */
-	if (iscalled)
+	if (is_called)
 	{
 		elm->last = next;		/* last returned number */
 		elm->last_valid = true;
@@ -1011,8 +1013,8 @@ do_setval(Oid relid, int64 next, bool iscalled)
 	START_CRIT_SECTION();
 
 	seq->last_value = next;		/* last fetched number */
-	seq->is_called = iscalled;
-	seq->log_cnt = 0;
+	seq->is_called = is_called;
+	seq->log_cnt = log_cnt;
 
 	MarkBufferDirty(buf);
 
@@ -1044,7 +1046,7 @@ do_setval(Oid relid, int64 next, bool iscalled)
 
 /*
  * Implement the 2 arg setval procedure.
- * See do_setval for discussion.
+ * See SetSequence for discussion.
  */
 Datum
 setval_oid(PG_FUNCTION_ARGS)
@@ -1052,14 +1054,14 @@ setval_oid(PG_FUNCTION_ARGS)
 	Oid			relid = PG_GETARG_OID(0);
 	int64		next = PG_GETARG_INT64(1);
 
-	do_setval(relid, next, true);
+	SetSequence(relid, next, SEQ_LOG_CNT_INVALID, true);
 
 	PG_RETURN_INT64(next);
 }
 
 /*
  * Implement the 3 arg setval procedure.
- * See do_setval for discussion.
+ * See SetSequence for discussion.
  */
 Datum
 setval3_oid(PG_FUNCTION_ARGS)
@@ -1068,7 +1070,7 @@ setval3_oid(PG_FUNCTION_ARGS)
 	int64		next = PG_GETARG_INT64(1);
 	bool		iscalled = PG_GETARG_BOOL(2);
 
-	do_setval(relid, next, iscalled);
+	SetSequence(relid, next, SEQ_LOG_CNT_INVALID, iscalled);
 
 	PG_RETURN_INT64(next);
 }
@@ -1889,6 +1891,10 @@ pg_sequence_last_value(PG_FUNCTION_ARGS)
 /*
  * Return the current on-disk state of the sequence.
  *
+ * The page LSN will be used in logical replication of sequences to record the
+ * LSN of the sequence page in the pg_subscription_rel system catalog.  It
+ * reflects the LSN of the remote sequence at the time it was synchronized.
+ *
  * Note: This is roughly equivalent to selecting the data from the sequence,
  * except that it also returns the page LSN.
  */
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 32c77ad372c..2b5dff21e44 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1068,7 +1068,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 					sub_remove_rels[remove_rel_len].relid = relid;
 					sub_remove_rels[remove_rel_len++].state = state;
 
-					logicalrep_worker_stop(sub->oid, relid);
+					logicalrep_worker_stop(sub->oid, relid, WORKERTYPE_TABLESYNC);
 
 					/*
 					 * For READY state, we would have already dropped the
@@ -1870,7 +1870,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	{
 		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-		logicalrep_worker_stop(w->subid, w->relid);
+		logicalrep_worker_stop(w->subid, w->relid, w->type);
 	}
 	list_free(subworkers);
 
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 116ddf7b835..81e0e369fb0 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -131,7 +131,10 @@ static const struct
 		"ParallelApplyWorkerMain", ParallelApplyWorkerMain
 	},
 	{
-		"TablesyncWorkerMain", TablesyncWorkerMain
+		"TableSyncWorkerMain", TableSyncWorkerMain
+	},
+	{
+		"SequenceSyncWorkerMain", SequenceSyncWorkerMain
 	}
 };
 
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c62c8c67521..c719af1f8a9 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -26,6 +26,7 @@ OBJS = \
 	proto.o \
 	relation.o \
 	reorderbuffer.o \
+	sequencesync.o \
 	slotsync.o \
 	snapbuild.o \
 	syncutils.o \
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 1c3c051403d..aa8f99e29aa 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -226,19 +226,18 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 
 /*
  * Walks the workers array and searches for one that matches given
- * subscription id and relid.
- *
- * We are only interested in the leader apply worker or table sync worker.
+ * subscription id, relid and type.
  */
 LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid subid, Oid relid, LogicalRepWorkerType wtype,
+					   bool only_running)
 {
 	int			i;
 	LogicalRepWorker *res = NULL;
 
 	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
-	/* Search for attached worker for a given subscription id. */
+	/* Search for the attached worker matching the specified criteria. */
 	for (i = 0; i < max_logical_replication_workers; i++)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
@@ -248,7 +247,7 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 			continue;
 
 		if (w->in_use && w->subid == subid && w->relid == relid &&
-			(!only_running || w->proc))
+			w->type == wtype && (!only_running || w->proc))
 		{
 			res = w;
 			break;
@@ -308,6 +307,7 @@ logicalrep_worker_launch(LogicalRepWorkerType wtype,
 	int			nparallelapplyworkers;
 	TimestampTz now;
 	bool		is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
+	bool		is_sequencesync_worker = (wtype == WORKERTYPE_SEQUENCESYNC);
 	bool		is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
 
 	/*----------
@@ -393,7 +393,8 @@ retry:
 	 * sync worker limit per subscription. So, just return silently as we
 	 * might get here because of an otherwise harmless race condition.
 	 */
-	if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
+	if ((is_tablesync_worker || is_sequencesync_worker) &&
+		nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
@@ -479,8 +480,16 @@ retry:
 			memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
 			break;
 
+		case WORKERTYPE_SEQUENCESYNC:
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "SequenceSyncWorkerMain");
+			snprintf(bgw.bgw_name, BGW_MAXLEN,
+					 "logical replication sequencesync worker for subscription %u",
+					 subid);
+			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication sequencesync worker");
+			break;
+
 		case WORKERTYPE_TABLESYNC:
-			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
+			snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TableSyncWorkerMain");
 			snprintf(bgw.bgw_name, BGW_MAXLEN,
 					 "logical replication tablesync worker for subscription %u sync %u",
 					 subid,
@@ -603,13 +612,13 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
  * Stop the logical replication worker for subid/relid, if any.
  */
 void
-logicalrep_worker_stop(Oid subid, Oid relid)
+logicalrep_worker_stop(Oid subid, Oid relid, LogicalRepWorkerType wtype)
 {
 	LogicalRepWorker *worker;
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-	worker = logicalrep_worker_find(subid, relid, false);
+	worker = logicalrep_worker_find(subid, relid, wtype, false);
 
 	if (worker)
 	{
@@ -676,7 +685,7 @@ logicalrep_worker_wakeup(Oid subid, Oid relid)
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-	worker = logicalrep_worker_find(subid, relid, true);
+	worker = logicalrep_worker_find(subid, relid, WORKERTYPE_APPLY, true);
 
 	if (worker)
 		logicalrep_worker_wakeup_ptr(worker);
@@ -806,6 +815,37 @@ logicalrep_launcher_onexit(int code, Datum arg)
 	LogicalRepCtx->launcher_pid = 0;
 }
 
+/*
+ * Set the sequencesync worker failure time.
+ */
+void
+logicalrep_seqsyncworker_set_failuretime()
+{
+	LogicalRepWorker *worker;
+
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+	worker = logicalrep_worker_find(MyLogicalRepWorker->subid, InvalidOid,
+									WORKERTYPE_APPLY, true);
+	if (worker)
+		worker->sequencesync_failure_time = GetCurrentTimestamp();
+
+	LWLockRelease(LogicalRepWorkerLock);
+}
+
+/*
+ * Update the failure time of the sequencesync worker in the subscription's
+ * apply worker.
+ *
+ * This function is invoked when the sequencesync worker exits due to a
+ * failure.
+ */
+void
+logicalrep_seqsyncworker_failure(int code, Datum arg)
+{
+	logicalrep_seqsyncworker_set_failuretime();
+}
+
 /*
  * Cleanup function.
  *
@@ -854,7 +894,7 @@ logicalrep_sync_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (isTablesyncWorker(w) && w->subid == subid)
+		if (w->subid == subid && (isTableSyncWorker(w) || isSequenceSyncWorker(w)))
 			res++;
 	}
 
@@ -1169,7 +1209,7 @@ ApplyLauncherMain(Datum main_arg)
 				continue;
 
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-			w = logicalrep_worker_find(sub->oid, InvalidOid, false);
+			w = logicalrep_worker_find(sub->oid, InvalidOid, WORKERTYPE_APPLY, false);
 			LWLockRelease(LogicalRepWorkerLock);
 
 			if (w != NULL)
@@ -1305,7 +1345,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 		worker_pid = worker.proc->pid;
 
 		values[0] = ObjectIdGetDatum(worker.subid);
-		if (isTablesyncWorker(&worker))
+		if (isTableSyncWorker(&worker))
 			values[1] = ObjectIdGetDatum(worker.relid);
 		else
 			nulls[1] = true;
@@ -1345,6 +1385,9 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 			case WORKERTYPE_PARALLEL_APPLY:
 				values[9] = CStringGetTextDatum("parallel apply");
 				break;
+			case WORKERTYPE_SEQUENCESYNC:
+				values[9] = CStringGetTextDatum("sequence synchronization");
+				break;
 			case WORKERTYPE_TABLESYNC:
 				values[9] = CStringGetTextDatum("table synchronization");
 				break;
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index 9283e996ef4..a2268d8361e 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -12,6 +12,7 @@ backend_sources += files(
   'proto.c',
   'relation.c',
   'reorderbuffer.c',
+  'sequencesync.c',
   'slotsync.c',
   'snapbuild.c',
   'syncutils.c',
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
new file mode 100644
index 00000000000..f402f9aca2f
--- /dev/null
+++ b/src/backend/replication/logical/sequencesync.c
@@ -0,0 +1,628 @@
+/*-------------------------------------------------------------------------
+ * sequencesync.c
+ *	  PostgreSQL logical replication: sequence synchronization
+ *
+ * Copyright (c) 2025, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/sequencesync.c
+ *
+ * NOTES
+ *	  This file contains code for sequence synchronization for
+ *	  logical replication.
+ *
+ * Sequences to be synchronized by the sequencesync worker will
+ * be added to pg_subscription_rel in INIT state when one of the following
+ * commands is executed:
+ * CREATE SUBSCRIPTION
+ * ALTER SUBSCRIPTION ... REFRESH PUBLICATION
+ * ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES
+ *
+ * The apply worker will periodically check if there are any sequences in INIT
+ * state and will start a sequencesync worker if needed.
+ *
+ * The sequencesync worker retrieves the sequences to be synchronized from the
+ * pg_subscription_rel catalog table.  It synchronizes multiple sequences per
+ * single transaction by fetching the sequence value and page LSN from the
+ * remote publisher and updating them in the local subscriber sequence.  After
+ * synchronization, it sets the sequence state to READY.
+ *
+ * So the state progression is always just: INIT -> READY.
+ *
+ * To avoid creating too many transactions, up to MAX_SEQUENCES_SYNC_PER_BATCH
+ * (100) sequences are synchronized per transaction. The locks on the sequence
+ * relation will be periodically released at each transaction commit.
+ *
+ * XXX: An alternative design was considered where the launcher process would
+ * periodically check for sequences that need syncing and then start the
+ * sequencesync worker. However, the approach of having the apply worker
+ * manage the sequencesync worker was chosen for the following reasons:
+ * a) It avoids overloading the launcher, which handles various other
+ *    subscription requests.
+ * b) It offers a more straightforward path for extending support for
+ *    incremental sequence synchronization.
+ * c) It utilizes the existing tablesync worker code to start the sequencesync
+ *    process, thus preventing code duplication in the launcher.
+ * d) It simplifies code maintenance by consolidating changes to a single
+ *    location rather than multiple components.
+ * e) The apply worker can access the sequences that need to be synchronized
+ *    from the pg_subscription_rel system catalog. Whereas the launcher process
+ *    operates without direct database access so would need a framework to
+ *    establish connections with the databases to retrieve the sequences for
+ *    synchronization.
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/table.h"
+#include "catalog/pg_sequence.h"
+#include "catalog/pg_subscription_rel.h"
+#include "commands/sequence.h"
+#include "pgstat.h"
+#include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
+#include "replication/worker_internal.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/catcache.h"
+#include "utils/lsyscache.h"
+#include "utils/pg_lsn.h"
+#include "utils/rls.h"
+#include "utils/syscache.h"
+#include "utils/usercontext.h"
+
+List	   *sequence_states_not_ready = NIL;
+
+/*
+ * Handle sequence synchronization cooperation from the apply worker.
+ *
+ * Walk over all subscription sequences that are individually tracked by the
+ * apply process (currently, all that have state SUBREL_STATE_INIT) and manage
+ * synchronization for them.
+ *
+ * If a sequencesync worker is running already, there is no need to start a new
+ * one; the existing sequencesync worker will synchronize all the sequences. If
+ * there are still any sequences to be synced after the sequencesync worker
+ * exited, then a new sequencesync worker can be started in the next iteration.
+ */
+void
+ProcessSyncingSequencesForApply(void)
+{
+	bool		started_tx = false;
+
+	Assert(!IsTransactionState());
+
+	/* Start the sequencesync worker if needed, and there is not one already. */
+	foreach_ptr(SubscriptionRelState, rstate, sequence_states_not_ready)
+	{
+		LogicalRepWorker *sequencesync_worker;
+		int			nsyncworkers;
+
+		if (!started_tx)
+		{
+			StartTransactionCommand();
+			started_tx = true;
+		}
+
+		Assert(get_rel_relkind(rstate->relid) == RELKIND_SEQUENCE);
+
+		if (rstate->state != SUBREL_STATE_INIT)
+			continue;
+
+		/*
+		 * Check if there is a sequencesync worker already running?
+		 */
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+		sequencesync_worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+													 InvalidOid,
+													 WORKERTYPE_SEQUENCESYNC,
+													 true);
+		if (sequencesync_worker)
+		{
+			/* Now safe to release the LWLock */
+			LWLockRelease(LogicalRepWorkerLock);
+			break;
+		}
+
+		/*
+		 * Count running sync workers for this subscription, while we have the
+		 * lock.
+		 */
+		nsyncworkers = logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
+
+		/* Now safe to release the LWLock */
+		LWLockRelease(LogicalRepWorkerLock);
+
+		/*
+		 * If there is a free sync worker slot, start a new sequencesync
+		 * worker, and break from the loop.
+		 */
+		if (nsyncworkers < max_sync_workers_per_subscription)
+		{
+			TimestampTz now = GetCurrentTimestamp();
+
+			/*
+			 * To prevent starting the sequencesync worker at a high frequency
+			 * after a failure, we store its last failure time. We start the
+			 * sequencesync worker again after waiting at least
+			 * wal_retrieve_retry_interval.
+			 */
+			if (!MyLogicalRepWorker->sequencesync_failure_time ||
+				TimestampDifferenceExceeds(MyLogicalRepWorker->sequencesync_failure_time,
+										   now, wal_retrieve_retry_interval))
+			{
+				MyLogicalRepWorker->sequencesync_failure_time = 0;
+
+				logicalrep_worker_launch(WORKERTYPE_SEQUENCESYNC,
+										 MyLogicalRepWorker->dbid,
+										 MySubscription->oid,
+										 MySubscription->name,
+										 MyLogicalRepWorker->userid,
+										 InvalidOid,
+										 DSM_HANDLE_INVALID);
+				break;
+			}
+		}
+	}
+
+	if (started_tx)
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(true);
+	}
+}
+
+/*
+ * report_error_sequences
+ *
+ * Reports discrepancies in sequence data between the publisher and subscriber.
+ * It identifies sequences that are missing on the publisher, as well as
+ * sequences that exist on both sides but have mismatched values.
+ */
+static void
+report_error_sequences(StringInfo missing_seqs, StringInfo mismatched_seqs)
+{
+	StringInfo	combined_error_msg = makeStringInfo();
+
+	appendStringInfo(combined_error_msg, "logical replication sequence synchronization failed for subscription \"%s\":",
+					 MySubscription->name);
+
+	if (missing_seqs->len)
+		appendStringInfo(combined_error_msg, " sequences (%s) are missing on the publisher.",
+						 missing_seqs->data);
+
+	if (mismatched_seqs->len)
+	{
+		/* Add a separator if both types of errors exist */
+		if (missing_seqs->len)
+			appendStringInfoString(combined_error_msg, " Additionally,");
+
+		appendStringInfo(combined_error_msg, " parameters differ for the remote and local sequences (%s)",
+						 mismatched_seqs->data);
+	}
+
+	ereport(ERROR, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+			errmsg("%s", combined_error_msg->data));
+}
+
+/*
+ * Copy existing data of sequences from the publisher.
+ *
+ * Fetch the sequence value from the publisher and set the subscriber sequence
+ * with the same value. Caller is responsible for locking the local relation.
+ */
+static void
+copy_sequences(WalReceiverConn *conn, List *remotesequences, Oid subid)
+{
+	int			total_seqs = list_length(remotesequences);
+	int			current_index = 0;
+	StringInfo	mismatched_seqs = makeStringInfo();
+	StringInfo	missing_seqs = makeStringInfo();
+
+	ereport(LOG,
+			errmsg("logical replication sequence synchronization for subscription \"%s\" - total unsynchronized: %d",
+				   MySubscription->name, total_seqs));
+
+	/*
+	 * We batch synchronize multiple sequences per transaction, because the
+	 * alternative of synchronizing each sequence individually incurs overhead
+	 * of starting and committing transactions repeatedly. On the other hand,
+	 * we want to avoid keeping this batch transaction open for extended
+	 * periods so it is currently limited to 100 sequences per batch.
+	 */
+#define MAX_SEQUENCES_SYNC_PER_BATCH 100
+
+	while (current_index < total_seqs)
+	{
+#define REMOTE_SEQ_COL_COUNT 12
+		Oid			seqRow[REMOTE_SEQ_COL_COUNT] = {TEXTOID, TEXTOID, LSNOID, INT8OID,
+		INT8OID, BOOLOID, OIDOID, INT8OID, INT8OID, INT8OID, INT8OID, BOOLOID};
+
+		int			batch_size = Min(MAX_SEQUENCES_SYNC_PER_BATCH, total_seqs - current_index);
+		int			batch_succeeded_count = 0;
+		int			batch_mismatched_count = 0;
+		StringInfo	seqstr = makeStringInfo();
+		StringInfo	cmd = makeStringInfo();
+		WalRcvExecResult *res;
+		TupleTableSlot *slot;
+
+		StartTransactionCommand();
+
+		/*
+		 * Prepare the string of current batch sequences to fetch from the
+		 * publisher.
+		 */
+		for (int i = 0; i < batch_size; i++)
+		{
+			LogicalRepSequenceInfo *seqinfo = lfirst(list_nth_cell(remotesequences, current_index + i));
+
+			if (seqstr->len > 0)
+				appendStringInfoString(seqstr, ", ");
+
+			appendStringInfo(seqstr, "(\'%s\', \'%s\')", seqinfo->nspname,
+							 seqinfo->seqname);
+		}
+
+		initStringInfo(cmd);
+		appendStringInfo(cmd,
+						 "SELECT s.schname, s.seqname, ps.*, seq.seqtypid,\n"
+						 "       seq.seqstart, seq.seqincrement, seq.seqmin,\n"
+						 "       seq.seqmax, seq.seqcycle\n"
+						 "FROM ( VALUES %s ) AS s (schname, seqname)\n"
+						 "JOIN LATERAL pg_sequence_state(s.schname, s.seqname) ps ON true\n"
+						 "JOIN pg_namespace n ON n.nspname = s.schname\n"
+						 "JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = s.seqname\n"
+						 "JOIN pg_sequence seq ON seq.seqrelid = c.oid\n",
+						 seqstr->data);
+
+		res = walrcv_exec(conn, cmd->data, lengthof(seqRow), seqRow);
+		if (res->status != WALRCV_OK_TUPLES)
+			ereport(ERROR,
+					errcode(ERRCODE_CONNECTION_FAILURE),
+					errmsg("could not receive list of sequence information from the publisher: %s",
+						   res->err));
+
+		destroyStringInfo(seqstr);
+
+		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+		while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+		{
+			int			col = 0;
+			bool		isnull;
+			char	   *nspname;
+			char	   *seqname;
+			XLogRecPtr	page_lsn;
+			int64		last_value;
+			int64		log_cnt;
+			bool		is_called;
+			Oid			seqtypid;
+			int64		seqstart;
+			int64		seqmin;
+			int64		seqmax;
+			int64		seqincrement;
+			bool		seqcycle;
+			HeapTuple	tup;
+			Form_pg_sequence seqform;
+			LogicalRepSequenceInfo *seqinfo = NULL;
+
+			CHECK_FOR_INTERRUPTS();
+
+			nspname = TextDatumGetCString(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqname = TextDatumGetCString(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			page_lsn = DatumGetLSN(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			last_value = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			log_cnt = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			is_called = DatumGetBool(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqtypid = DatumGetObjectId(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqstart = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqincrement = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqmin = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqmax = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			seqcycle = DatumGetBool(slot_getattr(slot, ++col, &isnull));
+			Assert(!isnull);
+
+			/* Sanity check */
+			Assert(col == REMOTE_SEQ_COL_COUNT);
+
+			/* Retrieve the sequence object fetched from the publisher */
+			for (int i = 0; i < batch_size; i++)
+			{
+				LogicalRepSequenceInfo *sequence_info = lfirst(list_nth_cell(remotesequences, current_index + i));
+
+				if (!strcmp(sequence_info->nspname, nspname) &&
+					!strcmp(sequence_info->seqname, seqname))
+					seqinfo = sequence_info;
+			}
+
+			Assert(seqinfo);
+
+			seqinfo->remote_seq_fetched = true;
+
+			/* Get the local sequence */
+			tup = SearchSysCache1(SEQRELID, ObjectIdGetDatum(seqinfo->localrelid));
+			if (!HeapTupleIsValid(tup))
+				elog(ERROR, "cache lookup failed for sequence \"%s.%s\"",
+					 seqinfo->nspname, seqinfo->seqname);
+
+			seqform = (Form_pg_sequence) GETSTRUCT(tup);
+
+			/* Update the sequence only if the parameters are identical */
+			if (seqform->seqtypid == seqtypid &&
+				seqform->seqmin == seqmin && seqform->seqmax == seqmax &&
+				seqform->seqcycle == seqcycle &&
+				seqform->seqstart == seqstart &&
+				seqform->seqincrement == seqincrement)
+			{
+				SetSequence(seqinfo->localrelid, last_value, log_cnt, is_called);
+
+				UpdateSubscriptionRelState(subid, seqinfo->localrelid,
+										   SUBREL_STATE_READY, page_lsn);
+				if (message_level_is_interesting(DEBUG1))
+					ereport(DEBUG1,
+							errmsg_internal("logical replication synchronization for subscription \"%s\", sequence \"%s\" has finished",
+											MySubscription->name,
+											seqinfo->seqname));
+
+				batch_succeeded_count++;
+			}
+			else
+			{
+				if (mismatched_seqs->len)
+					appendStringInfoString(mismatched_seqs, ", ");
+
+				appendStringInfo(mismatched_seqs, "\"%s.%s\"",
+								 seqinfo->nspname, seqinfo->seqname);
+				batch_mismatched_count++;
+			}
+
+			ReleaseSysCache(tup);
+		}
+
+		ereport(LOG,
+				errmsg("logical replication sequence synchronization for subscription \"%s\" - batch #%d = %d attempted, %d succeeded, %d mismatched, %d missing",
+					   MySubscription->name, (current_index / MAX_SEQUENCES_SYNC_PER_BATCH) + 1, batch_size,
+					   batch_succeeded_count, batch_mismatched_count,
+					   batch_size - (batch_succeeded_count + batch_mismatched_count)));
+
+		ExecDropSingleTupleTableSlot(slot);
+
+		walrcv_clear_result(res);
+
+		destroyStringInfo(cmd);
+
+		/* Commit this batch, and prepare for next batch */
+		CommitTransactionCommand();
+
+		/*
+		 * Sequence synchronization for this batch was incomplete because some
+		 * sequences are missing on the publisher. Identify the missing
+		 * sequences.
+		 */
+		if ((batch_succeeded_count + batch_mismatched_count) < batch_size)
+		{
+			for (int i = 0; i < batch_size; i++)
+			{
+				LogicalRepSequenceInfo *seqinfo = lfirst(list_nth_cell(remotesequences, current_index + i));
+
+				if (!seqinfo->remote_seq_fetched)
+				{
+					if (missing_seqs->len)
+						appendStringInfoString(missing_seqs, ", ");
+
+					appendStringInfo(missing_seqs, "\"%s.%s\"",
+									 seqinfo->nspname, seqinfo->seqname);
+				}
+			}
+		}
+
+		/*
+		 * current_indexes is not incremented sequentially because some
+		 * sequences may be missing, and the number of fetched rows may not
+		 * match the batch size.
+		 */
+		current_index += batch_size;
+	}
+
+	/*
+	 * Raise an error if any sequences are missing on the remote server, or if
+	 * the local and remote sequence parameters do not match.
+	 */
+	if (missing_seqs->len || mismatched_seqs->len)
+		report_error_sequences(missing_seqs, mismatched_seqs);
+
+	destroyStringInfo(missing_seqs);
+	destroyStringInfo(mismatched_seqs);
+}
+
+/*
+ * Start syncing the sequences in the sequencesync worker.
+ */
+static void
+LogicalRepSyncSequences(void)
+{
+	char	   *err;
+	bool		must_use_password;
+	List	   *sequences;
+	List	   *sequences_not_synced = NIL;
+	AclResult	aclresult;
+	UserContext ucxt;
+	bool		run_as_owner = false;
+	int			seq_count;
+	Oid			subid = MyLogicalRepWorker->subid;
+	MemoryContext oldctx;
+	StringInfoData app_name;
+	List	   *remotesequences = NIL;
+	char	   *nspname;
+	char	   *seqname;
+	LogicalRepSequenceInfo *seq_info;
+
+	StartTransactionCommand();
+
+	/* Get the sequences that should be synchronized. */
+	sequences = GetSubscriptionRelations(subid, false, true, true);
+
+	/* Allocate the tracking info in a permanent memory context. */
+	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+	foreach_ptr(SubscriptionRelState, seq_state, sequences)
+	{
+		SubscriptionRelState *rstate = palloc(sizeof(SubscriptionRelState));
+
+		memcpy(rstate, seq_state, sizeof(SubscriptionRelState));
+		sequences_not_synced = lappend(sequences_not_synced, rstate);
+	}
+	MemoryContextSwitchTo(oldctx);
+
+	CommitTransactionCommand();
+
+	/* Is the use of a password mandatory? */
+	must_use_password = MySubscription->passwordrequired &&
+		!MySubscription->ownersuperuser;
+
+	initStringInfo(&app_name);
+	appendStringInfo(&app_name, "%s_%s", MySubscription->name, "sequencesync worker");
+
+	/*
+	 * Establish the connection to the publisher for sequence synchronization.
+	 */
+	LogRepWorkerWalRcvConn =
+		walrcv_connect(MySubscription->conninfo, true, true,
+					   must_use_password,
+					   app_name.data, &err);
+	if (LogRepWorkerWalRcvConn == NULL)
+		ereport(ERROR,
+				errcode(ERRCODE_CONNECTION_FAILURE),
+				errmsg("could not connect to the publisher: %s", err));
+
+	pfree(app_name.data);
+
+	seq_count = list_length(sequences_not_synced);
+
+	StartTransactionCommand();
+	foreach_ptr(SubscriptionRelState, seqinfo, sequences_not_synced)
+	{
+		Relation	sequence_rel;
+
+		CHECK_FOR_INTERRUPTS();
+
+		sequence_rel = table_open(seqinfo->relid, RowExclusiveLock);
+		seqname = RelationGetRelationName(sequence_rel);
+		nspname = get_namespace_name(RelationGetNamespace(sequence_rel));
+
+		/*
+		 * Make sure that the copy command runs as the sequence owner, unless
+		 * the user has opted out of that behaviour.
+		 */
+		run_as_owner = MySubscription->runasowner;
+		if (!run_as_owner)
+			SwitchToUntrustedUser(sequence_rel->rd_rel->relowner, &ucxt);
+
+		/*
+		 * Check that our sequencesync worker has permission to insert into
+		 * the target sequence.
+		 */
+		aclresult = pg_class_aclcheck(RelationGetRelid(sequence_rel), GetUserId(),
+									  ACL_INSERT);
+		if (aclresult != ACLCHECK_OK)
+			aclcheck_error(aclresult,
+						   get_relkind_objtype(sequence_rel->rd_rel->relkind),
+						   seqname);
+
+		/* Allocate the tracking info in a permanent memory context. */
+		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+
+		seq_info = (LogicalRepSequenceInfo *) palloc(sizeof(LogicalRepSequenceInfo));
+		seq_info->seqname = pstrdup(seqname);
+		seq_info->nspname = pstrdup(nspname);
+		seq_info->localrelid = seqinfo->relid;
+		seq_info->remote_seq_fetched = false;
+		remotesequences = lappend(remotesequences, seq_info);
+
+		MemoryContextSwitchTo(oldctx);
+
+		table_close(sequence_rel, NoLock);
+	}
+
+	CommitTransactionCommand();
+
+	copy_sequences(LogRepWorkerWalRcvConn, remotesequences, subid);
+
+	list_free_deep(sequences_not_synced);
+	if (!run_as_owner && seq_count)
+		RestoreUserContext(&ucxt);
+}
+
+/*
+ * Execute the initial sync with error handling. Disable the subscription,
+ * if required.
+ *
+ * Allocate the slot name in long-lived context on return. Note that we don't
+ * handle FATAL errors which are probably because of system resource error and
+ * are not repeatable.
+ */
+static void
+start_sequence_sync()
+{
+	Assert(am_sequencesync_worker());
+
+	PG_TRY();
+	{
+		/* Call initial sync. */
+		LogicalRepSyncSequences();
+	}
+	PG_CATCH();
+	{
+		if (MySubscription->disableonerr)
+			DisableSubscriptionAndExit();
+		else
+		{
+			/*
+			 * Report the worker failed during sequence synchronization. Abort
+			 * the current transaction so that the stats message is sent in an
+			 * idle state.
+			 */
+			AbortOutOfAnyTransaction();
+			pgstat_report_subscription_error(MySubscription->oid,
+											 WORKERTYPE_SEQUENCESYNC);
+
+			PG_RE_THROW();
+		}
+	}
+	PG_END_TRY();
+}
+
+/* Logical Replication sequencesync worker entry point */
+void
+SequenceSyncWorkerMain(Datum main_arg)
+{
+	int			worker_slot = DatumGetInt32(main_arg);
+
+	SetupApplyOrSyncWorker(worker_slot);
+
+	start_sequence_sync();
+
+	FinishSyncWorker(WORKERTYPE_SEQUENCESYNC);
+}
diff --git a/src/backend/replication/logical/syncutils.c b/src/backend/replication/logical/syncutils.c
index db15051f47b..5f5770a3908 100644
--- a/src/backend/replication/logical/syncutils.c
+++ b/src/backend/replication/logical/syncutils.c
@@ -50,8 +50,10 @@ static SyncingRelationsState relation_states_validity = SYNC_RELATIONS_STATE_NEE
  * Exit routine for synchronization worker.
  */
 pg_noreturn void
-FinishSyncWorker(void)
+FinishSyncWorker(LogicalRepWorkerType wtype)
 {
+	Assert(wtype == WORKERTYPE_TABLESYNC || wtype == WORKERTYPE_SEQUENCESYNC);
+
 	/*
 	 * Commit any outstanding transaction. This is the usual case, unless
 	 * there was nothing to do for the table.
@@ -66,15 +68,24 @@ FinishSyncWorker(void)
 	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))));
+	if (wtype == WORKERTYPE_TABLESYNC)
+		ereport(LOG,
+				errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
+					   MySubscription->name,
+					   get_rel_name(MyLogicalRepWorker->relid)));
+	else
+		ereport(LOG,
+				errmsg("logical replication sequence synchronization worker for subscription \"%s\" has finished",
+					   MySubscription->name));
 	CommitTransactionCommand();
 
 	/* Find the leader apply worker and signal it. */
 	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
 
+	/* This is a clean exit, so no need for any sequence failure logic. */
+	if (wtype == WORKERTYPE_SEQUENCESYNC)
+		cancel_before_shmem_exit(logicalrep_seqsyncworker_failure, 0);
+
 	/* Stop gracefully */
 	proc_exit(0);
 }
@@ -89,7 +100,9 @@ SyncInvalidateRelationStates(Datum arg, int cacheid, uint32 hashvalue)
 }
 
 /*
- * Process possible state change(s) of relations that are being synchronized.
+ * Process possible state change(s) of relations that are being synchronized
+ * and start new tablesync workers for the newly added tables. Also, start a
+ * new sequencesync worker for the newly added sequences.
  */
 void
 ProcessSyncingRelations(XLogRecPtr current_lsn)
@@ -109,7 +122,19 @@ ProcessSyncingRelations(XLogRecPtr current_lsn)
 			break;
 
 		case WORKERTYPE_APPLY:
+			/*
+			 * We need up-to-date sync state info for subscription tables and
+			 * sequences here.
+			 */
+			SyncFetchRelationStates();
+
 			ProcessSyncingTablesForApply(current_lsn);
+			ProcessSyncingSequencesForApply();
+			break;
+
+		case WORKERTYPE_SEQUENCESYNC:
+			/* Should never happen. */
+			Assert(0);
 			break;
 
 		case WORKERTYPE_UNKNOWN:
@@ -121,17 +146,22 @@ ProcessSyncingRelations(XLogRecPtr current_lsn)
 /*
  * 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.
+ * The pg_subscription_rel catalog is shared by tables and sequences. Changes
+ * to either sequences or tables can affect the validity of relation states, so
+ * we update both table_states_not_ready and sequence_states_not_ready
+ * simultaneously to ensure consistency.
  *
- * Note: If this function started the transaction (indicated by the parameter)
- * then it is the caller's responsibility to commit it.
+ * Returns true if subscription has 1 or more tables, else false.
  */
 bool
-SyncFetchRelationStates(bool *started_tx)
+SyncFetchRelationStates()
 {
+	/*
+	 * has_subtables is declared as static, since the same value can be used
+	 * until the system table is invalidated.
+	 */
 	static bool has_subtables = false;
-
-	*started_tx = false;
+	bool		started_tx = false;
 
 	if (relation_states_validity != SYNC_RELATIONS_STATE_VALID)
 	{
@@ -144,12 +174,14 @@ SyncFetchRelationStates(bool *started_tx)
 
 		/* Clean the old lists. */
 		list_free_deep(table_states_not_ready);
+		list_free_deep(sequence_states_not_ready);
 		table_states_not_ready = NIL;
+		sequence_states_not_ready = NIL;
 
 		if (!IsTransactionState())
 		{
 			StartTransactionCommand();
-			*started_tx = true;
+			started_tx = true;
 		}
 
 		/* Fetch tables and sequences that are in non-ready state. */
@@ -162,7 +194,11 @@ SyncFetchRelationStates(bool *started_tx)
 		{
 			rstate = palloc(sizeof(SubscriptionRelState));
 			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
-			table_states_not_ready = lappend(table_states_not_ready, rstate);
+
+			if (get_rel_relkind(rstate->relid) == RELKIND_SEQUENCE)
+				sequence_states_not_ready = lappend(sequence_states_not_ready, rstate);
+			else
+				table_states_not_ready = lappend(table_states_not_ready, rstate);
 		}
 		MemoryContextSwitchTo(oldctx);
 
@@ -187,5 +223,11 @@ SyncFetchRelationStates(bool *started_tx)
 			relation_states_validity = SYNC_RELATIONS_STATE_VALID;
 	}
 
+	if (started_tx)
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(true);
+	}
+
 	return has_subtables;
 }
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f6111f94340..bea492d4b38 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -161,7 +161,7 @@ wait_for_table_state_change(Oid relid, char expected_state)
 		/* Check if the sync worker is still running and bail if not. */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 		worker = logicalrep_worker_find(MyLogicalRepWorker->subid, relid,
-										false);
+										WORKERTYPE_TABLESYNC, false);
 		LWLockRelease(LogicalRepWorkerLock);
 		if (!worker)
 			break;
@@ -208,7 +208,7 @@ wait_for_worker_state_change(char expected_state)
 		 */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 		worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-										InvalidOid, false);
+										InvalidOid, WORKERTYPE_APPLY, false);
 		if (worker && worker->proc)
 			logicalrep_worker_wakeup_ptr(worker);
 		LWLockRelease(LogicalRepWorkerLock);
@@ -334,7 +334,7 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn)
 		 */
 		replorigin_drop_by_name(originname, true, false);
 
-		FinishSyncWorker();
+		FinishSyncWorker(WORKERTYPE_TABLESYNC);
 	}
 	else
 		SpinLockRelease(&MyLogicalRepWorker->relmutex);
@@ -376,9 +376,6 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
 
 	Assert(!IsTransactionState());
 
-	/* We need up-to-date sync state info for subscription tables here. */
-	SyncFetchRelationStates(&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
@@ -411,6 +408,14 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
 	{
 		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 
+		if (!started_tx)
+		{
+			StartTransactionCommand();
+			started_tx = true;
+		}
+
+		Assert(get_rel_relkind(rstate->relid) != RELKIND_SEQUENCE);
+
 		if (rstate->state == SUBREL_STATE_SYNCDONE)
 		{
 			/*
@@ -424,11 +429,6 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
 
 				rstate->state = SUBREL_STATE_READY;
 				rstate->lsn = current_lsn;
-				if (!started_tx)
-				{
-					StartTransactionCommand();
-					started_tx = true;
-				}
 
 				/*
 				 * Remove the tablesync origin tracking if exists.
@@ -465,8 +465,8 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-												rstate->relid, false);
-
+												rstate->relid,
+												WORKERTYPE_TABLESYNC, true);
 			if (syncworker)
 			{
 				/* Found one, update our copy of its state */
@@ -1243,7 +1243,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 		case SUBREL_STATE_SYNCDONE:
 		case SUBREL_STATE_READY:
 		case SUBREL_STATE_UNKNOWN:
-			FinishSyncWorker(); /* doesn't return */
+			FinishSyncWorker(WORKERTYPE_TABLESYNC); /* doesn't return */
 	}
 
 	/* Calculate the name of the tablesync slot. */
@@ -1516,7 +1516,8 @@ start_table_sync(XLogRecPtr *origin_startpos, char **slotname)
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
-			pgstat_report_subscription_error(MySubscription->oid, false);
+			pgstat_report_subscription_error(MySubscription->oid,
+											 WORKERTYPE_TABLESYNC);
 
 			PG_RE_THROW();
 		}
@@ -1561,7 +1562,7 @@ run_tablesync_worker()
 
 /* Logical Replication Tablesync worker entry point */
 void
-TablesyncWorkerMain(Datum main_arg)
+TableSyncWorkerMain(Datum main_arg)
 {
 	int			worker_slot = DatumGetInt32(main_arg);
 
@@ -1569,7 +1570,7 @@ TablesyncWorkerMain(Datum main_arg)
 
 	run_tablesync_worker();
 
-	FinishSyncWorker();
+	FinishSyncWorker(WORKERTYPE_TABLESYNC);
 }
 
 /*
@@ -1583,23 +1584,16 @@ TablesyncWorkerMain(Datum main_arg)
 bool
 AllTablesyncsReady(void)
 {
-	bool		started_tx = false;
-	bool		has_subrels = false;
+	bool		has_tables;
 
 	/* We need up-to-date sync state info for subscription tables here. */
-	has_subrels = SyncFetchRelationStates(&started_tx);
-
-	if (started_tx)
-	{
-		CommitTransactionCommand();
-		pgstat_report_stat(true);
-	}
+	has_tables = SyncFetchRelationStates();
 
 	/*
 	 * 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);
+	return has_tables && (table_states_not_ready == NIL);
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f730578c219..c8e38973fc5 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -489,6 +489,11 @@ should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 					(rel->state == SUBREL_STATE_SYNCDONE &&
 					 rel->statelsn <= remote_final_lsn));
 
+		case WORKERTYPE_SEQUENCESYNC:
+			/* Should never happen. */
+			Assert(0);
+			break;
+
 		case WORKERTYPE_UNKNOWN:
 			/* Should never happen. */
 			elog(ERROR, "Unknown worker type");
@@ -1029,7 +1034,10 @@ apply_handle_commit(StringInfo s)
 
 	apply_handle_commit_internal(&commit_data);
 
-	/* Process any tables that are being synchronized in parallel. */
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
 	ProcessSyncingRelations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
@@ -1151,7 +1159,10 @@ apply_handle_prepare(StringInfo s)
 
 	in_remote_transaction = false;
 
-	/* Process any tables that are being synchronized in parallel. */
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
 	ProcessSyncingRelations(prepare_data.end_lsn);
 
 	/*
@@ -1207,7 +1218,10 @@ apply_handle_commit_prepared(StringInfo s)
 	store_flush_position(prepare_data.end_lsn, XactLastCommitEnd);
 	in_remote_transaction = false;
 
-	/* Process any tables that are being synchronized in parallel. */
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
 	ProcessSyncingRelations(prepare_data.end_lsn);
 
 	clear_subscription_skip_lsn(prepare_data.end_lsn);
@@ -1273,7 +1287,10 @@ apply_handle_rollback_prepared(StringInfo s)
 	store_flush_position(rollback_data.rollback_end_lsn, InvalidXLogRecPtr);
 	in_remote_transaction = false;
 
-	/* Process any tables that are being synchronized in parallel. */
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
 	ProcessSyncingRelations(rollback_data.rollback_end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
@@ -1408,7 +1425,10 @@ apply_handle_stream_prepare(StringInfo s)
 
 	pgstat_report_stat(false);
 
-	/* Process any tables that are being synchronized in parallel. */
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
 	ProcessSyncingRelations(prepare_data.end_lsn);
 
 	/*
@@ -2250,7 +2270,10 @@ apply_handle_stream_commit(StringInfo s)
 			break;
 	}
 
-	/* Process any tables that are being synchronized in parallel. */
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
 	ProcessSyncingRelations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
@@ -3727,7 +3750,10 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 			AcceptInvalidationMessages();
 			maybe_reread_subscription();
 
-			/* Process any table synchronization changes. */
+			/*
+			 * Process any tables that are being synchronized in parallel and
+			 * any newly added relations.
+			 */
 			ProcessSyncingRelations(last_received);
 		}
 
@@ -4536,7 +4562,8 @@ start_apply(XLogRecPtr origin_startpos)
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
-			pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+			pgstat_report_subscription_error(MySubscription->oid,
+											 WORKERTYPE_APPLY);
 
 			PG_RE_THROW();
 		}
@@ -4656,8 +4683,8 @@ run_apply_worker()
 }
 
 /*
- * Common initialization for leader apply worker, parallel apply worker and
- * tablesync worker.
+ * Common initialization for leader apply worker, parallel apply worker,
+ * tablesync worker and sequencesync worker.
  *
  * Initialize the database connection, in-memory subscription and necessary
  * config options.
@@ -4736,6 +4763,10 @@ InitializeLogRepWorker(void)
 				(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
 						MySubscription->name,
 						get_rel_name(MyLogicalRepWorker->relid))));
+	else if (am_sequencesync_worker())
+		ereport(LOG,
+				(errmsg("logical replication sequence synchronization worker for subscription \"%s\" has started",
+						MySubscription->name)));
 	else
 		ereport(LOG,
 				(errmsg("logical replication apply worker for subscription \"%s\" has started",
@@ -4755,14 +4786,17 @@ replorigin_reset(int code, Datum arg)
 	replorigin_session_origin_timestamp = 0;
 }
 
-/* Common function to setup the leader apply or tablesync worker. */
+/*
+ * Common function to setup the leader apply, tablesync worker and sequencesync
+ * worker.
+ */
 void
 SetupApplyOrSyncWorker(int worker_slot)
 {
 	/* Attach to slot */
 	logicalrep_worker_attach(worker_slot);
 
-	Assert(am_tablesync_worker() || am_leader_apply_worker());
+	Assert(am_tablesync_worker() || am_sequencesync_worker() || am_leader_apply_worker());
 
 	/* Setup signal handling */
 	pqsignal(SIGHUP, SignalHandlerForConfigReload);
@@ -4807,6 +4841,9 @@ SetupApplyOrSyncWorker(int worker_slot)
 	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
 								  SyncInvalidateRelationStates,
 								  (Datum) 0);
+
+	if (am_sequencesync_worker())
+		before_shmem_exit(logicalrep_seqsyncworker_failure, (Datum) 0);
 }
 
 /* Logical Replication Apply worker entry point */
@@ -4833,6 +4870,10 @@ ApplyWorkerMain(Datum main_arg)
 void
 DisableSubscriptionAndExit(void)
 {
+	LogicalRepWorkerType wtype = am_tablesync_worker() ? WORKERTYPE_TABLESYNC :
+		(am_sequencesync_worker()) ? WORKERTYPE_SEQUENCESYNC :
+		WORKERTYPE_APPLY;
+
 	/*
 	 * Emit the error message, and recover from the error state to an idle
 	 * state
@@ -4845,9 +4886,11 @@ DisableSubscriptionAndExit(void)
 
 	RESUME_INTERRUPTS();
 
-	/* Report the worker failed during either table synchronization or apply */
-	pgstat_report_subscription_error(MyLogicalRepWorker->subid,
-									 !am_tablesync_worker());
+	/*
+	 * Report the worker failed during either sequence synchronization or
+	 * table synchronization or apply.
+	 */
+	pgstat_report_subscription_error(MyLogicalRepWorker->subid, wtype);
 
 	/* Disable the subscription */
 	StartTransactionCommand();
diff --git a/src/backend/utils/activity/pgstat_subscription.c b/src/backend/utils/activity/pgstat_subscription.c
index f9a1c831a07..002d630d4ae 100644
--- a/src/backend/utils/activity/pgstat_subscription.c
+++ b/src/backend/utils/activity/pgstat_subscription.c
@@ -17,6 +17,7 @@
 
 #include "postgres.h"
 
+#include "replication/worker_internal.h"
 #include "utils/pgstat_internal.h"
 
 
@@ -24,7 +25,7 @@
  * Report a subscription error.
  */
 void
-pgstat_report_subscription_error(Oid subid, bool is_apply_error)
+pgstat_report_subscription_error(Oid subid, LogicalRepWorkerType wtype)
 {
 	PgStat_EntryRef *entry_ref;
 	PgStat_BackendSubEntry *pending;
@@ -33,10 +34,25 @@ pgstat_report_subscription_error(Oid subid, bool is_apply_error)
 										  InvalidOid, subid, NULL);
 	pending = entry_ref->pending;
 
-	if (is_apply_error)
-		pending->apply_error_count++;
-	else
-		pending->sync_error_count++;
+	switch (wtype)
+	{
+		case WORKERTYPE_APPLY:
+			pending->apply_error_count++;
+			break;
+
+		case WORKERTYPE_SEQUENCESYNC:
+			pending->sequence_sync_error_count++;
+			break;
+
+		case WORKERTYPE_TABLESYNC:
+			pending->sync_error_count++;
+			break;
+
+		default:
+			/* Should never happen. */
+			Assert(0);
+			break;
+	}
 }
 
 /*
@@ -115,6 +131,7 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
 
 #define SUB_ACC(fld) shsubent->stats.fld += localent->fld
 	SUB_ACC(apply_error_count);
+	SUB_ACC(sequence_sync_error_count);
 	SUB_ACC(sync_error_count);
 	for (int i = 0; i < CONFLICT_NUM_TYPES; i++)
 		SUB_ACC(conflict_count[i]);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 1c12ddbae49..ab061d0ba9b 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -2171,7 +2171,7 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS	11
+#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS	12
 	Oid			subid = PG_GETARG_OID(0);
 	TupleDesc	tupdesc;
 	Datum		values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0};
@@ -2189,23 +2189,25 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 					   OIDOID, -1, 0);
 	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "apply_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sequence_sync_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "confl_insert_exists",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "sync_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_update_origin_differs",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_insert_exists",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_exists",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_origin_differs",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_missing",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_exists",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_delete_origin_differs",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_update_missing",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_delete_missing",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_delete_origin_differs",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_multiple_unique_conflicts",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_delete_missing",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "stats_reset",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "confl_multiple_unique_conflicts",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 12, "stats_reset",
 					   TIMESTAMPTZOID, -1, 0);
 	BlessTupleDesc(tupdesc);
 
@@ -2222,6 +2224,9 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 	/* apply_error_count */
 	values[i++] = Int64GetDatum(subentry->apply_error_count);
 
+	/* sequence_sync_error_count */
+	values[i++] = Int64GetDatum(subentry->sequence_sync_error_count);
+
 	/* sync_error_count */
 	values[i++] = Int64GetDatum(subentry->sync_error_count);
 
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index f04bfedb2fd..ff424159e54 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -3356,7 +3356,7 @@ struct config_int ConfigureNamesInt[] =
 		{"max_sync_workers_per_subscription",
 			PGC_SIGHUP,
 			REPLICATION_SUBSCRIBERS,
-			gettext_noop("Maximum number of table synchronization workers per subscription."),
+			gettext_noop("Maximum number of workers per subscription for synchronizing tables and sequences."),
 			NULL,
 		},
 		&max_sync_workers_per_subscription,
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fa824499fa2..8eb7b0c061c 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5695,9 +5695,9 @@
 { oid => '6231', descr => 'statistics: information about subscription stats',
   proname => 'pg_stat_get_subscription_stats', provolatile => 's',
   proparallel => 'r', prorettype => 'record', proargtypes => 'oid',
-  proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{subid,subid,apply_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}',
+  proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{subid,subid,apply_error_count,sequence_sync_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}',
   prosrc => 'pg_stat_get_subscription_stats' },
 { oid => '6118', descr => 'statistics: information about subscription',
   proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index a541f4843bd..49af743b20d 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -82,6 +82,14 @@ typedef struct SubscriptionRelState
 	char		state;
 } SubscriptionRelState;
 
+typedef struct LogicalRepSequenceInfo
+{
+	char	   *seqname;
+	char	   *nspname;
+	Oid			localrelid;
+	bool		remote_seq_fetched;
+} LogicalRepSequenceInfo;
+
 extern void AddSubscriptionRelState(Oid subid, Oid relid, char state,
 									XLogRecPtr sublsn, bool retain_lock);
 extern void UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 9ac0b67683d..3aec610028f 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -45,6 +45,8 @@ typedef FormData_pg_sequence_data *Form_pg_sequence_data;
 /* XLOG stuff */
 #define XLOG_SEQ_LOG			0x00
 
+#define SEQ_LOG_CNT_INVALID		0
+
 typedef struct xl_seq_rec
 {
 	RelFileLocator locator;
@@ -60,6 +62,7 @@ extern ObjectAddress AlterSequence(ParseState *pstate, AlterSeqStmt *stmt);
 extern void SequenceChangePersistence(Oid relid, char newrelpersistence);
 extern void DeleteSequenceTuple(Oid relid);
 extern void ResetSequence(Oid seq_relid);
+extern void SetSequence(Oid relid, int64 next, int64 log_cnt, bool is_called);
 extern void ResetSequenceCaches(void);
 
 extern void seq_redo(XLogReaderState *record);
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 378f2f2c2ba..03d4df572f4 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -15,6 +15,7 @@
 #include "portability/instr_time.h"
 #include "postmaster/pgarch.h"	/* for MAX_XFN_CHARS */
 #include "replication/conflict.h"
+#include "replication/worker_internal.h"
 #include "utils/backend_progress.h" /* for backward compatibility */	/* IWYU pragma: export */
 #include "utils/backend_status.h"	/* for backward compatibility */	/* IWYU pragma: export */
 #include "utils/pgstat_kind.h"
@@ -107,6 +108,7 @@ typedef struct PgStat_FunctionCallUsage
 typedef struct PgStat_BackendSubEntry
 {
 	PgStat_Counter apply_error_count;
+	PgStat_Counter sequence_sync_error_count;
 	PgStat_Counter sync_error_count;
 	PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
 } PgStat_BackendSubEntry;
@@ -413,6 +415,7 @@ typedef struct PgStat_SLRUStats
 typedef struct PgStat_StatSubEntry
 {
 	PgStat_Counter apply_error_count;
+	PgStat_Counter sequence_sync_error_count;
 	PgStat_Counter sync_error_count;
 	PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
 	TimestampTz stat_reset_timestamp;
@@ -763,7 +766,8 @@ extern PgStat_SLRUStats *pgstat_fetch_slru(void);
  * Functions in pgstat_subscription.c
  */
 
-extern void pgstat_report_subscription_error(Oid subid, bool is_apply_error);
+extern void pgstat_report_subscription_error(Oid subid,
+											 LogicalRepWorkerType wtype);
 extern void pgstat_report_subscription_conflict(Oid subid, ConflictType type);
 extern void pgstat_create_subscription(Oid subid);
 extern void pgstat_drop_subscription(Oid subid);
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 88912606e4d..56fa79b648e 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -18,7 +18,8 @@ extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending;
 
 extern void ApplyWorkerMain(Datum main_arg);
 extern void ParallelApplyWorkerMain(Datum main_arg);
-extern void TablesyncWorkerMain(Datum main_arg);
+extern void TableSyncWorkerMain(Datum main_arg);
+extern void SequenceSyncWorkerMain(Datum main_arg);
 
 extern bool IsLogicalWorker(void);
 extern bool IsLogicalParallelApplyWorker(void);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 5394fbc4afe..7590df29910 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -30,6 +30,7 @@ typedef enum LogicalRepWorkerType
 {
 	WORKERTYPE_UNKNOWN = 0,
 	WORKERTYPE_TABLESYNC,
+	WORKERTYPE_SEQUENCESYNC,
 	WORKERTYPE_APPLY,
 	WORKERTYPE_PARALLEL_APPLY,
 } LogicalRepWorkerType;
@@ -92,6 +93,8 @@ typedef struct LogicalRepWorker
 	TimestampTz last_recv_time;
 	XLogRecPtr	reply_lsn;
 	TimestampTz reply_time;
+
+	TimestampTz sequencesync_failure_time;
 } LogicalRepWorker;
 
 /*
@@ -238,9 +241,11 @@ extern PGDLLIMPORT bool in_remote_transaction;
 extern PGDLLIMPORT bool InitializingApplyWorker;
 
 extern PGDLLIMPORT List *table_states_not_ready;
+extern PGDLLIMPORT List *sequence_states_not_ready;
 
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
+												LogicalRepWorkerType wtype,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running,
 									 bool acquire_lock);
@@ -248,13 +253,17 @@ 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);
+extern void logicalrep_worker_stop(Oid subid, Oid relid,
+								   LogicalRepWorkerType wtype);
 extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
 extern int	logicalrep_sync_worker_count(Oid subid);
 
+extern void logicalrep_seqsyncworker_set_failuretime(void);
+extern void logicalrep_seqsyncworker_failure(int code, Datum arg);
+
 extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 											   char *originname, Size szoriginname);
 
@@ -263,12 +272,13 @@ extern void UpdateTwoPhaseState(Oid suboid, char new_state);
 
 extern void ProcessSyncingTablesForSync(XLogRecPtr current_lsn);
 extern void ProcessSyncingTablesForApply(XLogRecPtr current_lsn);
+extern void ProcessSyncingSequencesForApply(void);
 
-pg_noreturn extern void FinishSyncWorker(void);
+pg_noreturn extern void FinishSyncWorker(LogicalRepWorkerType wtype);
 extern void SyncInvalidateRelationStates(Datum arg, int cacheid,
 										 uint32 hashvalue);
 extern void ProcessSyncingRelations(XLogRecPtr current_lsn);
-extern bool SyncFetchRelationStates(bool *started_tx);
+extern bool SyncFetchRelationStates(void);
 
 extern void stream_start_internal(TransactionId xid, bool first_segment);
 extern void stream_stop_internal(TransactionId xid);
@@ -333,15 +343,25 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
+#define isApplyWorker(worker) ((worker)->in_use && \
+							   (worker)->type == WORKERTYPE_APPLY)
 #define isParallelApplyWorker(worker) ((worker)->in_use && \
 									   (worker)->type == WORKERTYPE_PARALLEL_APPLY)
-#define isTablesyncWorker(worker) ((worker)->in_use && \
+#define isTableSyncWorker(worker) ((worker)->in_use && \
 								   (worker)->type == WORKERTYPE_TABLESYNC)
+#define isSequenceSyncWorker(worker) ((worker)->in_use && \
+									  (worker)->type == WORKERTYPE_SEQUENCESYNC)
 
 static inline bool
 am_tablesync_worker(void)
 {
-	return isTablesyncWorker(MyLogicalRepWorker);
+	return isTableSyncWorker(MyLogicalRepWorker);
+}
+
+static inline bool
+am_sequencesync_worker(void)
+{
+	return isSequenceSyncWorker(MyLogicalRepWorker);
 }
 
 static inline bool
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index 586ffba434e..a6c267a8a2c 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -42,6 +42,7 @@ tests += {
       't/033_run_as_table_owner.pl',
       't/034_temporal.pl',
       't/035_conflicts.pl',
+      't/036_sequences.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/036_sequences.pl b/src/test/subscription/t/036_sequences.pl
new file mode 100644
index 00000000000..979756894b7
--- /dev/null
+++ b/src/test/subscription/t/036_sequences.pl
@@ -0,0 +1,226 @@
+
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# This tests that sequences are synced correctly to the subscriber
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+
+# Avoid checkpoint during the test, otherwise, extra values will be fetched for
+# the sequences which will cause the test to fail randomly.
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf', 'checkpoint_timeout = 1h');
+$node_publisher->start;
+
+# Initialize subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Setup structure on the publisher
+my $ddl = qq(
+	CREATE TABLE regress_seq_test (v BIGINT);
+	CREATE SEQUENCE regress_s1;
+);
+$node_publisher->safe_psql('postgres', $ddl);
+
+# Setup the same structure on the subscriber, plus some extra sequences that
+# we'll create on the publisher later
+$ddl = qq(
+	CREATE TABLE regress_seq_test (v BIGINT);
+	CREATE SEQUENCE regress_s1;
+	CREATE SEQUENCE regress_s2;
+	CREATE SEQUENCE regress_s3;
+	CREATE SEQUENCE regress_s4
+);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Insert initial test data
+$node_publisher->safe_psql(
+	'postgres', qq(
+	-- generate a number of values using the sequence
+	INSERT INTO regress_seq_test SELECT nextval('regress_s1') FROM generate_series(1,100);
+));
+
+# Setup logical replication pub/sub
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION regress_seq_pub FOR ALL SEQUENCES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION regress_seq_sub CONNECTION '$publisher_connstr' PUBLICATION regress_seq_pub"
+);
+
+# Wait for initial sync to finish
+my $synced_query =
+  "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check the initial data on subscriber
+my $result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s1;
+));
+is($result, '100|32|t', 'initial test data replicated');
+
+##########
+## ALTER SUBSCRIPTION ... REFRESH PUBLICATION should cause sync of new
+# sequences of the publisher, but changes to existing sequences should
+# not be synced.
+##########
+
+# Create a new sequence 'regress_s2', and update existing sequence 'regress_s1'
+$node_publisher->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s2;
+	INSERT INTO regress_seq_test SELECT nextval('regress_s2') FROM generate_series(1,100);
+
+    -- Existing sequence
+	INSERT INTO regress_seq_test SELECT nextval('regress_s1') FROM generate_series(1,100);
+));
+
+# Do ALTER SUBSCRIPTION ... REFRESH PUBLICATION
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION
+));
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+$result = $node_publisher->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s1;
+));
+is($result, '200|31|t', 'Check sequence value in the publisher');
+
+# Check - existing sequence is not synced
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s1;
+));
+is($result, '100|32|t',
+	'REFRESH PUBLICATION does not sync existing sequence');
+
+# Check - newly published sequence is synced
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s2;
+));
+is($result, '100|32|t',
+	'REFRESH PUBLICATION will sync newly published sequence');
+
+##########
+## ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES should cause sync of
+# new sequences of the publisher, and changes to existing sequences should
+# also be synced.
+##########
+
+# Create a new sequence 'regress_s3', and update the existing sequence
+# 'regress_s2'.
+$node_publisher->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s3;
+	INSERT INTO regress_seq_test SELECT nextval('regress_s3') FROM generate_series(1,100);
+
+	-- Existing sequence
+	INSERT INTO regress_seq_test SELECT nextval('regress_s2') FROM generate_series(1,100);
+));
+
+# Do ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION SEQUENCES
+));
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check - existing sequences are synced
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s1;
+));
+is($result, '200|31|t',
+	'REFRESH PUBLICATION SEQUENCES will sync existing sequences');
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s2;
+));
+is($result, '200|31|t',
+	'REFRESH PUBLICATION SEQUENCES will sync existing sequences');
+
+# Check - newly published sequence is synced
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s3;
+));
+is($result, '100|32|t',
+	'REFRESH PUBLICATION SEQUENCES will sync newly published sequence');
+
+##########
+## ALTER SUBSCRIPTION ... REFRESH PUBLICATION with (copy_data = off) should
+# not update the sequence values for the new sequence.
+##########
+
+# Create a new sequence 'regress_s4'
+$node_publisher->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s4;
+	INSERT INTO regress_seq_test SELECT nextval('regress_s4') FROM generate_series(1,100);
+));
+
+# Do ALTER SUBSCRIPTION ... REFRESH PUBLICATION
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION with (copy_data = false);
+));
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+$result = $node_publisher->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s4;
+));
+is($result, '100|32|t', 'Check sequence value in the publisher');
+
+# Check - newly published sequence values are not updated
+$result = $node_subscriber->safe_psql(
+	'postgres', qq(
+	SELECT last_value, log_cnt, is_called FROM regress_s4;
+));
+is($result, '1|0|f',
+	'REFRESH PUBLICATION will not sync newly published sequence with copy_data as off');
+
+##########
+# ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES should throw an error
+# for sequence definition not matching between the publisher and the subscriber.
+##########
+
+# Create a new sequence 'regress_s5' whose START value is not the same in the
+# publisher and subscriber.
+$node_publisher->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s5 START 1 INCREMENT 2;
+));
+
+$node_subscriber->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s5 START 10 INCREMENT 2;
+));
+
+my $log_offset = -s $node_subscriber->logfile;
+
+# Do ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES
+$node_subscriber->safe_psql(
+	'postgres', "
+    ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION SEQUENCES"
+);
+
+# Confirm that the warning for parameters differing is logged.
+$node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? logical replication sequence synchronization failed for subscription "regress_seq_sub": parameters differ for the remote and local sequences \("public.regress_s5"\)/,
+	$log_offset);
+done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 36af636a4c0..4f0d87b8c00 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1623,6 +1623,7 @@ LogicalRepRelId
 LogicalRepRelMapEntry
 LogicalRepRelation
 LogicalRepRollbackPreparedTxnData
+LogicalRepSequenceInfo
 LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
-- 
2.34.1

