v20240725-0003-Enhance-sequence-synchronization-during-su.patch
text/x-patch
Filename: v20240725-0003-Enhance-sequence-synchronization-during-su.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 v20240725-0003
Subject: Enhance sequence synchronization during subscription management
| File | + | − |
|---|---|---|
| doc/src/sgml/catalogs.sgml | 7 | 6 |
| 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 | 34 | 5 |
| doc/src/sgml/system-views.sgml | 67 | 0 |
| src/backend/catalog/pg_publication.c | 46 | 0 |
| src/backend/catalog/pg_subscription.c | 30 | 5 |
| src/backend/catalog/system_views.sql | 10 | 0 |
| src/backend/commands/sequence.c | 2 | 2 |
| src/backend/commands/subscriptioncmds.c | 172 | 32 |
| 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 | 95 | 5 |
| src/backend/replication/logical/Makefile | 1 | 0 |
| src/backend/replication/logical/meson.build | 1 | 0 |
| src/backend/replication/logical/sequencesync.c | 356 | 0 |
| src/backend/replication/logical/tablesync.c | 143 | 20 |
| 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.h | 6 | 0 |
| src/include/catalog/pg_subscription_rel.h | 9 | 1 |
| src/include/commands/sequence.h | 1 | 1 |
| src/include/nodes/parsenodes.h | 2 | 1 |
| src/include/replication/logicalworker.h | 2 | 1 |
| src/include/replication/worker_internal.h | 21 | 2 |
| src/test/regress/expected/rules.out | 8 | 0 |
| src/test/subscription/t/034_sequences.pl | 154 | 0 |
| src/tools/pgindent/typedefs.list | 2 | 0 |
From 130f4be8fc6f2523f57eaab63216a239db68443c Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Wed, 24 Jul 2024 11:24:57 +0530
Subject: [PATCH v20240725 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 | 13 +-
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 | 39 +-
doc/src/sgml/system-views.sgml | 67 ++++
src/backend/catalog/pg_publication.c | 46 +++
src/backend/catalog/pg_subscription.c | 35 +-
src/backend/catalog/system_views.sql | 10 +
src/backend/commands/sequence.c | 4 +-
src/backend/commands/subscriptioncmds.c | 204 ++++++++--
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 | 100 ++++-
src/backend/replication/logical/meson.build | 1 +
.../replication/logical/sequencesync.c | 356 ++++++++++++++++++
src/backend/replication/logical/tablesync.c | 163 +++++++-
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.h | 6 +
src/include/catalog/pg_subscription_rel.h | 10 +-
src/include/commands/sequence.h | 2 +-
src/include/nodes/parsenodes.h | 3 +-
src/include/replication/logicalworker.h | 3 +-
src/include/replication/worker_internal.h | 23 +-
src/test/regress/expected/rules.out | 8 +
src/test/subscription/t/034_sequences.pl | 154 ++++++++
src/tools/pgindent/typedefs.list | 2 +
31 files changed, 1218 insertions(+), 95 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..19d04b107e 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8103,14 +8103,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<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.
+ state for each replicated tables and sequences in 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
+ 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>.
</para>
@@ -8145,7 +8146,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 3dec0b7cfe..2bb4660336 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 ccdd24312b..1d3bc6a285 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1984,8 +1984,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 6af6d0d2c8..d3029b9e91 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>] [, ... ] )
@@ -153,30 +154,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
@@ -195,6 +211,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 bdc34cf94e..b893fc2d90 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>
@@ -2147,6 +2152,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..5610c0749c 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"
@@ -493,12 +494,19 @@ 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.
+ * If rel_type is SUB_REL_KIND_SEQUENCE, get only the sequences. If rel_type is
+ * SUB_REL_KIND_TABLE, get only the tables. If rel_type is SUB_REL_KIND_ALL,
+ * get both tables and sequences.
+ * If not_all_relations is true for SUB_REL_KIND_TABLE and SUB_REL_KIND_ALL,
+ * return only the relations that are not in a ready state, otherwise return all
+ * the relations of the subscription. If not_all_relations is true for
+ * SUB_REL_KIND_SEQUENCE, return only the sequences that are in init state,
+ * otherwise return all the sequences of the subscription.
+ * The returned list is palloc'ed in the current memory context.
*/
List *
-GetSubscriptionRelations(Oid subid, bool not_ready)
+GetSubscriptionRelations(Oid subid, SubscriptionRelKind rel_type,
+ bool not_all_relations)
{
List *res = NIL;
Relation rel;
@@ -514,11 +522,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 (rel_type != SUB_REL_KIND_SEQUENCE && not_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 (rel_type == SUB_REL_KIND_SEQUENCE && not_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 +544,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 (rel_type == SUB_REL_KIND_TABLE && subreltype == RELKIND_SEQUENCE)
+ continue;
+
+ /* If only sequences were requested, skip the tables */
+ if (rel_type == SUB_REL_KIND_SEQUENCE && subreltype != RELKIND_SEQUENCE)
+ 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 bff990afa7..a3e7c791b2 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -342,7 +342,7 @@ ResetSequence(Oid seq_relid)
* logical replication.
*/
void
-SetSequenceLastValue(Oid seq_relid, int64 new_last_value)
+SetSequenceLastValue(Oid seq_relid, int64 new_last_value, int64 log_cnt)
{
SeqTable elm;
Relation seqrel;
@@ -370,7 +370,7 @@ SetSequenceLastValue(Oid seq_relid, int64 new_last_value)
seq->last_value = new_last_value;
seq->is_called = true;
- seq->log_cnt = 0;
+ seq->log_cnt = log_cnt;
MarkBufferDirty(buf);
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index d124bfe55c..3318598f08 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,25 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
return myself;
}
+/*
+ * Update the subscription to refresh both the publication and the publication
+ * objects associated with the subscription.
+ *
+ * If the copy_data parameter is true, the function will set the state
+ * to "init"; otherwise, it will set the state to "ready". When the
+ * validate_publications is provided with a publication list, the function
+ * checks that the specified publications exist on the publisher. If
+ * refresh_all_sequences is true, it will mark all sequences with "init" state
+ * for re-synchronization; otherwise, only the newly added relations 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_all_sequences)
{
char *err;
- List *pubrel_names;
+ List *pubrel_names = NIL;
List *subrel_states;
Oid *subrel_local_oids;
Oid *pubrel_local_oids;
@@ -885,14 +917,21 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
PG_TRY();
{
+ SubscriptionRelKind reltype = refresh_all_sequences ?
+ SUB_REL_KIND_SEQUENCE : SUB_REL_KIND_ALL;
+
if (validate_publications)
check_publications(wrconn, validate_publications);
- /* Get the table list from publisher. */
- pubrel_names = fetch_table_list(wrconn, sub->publications);
+ if (reltype == SUB_REL_KIND_ALL)
+ /* Get the table list from publisher. */
+ pubrel_names = fetch_table_list(wrconn, sub->publications);
+
+ /* Get the sequence list from publisher. */
+ 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, reltype, false);
subrel_count = list_length(subrel_states);
/*
@@ -911,9 +950,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_all_sequences)
+ 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
@@ -973,6 +1013,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,30 +1035,37 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
/* Last known rel state. */
state = GetSubscriptionRelState(sub->oid, relid, &statelsn);
+ RemoveSubscriptionRel(sub->oid, relid);
+
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);
+ /*
+ * Since one sequence sync workers synchronizes all the
+ * sequences, stop the worker only if relation kind is not
+ * sequence.
+ */
+ if (relkind != RELKIND_SEQUENCE)
+ logicalrep_worker_stop(sub->oid, relid);
/*
* For READY state, we would have already dropped the
* tablesync origin.
*/
- if (state != SUBREL_STATE_READY)
+ if (state != SUBREL_STATE_READY && relkind != RELKIND_SEQUENCE)
{
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.
+ * 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));
@@ -1025,10 +1073,25 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
}
ereport(DEBUG1,
- (errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"",
+ (errmsg_internal("%s \"%s.%s\" removed from subscription \"%s\"",
+ get_namespace_name(get_rel_namespace(relid)),
+ get_rel_name(relid),
+ sub->name,
+ get_rel_relkind(relid) == RELKIND_SEQUENCE ? "sequence" : "table")));
+ }
+ /*
+ * In case of REFRESH PUBLICATION SEQUENCES, the existing sequences
+ * should be re-synchronized.
+ */
+ else if (refresh_all_sequences)
+ {
+ 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 +1102,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 +1490,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 +1505,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
sub->publications = stmt->publication;
AlterSubscription_refresh(sub, opts.copy_data,
- stmt->publication);
+ stmt->publication, false);
}
break;
@@ -1479,8 +1545,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 +1564,27 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
sub->publications = publist;
AlterSubscription_refresh(sub, opts.copy_data,
- validate_publications);
+ validate_publications, 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, true);
+
+ break;
+ }
+
+ case ALTER_SUBSCRIPTION_REFRESH_PUBLICATION:
{
if (!sub->enabled)
ereport(ERROR,
@@ -1539,7 +1619,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, false);
break;
}
@@ -1804,7 +1884,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, SUB_REL_KIND_TABLE, true);
foreach(lc, rstates)
{
SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
@@ -2162,11 +2242,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 +2420,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 585f61e414..3f66ddd616 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..76b934234b 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -237,7 +237,8 @@ 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.
+ * We are only interested in the leader apply worker, table sync worker and
+ * sequence sync worker.
*/
LogicalRepWorker *
logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
@@ -267,6 +268,38 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
return res;
}
+/*
+ * Walks the workers array and searches for one that matches given
+ * subscription id.
+ *
+ * We are only interested in the sequence sync worker.
+ */
+LogicalRepWorker *
+logicalrep_sequence_sync_worker_find(Oid subid, bool only_running)
+{
+ LogicalRepWorker *res = NULL;
+
+ Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+ /* Search for attached worker for a given subscription id. */
+ for (int i = 0; i < max_logical_replication_workers; i++)
+ {
+ LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+ /* Skip non sequence sync workers. */
+ if (!isSequenceSyncWorker(w))
+ continue;
+
+ if (w->in_use && w->subid == subid && (only_running && w->proc))
+ {
+ res = w;
+ break;
+ }
+ }
+
+ return res;
+}
+
/*
* Similar to logicalrep_worker_find(), but returns a list of all workers for
* the subscription, instead of just one.
@@ -297,6 +330,26 @@ logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
return res;
}
+/*
+ * Return the pid of the apply worker for one that matches given
+ * subscription id.
+ */
+static LogicalRepWorker *
+logicalrep_apply_worker_find(Oid subid, bool only_running)
+{
+ Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+ for (int i = 0; i < max_logical_replication_workers; i++)
+ {
+ LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+ if (isApplyWorker(w) && w->subid == subid && only_running && w->proc)
+ return w;
+ }
+
+ return NULL;
+}
+
/*
* Start new logical replication background worker, if possible.
*
@@ -317,6 +370,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,11 +378,15 @@ 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));
+ if (is_sequencesync_worker)
+ Assert(!OidIsValid(relid));
+
ereport(DEBUG1,
(errmsg_internal("starting logical replication worker for subscription \"%s\"",
subname)));
@@ -402,7 +460,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 +548,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 +556,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");
@@ -815,6 +882,26 @@ logicalrep_launcher_onexit(int code, Datum arg)
LogicalRepCtx->launcher_pid = 0;
}
+/*
+ * Update the failure time for 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_apply_worker_find(MyLogicalRepWorker->subid, true);
+ if (worker)
+ worker->sequencesync_failure_time = GetCurrentTimestamp();
+
+ LWLockRelease(LogicalRepWorkerLock);
+}
+
/*
* Cleanup function.
*
@@ -863,7 +950,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++;
}
@@ -1314,7 +1401,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 +1444,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..d8b007f9e9
--- /dev/null
+++ b/src/backend/replication/logical/sequencesync.c
@@ -0,0 +1,356 @@
+/*-------------------------------------------------------------------------
+ * 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
+ *
+ * Fetch sequence data (current state) from the remote node, including
+ * the latest sequence value from the publisher and the Page LSN for the
+ * sequence.
+ */
+static int64
+fetch_remote_sequence_data(WalReceiverConn *conn, Oid remoteid,
+ int64 *log_cnt, XLogRecPtr *lsn)
+{
+ WalRcvExecResult *res;
+ StringInfoData cmd;
+ TupleTableSlot *slot;
+ Oid tableRow[3] = {INT8OID, INT8OID, LSNOID};
+ int64 value = (Datum) 0;
+
+ initStringInfo(&cmd);
+
+ appendStringInfo(&cmd, "SELECT last_value, log_cnt, page_lsn "
+ "FROM pg_sequence_state(%d)", remoteid);
+
+ res = walrcv_exec(conn, cmd.data, 3, 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);
+ while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ bool isnull;
+
+ value = DatumGetInt64(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+
+ *log_cnt = DatumGetInt64(slot_getattr(slot, 2, &isnull));
+ Assert(!isnull);
+
+ *lsn = DatumGetInt64(slot_getattr(slot, 3, &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
+ * withe the retreived 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);
+
+ /* 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, &log_cnt, &lsn);
+
+ SetSequenceLastValue(RelationGetRelid(rel), sequence_value, 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
+
+ /* Get the sequences that should be synchronized. */
+ StartTransactionCommand();
+ sequences = GetSubscriptionRelations(subid, SUB_REL_KIND_SEQUENCE, true);
+
+ /* Allocate the tracking info in a permanent memory context. */
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ foreach_ptr(SubscriptionRelState, seq_state, sequences)
+ {
+ SubscriptionRelState *rstate = palloc(sizeof(SubscriptionRelState));
+
+ memcpy(rstate, seq_state, sizeof(SubscriptionRelState));
+ sequences_not_synced = lappend(sequences_not_synced, rstate);
+ }
+ MemoryContextSwitchTo(oldctx);
+
+
+ CommitTransactionCommand();
+
+ /* Is the use of a password mandatory? */
+ must_use_password = MySubscription->passwordrequired &&
+ !MySubscription->ownersuperuser;
+
+ 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);
+
+ /*
+ * Verify whether the current batch of sequences is synchronized or if
+ * there are no remaining sequences to synchronize.
+ */
+ if ((((curr_seq + 1) % MAX_SEQUENCES_SYNC_PER_BATCH) == 0) ||
+ (curr_seq + 1) == seq_count)
+ {
+ /* Obtain the starting index of the current batch. */
+ int i = curr_seq - (curr_seq % 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)));
+ }
+
+ CommitTransactionCommand();
+ start_txn = true;
+ }
+
+ curr_seq++;
+ }
+
+ 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..313e5eb357 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -132,16 +132,16 @@ typedef enum
static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
static List *table_states_not_ready = NIL;
-static bool FetchTableStates(bool *started_tx);
+static bool FetchTableStates(bool *started_tx, SubscriptionRelKind rel_type);
static StringInfo copybuf = NULL;
/*
* Exit routine for synchronization worker.
*/
-static void
+void
pg_attribute_noreturn()
-finish_sync_worker(void)
+finish_sync_worker(LogicalRepWorkerType wtype)
{
/*
* Commit any outstanding transaction. This is the usual case, unless
@@ -157,15 +157,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);
+ /* No need to set the sequence failure time when it is a clean exit */
+ if (wtype == WORKERTYPE_SEQUENCESYNC)
+ cancel_before_shmem_exit(logicalrep_seqsyncworker_failuretime, 0);
+
/* Stop gracefully */
proc_exit(0);
}
@@ -387,7 +396,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);
@@ -430,7 +439,7 @@ 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);
+ FetchTableStates(&started_tx, SUB_REL_KIND_ALL);
/*
* Prepare a hash table for tracking last start times of workers, to avoid
@@ -463,6 +472,17 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
foreach(lc, table_states_not_ready)
{
SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
+ char relkind;
+
+ if (!started_tx)
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
+
+ relkind = get_rel_relkind(rstate->relid);
+ if (relkind == RELKIND_SEQUENCE)
+ continue;
if (rstate->state == SUBREL_STATE_SYNCDONE)
{
@@ -477,11 +497,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.
@@ -660,6 +675,108 @@ 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());
+
+ /* We need up-to-date sync state info for subscription sequences here. */
+ FetchTableStates(&started_tx, SUB_REL_KIND_ALL);
+
+ /*
+ * Start sequence sync worker if there is not one already.
+ */
+ foreach_ptr(SubscriptionRelState, rstate, table_states_not_ready)
+ {
+ LogicalRepWorker *syncworker;
+ char relkind;
+
+ if (!started_tx)
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
+
+ relkind = get_rel_relkind(rstate->relid);
+ if (relkind != RELKIND_SEQUENCE || rstate->state != SUBREL_STATE_INIT)
+ continue;
+
+ /*
+ * Check if there is a sequence worker already running?
+ */
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+ syncworker = logicalrep_sequence_sync_worker_find(MyLogicalRepWorker->subid,
+ 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.
*/
@@ -683,6 +800,12 @@ process_syncing_tables(XLogRecPtr current_lsn)
case WORKERTYPE_APPLY:
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 +1443,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. */
@@ -1570,7 +1693,7 @@ copy_table_done:
* then it is the caller's responsibility to commit it.
*/
static bool
-FetchTableStates(bool *started_tx)
+FetchTableStates(bool *started_tx, SubscriptionRelKind rel_type)
{
static bool has_subrels = false;
@@ -1596,7 +1719,7 @@ FetchTableStates(bool *started_tx)
}
/* Fetch all non-ready tables. */
- rstates = GetSubscriptionRelations(MySubscription->oid, true);
+ rstates = GetSubscriptionRelations(MySubscription->oid, rel_type, true);
/* Allocate the tracking info in a permanent memory context. */
oldctx = MemoryContextSwitchTo(CacheMemoryContext);
@@ -1709,7 +1832,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 +1840,7 @@ TablesyncWorkerMain(Datum main_arg)
run_tablesync_worker();
- finish_sync_worker();
+ finish_sync_worker(WORKERTYPE_TABLESYNC);
}
/*
@@ -1735,7 +1858,7 @@ AllTablesyncsReady(void)
bool has_subrels = false;
/* We need up-to-date sync state info for subscription tables here. */
- has_subrels = FetchTableStates(&started_tx);
+ has_subrels = FetchTableStates(&started_tx, SUB_REL_KIND_TABLE);
if (started_tx)
{
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ec96b5fe85..e60e52dff2 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 (isSequenceSyncWorker(MyLogicalRepWorker))
+ 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 1a949966e0..e09e321ab1 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11968,6 +11968,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.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a2..8c96f0ce72 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -159,6 +159,12 @@ typedef struct Subscription
* specified origin */
} Subscription;
+typedef struct SubscriptionSeqInfo
+{
+ Oid seqid;
+ XLogRecPtr lsn;
+} SubscriptionSeqInfo;
+
/* Disallow streaming in-progress transactions. */
#define LOGICALREP_STREAM_OFF 'f'
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 8244ad537a..5584136a35 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -82,6 +82,13 @@ typedef struct SubscriptionRelState
char state;
} SubscriptionRelState;
+typedef enum
+{
+ SUB_REL_KIND_TABLE,
+ SUB_REL_KIND_SEQUENCE,
+ SUB_REL_KIND_ALL,
+} SubscriptionRelKind;
+
extern void AddSubscriptionRelState(Oid subid, Oid relid, char state,
XLogRecPtr sublsn, bool retain_lock);
extern void UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
@@ -90,6 +97,7 @@ 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, SubscriptionRelKind reltype,
+ bool not_ready);
#endif /* PG_SUBSCRIPTION_REL_H */
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 003f2e3413..a302890156 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -60,7 +60,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 SetSequenceLastValue(Oid seq_relid, int64 new_last_value);
+extern void SetSequenceLastValue(Oid seq_relid, int64 new_last_value, 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..adb1c6e32b 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;
/*
@@ -242,6 +245,8 @@ extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running,
bool acquire_lock);
+extern LogicalRepWorker *logicalrep_sequence_sync_worker_find(Oid subid,
+ bool only_running);
extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
Oid dbid, Oid subid, const char *subname,
Oid userid, Oid relid,
@@ -253,6 +258,10 @@ 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 4c789279e5..bd5efd5d27 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1442,6 +1442,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..adf7b0cfc1
--- /dev/null
+++ b/src/test/subscription/t/034_sequences.pl
@@ -0,0 +1,154 @@
+
+# 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();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 90326c6e53..a875c196a8 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2772,7 +2772,9 @@ SubscriptingRef
SubscriptingRefState
Subscription
SubscriptionInfo
+SubscriptionRelKind
SubscriptionRelState
+SubscriptionSeqInfo
SummarizerReadLocalXLogPrivate
SupportRequestCost
SupportRequestIndexCondition
--
2.34.1