v20240730_2-0003-Enhance-sequence-synchronization-during-.patch
text/x-patch
Filename: v20240730_2-0003-Enhance-sequence-synchronization-during-.patch
Type: text/x-patch
Part: 2
Message:
Re: Logical Replication of sequences
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: format-patch
Series: patch v20240730-0003
Subject: Enhance sequence synchronization during subscription management
| File | + | − |
|---|---|---|
| doc/src/sgml/catalogs.sgml | 10 | 8 |
| doc/src/sgml/config.sgml | 2 | 2 |
| doc/src/sgml/logical-replication.sgml | 2 | 2 |
| doc/src/sgml/monitoring.sgml | 3 | 2 |
| doc/src/sgml/ref/alter_subscription.sgml | 35 | 5 |
| doc/src/sgml/system-views.sgml | 67 | 0 |
| src/backend/catalog/pg_publication.c | 46 | 0 |
| src/backend/catalog/pg_subscription.c | 47 | 10 |
| src/backend/catalog/system_views.sql | 10 | 0 |
| src/backend/commands/sequence.c | 8 | 6 |
| src/backend/commands/subscriptioncmds.c | 198 | 42 |
| src/backend/executor/execReplication.c | 3 | 1 |
| src/backend/parser/gram.y | 10 | 1 |
| src/backend/postmaster/bgworker.c | 4 | 1 |
| src/backend/replication/logical/launcher.c | 51 | 12 |
| src/backend/replication/logical/Makefile | 1 | 0 |
| src/backend/replication/logical/meson.build | 1 | 0 |
| src/backend/replication/logical/sequencesync.c | 368 | 0 |
| src/backend/replication/logical/tablesync.c | 183 | 44 |
| src/backend/replication/logical/worker.c | 19 | 4 |
| src/bin/psql/tab-complete.c | 1 | 1 |
| src/include/catalog/pg_proc.dat | 5 | 0 |
| src/include/catalog/pg_subscription_rel.h | 3 | 1 |
| src/include/commands/sequence.h | 3 | 0 |
| src/include/nodes/parsenodes.h | 2 | 1 |
| src/include/replication/logicalworker.h | 2 | 1 |
| src/include/replication/worker_internal.h | 22 | 3 |
| src/test/regress/expected/rules.out | 8 | 0 |
| src/test/subscription/t/034_sequences.pl | 153 | 0 |
From 610db32f437935a543b38fea5986b65ed5f9f761 Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Mon, 29 Jul 2024 12:37:53 +0530
Subject: [PATCH v20240730_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 | 18 +-
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 | 40 +-
doc/src/sgml/system-views.sgml | 67 ++++
src/backend/catalog/pg_publication.c | 46 +++
src/backend/catalog/pg_subscription.c | 57 ++-
src/backend/catalog/system_views.sql | 10 +
src/backend/commands/sequence.c | 14 +-
src/backend/commands/subscriptioncmds.c | 240 ++++++++++--
src/backend/executor/execReplication.c | 4 +-
src/backend/parser/gram.y | 11 +-
src/backend/postmaster/bgworker.c | 5 +-
src/backend/replication/logical/Makefile | 1 +
src/backend/replication/logical/launcher.c | 63 ++-
src/backend/replication/logical/meson.build | 1 +
.../replication/logical/sequencesync.c | 368 ++++++++++++++++++
src/backend/replication/logical/tablesync.c | 227 ++++++++---
src/backend/replication/logical/worker.c | 23 +-
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 | 25 +-
src/test/regress/expected/rules.out | 8 +
src/test/subscription/t/034_sequences.pl | 153 ++++++++
29 files changed, 1267 insertions(+), 147 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..22b2a93535 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8102,16 +8102,18 @@ SCRAM-SHA-256$<replaceable><iteration count></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"><command>ALTER SUBSCRIPTION ... REFRESH
+ PUBLICATION</command></link> or <link linkend="sql-altersubscription">
+ <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES</command></link>.
</para>
<table>
@@ -8145,7 +8147,7 @@ SCRAM-SHA-256$<replaceable><iteration count></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 57cd7bb972..699dd22b7c 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..7868cad75f 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
@@ -200,6 +217,19 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
</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.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry id="sql-altersubscription-params-enable">
<term><literal>ENABLE</literal></term>
<listitem>
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..bc6d18b8b2 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,19 @@ 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);
+
+ /* If even a single tuple exists then the subscription has tables. */
+ if (get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE)
+ {
+ has_subrels = true;
+ break;
+ }
+ }
/* Cleanup */
systable_endscan(scan);
@@ -493,12 +508,17 @@ 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.
+ * all_relations:
+ * If returning sequences, if all_relations=true get all sequences,
+ * otherwise only get sequences that are in 'init' state.
+ * If returning tables, if all_relation=true get all tables, otherwise
+ * only get tables that have not reached 'READY' 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_relations)
{
List *res = NIL;
Relation rel;
@@ -514,11 +534,18 @@ GetSubscriptionRelations(Oid subid, bool not_ready)
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(subid));
- if (not_ready)
+ /* Get the relations that are not in ready state */
+ if (get_tables && !all_relations)
ScanKeyInit(&skey[nkeys++],
Anum_pg_subscription_rel_srsubstate,
BTEqualStrategyNumber, F_CHARNE,
CharGetDatum(SUBREL_STATE_READY));
+ /* Get the sequences that are in init state */
+ else if (get_sequences && !all_relations)
+ ScanKeyInit(&skey[nkeys++],
+ Anum_pg_subscription_rel_srsubstate,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(SUBREL_STATE_INIT));
scan = systable_beginscan(rel, InvalidOid, false,
NULL, nkeys, skey);
@@ -529,8 +556,18 @@ GetSubscriptionRelations(Oid subid, bool not_ready)
SubscriptionRelState *relstate;
Datum d;
bool isnull;
+ char subreltype;
subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
+ subreltype = get_rel_relkind(subrel->srrelid);
+
+ /* If only tables were requested, skip the sequences */
+ if (subreltype == RELKIND_SEQUENCE && !get_sequences)
+ continue;
+
+ /* If only sequences were requested, skip the tables */
+ if (subreltype != RELKIND_SEQUENCE && !get_tables)
+ continue;
relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
relstate->relid = subrel->srrelid;
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 32096725e2..e8bd53cca1 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.
+ *
+ * logcnt is currently used only by sequence syncworker to set the log_cnt for
+ * sequences while synchronization of sequence values from the publisher.
*/
-static void
-do_setval(Oid relid, int64 next, bool iscalled)
+void
+do_setval(Oid relid, int64 next, bool iscalled, int64 logcnt)
{
SeqTable elm;
Relation seqrel;
@@ -1013,7 +1015,7 @@ do_setval(Oid relid, int64 next, bool iscalled)
seq->last_value = next; /* last fetched number */
seq->is_called = iscalled;
- seq->log_cnt = 0;
+ seq->log_cnt = (logcnt == SEQ_LOG_CNT_INVALID) ? 0: logcnt;
MarkBufferDirty(buf);
@@ -1053,7 +1055,7 @@ setval_oid(PG_FUNCTION_ARGS)
Oid relid = PG_GETARG_OID(0);
int64 next = PG_GETARG_INT64(1);
- do_setval(relid, next, true);
+ do_setval(relid, next, true, SEQ_LOG_CNT_INVALID);
PG_RETURN_INT64(next);
}
@@ -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);
+ do_setval(relid, next, iscalled, SEQ_LOG_CNT_INVALID);
PG_RETURN_INT64(next);
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index d124bfe55c..984f72dc5a 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -103,6 +103,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,
@@ -751,6 +752,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
PG_TRY();
{
+ List *sequences;
+
check_publications(wrconn, publications);
check_publications_origin(wrconn, publications, opts.copy_data,
opts.origin, NULL, 0, stmt->subname);
@@ -781,6 +784,22 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
InvalidXLogRecPtr, true);
}
+ /* Add the sequences in init state */
+ sequences = fetch_sequence_list(wrconn, publications);
+ foreach_ptr(RangeVar, rv, sequences)
+ {
+ 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);
+ }
+
/*
* If requested, create permanent slot for the subscription. We
* won't use the initial snapshot for anything, so no need to
@@ -847,12 +866,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 'all_relations' is true, it will mark all objects with "init" state
+ * for re-synchronization; otherwise, only the newly added tables and
+ * sequences will be updated 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 all_relations)
{
char *err;
- List *pubrel_names;
+ List *pubrel_names = NIL;
List *subrel_states;
Oid *subrel_local_oids;
Oid *pubrel_local_oids;
@@ -889,10 +931,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 +959,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
@@ -951,7 +1000,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\"",
+ get_rel_relkind(relid) == RELKIND_SEQUENCE ? "sequence" : "table",
rv->schemaname, rv->relname, sub->name)));
}
}
@@ -973,6 +1023,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
{
char state;
XLogRecPtr statelsn;
+ char relkind = get_rel_relkind(relid);
/*
* Lock pg_subscription_rel with AccessExclusiveLock to
@@ -994,41 +1045,67 @@ 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 sequence-sync 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\"",
+ get_rel_relkind(relid) == RELKIND_SEQUENCE ? "sequence" : "table",
+ get_namespace_name(get_rel_namespace(relid)),
+ get_rel_name(relid),
+ sub->name)));
+ }
+
+ /*
+ * If all the relations should be re-synchronized, then set the
+ * state to init for re-synchronization. This is currently
+ * supported only for sequences.
+ */
+ else if (all_relations)
+ {
+ ereport(DEBUG1,
+ (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);
}
}
@@ -1039,6 +1116,9 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
*/
for (off = 0; off < remove_rel_len; off++)
{
+ 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 +1504,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 +1519,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 +1560,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 +1579,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 +1635,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 +1877,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
{
LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
- logicalrep_worker_stop(w->subid, w->relid);
+ logicalrep_worker_stop(w->subid, w->relid, w->type);
}
list_free(subworkers);
@@ -1804,7 +1900,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 +2258,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 +2436,62 @@ 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[2] = {TEXTOID, TEXTOID};
+ List *seqlist = NIL;
+
+ Assert(list_length(publications) > 0);
+
+ initStringInfo(&cmd);
+ appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.sequencename\n"
+ " FROM pg_catalog.pg_publication_sequences s\n"
+ " WHERE s.pubname IN (");
+ get_publications_str(publications, &cmd, true);
+ appendStringInfoChar(&cmd, ')');
+
+ res = walrcv_exec(wrconn, cmd.data, 2, 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;
+ bool isnull;
+ RangeVar *rv;
+
+ nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+ relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+ Assert(!isnull);
+
+ rv = makeRangeVar(nspname, relname, -1);
+ 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/launcher.c b/src/backend/replication/logical/launcher.c
index c566d50a07..04d76e7f54 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -235,12 +235,14 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
/*
* Walks the workers array and searches for one that matches given
- * subscription id and relid.
+ * subscription id, relid and type.
*
- * We are only interested in the leader apply worker or table sync worker.
+ * We are only interested in the leader apply worker, table sync worker, or
+ * sequence sync worker.
*/
LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid subid, Oid relid, LogicalRepWorkerType type,
+ bool only_running)
{
int i;
LogicalRepWorker *res = NULL;
@@ -257,7 +259,7 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
continue;
if (w->in_use && w->subid == subid && w->relid == relid &&
- (!only_running || w->proc))
+ w->type == type && (!only_running || w->proc))
{
res = w;
break;
@@ -317,6 +319,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);
/*----------
@@ -324,10 +327,12 @@ logicalrep_worker_launch(LogicalRepWorkerType wtype,
* - must be valid worker type
* - tablesync workers are only ones to have relid
* - parallel apply worker is the only kind of subworker
+ * - sequencesync workers will not have relid
*/
Assert(wtype != WORKERTYPE_UNKNOWN);
Assert(is_tablesync_worker == OidIsValid(relid));
Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
+ Assert(!is_sequencesync_worker || !OidIsValid(relid));
ereport(DEBUG1,
(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -402,7 +407,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 +495,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 +503,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 +626,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 type)
{
LogicalRepWorker *worker;
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
- worker = logicalrep_worker_find(subid, relid, false);
+ worker = logicalrep_worker_find(subid, relid, type, false);
if (worker)
{
@@ -685,7 +699,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 +829,28 @@ logicalrep_launcher_onexit(int code, Datum arg)
LogicalRepCtx->launcher_pid = 0;
}
+/*
+ * Update the failure time of the sequence sync worker in the subscription's
+ * apply worker.
+ *
+ * This function is invoked when the sequence sync 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 +899,7 @@ logicalrep_sync_worker_count(Oid subid)
{
LogicalRepWorker *w = &LogicalRepCtx->workers[i];
- if (isTablesyncWorker(w) && w->subid == subid)
+ if (isTableSyncWorker(w) && w->subid == subid)
res++;
}
@@ -1178,7 +1214,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 +1350,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 +1393,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..fc36bf9ef8
--- /dev/null
+++ b/src/backend/replication/logical/sequencesync.c
@@ -0,0 +1,368 @@
+/*-------------------------------------------------------------------------
+ * 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.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#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/rls.h"
+#include "utils/usercontext.h"
+
+/*
+ * fetch_remote_sequence_data
+ *
+ * Retrieve the last_value, log_cnt, page_lsn and is_called of the sequence
+ * from the remote node. The last_value will be returned directly, while
+ * log_cnt, is_called and page_lsn will be provided through 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 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)));
+
+ 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 = DatumGetInt64(slot_getattr(slot, 4, &isnull));
+ Assert(!isnull);
+
+ ExecDropSingleTupleTableSlot(slot);
+
+ walrcv_clear_result(res);
+
+ return value;
+}
+
+/*
+ * Copy existing data of a sequence from publisher.
+ *
+ * Fetch the sequence value from the publisher and set the subscriber sequence
+ * with the retrieved value. Caller is responsible for locking the local
+ * relation.
+ */
+static XLogRecPtr
+copy_sequence(WalReceiverConn *conn, Relation rel)
+{
+ StringInfoData cmd;
+ int64 sequence_value;
+ int64 log_cnt;
+ XLogRecPtr 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);
+ bool is_called;
+
+ /* 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);
+
+ sequence_value = fetch_remote_sequence_data(conn, remoteid, nspname,
+ relname, &log_cnt, &is_called,
+ &lsn);
+
+ do_setval(RelationGetRelid(rel), sequence_value, is_called, log_cnt);
+
+ /* return the LSN when the sequence state was set */
+ return 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 sequence sync 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 sequence sync 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..3e162b9007 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);
@@ -280,7 +292,7 @@ wait_for_worker_state_change(char expected_state)
void
invalidate_syncing_table_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)
{
@@ -660,6 +673,106 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
}
}
+/*
+ * 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 there is a sequence synchronization worker running already, no need to
+ * start a new one; the existing sequence sync worker will synchronize all the
+ * sequences. If there are still any sequences to be synced after the sequence
+ * sync worker exited, then a new sequence sync worker can be started in the
+ * next iteration. To prevent starting the sequence sync 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 sequence sync worker if there is not one already.
+ */
+ foreach_ptr(SubscriptionRelState, rstate, sequence_states_not_ready)
+ {
+ LogicalRepWorker *syncworker;
+
+ 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;
+ }
+ else
+ {
+ /*
+ * Count running sync workers for this subscription, while we have
+ * the lock.
+ */
+ int nsyncworkers =
+ logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
+
+ /* Now safe to release the LWLock */
+ LWLockRelease(LogicalRepWorkerLock);
+
+ /*
+ * If there are free sync worker slot(s), start a new 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.
*/
@@ -682,7 +795,19 @@ 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 +1445,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 +1689,50 @@ 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 into table_states_not_ready and sequences
+ * that are not ready into sequence_states_not_ready. The pg_subscription_rel
+ * table is shared between sequences and tables. Because changes to either
+ * sequences or relations can affect the validity of relation states, we update
+ * both table_states_not_ready and sequence_states_not_ready simultaneously
+ * to ensure consistency, rather than updating them separately. Returns true if
+ * subscription has 1 or more tables, else false.
*
* Note: If this function started the transaction (indicated by the parameter)
* then it is the caller's responsibility to commit it.
*/
static bool
-FetchTableStates(bool *started_tx)
+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 the tables that are in non-ready state and the 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 +1740,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 +1765,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 +1855,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 +1863,7 @@ TablesyncWorkerMain(Datum main_arg)
run_tablesync_worker();
- finish_sync_worker();
+ finish_sync_worker(WORKERTYPE_TABLESYNC);
}
/*
@@ -1731,17 +1877,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 ec96b5fe85..7491afdb48 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");
@@ -4531,8 +4536,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 +4616,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 +4628,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);
@@ -4658,6 +4670,9 @@ SetupApplyOrSyncWorker(int worker_slot)
CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
invalidate_syncing_table_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 be0ed1fc27..0c5601af82 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 5b250bad78..23445d7aa9 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11995,6 +11995,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..71d8c76235 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 (-1)
+
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 do_setval(Oid relid, int64 next, bool iscalled, int64 logcnt);
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..8a12ecb1fe 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 type,
bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running,
bool acquire_lock);
@@ -246,13 +250,18 @@ 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);
@@ -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/t/034_sequences.pl b/src/test/subscription/t/034_sequences.pl
new file mode 100644
index 0000000000..d7cee9dcb1
--- /dev/null
+++ b/src/test/subscription/t/034_sequences.pl
@@ -0,0 +1,153 @@
+
+# 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',
+ 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');
+
+done_testing();
--
2.34.1