v20240805_2-0003-Enhance-sequence-synchronization-during-.patch

From 9a6060feabf73c9c002c56506e66caadb9b65f77 Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Mon, 5 Aug 2024 12:18:31 +0530
Subject: [PATCH v20240805_2 3/3] Enhance sequence synchronization during
 subscription management

This commit introduces sequence synchronization:
1) During subscription creation:
   - The subscriber retrieves sequences associated with publications.
   - Sequences  are added in 'init' state to pg_subscription_rel table.
   - A new sequence synchronization worker handles synchronization in
     batches of 100 sequences:
     a) Retrieves sequence values using pg_sequence_state from the publisher.
     b) Sets sequence values accordingly.
     c) Updates sequence state to 'READY'.
     d) Commits batches of 100 synchronized sequences.

2) Refreshing sequences:
   - Refreshing sequences occurs with
        ALTER SUBSCRIPTION ... REFRESH PUBLICATION (no syntax change).
   - Stale sequences are removed from pg_subscription_rel.
   - Newly added sequences in the publisher are added in 'init'
     state to pg_subscription_rel.
   - Initiates sequence synchronization for all sequences by sequence
     sync worker as listed in subscription creation process.
   - Sequence synchronization occurs for newly added sequences only.

3) Introduce new command for refreshing all sequences:
   - ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES.
   - Removes stale sequences and adds newly added sequences from
     the publisher to pg_subscription_rel.
   - Resets all sequences in pg_subscription_rel to 'init' state.
   - Initiates sequence synchronization for all sequences by sequence
     sync worker as listed in subscription creation process.
---
 doc/src/sgml/catalogs.sgml                    |  19 +-
 doc/src/sgml/config.sgml                      |   4 +-
 doc/src/sgml/logical-replication.sgml         |   4 +-
 doc/src/sgml/monitoring.sgml                  |   5 +-
 doc/src/sgml/ref/alter_subscription.sgml      |  50 ++-
 doc/src/sgml/ref/create_subscription.sgml     |   8 +
 doc/src/sgml/system-views.sgml                |  67 +++
 src/backend/catalog/pg_publication.c          |  46 ++
 src/backend/catalog/pg_subscription.c         |  81 +++-
 src/backend/catalog/system_views.sql          |  10 +
 src/backend/commands/sequence.c               |  31 +-
 src/backend/commands/subscriptioncmds.c       | 315 +++++++++++---
 src/backend/executor/execReplication.c        |   4 +-
 src/backend/parser/gram.y                     |  11 +-
 src/backend/postmaster/bgworker.c             |   5 +-
 src/backend/replication/logical/Makefile      |   1 +
 .../replication/logical/applyparallelworker.c |   2 +-
 src/backend/replication/logical/launcher.c    |  71 ++-
 src/backend/replication/logical/meson.build   |   1 +
 .../replication/logical/sequencesync.c        | 404 ++++++++++++++++++
 src/backend/replication/logical/tablesync.c   | 234 +++++++---
 src/backend/replication/logical/worker.c      |  74 +++-
 src/bin/psql/tab-complete.c                   |   2 +-
 src/include/catalog/pg_proc.dat               |   5 +
 src/include/catalog/pg_subscription_rel.h     |   4 +-
 src/include/commands/sequence.h               |   3 +
 src/include/nodes/parsenodes.h                |   3 +-
 src/include/replication/logicalworker.h       |   3 +-
 src/include/replication/worker_internal.h     |  31 +-
 src/test/regress/expected/rules.out           |   8 +
 src/test/subscription/meson.build             |   1 +
 src/test/subscription/t/034_sequences.pl      | 178 ++++++++
 32 files changed, 1485 insertions(+), 200 deletions(-)
 create mode 100644 src/backend/replication/logical/sequencesync.c
 create mode 100644 src/test/subscription/t/034_sequences.pl

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index b654fae1b2..16c427ec3a 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8102,16 +8102,19 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
   </indexterm>
 
   <para>
-   The catalog <structname>pg_subscription_rel</structname> contains the
-   state for each replicated relation in each subscription.  This is a
-   many-to-many mapping.
+   The catalog <structname>pg_subscription_rel</structname> stores the
+   state of each replicated table and sequence for each subscription.  This
+   is a many-to-many mapping.
   </para>
 
   <para>
-   This catalog only contains tables known to the subscription after running
-   either <link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link> or
-   <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ... REFRESH
-   PUBLICATION</command></link>.
+   This catalog only contains tables and sequences known to the subscription
+   after running either
+   <link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link> or
+  <link linkend="sql-altersubscription-params-refresh-publication">
+   <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command></link> or
+   <link linkend="sql-altersubscription-params-refresh-publication-sequences">
+   <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES</command></link>.
   </para>
 
   <table>
@@ -8145,7 +8148,7 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
        (references <link linkend="catalog-pg-class"><structname>pg_class</structname></link>.<structfield>oid</structfield>)
       </para>
       <para>
-       Reference to relation
+       Reference to table or sequence
       </para></entry>
      </row>
 
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index a65839a670..3c550ae975 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5206,8 +5206,8 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       <listitem>
        <para>
         Specifies maximum number of logical replication workers. This includes
-        leader apply workers, parallel apply workers, and table synchronization
-        workers.
+        leader apply workers, parallel apply workers, table synchronization
+        workers and a sequence synchronization worker.
        </para>
        <para>
         Logical replication workers are taken from the pool defined by
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index a23a3d57e2..1d3318a4f4 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1999,8 +1999,8 @@ CONTEXT:  processing remote data for replication origin "pg_16395" during "INSER
    <para>
     <link linkend="guc-max-logical-replication-workers"><varname>max_logical_replication_workers</varname></link>
     must be set to at least the number of subscriptions (for leader apply
-    workers), plus some reserve for the table synchronization workers and
-    parallel apply workers.
+    workers), plus some reserve for the parallel apply workers, table synchronization workers, and a sequence
+    synchronization worker.
    </para>
 
    <para>
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 55417a6fa9..5fbb0c9c45 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2017,8 +2017,9 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        Type of the subscription worker process.  Possible types are
-       <literal>apply</literal>, <literal>parallel apply</literal>, and
-       <literal>table synchronization</literal>.
+       <literal>apply</literal>, <literal>parallel apply</literal>,
+       <literal>table synchronization</literal>, and
+       <literal>sequence synchronization</literal>.
       </para></entry>
      </row>
 
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index fdc648d007..eb7d544116 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -26,6 +26,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SET PUBLICA
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> ADD PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...] [ WITH ( <replaceable class="parameter">publication_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> DROP PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...] [ WITH ( <replaceable class="parameter">publication_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> REFRESH PUBLICATION [ WITH ( <replaceable class="parameter">refresh_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
+ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> REFRESH PUBLICATION SEQUENCES
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> ENABLE
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> DISABLE
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SET ( <replaceable class="parameter">subscription_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )
@@ -67,6 +68,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
 
   <para>
    Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>,
+   <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES</command>,
    <command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
    with <literal>refresh</literal> option as <literal>true</literal>,
    <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
@@ -158,30 +160,45 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
     <term><literal>REFRESH PUBLICATION</literal></term>
     <listitem>
      <para>
-      Fetch missing table information from publisher.  This will start
+      Fetch missing table information from the publisher.  This will start
       replication of tables that were added to the subscribed-to publications
       since <link linkend="sql-createsubscription">
       <command>CREATE SUBSCRIPTION</command></link> or
       the last invocation of <command>REFRESH PUBLICATION</command>.
      </para>
 
+     <para>
+      Also, fetch missing sequence information from the publisher.
+     </para>
+
+     <para>
+      The system catalog <link linkend="catalog-pg-subscription-rel">pg_subscription_rel</link>
+      is updated to record all tables and sequences known to the subscription,
+      that are still part of the publication.
+     </para>
+
      <para>
       <replaceable>refresh_option</replaceable> specifies additional options for the
-      refresh operation.  The supported options are:
+      refresh operation.  The only supported option is:
 
       <variablelist>
        <varlistentry>
         <term><literal>copy_data</literal> (<type>boolean</type>)</term>
         <listitem>
          <para>
-          Specifies whether to copy pre-existing data in the publications
-          that are being subscribed to when the replication starts.
-          The default is <literal>true</literal>.
+          Specifies whether to copy pre-existing data for tables and synchronize
+          sequences in the publications that are being subscribed to when the replication
+          starts. The default is <literal>true</literal>.
          </para>
          <para>
           Previously subscribed tables are not copied, even if a table's row
           filter <literal>WHERE</literal> clause has since been modified.
          </para>
+         <para>
+          Previously subscribed sequences are not re-synchronized. To do that,
+          see <link linkend="sql-altersubscription-params-refresh-publication-sequences">
+          <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES</command></link>
+         </para>
          <para>
           See <xref linkend="sql-createsubscription-notes"/> for details of
           how <literal>copy_data = true</literal> can interact with the
@@ -194,12 +211,35 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
           parameter of <command>CREATE SUBSCRIPTION</command> for details about
           copying pre-existing data in binary format.
          </para>
+         <para>
+          See the <link linkend="sql-createsubscription-params-with-copy-data"><literal>copy_data</literal></link>
+          on how to handle the warnings regarding the difference in sequence
+          definition between the publisher and the subscriber.
+         </para>
         </listitem>
        </varlistentry>
       </variablelist></para>
     </listitem>
    </varlistentry>
 
+   <varlistentry id="sql-altersubscription-params-refresh-publication-sequences">
+    <term><literal>REFRESH PUBLICATION SEQUENCES</literal></term>
+    <listitem>
+     <para>
+      Fetch missing sequence information from the publisher, then re-synchronize
+      sequence data with the publisher. Unlike <link linkend="sql-altersubscription-params-refresh-publication">
+      <literal>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</literal></link> which
+      only synchronizes newly added sequences, <literal>REFRESH PUBLICATION SEQUENCES</literal>
+      will re-synchronize the sequence data for all subscribed sequences. The
+      sequence definition can differ between the publisher and the subscriber,
+      this is detected and a WARNING is logged to the user, but the warning is
+      only an indication of a potential problem; it is recommended to alter the
+      sequence to keep the sequence option same as the publisher and execute
+      the command again.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry id="sql-altersubscription-params-enable">
     <term><literal>ENABLE</literal></term>
     <listitem>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 740b7d9421..de3bdb8ed1 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -263,6 +263,14 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
           <literal>copy_data = true</literal> can interact with the
           <literal>origin</literal> parameter.
          </para>
+         <para>
+          The sequence definition can differ between the publisher and the
+          subscriber, this is detected and a WARNING is logged to the user, but
+          the warning is only an indication of a potential problem; it is
+          recommended to alter the sequence to keep the sequence option same as
+          the publisher and execute <link linkend="sql-altersubscription-params-refresh-publication-sequences">
+          <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES</command></link>.
+         </para>
         </listitem>
        </varlistentry>
 
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index a0b692bf1e..8373ebc5b0 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -126,6 +126,11 @@
       <entry>prepared transactions</entry>
      </row>
 
+     <row>
+      <entry><link linkend="view-pg-publication-sequences"><structname>pg_publication_sequences</structname></link></entry>
+      <entry>publications and information of their associated sequences</entry>
+     </row>
+
      <row>
       <entry><link linkend="view-pg-publication-tables"><structname>pg_publication_tables</structname></link></entry>
       <entry>publications and information of their associated tables</entry>
@@ -2185,6 +2190,68 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
 
  </sect1>
 
+ <sect1 id="view-pg-publication-sequences">
+  <title><structname>pg_publication_sequences</structname></title>
+
+  <indexterm zone="view-pg-publication-sequences">
+   <primary>pg_publication_sequences</primary>
+  </indexterm>
+
+  <para>
+   The view <structname>pg_publication_sequences</structname> provides
+   information about the mapping between publications and information of
+   sequences they contain.
+  </para>
+
+  <table>
+   <title><structname>pg_publication_sequences</structname> Columns</title>
+   <tgroup cols="1">
+    <thead>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       Column Type
+      </para>
+      <para>
+       Description
+      </para></entry>
+     </row>
+    </thead>
+
+    <tbody>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>pubname</structfield> <type>name</type>
+       (references <link linkend="catalog-pg-publication"><structname>pg_publication</structname></link>.<structfield>pubname</structfield>)
+      </para>
+      <para>
+       Name of publication
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>schemaname</structfield> <type>name</type>
+       (references <link linkend="catalog-pg-namespace"><structname>pg_namespace</structname></link>.<structfield>nspname</structfield>)
+      </para>
+      <para>
+       Name of schema containing sequence
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>sequencename</structfield> <type>name</type>
+       (references <link linkend="catalog-pg-class"><structname>pg_class</structname></link>.<structfield>relname</structfield>)
+      </para>
+      <para>
+       Name of sequence
+      </para></entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+ </sect1>
+
  <sect1 id="view-pg-publication-tables">
   <title><structname>pg_publication_tables</structname></title>
 
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index acfac67f8c..980e5574a4 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -1292,3 +1292,49 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
 
 	SRF_RETURN_DONE(funcctx);
 }
+
+/*
+ * Returns Oids of sequences in a publication.
+ */
+Datum
+pg_get_publication_sequences(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx;
+	char	   *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	Publication *publication;
+	List	   *sequences = NIL;
+
+	/* stuff done only on the first call of the function */
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext oldcontext;
+
+		/* create a function context for cross-call persistence */
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		/* switch to memory context appropriate for multiple function calls */
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		publication = GetPublicationByName(pubname, false);
+
+		if (publication->allsequences)
+			sequences = GetAllSequencesPublicationRelations();
+
+		funcctx->user_fctx = (void *) sequences;
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	/* stuff done on every call of the function */
+	funcctx = SRF_PERCALL_SETUP();
+	sequences = (List *) funcctx->user_fctx;
+
+	if (funcctx->call_cntr < list_length(sequences))
+	{
+		Oid			relid = list_nth_oid(sequences, funcctx->call_cntr);
+
+		SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid));
+	}
+
+	SRF_RETURN_DONE(funcctx);
+}
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..8a2161e259 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -27,6 +27,7 @@
 #include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
+#include "utils/memutils.h"
 #include "utils/lsyscache.h"
 #include "utils/pg_lsn.h"
 #include "utils/rel.h"
@@ -431,7 +432,9 @@ RemoveSubscriptionRel(Oid subid, Oid relid)
 		 * leave tablesync slots or origins in the system when the
 		 * corresponding table is dropped.
 		 */
-		if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
+		if (!OidIsValid(subid) &&
+			get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE &&
+			subrel->srsubstate != SUBREL_STATE_READY)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
@@ -457,7 +460,7 @@ 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.
@@ -468,7 +471,8 @@ HasSubscriptionRelations(Oid subid)
 	Relation	rel;
 	ScanKeyData skey[1];
 	SysScanDesc scan;
-	bool		has_subrels;
+	HeapTuple	tup;
+	bool		has_subrels = false;
 
 	rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
@@ -480,8 +484,22 @@ HasSubscriptionRelations(Oid subid)
 	scan = systable_beginscan(rel, InvalidOid, false,
 							  NULL, 1, skey);
 
-	/* If even a single tuple exists then the subscription has tables. */
-	has_subrels = HeapTupleIsValid(systable_getnext(scan));
+	while (HeapTupleIsValid(tup = systable_getnext(scan)))
+	{
+		Form_pg_subscription_rel subrel;
+
+		subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
+
+		/*
+		 * Skip sequence tuples. If even a single table tuple exists then the
+		 * subscription has tables.
+		 */
+		if (get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE)
+		{
+			has_subrels = true;
+			break;
+		}
+	}
 
 	/* Cleanup */
 	systable_endscan(scan);
@@ -493,35 +511,41 @@ HasSubscriptionRelations(Oid subid)
 /*
  * Get the relations for the subscription.
  *
- * If not_ready is true, return only the relations that are not in a ready
- * state, otherwise return all the relations of the subscription.  The
- * returned list is palloc'ed in the current memory context.
+ * get_tables: get relations for tables of the subscription.
+ *
+ * get_sequences: get relations for sequences of the subscription.
+ *
+ * all_states:
+ * If getting tables, if all_states is true get all tables, otherwise
+ * only get tables that have not reached 'READY' state.
+ * If getting sequences, if all_states is true get all sequences,
+ * otherwise only get sequences that are in 'init' state.
+ *
+ * The returned list is palloc'ed in the current memory context.
  */
 List *
-GetSubscriptionRelations(Oid subid, bool not_ready)
+GetSubscriptionRelations(Oid subid, bool get_tables, bool get_sequences,
+						 bool all_states)
 {
 	List	   *res = NIL;
 	Relation	rel;
 	HeapTuple	tup;
 	int			nkeys = 0;
-	ScanKeyData skey[2];
+	ScanKeyData skey;
 	SysScanDesc scan;
 
+	/* One or both of 'get_tables' and 'get_sequences' must be true. */
+	Assert(get_tables || get_sequences);
+
 	rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
-	ScanKeyInit(&skey[nkeys++],
+	ScanKeyInit(&skey,
 				Anum_pg_subscription_rel_srsubid,
 				BTEqualStrategyNumber, F_OIDEQ,
 				ObjectIdGetDatum(subid));
 
-	if (not_ready)
-		ScanKeyInit(&skey[nkeys++],
-					Anum_pg_subscription_rel_srsubstate,
-					BTEqualStrategyNumber, F_CHARNE,
-					CharGetDatum(SUBREL_STATE_READY));
-
 	scan = systable_beginscan(rel, InvalidOid, false,
-							  NULL, nkeys, skey);
+							  NULL, nkeys, &skey);
 
 	while (HeapTupleIsValid(tup = systable_getnext(scan)))
 	{
@@ -532,6 +556,27 @@ GetSubscriptionRelations(Oid subid, bool not_ready)
 
 		subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
 
+		if (get_rel_relkind(subrel->srrelid) == RELKIND_SEQUENCE)
+		{
+			/* Skip sequences if they were not requested */
+			if (!get_sequences)
+				continue;
+
+			/* Skip all non-init sequences if not all_states were requested */
+			if (!all_states && (subrel->srsubstate != SUBREL_STATE_INIT))
+				continue;
+		}
+		else
+		{
+			/* Skip tables if they were not requested */
+			if (!get_tables)
+				continue;
+
+			/* Skip all ready tables if not all_states were requested */
+			if (!all_states && (subrel->srsubstate == SUBREL_STATE_READY))
+				continue;
+		}
+
 		relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
 		relstate->relid = subrel->srrelid;
 		relstate->state = subrel->srsubstate;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 19cabc9a47..a6475af855 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -394,6 +394,16 @@ CREATE VIEW pg_publication_tables AS
          pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)
     WHERE C.oid = GPT.relid;
 
+CREATE VIEW pg_publication_sequences AS
+    SELECT
+        P.pubname AS pubname,
+        N.nspname AS schemaname,
+        C.relname AS sequencename
+    FROM pg_publication P,
+         LATERAL pg_get_publication_sequences(P.pubname) GPS,
+         pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)
+    WHERE C.oid = GPS.relid;
+
 CREATE VIEW pg_locks AS
     SELECT * FROM pg_lock_status() AS L;
 
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index a23d2c87fd..ec7d5bbba1 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -111,7 +111,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);
 
 
@@ -942,9 +941,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, bool is_called, int64 log_cnt)
 {
 	SeqTable	elm;
 	Relation	seqrel;
@@ -995,7 +997,7 @@ do_setval(Oid relid, int64 next, bool iscalled)
 						(long long) minv, (long long) maxv)));
 
 	/* Set the currval() state only if iscalled = true */
-	if (iscalled)
+	if (is_called)
 	{
 		elm->last = next;		/* last returned number */
 		elm->last_valid = true;
@@ -1012,8 +1014,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,8 +1046,8 @@ do_setval(Oid relid, int64 next, bool iscalled)
 }
 
 /*
- * Implement the 2 arg setval procedure.
- * See do_setval for discussion.
+ * Implement the 2 arg set sequence procedure.
+ * See SetSequence for discussion.
  */
 Datum
 setval_oid(PG_FUNCTION_ARGS)
@@ -1053,14 +1055,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, true, SEQ_LOG_CNT_INVALID);
 
 	PG_RETURN_INT64(next);
 }
 
 /*
- * Implement the 3 arg setval procedure.
- * See do_setval for discussion.
+ * Implement the 3 arg set sequence procedure.
+ * See SetSequence for discussion.
  */
 Datum
 setval3_oid(PG_FUNCTION_ARGS)
@@ -1069,7 +1071,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, iscalled, SEQ_LOG_CNT_INVALID);
 
 	PG_RETURN_INT64(next);
 }
@@ -1896,6 +1898,11 @@ pg_sequence_last_value(PG_FUNCTION_ARGS)
 /*
  * Return the current on-disk state of the sequence.
  *
+ * The page_lsn allows the user to determine if the sequence has been updated
+ * since the last synchronization with the subscriber. This is done by
+ * comparing the current page_lsn with the value stored in pg_subscription_rel
+ * from the last synchronization.
+ *
  * 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 d124bfe55c..8c7796d052 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -26,6 +26,7 @@
 #include "catalog/objectaddress.h"
 #include "catalog/pg_authid_d.h"
 #include "catalog/pg_database_d.h"
+#include "catalog/pg_sequence.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_type.h"
@@ -103,6 +104,7 @@ typedef struct SubOpts
 } SubOpts;
 
 static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *fetch_sequence_list(WalReceiverConn *wrconn, List *publications);
 static void check_publications_origin(WalReceiverConn *wrconn,
 									  List *publications, bool copydata,
 									  char *origin, Oid *subrel_local_oids,
@@ -734,9 +736,6 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	{
 		char	   *err;
 		WalReceiverConn *wrconn;
-		List	   *tables;
-		ListCell   *lc;
-		char		table_state;
 		bool		must_use_password;
 
 		/* Try to connect to the publisher. */
@@ -751,6 +750,10 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 
 		PG_TRY();
 		{
+			List	   *tables;
+			List	   *sequences;
+			char		table_state;
+
 			check_publications(wrconn, publications);
 			check_publications_origin(wrconn, publications, opts.copy_data,
 									  opts.origin, NULL, 0, stmt->subname);
@@ -766,9 +769,24 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 			 * info.
 			 */
 			tables = fetch_table_list(wrconn, publications);
-			foreach(lc, tables)
+			foreach_ptr(RangeVar, rv, tables)
+			{
+				Oid			relid;
+
+				relid = RangeVarGetRelid(rv, AccessShareLock, false);
+
+				/* Check for supported relkind. */
+				CheckSubscriptionRelkind(get_rel_relkind(relid),
+										 rv->schemaname, rv->relname);
+
+				AddSubscriptionRelState(subid, relid, table_state,
+										InvalidXLogRecPtr, true);
+			}
+
+			/* Add the sequences in init state */
+			sequences = fetch_sequence_list(wrconn, publications);
+			foreach_ptr(RangeVar, rv, sequences)
 			{
-				RangeVar   *rv = (RangeVar *) lfirst(lc);
 				Oid			relid;
 
 				relid = RangeVarGetRelid(rv, AccessShareLock, false);
@@ -847,12 +865,35 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	return myself;
 }
 
+/*
+ * Update the subscription to refresh both the publication and the publication
+ * objects associated with the subscription.
+ *
+ * If 'copy_data' parameter is true, the function will set the state
+ * to "init"; otherwise, it will set the state to "ready".
+ *
+ * When 'validate_publications' is provided with a publication list, the
+ * function checks that the specified publications exist on the publisher.
+ *
+ * If 'refresh_tables' is true, update the subscription by adding or removing
+ * tables that have been added or removed since the last subscription creation
+ * or refresh publication.
+ *
+ * If 'refresh_sequences' is true, update the subscription by adding or removing
+ * sequences that have been added or removed since the last subscription
+ * creation or publication refresh.
+ *
+ * If 'resync_all_sequences' is true, mark all objects with "init" state
+ * for re-synchronization; otherwise, only update the newly added tables and
+ * sequences based on the copy_data parameter.
+ */
 static void
 AlterSubscription_refresh(Subscription *sub, bool copy_data,
-						  List *validate_publications)
+						  List *validate_publications, bool refresh_tables,
+						  bool refresh_sequences, bool resync_all_sequences)
 {
 	char	   *err;
-	List	   *pubrel_names;
+	List	   *pubrel_names = NIL;
 	List	   *subrel_states;
 	Oid		   *subrel_local_oids;
 	Oid		   *pubrel_local_oids;
@@ -870,6 +911,12 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 	WalReceiverConn *wrconn;
 	bool		must_use_password;
 
+	/* resync_all_sequences cannot be specified with refresh_tables */
+	Assert(!(resync_all_sequences && refresh_tables));
+
+	/* resync_all_sequences cannot be specified with copy_data as false */
+	Assert(!(resync_all_sequences && !copy_data));
+
 	/* Load the library providing us libpq calls. */
 	load_file("libpqwalreceiver", false);
 
@@ -889,10 +936,16 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 			check_publications(wrconn, validate_publications);
 
 		/* Get the table list from publisher. */
-		pubrel_names = fetch_table_list(wrconn, sub->publications);
+		if (refresh_tables)
+			pubrel_names = fetch_table_list(wrconn, sub->publications);
+
+		/* Get the sequence list from publisher. */
+		if (refresh_sequences)
+			pubrel_names = list_concat(pubrel_names,
+									   fetch_sequence_list(wrconn, sub->publications));
 
 		/* Get local table list. */
-		subrel_states = GetSubscriptionRelations(sub->oid, false);
+		subrel_states = GetSubscriptionRelations(sub->oid, refresh_tables, refresh_sequences, true);
 		subrel_count = list_length(subrel_states);
 
 		/*
@@ -911,9 +964,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 		qsort(subrel_local_oids, subrel_count,
 			  sizeof(Oid), oid_cmp);
 
-		check_publications_origin(wrconn, sub->publications, copy_data,
-								  sub->origin, subrel_local_oids,
-								  subrel_count, sub->name);
+		if (refresh_tables)
+			check_publications_origin(wrconn, sub->publications, copy_data,
+									  sub->origin, subrel_local_oids,
+									  subrel_count, sub->name);
 
 		/*
 		 * Rels that we want to remove from subscription and drop any slots
@@ -935,12 +989,13 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 		{
 			RangeVar   *rv = (RangeVar *) lfirst(lc);
 			Oid			relid;
+			char		relkind;
 
 			relid = RangeVarGetRelid(rv, AccessShareLock, false);
 
 			/* Check for supported relkind. */
-			CheckSubscriptionRelkind(get_rel_relkind(relid),
-									 rv->schemaname, rv->relname);
+			relkind = get_rel_relkind(relid);
+			CheckSubscriptionRelkind(relkind, rv->schemaname, rv->relname);
 
 			pubrel_local_oids[off++] = relid;
 
@@ -951,7 +1006,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 										copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
 										InvalidXLogRecPtr, true);
 				ereport(DEBUG1,
-						(errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
+						(errmsg_internal("%s \"%s.%s\" added to subscription \"%s\"",
+										 relkind == RELKIND_SEQUENCE ? "sequence" : "table",
 										 rv->schemaname, rv->relname, sub->name)));
 			}
 		}
@@ -968,11 +1024,31 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 		{
 			Oid			relid = subrel_local_oids[off];
 
-			if (!bsearch(&relid, pubrel_local_oids,
-						 list_length(pubrel_names), sizeof(Oid), oid_cmp))
+			if (bsearch(&relid, pubrel_local_oids,
+						list_length(pubrel_names), sizeof(Oid), oid_cmp))
+			{
+				/*
+				 * The resync_all_sequences flag will only be set to true for
+				 * the REFRESH PUBLICATION SEQUENCES command, indicating tha
+				 * the existing sequences need to be re-synchronized by
+				 * resetting the relation to its initial state.
+				 */
+				if (resync_all_sequences)
+				{
+					ereport(LOG,
+							(errmsg_internal("sequence \"%s.%s\" of subscription \"%s\" set to INIT state",
+											 get_namespace_name(get_rel_namespace(relid)),
+											 get_rel_name(relid),
+											 sub->name)));
+					UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_INIT,
+											   InvalidXLogRecPtr);
+				}
+			}
+			else
 			{
 				char		state;
 				XLogRecPtr	statelsn;
+				char		relkind = get_rel_relkind(relid);
 
 				/*
 				 * Lock pg_subscription_rel with AccessExclusiveLock to
@@ -994,38 +1070,48 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 				/* Last known rel state. */
 				state = GetSubscriptionRelState(sub->oid, relid, &statelsn);
 
-				sub_remove_rels[remove_rel_len].relid = relid;
-				sub_remove_rels[remove_rel_len++].state = state;
-
 				RemoveSubscriptionRel(sub->oid, relid);
 
-				logicalrep_worker_stop(sub->oid, relid);
+				sub_remove_rels[remove_rel_len].relid = relid;
+				sub_remove_rels[remove_rel_len++].state = state;
 
 				/*
-				 * For READY state, we would have already dropped the
-				 * tablesync origin.
+				 * A single sequencesync worker synchronizes all sequences, so
+				 * only stop workers when relation kind is not sequence.
 				 */
-				if (state != SUBREL_STATE_READY)
+				if (relkind != RELKIND_SEQUENCE)
 				{
-					char		originname[NAMEDATALEN];
+					logicalrep_worker_stop(sub->oid, relid, WORKERTYPE_TABLESYNC);
 
 					/*
-					 * Drop the tablesync's origin tracking if exists.
-					 *
-					 * It is possible that the origin is not yet created for
-					 * tablesync worker, this can happen for the states before
-					 * SUBREL_STATE_FINISHEDCOPY. The tablesync worker or
-					 * apply worker can also concurrently try to drop the
-					 * origin and by this time the origin might be already
-					 * removed. For these reasons, passing missing_ok = true.
+					 * For READY state, we would have already dropped the
+					 * tablesync origin.
 					 */
-					ReplicationOriginNameForLogicalRep(sub->oid, relid, originname,
-													   sizeof(originname));
-					replorigin_drop_by_name(originname, true, false);
+					if (state != SUBREL_STATE_READY)
+					{
+						char		originname[NAMEDATALEN];
+
+						/*
+						 * Drop the tablesync's origin tracking if exists.
+						 *
+						 * It is possible that the origin is not yet created
+						 * for tablesync worker, this can happen for the
+						 * states before SUBREL_STATE_FINISHEDCOPY. The
+						 * tablesync worker or apply worker can also
+						 * concurrently try to drop the origin and by this
+						 * time the origin might be already removed. For these
+						 * reasons, passing missing_ok = true.
+						 */
+						ReplicationOriginNameForLogicalRep(sub->oid, relid,
+														   originname,
+														   sizeof(originname));
+						replorigin_drop_by_name(originname, true, false);
+					}
 				}
 
 				ereport(DEBUG1,
-						(errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"",
+						(errmsg_internal("%s \"%s.%s\" removed from subscription \"%s\"",
+										 relkind == RELKIND_SEQUENCE ? "sequence" : "table",
 										 get_namespace_name(get_rel_namespace(relid)),
 										 get_rel_name(relid),
 										 sub->name)));
@@ -1039,6 +1125,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 		 */
 		for (off = 0; off < remove_rel_len; off++)
 		{
+			/* Skip relations belonging to sequences. */
+			if (get_rel_relkind(sub_remove_rels[off].relid) == RELKIND_SEQUENCE)
+				continue;
+
 			if (sub_remove_rels[off].state != SUBREL_STATE_READY &&
 				sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE)
 			{
@@ -1424,8 +1514,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));
 
 					/*
-					 * See ALTER_SUBSCRIPTION_REFRESH for details why this is
-					 * not allowed.
+					 * See ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for details
+					 * why this is not allowed.
 					 */
 					if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
 						ereport(ERROR,
@@ -1439,7 +1529,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					sub->publications = stmt->publication;
 
 					AlterSubscription_refresh(sub, opts.copy_data,
-											  stmt->publication);
+											  stmt->publication, true, true,
+											  false);
 				}
 
 				break;
@@ -1479,8 +1570,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 										 "ALTER SUBSCRIPTION ... DROP PUBLICATION ... WITH (refresh = false)")));
 
 					/*
-					 * See ALTER_SUBSCRIPTION_REFRESH for details why this is
-					 * not allowed.
+					 * See ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for details
+					 * why this is not allowed.
 					 */
 					if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
 						ereport(ERROR,
@@ -1498,13 +1589,28 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					sub->publications = publist;
 
 					AlterSubscription_refresh(sub, opts.copy_data,
-											  validate_publications);
+											  validate_publications, true, true,
+											  false);
 				}
 
 				break;
 			}
 
-		case ALTER_SUBSCRIPTION_REFRESH:
+		case ALTER_SUBSCRIPTION_REFRESH_PUBLICATION_SEQUENCES:
+			{
+				if (!sub->enabled)
+					ereport(ERROR,
+							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							 errmsg("ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES is not allowed for disabled subscriptions")));
+
+				PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES");
+
+				AlterSubscription_refresh(sub, true, NULL, false, true, true);
+
+				break;
+			}
+
+		case ALTER_SUBSCRIPTION_REFRESH_PUBLICATION:
 			{
 				if (!sub->enabled)
 					ereport(ERROR,
@@ -1539,7 +1645,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 				PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");
 
-				AlterSubscription_refresh(sub, opts.copy_data, NULL);
+				AlterSubscription_refresh(sub, opts.copy_data, NULL, true, true, false);
 
 				break;
 			}
@@ -1781,7 +1887,11 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	{
 		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-		logicalrep_worker_stop(w->subid, w->relid);
+		/* Worker might have exited because of an error */
+		if (w->type == WORKERTYPE_UNKNOWN)
+			continue;
+
+		logicalrep_worker_stop(w->subid, w->relid, w->type);
 	}
 	list_free(subworkers);
 
@@ -1804,7 +1914,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	 * the apply and tablesync workers and they can't restart because of
 	 * exclusive lock on the subscription.
 	 */
-	rstates = GetSubscriptionRelations(subid, true);
+	rstates = GetSubscriptionRelations(subid, true, false, false);
 	foreach(lc, rstates)
 	{
 		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
@@ -2162,11 +2272,15 @@ check_publications_origin(WalReceiverConn *wrconn, List *publications,
 	for (i = 0; i < subrel_count; i++)
 	{
 		Oid			relid = subrel_local_oids[i];
-		char	   *schemaname = get_namespace_name(get_rel_namespace(relid));
-		char	   *tablename = get_rel_name(relid);
 
-		appendStringInfo(&cmd, "AND NOT (N.nspname = '%s' AND C.relname = '%s')\n",
-						 schemaname, tablename);
+		if (get_rel_relkind(relid) != RELKIND_SEQUENCE)
+		{
+			char	   *schemaname = get_namespace_name(get_rel_namespace(relid));
+			char	   *tablename = get_rel_name(relid);
+
+			appendStringInfo(&cmd, "AND NOT (N.nspname = '%s' AND C.relname = '%s')\n",
+							 schemaname, tablename);
+		}
 	}
 
 	res = walrcv_exec(wrconn, cmd.data, 1, tableRow);
@@ -2336,6 +2450,105 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
 	return tablelist;
 }
 
+/*
+ * Get the list of sequences which belong to specified publications on the
+ * publisher connection.
+ */
+static List *
+fetch_sequence_list(WalReceiverConn *wrconn, List *publications)
+{
+	WalRcvExecResult *res;
+	StringInfoData cmd;
+	TupleTableSlot *slot;
+	Oid			tableRow[8] = {TEXTOID, TEXTOID, OIDOID, INT8OID, INT8OID, INT8OID, INT8OID, BOOLOID};
+	List	   *seqlist = NIL;
+
+	Assert(list_length(publications) > 0);
+
+	initStringInfo(&cmd);
+
+	appendStringInfoString(&cmd, "SELECT DISTINCT n.nspname, c.relname, s.seqtypid, s.seqmin, s.seqmax, s.seqstart, s.seqincrement, s.seqcycle"
+						   " FROM pg_publication p, LATERAL pg_get_publication_sequences(p.pubname::text) gps(relid),"
+						   " pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_sequence s ON c.oid = s.seqrelid"
+						   " WHERE c.oid = gps.relid AND p.pubname IN (");
+	get_publications_str(publications, &cmd, true);
+	appendStringInfoChar(&cmd, ')');
+
+	res = walrcv_exec(wrconn, cmd.data, 8, tableRow);
+	pfree(cmd.data);
+
+	if (res->status != WALRCV_OK_TUPLES)
+		ereport(ERROR,
+				(errmsg("could not receive list of sequences from the publisher: %s",
+						res->err)));
+
+	/* Process sequences. */
+	slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+	while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+	{
+		char	   *nspname;
+		char	   *relname;
+		Oid			seqtypid;
+		int64		seqmin;
+		int64		seqmax;
+		int64		seqstart;
+		int64		seqincrement;
+		bool		seqcycle;
+		bool		isnull;
+		RangeVar   *rv;
+		Oid			relid;
+		HeapTuple	tup;
+		Form_pg_sequence seqform;
+
+		nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+		Assert(!isnull);
+		relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+		Assert(!isnull);
+		seqtypid = DatumGetObjectId(slot_getattr(slot, 3, &isnull));
+		Assert(!isnull);
+		seqmin = DatumGetInt64(slot_getattr(slot, 4, &isnull));
+		Assert(!isnull);
+		seqmax = DatumGetInt64(slot_getattr(slot, 5, &isnull));
+		Assert(!isnull);
+		seqstart = DatumGetInt64(slot_getattr(slot, 6, &isnull));
+		Assert(!isnull);
+		seqincrement = DatumGetInt64(slot_getattr(slot, 7, &isnull));
+		Assert(!isnull);
+		seqcycle = DatumGetBool(slot_getattr(slot, 8, &isnull));
+		Assert(!isnull);
+
+		rv = makeRangeVar(nspname, relname, -1);
+		relid = RangeVarGetRelid(rv, AccessShareLock, false);
+
+		/* Get the local sequence */
+		tup = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
+		if (!HeapTupleIsValid(tup))
+			elog(ERROR, "cache lookup failed for sequence \"%s.%s\"",
+				 get_namespace_name(get_rel_namespace(relid)), get_rel_name(relid));
+
+		seqform = (Form_pg_sequence) GETSTRUCT(tup);
+
+		if (seqform->seqtypid != seqtypid || seqform->seqmin != seqmin ||
+			seqform->seqmax != seqmax || seqform->seqstart != seqstart ||
+			seqform->seqincrement != seqincrement ||
+			seqform->seqcycle != seqcycle)
+			ereport(WARNING,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("Sequence option in remote and local is not same for \"%s.%s\"",
+						   get_namespace_name(get_rel_namespace(relid)), get_rel_name(relid)),
+					errhint("Alter/Re-create the sequence using the same options as in remote."));
+
+		ReleaseSysCache(tup);
+		seqlist = lappend(seqlist, rv);
+		ExecClearTuple(slot);
+	}
+
+	ExecDropSingleTupleTableSlot(slot);
+	walrcv_clear_result(res);
+
+	return seqlist;
+}
+
 /*
  * This is to report the connection failure while dropping replication slots.
  * Here, we report the WARNING for all tablesync slots so that user can drop
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index d0a89cd577..fdf69e4f28 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -739,7 +739,9 @@ void
 CheckSubscriptionRelkind(char relkind, const char *nspname,
 						 const char *relname)
 {
-	if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE)
+	if (relkind != RELKIND_RELATION &&
+		relkind != RELKIND_PARTITIONED_TABLE &&
+		relkind != RELKIND_SEQUENCE)
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("cannot use relation \"%s.%s\" as logical replication target",
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 9b3cad1cac..28b772df32 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -10835,11 +10835,20 @@ AlterSubscriptionStmt:
 					AlterSubscriptionStmt *n =
 						makeNode(AlterSubscriptionStmt);
 
-					n->kind = ALTER_SUBSCRIPTION_REFRESH;
+					n->kind = ALTER_SUBSCRIPTION_REFRESH_PUBLICATION;
 					n->subname = $3;
 					n->options = $6;
 					$$ = (Node *) n;
 				}
+			| ALTER SUBSCRIPTION name REFRESH PUBLICATION SEQUENCES
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+
+					n->kind = ALTER_SUBSCRIPTION_REFRESH_PUBLICATION_SEQUENCES;
+					n->subname = $3;
+					$$ = (Node *) n;
+				}
 			| ALTER SUBSCRIPTION name ADD_P PUBLICATION name_list opt_definition
 				{
 					AlterSubscriptionStmt *n =
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 77707bb384..f8dd93a83a 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 ba03eeff1c..7621fa8aed 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -25,6 +25,7 @@ OBJS = \
 	proto.o \
 	relation.o \
 	reorderbuffer.o \
+	sequencesync.o \
 	slotsync.o \
 	snapbuild.o \
 	tablesync.o \
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index e7f7d4c5e4..cbe7c814ad 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -962,7 +962,7 @@ ParallelApplyWorkerMain(Datum main_arg)
 	 * the subscription relation state.
 	 */
 	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
-								  invalidate_syncing_table_states,
+								  invalidate_syncing_relation_states,
 								  (Datum) 0);
 
 	set_apply_error_context_origin(originname);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index c566d50a07..36c9952994 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -235,29 +235,28 @@ 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(wtype == WORKERTYPE_TABLESYNC ||
+		   wtype == WORKERTYPE_SEQUENCESYNC ||
+		   wtype == WORKERTYPE_APPLY);
+
 	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];
 
-		/* Skip parallel apply workers. */
-		if (isParallelApplyWorker(w))
-			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;
@@ -317,6 +316,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);
 
 	/*----------
@@ -402,7 +402,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;
@@ -489,7 +490,7 @@ retry:
 			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,
@@ -497,6 +498,14 @@ retry:
 			snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
 			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_UNKNOWN:
 			/* Should never happen. */
 			elog(ERROR, "unknown worker type");
@@ -612,13 +621,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)
 	{
@@ -685,7 +694,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);
@@ -815,6 +824,28 @@ logicalrep_launcher_onexit(int code, Datum arg)
 	LogicalRepCtx->launcher_pid = 0;
 }
 
+/*
+ * 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_failuretime(int code, Datum arg)
+{
+	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);
+}
+
 /*
  * Cleanup function.
  *
@@ -863,7 +894,8 @@ logicalrep_sync_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (isTablesyncWorker(w) && w->subid == subid)
+		if ((isTableSyncWorker(w) || isSequenceSyncWorker(w)) &&
+			w->subid == subid)
 			res++;
 	}
 
@@ -1178,7 +1210,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)
@@ -1314,7 +1346,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;
@@ -1357,6 +1389,9 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 			case WORKERTYPE_TABLESYNC:
 				values[9] = CStringGetTextDatum("table synchronization");
 				break;
+			case WORKERTYPE_SEQUENCESYNC:
+				values[9] = CStringGetTextDatum("sequence synchronization");
+				break;
 			case WORKERTYPE_UNKNOWN:
 				/* Should never happen. */
 				elog(ERROR, "unknown worker type");
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index 3dec36a6de..1711fc3248 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -11,6 +11,7 @@ backend_sources += files(
   'proto.c',
   'relation.c',
   'reorderbuffer.c',
+  'sequencesync.c',
   'slotsync.c',
   'snapbuild.c',
   'tablesync.c',
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
new file mode 100644
index 0000000000..2a5c8c5939
--- /dev/null
+++ b/src/backend/replication/logical/sequencesync.c
@@ -0,0 +1,404 @@
+/*-------------------------------------------------------------------------
+ * sequencesync.c
+ *	  PostgreSQL logical replication: initial sequence synchronization
+ *
+ * Copyright (c) 2024, 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 SEQUENCE
+ *
+ * Sequencesync worker will get the sequences that should be synchronized from
+ * pg_subscription_rel catalog table. It synchronizes
+ * MAX_SEQUENCES_SYNC_PER_BATCH (100) sequences within a single transaction by
+ * getting the sequence value from the remote publisher and updating it to the
+ * local subscriber sequence and updates the seqeunce state to READY. It also
+ * updates the remote sequence's lsn to pg_subscription_rel which can be
+ * later used to compare it with the pg_sequence_state page_lsn value to
+ * identify if sequence is changed since the last synchronization.
+ *
+ * The sequencesync worker retrieves the sequences that need to be synchronized
+ * from the pg_subscription_rel catalog table. It synchronizes up to
+ * MAX_SEQUENCES_SYNC_PER_BATCH (100) sequences in a single transaction by
+ * fetching the sequence values and the sequence's page_lsn from the remote
+ * publisher and updating them in the local subscriber sequence. After
+ * synchronization, it sets the sequence state to READY. This LSN can later be
+ * compared with the pg_sequence_state page LSN value to determine if the
+ * sequence has changed since the last synchronization.
+ *
+ * So the state progression is always just: INIT -> READY.
+ *
+ * Here MAX_SEQUENCES_SYNC_PER_BATCH (100) sequences are synchronized within a
+ * single transaction  to avoid creating a lot of transactions and also the
+ * locks on the sequence relation will be periodically released during the
+ * commit transaction.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/table.h"
+#include "catalog/pg_subscription_rel.h"
+#include "commands/sequence.h"
+#include "pgstat.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/usercontext.h"
+
+/*
+ * fetch_remote_sequence_data
+ *
+ * Retrieve sequence data (last_value, log_cnt, page_lsn and is_called)
+ * from the remote node.
+ *
+ * The sequence last_value will be returned directly, while
+ * log_cnt, is_called and page_lsn will be returned via the output
+ * parameters log_cnt, is_called and lsn, respectively.
+ */
+static int64
+fetch_remote_sequence_data(WalReceiverConn *conn, Oid remoteid, char *nspname,
+						   char *relname, int64 *log_cnt, bool *is_called,
+						   XLogRecPtr *lsn)
+{
+	WalRcvExecResult *res;
+	StringInfoData cmd;
+	TupleTableSlot *slot;
+	Oid			tableRow[4] = {INT8OID, INT8OID, BOOLOID, LSNOID};
+	int64		last_value = (Datum) 0;
+	bool		isnull;
+
+	initStringInfo(&cmd);
+
+	appendStringInfo(&cmd, "SELECT last_value, log_cnt, is_called, page_lsn "
+					 "FROM pg_sequence_state(%d)", remoteid);
+
+	res = walrcv_exec(conn, cmd.data, 4, tableRow);
+	pfree(cmd.data);
+
+	if (res->status != WALRCV_OK_TUPLES)
+		ereport(ERROR,
+				(errmsg("could not receive sequence list from the publisher: %s",
+						res->err)));
+
+	/* Process the sequence. */
+	slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+	if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("sequence \"%s.%s\" not found on publisher",
+						nspname, relname)));
+
+	last_value = DatumGetInt64(slot_getattr(slot, 1, &isnull));
+	Assert(!isnull);
+
+	*log_cnt = DatumGetInt64(slot_getattr(slot, 2, &isnull));
+	Assert(!isnull);
+
+	*is_called = DatumGetBool(slot_getattr(slot, 3, &isnull));
+	Assert(!isnull);
+
+	*lsn = DatumGetLSN(slot_getattr(slot, 4, &isnull));
+	Assert(!isnull);
+
+	ExecDropSingleTupleTableSlot(slot);
+
+	walrcv_clear_result(res);
+
+	return last_value;
+}
+
+/*
+ * Copy existing data of a sequence from 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 XLogRecPtr
+copy_sequence(WalReceiverConn *conn, Relation rel)
+{
+	StringInfoData cmd;
+	int64		seq_last_value;
+	int64		seq_log_cnt;
+	bool		seq_is_called;
+	XLogRecPtr	seq_lsn = InvalidXLogRecPtr;
+	WalRcvExecResult *res;
+	Oid			tableRow[] = {OIDOID, CHAROID};
+	TupleTableSlot *slot;
+	LogicalRepRelId remoteid;	/* unique id of the relation */
+	char		relkind PG_USED_FOR_ASSERTS_ONLY;
+	bool		isnull;
+	char	   *nspname = get_namespace_name(RelationGetNamespace(rel));
+	char	   *relname = RelationGetRelationName(rel);
+
+	/* Fetch Oid. */
+	initStringInfo(&cmd);
+	appendStringInfo(&cmd, "SELECT c.oid, c.relkind"
+					 "  FROM pg_catalog.pg_class c"
+					 "  INNER JOIN pg_catalog.pg_namespace n"
+					 "        ON (c.relnamespace = n.oid)"
+					 " WHERE n.nspname = %s"
+					 "   AND c.relname = %s",
+					 quote_literal_cstr(nspname),
+					 quote_literal_cstr(relname));
+
+	res = walrcv_exec(conn, cmd.data,
+					  lengthof(tableRow), tableRow);
+	if (res->status != WALRCV_OK_TUPLES)
+		ereport(ERROR,
+				(errcode(ERRCODE_CONNECTION_FAILURE),
+				 errmsg("sequence \"%s.%s\" info could not be fetched from publisher: %s",
+						nspname, relname, res->err)));
+
+	slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+	if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("sequence \"%s.%s\" not found on publisher",
+						nspname, relname)));
+
+	remoteid = DatumGetObjectId(slot_getattr(slot, 1, &isnull));
+	Assert(!isnull);
+	relkind = DatumGetChar(slot_getattr(slot, 2, &isnull));
+	Assert(!isnull);
+	Assert(relkind == RELKIND_SEQUENCE);
+
+	ExecDropSingleTupleTableSlot(slot);
+	walrcv_clear_result(res);
+
+	seq_last_value = fetch_remote_sequence_data(conn, remoteid, nspname,
+												relname, &seq_log_cnt, &seq_is_called,
+												&seq_lsn);
+
+	SetSequence(RelationGetRelid(rel), seq_last_value, seq_is_called,
+				seq_log_cnt);
+
+	/* return the LSN when the sequence state was set */
+	return seq_lsn;
+}
+
+/*
+ * Start syncing the sequences in the sync worker.
+ */
+static void
+LogicalRepSyncSequences(void)
+{
+	char	   *err;
+	bool		must_use_password;
+	List	   *sequences;
+	List	   *sequences_not_synced = NIL;
+	char		slotname[NAMEDATALEN];
+	AclResult	aclresult;
+	UserContext ucxt;
+	bool		run_as_owner = false;
+	int			curr_seq = 0;
+	int			seq_count;
+	bool		start_txn = true;
+	Oid			subid = MyLogicalRepWorker->subid;
+	MemoryContext oldctx;
+
+/*
+ * Synchronizing each sequence individually incurs overhead from starting
+ * and committing a transaction repeatedly. Additionally, we want to avoid
+ * keeping transactions open for extended periods by setting excessively
+ * high values.
+ */
+#define MAX_SEQUENCES_SYNC_PER_BATCH 100
+
+	StartTransactionCommand();
+
+	/* Get the sequences that should be synchronized. */
+	sequences = GetSubscriptionRelations(subid, false, true, false);
+
+	/* 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;
+
+	snprintf(slotname, NAMEDATALEN, "pg_%u_sync_sequences_" UINT64_FORMAT,
+			 subid, GetSystemIdentifier());
+
+	/*
+	 * Here we use the slot name instead of the subscription name as the
+	 * application_name, so that it is different from the leader apply worker,
+	 * so that synchronous replication can distinguish them.
+	 */
+	LogRepWorkerWalRcvConn =
+		walrcv_connect(MySubscription->conninfo, true, true,
+					   must_use_password,
+					   slotname, &err);
+	if (LogRepWorkerWalRcvConn == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_CONNECTION_FAILURE),
+				 errmsg("could not connect to the publisher: %s", err)));
+
+	seq_count = list_length(sequences_not_synced);
+	foreach_ptr(SubscriptionRelState, seqinfo, sequences_not_synced)
+	{
+		Relation	sequence_rel;
+		XLogRecPtr	sequence_lsn;
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (start_txn)
+		{
+			StartTransactionCommand();
+			start_txn = false;
+		}
+
+		sequence_rel = table_open(seqinfo->relid, RowExclusiveLock);
+
+		/*
+		 * 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),
+						   RelationGetRelationName(sequence_rel));
+
+		/*
+		 * COPY FROM does not honor RLS policies.  That is not a problem for
+		 * subscriptions owned by roles with BYPASSRLS privilege (or
+		 * superuser, who has it implicitly), but other roles should not be
+		 * able to circumvent RLS.  Disallow logical replication into RLS
+		 * enabled relations for such roles.
+		 */
+		if (check_enable_rls(RelationGetRelid(sequence_rel), InvalidOid, false) == RLS_ENABLED)
+			ereport(ERROR,
+					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					errmsg("user \"%s\" cannot replicate into sequence with row-level security enabled: \"%s\"",
+						   GetUserNameFromId(GetUserId(), true),
+						   RelationGetRelationName(sequence_rel)));
+
+		sequence_lsn = copy_sequence(LogRepWorkerWalRcvConn, sequence_rel);
+
+		UpdateSubscriptionRelState(subid, seqinfo->relid, SUBREL_STATE_READY,
+								   sequence_lsn);
+
+		table_close(sequence_rel, NoLock);
+
+		curr_seq++;
+
+		/*
+		 * Have we reached the end of the current batch of sequences, or last
+		 * remaining sequences to synchronize?
+		 */
+		if (((curr_seq % MAX_SEQUENCES_SYNC_PER_BATCH) == 0) ||
+			curr_seq == seq_count)
+		{
+			/* Obtain the starting index of the current batch. */
+			int			i = (curr_seq - 1) - ((curr_seq - 1) % MAX_SEQUENCES_SYNC_PER_BATCH);
+
+			/* LOG all the sequences synchronized during current batch. */
+			for (; i < curr_seq; i++)
+			{
+				SubscriptionRelState *done_seq;
+
+				done_seq = (SubscriptionRelState *) lfirst(list_nth_cell(sequences_not_synced, i));
+				ereport(LOG,
+						errmsg("logical replication synchronization for subscription \"%s\", sequence \"%s\" has finished",
+							   get_subscription_name(subid, false), get_rel_name(done_seq->relid)));
+			}
+
+			/* Commit this batch, and prepare for next batch. */
+			CommitTransactionCommand();
+			start_txn = true;
+		}
+
+	}
+
+	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, false);
+
+			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();
+
+	finish_sync_worker(WORKERTYPE_SEQUENCESYNC);
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e03e761392..96ff253ab0 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -130,19 +130,22 @@ typedef enum
 	SYNC_TABLE_STATE_VALID,
 } SyncingTablesState;
 
-static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
+static SyncingTablesState relation_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
 static List *table_states_not_ready = NIL;
-static bool FetchTableStates(bool *started_tx);
+static List *sequence_states_not_ready = NIL;
+static bool FetchTableStates(void);
 
 static StringInfo copybuf = NULL;
 
 /*
  * Exit routine for synchronization worker.
  */
-static void
+void
 pg_attribute_noreturn()
-finish_sync_worker(void)
+finish_sync_worker(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.
@@ -157,15 +160,24 @@ finish_sync_worker(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 to set a sequence failure time. */
+	if (wtype == WORKERTYPE_SEQUENCESYNC)
+		cancel_before_shmem_exit(logicalrep_seqsyncworker_failuretime, 0);
+
 	/* Stop gracefully */
 	proc_exit(0);
 }
@@ -205,7 +217,7 @@ wait_for_relation_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;
@@ -252,7 +264,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);
@@ -278,9 +290,9 @@ wait_for_worker_state_change(char expected_state)
  * Callback from syscache invalidation.
  */
 void
-invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
+invalidate_syncing_relation_states(Datum arg, int cacheid, uint32 hashvalue)
 {
-	table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
+	relation_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
 }
 
 /*
@@ -387,7 +399,7 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 		 */
 		replorigin_drop_by_name(originname, true, false);
 
-		finish_sync_worker();
+		finish_sync_worker(WORKERTYPE_TABLESYNC);
 	}
 	else
 		SpinLockRelease(&MyLogicalRepWorker->relmutex);
@@ -429,9 +441,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 
 	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
@@ -464,6 +473,14 @@ process_syncing_tables_for_apply(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)
 		{
 			/*
@@ -477,11 +494,6 @@ process_syncing_tables_for_apply(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.
@@ -518,7 +530,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-												rstate->relid, false);
+												rstate->relid,
+												WORKERTYPE_TABLESYNC, true);
 
 			if (syncworker)
 			{
@@ -661,10 +674,108 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 }
 
 /*
- * Process possible state change(s) of tables that are being synchronized.
+ * 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.
+ * To prevent starting the sequencesync worker at a high frequency after a
+ * failure, we store its last failure time. We start the sync worker for the
+ * same relation after waiting at least wal_retrieve_retry_interval.
+ */
+static void
+process_syncing_sequences_for_apply()
+{
+	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 *syncworker;
+		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 sequence worker already running?
+		 */
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+		syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+											InvalidOid, WORKERTYPE_SEQUENCESYNC,
+											true);
+		if (syncworker)
+		{
+			/* 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 are free sync worker slot(s), start a new sequence sync
+		 * worker, and break from the loop.
+		 */
+		if (nsyncworkers < max_sync_workers_per_subscription)
+		{
+			TimestampTz now = GetCurrentTimestamp();
+
+			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);
+	}
+}
+
+/*
+ * Process possible state change(s) of tables that are being synchronized and
+ * start new tablesync worker and/or sequencesync worker for the newly added
+ * relations.
  */
 void
-process_syncing_tables(XLogRecPtr current_lsn)
+process_syncing_relations(XLogRecPtr current_lsn)
 {
 	switch (MyLogicalRepWorker->type)
 	{
@@ -682,7 +793,20 @@ process_syncing_tables(XLogRecPtr current_lsn)
 			break;
 
 		case WORKERTYPE_APPLY:
+
+			/*
+			 * We need up-to-date sync state info for subscription tables and
+			 * sequences here.
+			 */
+			FetchTableStates();
+
 			process_syncing_tables_for_apply(current_lsn);
+			process_syncing_sequences_for_apply();
+			break;
+
+		case WORKERTYPE_SEQUENCESYNC:
+			/* Should never happen. */
+			Assert(0);
 			break;
 
 		case WORKERTYPE_UNKNOWN:
@@ -1320,7 +1444,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 		case SUBREL_STATE_SYNCDONE:
 		case SUBREL_STATE_READY:
 		case SUBREL_STATE_UNKNOWN:
-			finish_sync_worker();	/* doesn't return */
+			finish_sync_worker(WORKERTYPE_TABLESYNC);	/* doesn't return */
 	}
 
 	/* Calculate the name of the tablesync slot. */
@@ -1564,39 +1688,48 @@ copy_table_done:
 /*
  * 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.
+ * Copy tables that are not READY state into table_states_not_ready, and sequences
+ * that have INIT state into sequence_states_not_ready. 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.
  */
 static bool
-FetchTableStates(bool *started_tx)
+FetchTableStates(void)
 {
 	static bool has_subrels = false;
+	bool		started_tx = false;
 
-	*started_tx = false;
-
-	if (table_states_validity != SYNC_TABLE_STATE_VALID)
+	if (relation_states_validity != SYNC_TABLE_STATE_VALID)
 	{
 		MemoryContext oldctx;
 		List	   *rstates;
 		ListCell   *lc;
 		SubscriptionRelState *rstate;
 
-		table_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED;
+		relation_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED;
 
 		/* 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 all non-ready tables. */
-		rstates = GetSubscriptionRelations(MySubscription->oid, true);
+		/*
+		 * Fetch tables that are in non-ready state, and sequences that are in
+		 * init state.
+		 */
+		rstates = GetSubscriptionRelations(MySubscription->oid, true, true,
+										   false);
 
 		/* Allocate the tracking info in a permanent memory context. */
 		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
@@ -1604,7 +1737,11 @@ FetchTableStates(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);
 
@@ -1625,8 +1762,14 @@ FetchTableStates(bool *started_tx)
 		 * 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;
+		if (relation_states_validity == SYNC_TABLE_STATE_REBUILD_STARTED)
+			relation_states_validity = SYNC_TABLE_STATE_VALID;
+	}
+
+	if (started_tx)
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(true);
 	}
 
 	return has_subrels;
@@ -1709,7 +1852,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);
 
@@ -1717,7 +1860,7 @@ TablesyncWorkerMain(Datum main_arg)
 
 	run_tablesync_worker();
 
-	finish_sync_worker();
+	finish_sync_worker(WORKERTYPE_TABLESYNC);
 }
 
 /*
@@ -1731,17 +1874,10 @@ TablesyncWorkerMain(Datum main_arg)
 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);
-	}
+	has_subrels = FetchTableStates();
 
 	/*
 	 * Return false when there are no tables in subscription or not all tables
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6dc54c7283..2e84b24617 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -486,6 +486,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");
@@ -1025,8 +1030,11 @@ 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);
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
+	process_syncing_relations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
@@ -1137,8 +1145,11 @@ 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);
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
+	process_syncing_relations(prepare_data.end_lsn);
 
 	/*
 	 * Since we have already prepared the transaction, in a case where the
@@ -1193,8 +1204,11 @@ 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_syncing_tables(prepare_data.end_lsn);
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
+	process_syncing_relations(prepare_data.end_lsn);
 
 	clear_subscription_skip_lsn(prepare_data.end_lsn);
 
@@ -1254,8 +1268,11 @@ apply_handle_rollback_prepared(StringInfo s)
 	store_flush_position(rollback_data.rollback_end_lsn, XactLastCommitEnd);
 	in_remote_transaction = false;
 
-	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(rollback_data.rollback_end_lsn);
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
+	process_syncing_relations(rollback_data.rollback_end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
@@ -1381,8 +1398,11 @@ 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);
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
+	process_syncing_relations(prepare_data.end_lsn);
 
 	/*
 	 * Similar to prepare case, the subskiplsn could be left in a case of
@@ -2223,8 +2243,11 @@ apply_handle_stream_commit(StringInfo s)
 			break;
 	}
 
-	/* Process any tables that are being synchronized in parallel. */
-	process_syncing_tables(commit_data.end_lsn);
+	/*
+	 * Process any tables that are being synchronized in parallel and any
+	 * newly added relations.
+	 */
+	process_syncing_relations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 
@@ -3621,8 +3644,11 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 			AcceptInvalidationMessages();
 			maybe_reread_subscription();
 
-			/* Process any table synchronization changes. */
-			process_syncing_tables(last_received);
+			/*
+			 * Process any tables that are being synchronized in parallel and
+			 * any newly added relations.
+			 */
+			process_syncing_relations(last_received);
 		}
 
 		/* Cleanup the memory. */
@@ -4531,8 +4557,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.
@@ -4611,6 +4637,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",
@@ -4619,14 +4649,17 @@ InitializeLogRepWorker(void)
 	CommitTransactionCommand();
 }
 
-/* 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);
@@ -4656,8 +4689,11 @@ SetupApplyOrSyncWorker(int worker_slot)
 	 * the subscription relation state.
 	 */
 	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
-								  invalidate_syncing_table_states,
+								  invalidate_syncing_relation_states,
 								  (Datum) 0);
+
+	if (am_sequencesync_worker())
+		before_shmem_exit(logicalrep_seqsyncworker_failuretime, (Datum) 0);
 }
 
 /* Logical Replication Apply worker entry point */
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 3d32ddbd7b..1a7aa21265 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1936,7 +1936,7 @@ psql_completion(const char *text, int start, int end)
 	/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION */
 	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
 			 TailMatches("REFRESH", "PUBLICATION"))
-		COMPLETE_WITH("WITH (");
+		COMPLETE_WITH("SEQUENCES", "WITH (");
 	/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION WITH ( */
 	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
 			 TailMatches("REFRESH", "PUBLICATION", "WITH", "("))
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 7997b841cb..899f0299b8 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12007,6 +12007,11 @@
   proargmodes => '{v,o,o,o,o}',
   proargnames => '{pubname,pubid,relid,attrs,qual}',
   prosrc => 'pg_get_publication_tables' },
+{ oid => '8000', descr => 'get OIDs of sequences in a publication',
+  proname => 'pg_get_publication_sequences', prorows => '1000', proretset => 't',
+  provolatile => 's', prorettype => 'oid', proargtypes => 'text',
+  proallargtypes => '{text,oid}', proargmodes => '{i,o}',
+  proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_sequences' },
 { oid => '6121',
   descr => 'returns whether a relation can be part of a publication',
   proname => 'pg_relation_is_publishable', provolatile => 's',
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 8244ad537a..58abed907a 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -90,6 +90,8 @@ extern char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn);
 extern void RemoveSubscriptionRel(Oid subid, Oid relid);
 
 extern bool HasSubscriptionRelations(Oid subid);
-extern List *GetSubscriptionRelations(Oid subid, bool not_ready);
+extern List *GetSubscriptionRelations(Oid subid, bool get_tables,
+									  bool get_sequences,
+									  bool all_relations);
 
 #endif							/* PG_SUBSCRIPTION_REL_H */
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index e88cbee3b5..adafa440e3 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, bool is_called, int64 log_cnt);
 extern void ResetSequenceCaches(void);
 
 extern void seq_redo(XLogReaderState *record);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 3a5f8279ed..346abdcef9 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4230,7 +4230,8 @@ typedef enum AlterSubscriptionType
 	ALTER_SUBSCRIPTION_SET_PUBLICATION,
 	ALTER_SUBSCRIPTION_ADD_PUBLICATION,
 	ALTER_SUBSCRIPTION_DROP_PUBLICATION,
-	ALTER_SUBSCRIPTION_REFRESH,
+	ALTER_SUBSCRIPTION_REFRESH_PUBLICATION,
+	ALTER_SUBSCRIPTION_REFRESH_PUBLICATION_SEQUENCES,
 	ALTER_SUBSCRIPTION_ENABLED,
 	ALTER_SUBSCRIPTION_SKIP,
 } AlterSubscriptionType;
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index a18d79d1b2..47a3326ad3 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 9646261d7e..6dff23fe6f 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;
 
 /*
@@ -239,6 +242,7 @@ extern PGDLLIMPORT bool InitializingApplyWorker;
 
 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);
@@ -246,22 +250,27 @@ 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 pg_attribute_noreturn() finish_sync_worker(LogicalRepWorkerType wtype);
+
+extern void logicalrep_seqsyncworker_failuretime(int code, Datum arg);
+
 extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
 											   char *originname, Size szoriginname);
 
 extern bool AllTablesyncsReady(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 void process_syncing_relations(XLogRecPtr current_lsn);
+extern void invalidate_syncing_relation_states(Datum arg, int cacheid,
+											   uint32 hashvalue);
 
 extern void stream_start_internal(TransactionId xid, bool first_segment);
 extern void stream_stop_internal(TransactionId xid);
@@ -326,15 +335,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/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 5201280669..358c76e78e 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1443,6 +1443,14 @@ pg_prepared_xacts| SELECT p.transaction,
    FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid)
      LEFT JOIN pg_authid u ON ((p.ownerid = u.oid)))
      LEFT JOIN pg_database d ON ((p.dbid = d.oid)));
+pg_publication_sequences| SELECT p.pubname,
+    n.nspname AS schemaname,
+    c.relname AS sequencename
+   FROM pg_publication p,
+    LATERAL pg_get_publication_sequences((p.pubname)::text) gps(relid),
+    (pg_class c
+     JOIN pg_namespace n ON ((n.oid = c.relnamespace)))
+  WHERE (c.oid = gps.relid);
 pg_publication_tables| SELECT p.pubname,
     n.nspname AS schemaname,
     c.relname AS tablename,
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..870b4175e8 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
       't/031_column_list.pl',
       't/032_subscribe_use_index.pl',
       't/033_run_as_table_owner.pl',
+      't/034_sequences.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/034_sequences.pl b/src/test/subscription/t/034_sequences.pl
new file mode 100644
index 0000000000..88f2705abe
--- /dev/null
+++ b/src/test/subscription/t/034_sequences.pl
@@ -0,0 +1,178 @@
+
+# Copyright (c) 2024, 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;
+);
+$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";
+
+# 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 syned
+$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 SEQUENCES should throw a warning
+# for sequence definition not matching between the publisher and the subscriber.
+
+# Create a new sequence 'regress_s4' whose START value is not the same in the
+# publisher and subscriber.
+$node_publisher->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s4 START 1 INCREMENT 2;
+));
+
+$node_subscriber->safe_psql(
+	'postgres', qq(
+	CREATE SEQUENCE regress_s4 START 10 INCREMENT 2;
+));
+
+($result, my $stdout, my $stderr) = $node_subscriber->psql(
+	'postgres', "
+        ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION SEQUENCES");
+like(
+	$stderr,
+	qr/WARNING: ( [A-Z0-9]+:)? Sequence option in remote and local is not same for "public.regress_s4"/,
+	"Refresh publication sequences should throw a warning if the sequence definition is not the same"
+);
+
+
+done_testing();
-- 
2.34.1