public inbox for [email protected]
help / color / mirror / Atom feedRe: [PATCH] Support automatic sequence replication
2+ messages / 2 participants
[nested] [flat]
* Re: [PATCH] Support automatic sequence replication
@ 2026-02-05 05:03 Ajin Cherian <[email protected]>
0 siblings, 1 reply; 2+ messages in thread
From: Ajin Cherian @ 2026-02-05 05:03 UTC (permalink / raw)
To: shveta malik <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>
On Tue, Feb 3, 2026 at 9:22 PM shveta malik <[email protected]> wrote:
>
> On Tue, Feb 3, 2026 at 9:18 AM Ajin Cherian <[email protected]> wrote:
> Thanks for the patch.
>
> +1 for the overall idea of patch that once a subscription is created
> which subscribes to sequences, a sequence sync worker is started which
> continuously syncs the sequences. This makes usage of REFRESH
> SEQUENCES redundant and thus it is removed. I am still reviewing the
> design choice here, and will post my comments soon (if any).
>
Thanks!
> By quick validation, few issues in current implementation:
>
> 1)
> If the sequence sync worker exits due to some issue (or killed or
> server restarts), sequence-sync worker is not started again by apply
> worker unless there is a sequence in INIT state i.e. synchronization
> of sequences in READY state stops. IIUC, the logic of
> ProcessSequencesForSync() needs to change to start seq sync worker
> irrespective of state of sequences.
>
Yes, I fixed this. I've changed FetchRelationStates to fetch sequences
in ANY state and not just ones in NON READY state.
> 2)
> There is some issue in how LOGs (DEBUGs) are getting generated.
>
> a) Even if there is no drift, it still keeps on dumping:
> "logical replication sequence synchronization for subscription "sub1"
> - total unsynchronized: 3"
>
Removed this.
> b)
> When there is a drift in say single sequence, it puts rest (which are
> in sync) to "missing" section:
> "logical replication sequence synchronization for subscription "sub1"
> - batch #1 = 3 attempted, 1 succeeded, 0 mismatched, 0 insufficient
> permission, 2 missing from publisher, 0 skipped"
>
Fixed, and added a new section called "no drift". Also now this debug
message is printed every time the worker attempts to synchronize
sequences. also mentioning the state of the sequences being synced.
> 3)
> If a sequence sync worker is taking a nap, and subscription is
> disabled or the server is stopped just before upgrade, how is the user
> supposed to know that sequences are synced at the end?
Well, one way is to wait for a debug message that says that all the
attempted sequences are in the "no drift" state. Also remote
sequence's LSN is updated in pg_subscription_rel for each sequence.
Let me know if you have anything more in mind. One option is to leave
the ALTER SUBSCRIPTION..REFRESH SEQUENCE in place, that will change
the state of all the sequences to the INIT state, and the user can
then wait for the sequences to change state to READY.
Attaching patch v3 addressing the above comments.
regards,
Ajin Cherian
Fujitsu Australia
Attachments:
[application/octet-stream] v3-0001-Support-automatic-sequence-replication.patch (38.8K, 2-v3-0001-Support-automatic-sequence-replication.patch)
download | inline diff:
From 24d719f882709b02f9cef6e76ac4b74f80cc2034 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Mon, 2 Feb 2026 21:41:54 +1100
Subject: [PATCH v3] Support automatic sequence replication
Logical replication sequences can drift between publisher and
subscriber as values are consumed independently on each node.
Previously, the sequence sync worker exited after the initial
synchronization, allowing sequences to diverge over time.
This change keeps the sequence sync worker running continuously
so it can monitor sequences and resynchronize them when drift
is detected. The worker uses an adaptive sleep interval:
it starts at 2 seconds, doubles up to a maximum of 30 seconds
when no drift is observed, and resets to the minimum interval
once drift is found.
Since synchronization is now continuous,
ALTER SUBSCRIPTION ... REFRESH SEQUENCES is no longer needed
and has been removed.
Sequences remain in the READY state during continuous
synchronization.
Author: Ajin Cherian [email protected]
---
doc/src/sgml/catalogs.sgml | 6 +-
doc/src/sgml/logical-replication.sgml | 56 +---
doc/src/sgml/ref/alter_subscription.sgml | 30 ---
src/backend/commands/sequence.c | 36 +++
src/backend/commands/subscriptioncmds.c | 65 -----
src/backend/parser/gram.y | 9 -
.../replication/logical/sequencesync.c | 242 +++++++++++++-----
src/backend/replication/logical/syncutils.c | 47 ++--
src/bin/psql/tab-complete.in.c | 4 +-
src/include/commands/sequence.h | 1 +
src/include/nodes/parsenodes.h | 1 -
src/test/subscription/t/026_stats.pl | 2 +
src/test/subscription/t/036_sequences.pl | 71 +----
13 files changed, 260 insertions(+), 310 deletions(-)
mode change 100644 => 100755 src/backend/replication/logical/sequencesync.c
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 332193565e2..ec51a87c253 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8261,11 +8261,9 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<para>
This catalog contains tables and sequences known to the subscription
after running:
- <link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link>,
+ <link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link>, or
<link linkend="sql-altersubscription-params-refresh-publication">
- <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command></link>, or
- <link linkend="sql-altersubscription-params-refresh-sequences">
- <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>.
+ <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command></link>.
</para>
<table>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 5028fe9af09..c1ed960cea8 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1775,20 +1775,14 @@ Publications:
to synchronize only newly added sequences.
</para>
</listitem>
- <listitem>
- <para>
- use <link linkend="sql-altersubscription-params-refresh-sequences">
- <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>
- to re-synchronize all sequences currently known to the subscription.
- </para>
- </listitem>
</itemizedlist>
</para>
<para>
A <firstterm>sequence synchronization worker</firstterm> will be started
- after executing any of the above subscriber commands, and will exit once the
- sequences are synchronized.
+ after executing any of the above subscriber commands, this worker will
+ remain running for the life of the subscription, and will periodically
+ synchronize all published sequences.
</para>
<para>
The ability to launch a sequence synchronization worker is limited by the
@@ -1815,21 +1809,7 @@ Publications:
</sect2>
<sect2 id="sequences-out-of-sync">
- <title>Refreshing Out-of-Sync Sequences</title>
- <para>
- Subscriber sequence values will become out of sync as the publisher
- advances them.
- </para>
- <para>
- To detect this, compare the
- <link linkend="catalog-pg-subscription-rel">pg_subscription_rel</link>.<structfield>srsublsn</structfield>
- on the subscriber with the <structfield>page_lsn</structfield> obtained
- from the <link linkend="func-pg-get-sequence-data"><function>pg_get_sequence_data</function></link>
- function for the sequence on the publisher. Then run
- <link linkend="sql-altersubscription-params-refresh-sequences">
- <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link> to
- re-synchronize if necessary.
- </para>
+ <title>Out-of-Sync Sequences</title>
<warning>
<para>
Each sequence caches a block of values (typically 32) in memory before
@@ -1961,16 +1941,6 @@ Publications:
------------+-----------+------------
610 | t | 0/017CEDF8
(1 row)
-</programlisting></para>
-
- <para>
- The difference between the sequence page LSNs on the publisher and the
- sequence page LSNs on the subscriber indicates that the sequences are out
- of sync. Re-synchronize all sequences known to the subscriber using
- <link linkend="sql-altersubscription-params-refresh-sequences">
- <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>.
-<programlisting>
-/* sub # */ ALTER SUBSCRIPTION sub1 REFRESH SEQUENCES;
</programlisting></para>
<para>
@@ -2333,24 +2303,6 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
</para>
</listitem>
- <listitem>
- <para>
- Incremental sequence changes are not replicated. Although the data in
- serial or identity columns backed by sequences will be replicated as part
- of the table, the sequences themselves do not replicate ongoing changes.
- On the subscriber, a sequence will retain the last value it synchronized
- from the publisher. If the subscriber is used as a read-only database,
- then this should typically not be a problem. If, however, some kind of
- switchover or failover to the subscriber database is intended, then the
- sequences would need to be updated to the latest values, either by
- executing <link linkend="sql-altersubscription-params-refresh-sequences">
- <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>
- or by copying the current data from the publisher (perhaps using
- <command>pg_dump</command>) or by determining a sufficiently high value
- from the tables themselves.
- </para>
- </listitem>
-
<listitem>
<para>
Replication of <command>TRUNCATE</command> commands is supported, but
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 27c06439f4f..c0fd426d712 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -26,7 +26,6 @@ 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 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>] [, ... ] )
@@ -190,11 +189,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
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,
- use <link linkend="sql-altersubscription-params-refresh-sequences">
- <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>.
- </para>
<para>
See <xref linkend="sequence-definition-mismatches"/> for recommendations on how
to handle any warnings about sequence definition differences between
@@ -219,30 +213,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
</listitem>
</varlistentry>
- <varlistentry id="sql-altersubscription-params-refresh-sequences">
- <term><literal>REFRESH SEQUENCES</literal></term>
- <listitem>
- <para>
- Re-synchronize sequence data with the publisher. Unlike
- <link linkend="sql-altersubscription-params-refresh-publication">
- <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command></link> which
- only has the ability to synchronize newly added sequences,
- <literal>REFRESH SEQUENCES</literal> will re-synchronize the sequence
- data for all currently subscribed sequences. It does not add or remove
- sequences from the subscription to match the publication.
- </para>
- <para>
- See <xref linkend="sequence-definition-mismatches"/> for
- recommendations on how to handle any warnings about sequence definition
- differences between the publisher and the subscriber.
- </para>
- <para>
- See <xref linkend="sequences-out-of-sync"/> for recommendations on how to
- identify and handle out-of-sync sequences.
- </para>
- </listitem>
- </varlistentry>
-
<varlistentry id="sql-altersubscription-params-enable">
<term><literal>ENABLE</literal></term>
<listitem>
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index e1b808bbb60..d6f9085e23c 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -929,6 +929,42 @@ lastval(PG_FUNCTION_ARGS)
PG_RETURN_INT64(result);
}
+/*
+ * Read the current sequence values (last_value and is_called)
+ *
+ * This is a read-only operation that acquires AccessShareLock on the sequence.
+ * Used by logical replication sequence synchronization to detect drift.
+ */
+void
+GetSequence(Oid relid, int64 *last_value, bool *is_called)
+{
+ Relation seqrel;
+ Buffer buf;
+ HeapTupleData seqtuple;
+ Form_pg_sequence_data seq;
+
+ /* Open and lock the sequence relation */
+ seqrel = table_open(relid, AccessShareLock);
+
+ /* Verify it's actually a sequence */
+ if (seqrel->rd_rel->relkind != RELKIND_SEQUENCE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a sequence",
+ RelationGetRelationName(seqrel))));
+
+ /* Read the sequence tuple */
+ seq = read_seq_tuple(seqrel, &buf, &seqtuple);
+
+ /* Extract the values */
+ *last_value = seq->last_value;
+ *is_called = seq->is_called;
+
+ /* Release buffer and close relation */
+ UnlockReleaseBuffer(buf);
+ table_close(seqrel, AccessShareLock);
+}
+
/*
* Main internal procedure that handles 2 & 3 arg forms of SETVAL.
*
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 0b3c8499b49..c1b903c5908 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1183,58 +1183,6 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
table_close(rel, NoLock);
}
-/*
- * Marks all sequences with INIT state.
- */
-static void
-AlterSubscription_refresh_seq(Subscription *sub)
-{
- char *err = NULL;
- WalReceiverConn *wrconn;
- bool must_use_password;
-
- /* Load the library providing us libpq calls. */
- load_file("libpqwalreceiver", false);
-
- /* Try to connect to the publisher. */
- must_use_password = sub->passwordrequired && !sub->ownersuperuser;
- wrconn = walrcv_connect(sub->conninfo, true, true, must_use_password,
- sub->name, &err);
- if (!wrconn)
- ereport(ERROR,
- errcode(ERRCODE_CONNECTION_FAILURE),
- errmsg("subscription \"%s\" could not connect to the publisher: %s",
- sub->name, err));
-
- PG_TRY();
- {
- List *subrel_states;
-
- check_publications_origin_sequences(wrconn, sub->publications, true,
- sub->origin, NULL, 0, sub->name);
-
- /* Get local sequence list. */
- subrel_states = GetSubscriptionRelations(sub->oid, false, true, false);
- foreach_ptr(SubscriptionRelState, subrel, subrel_states)
- {
- Oid relid = subrel->relid;
-
- UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_INIT,
- InvalidXLogRecPtr, false);
- 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));
- }
- }
- PG_FINALLY();
- {
- walrcv_disconnect(wrconn);
- }
- PG_END_TRY();
-}
-
/*
* Common checks for altering failover, two_phase, and retain_dead_tuples
* options.
@@ -1871,19 +1819,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
break;
}
- case ALTER_SUBSCRIPTION_REFRESH_SEQUENCES:
- {
- if (!sub->enabled)
- ereport(ERROR,
- errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("%s is not allowed for disabled subscriptions",
- "ALTER SUBSCRIPTION ... REFRESH SEQUENCES"));
-
- AlterSubscription_refresh_seq(sub);
-
- break;
- }
-
case ALTER_SUBSCRIPTION_SKIP:
{
parse_subscription_options(pstate, stmt->options, SUBOPT_LSN, &opts);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 713ee5c10a2..9d362127632 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11069,15 +11069,6 @@ AlterSubscriptionStmt:
n->options = $6;
$$ = (Node *) n;
}
- | ALTER SUBSCRIPTION name REFRESH SEQUENCES
- {
- AlterSubscriptionStmt *n =
- makeNode(AlterSubscriptionStmt);
-
- n->kind = ALTER_SUBSCRIPTION_REFRESH_SEQUENCES;
- n->subname = $3;
- $$ = (Node *) n;
- }
| ALTER SUBSCRIPTION name ADD_P PUBLICATION name_list opt_definition
{
AlterSubscriptionStmt *n =
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
old mode 100644
new mode 100755
index 165f909b3ba..efda42d8ce3
--- a/src/backend/replication/logical/sequencesync.c
+++ b/src/backend/replication/logical/sequencesync.c
@@ -19,10 +19,6 @@
* CREATE SUBSCRIPTION
* ALTER SUBSCRIPTION ... REFRESH PUBLICATION
*
- * Executing the following command resets all sequences in the subscription to
- * state INIT, triggering re-synchronization:
- * ALTER SUBSCRIPTION ... REFRESH SEQUENCES
- *
* The apply worker periodically scans pg_subscription_rel for sequences in
* INIT state. When such sequences are found, it spawns a sequencesync worker
* to handle synchronization.
@@ -36,8 +32,24 @@
* local subscriber, and finally marks each sequence as READY upon successful
* synchronization.
*
+ * The sequencesync worker then fetches all sequences that are
+ * in the READY state, queries the publisher for current sequence values, and
+ * updates any sequences that have drifted and then goes to sleep. The sleep
+ * interval starts at SEQSYNC_MIN_SLEEP_MS and doubles after each wake cycle
+ * (up to SEQSYNC_MAX_SLEEP_MS). When drift is detected, the interval resets to
+ * the minimum to ensure timely updates.
+ *
+ * After CREATE SUBSCRIPTION, sequences begin in the INIT state. Sequences
+ * added through ALTER SUBSCRIPTION.. REFRESH PUBLICATION also start in the INIT
+ * state. All INIT sequences are synchronized unconditionally, then transition
+ * to the READY state. Once in the READY state, sequences are checked for drift
+ * from the publisher and synchronized only when drift is detected.
+ *
* Sequence state transitions follow this pattern:
- * INIT -> READY
+ * INIT --> READY ->-+
+ * ^ | (check/synchronzize)
+ * | |
+ * +--<---+
*
* To avoid creating too many transactions, up to MAX_SEQUENCES_SYNC_PER_BATCH
* sequences are synchronized per transaction. The locks on the sequence
@@ -78,11 +90,19 @@ typedef enum CopySeqResult
COPYSEQ_SUCCESS,
COPYSEQ_MISMATCH,
COPYSEQ_INSUFFICIENT_PERM,
- COPYSEQ_SKIPPED
+ COPYSEQ_SKIPPED,
+ COPYSEQ_NOWORK,
} CopySeqResult;
+/* List of sequences that is currently being synced */
static List *seqinfos = NIL;
+/* Sleep intervals for sync */
+#define SEQSYNC_MIN_SLEEP_MS 2000 /* 2 seconds */
+#define SEQSYNC_MAX_SLEEP_MS 30000 /* 30 seconds */
+
+static long sleep_ms = SEQSYNC_MIN_SLEEP_MS;
+
/*
* Apply worker determines if sequence synchronization is needed.
*
@@ -325,11 +345,12 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
}
/*
- * Apply remote sequence state to local sequence and mark it as
- * synchronized (READY).
+ * Apply remote sequence state to local sequence. If we are doing this
+ * for sequences in the INIT state, move them to the READY state once
+ * synchronized.
*/
static CopySeqResult
-copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner)
+copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner, char relstate)
{
UserContext ucxt;
AclResult aclresult;
@@ -368,19 +389,47 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner)
/*
* Record the remote sequence's LSN in pg_subscription_rel and mark the
- * sequence as READY.
+ * sequence as READY if updating a sequence that is in INIT state.
*/
- UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY,
- seqinfo->page_lsn, false);
+ if (relstate == SUBREL_STATE_INIT)
+ UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY,
+ seqinfo->page_lsn, false);
return COPYSEQ_SUCCESS;
}
+/*
+ * check_sequence_drift
+ *
+ * Check if the remote sequence values differ from the local sequence.
+ * Returns true/false if any sequences drifted.
+ */
+static bool
+check_sequence_drift(LogicalRepSequenceInfo *seqinfo)
+{
+ int64 local_last_value;
+ bool local_is_called;
+
+ /* Get current local sequence state */
+ GetSequence(seqinfo->localrelid, &local_last_value, &local_is_called);
+
+ /* Check if values have drifted and return accordingly */
+ return (local_last_value != seqinfo->last_value ||
+ local_is_called != seqinfo->is_called);
+
+}
+
/*
* Copy existing data of sequences from the publisher.
+ *
+ * If relstate is SUBREL_STATE_READY, only synchronize sequences that
+ * have drifted from their publisher values. Otherwise, synchronize
+ * all sequences.
+ *
+ * Returns true/false if any sequences were actually copied.
*/
-static void
-copy_sequences(WalReceiverConn *conn)
+static bool
+copy_sequences(WalReceiverConn *conn, char relstate)
{
int cur_batch_base_index = 0;
int n_seqinfos = list_length(seqinfos);
@@ -390,13 +439,10 @@ copy_sequences(WalReceiverConn *conn)
StringInfo seqstr = makeStringInfo();
StringInfo cmd = makeStringInfo();
MemoryContext oldctx;
+ bool drift_detected = false;
#define MAX_SEQUENCES_SYNC_PER_BATCH 100
- elog(DEBUG1,
- "logical replication sequence synchronization for subscription \"%s\" - total unsynchronized: %d",
- MySubscription->name, n_seqinfos);
-
while (cur_batch_base_index < n_seqinfos)
{
Oid seqRow[REMOTE_SEQ_COL_COUNT] = {INT8OID, INT8OID,
@@ -406,6 +452,7 @@ copy_sequences(WalReceiverConn *conn)
int batch_mismatched_count = 0;
int batch_skipped_count = 0;
int batch_insuffperm_count = 0;
+ int batch_no_drift = 0;
int batch_missing_count;
Relation sequence_rel = NULL;
@@ -491,6 +538,7 @@ copy_sequences(WalReceiverConn *conn)
CopySeqResult sync_status;
LogicalRepSequenceInfo *seqinfo;
int seqidx;
+ bool should_sync = true;
CHECK_FOR_INTERRUPTS();
@@ -502,19 +550,34 @@ copy_sequences(WalReceiverConn *conn)
sync_status = get_and_validate_seq_info(slot, &sequence_rel,
&seqinfo, &seqidx);
- if (sync_status == COPYSEQ_SUCCESS)
+
+ /*
+ * For sequences in READY state, only sync if there's drift
+ */
+ if (relstate == SUBREL_STATE_READY && sync_status == COPYSEQ_SUCCESS)
+ {
+ should_sync = check_sequence_drift(seqinfo);
+ if (should_sync)
+ drift_detected = true;
+ }
+
+ if (sync_status == COPYSEQ_SUCCESS && should_sync)
sync_status = copy_sequence(seqinfo,
- sequence_rel->rd_rel->relowner);
+ sequence_rel->rd_rel->relowner,
+ relstate);
+ else if (sync_status == COPYSEQ_SUCCESS && !should_sync)
+ sync_status = COPYSEQ_NOWORK;
switch (sync_status)
{
case COPYSEQ_SUCCESS:
elog(DEBUG1,
- "logical replication synchronization for subscription \"%s\", sequence \"%s.%s\" has finished",
+ "logical replication sync for subscription \"%s\", sequence \"%s.%s\" has been updated",
MySubscription->name, seqinfo->nspname,
seqinfo->seqname);
batch_succeeded_count++;
break;
+
case COPYSEQ_MISMATCH:
/*
@@ -528,6 +591,7 @@ copy_sequences(WalReceiverConn *conn)
MemoryContextSwitchTo(oldctx);
batch_mismatched_count++;
break;
+
case COPYSEQ_INSUFFICIENT_PERM:
/*
@@ -541,6 +605,7 @@ copy_sequences(WalReceiverConn *conn)
MemoryContextSwitchTo(oldctx);
batch_insuffperm_count++;
break;
+
case COPYSEQ_SKIPPED:
/*
@@ -558,6 +623,15 @@ copy_sequences(WalReceiverConn *conn)
batch_skipped_count++;
}
break;
+
+ case COPYSEQ_NOWORK:
+ /* Nothing to do */
+ batch_no_drift++;
+ break;
+
+ default:
+ elog(ERROR, "unrecognized Sequence replication result: %d", (int) sync_status);
+
}
if (sequence_rel)
@@ -572,14 +646,16 @@ copy_sequences(WalReceiverConn *conn)
batch_missing_count = batch_size - (batch_succeeded_count +
batch_mismatched_count +
batch_insuffperm_count +
- batch_skipped_count);
+ batch_skipped_count +
+ batch_no_drift);
elog(DEBUG1,
- "logical replication sequence synchronization for subscription \"%s\" - batch #%d = %d attempted, %d succeeded, %d mismatched, %d insufficient permission, %d missing from publisher, %d skipped",
- MySubscription->name,
- (cur_batch_base_index / MAX_SEQUENCES_SYNC_PER_BATCH) + 1,
- batch_size, batch_succeeded_count, batch_mismatched_count,
- batch_insuffperm_count, batch_missing_count, batch_skipped_count);
+ "logical replication sequence synchronization for subscription \"%s\" - for sequences in %s state batch #%d = %d attempted, %d succeeded, %d mismatched, %d insufficient permission, %d missing from publisher, %d skipped, %d no drift",
+ MySubscription->name,
+ (relstate == 'r') ? "READY" : (relstate == 'i') ? "INIT" : "UNKNOWN",
+ (cur_batch_base_index / MAX_SEQUENCES_SYNC_PER_BATCH) + 1,
+ batch_size, batch_succeeded_count, batch_mismatched_count,
+ batch_insuffperm_count, batch_missing_count, batch_skipped_count, batch_no_drift);
/* Commit this batch, and prepare for next batch */
CommitTransactionCommand();
@@ -608,23 +684,28 @@ copy_sequences(WalReceiverConn *conn)
/* Report mismatches, permission issues, or missing sequences */
report_sequence_errors(mismatched_seqs_idx, insuffperm_seqs_idx,
missing_seqs_idx);
+
+ list_free(seqinfos);
+ seqinfos = NIL;
+
+ return drift_detected;
}
/*
* Identifies sequences that require synchronization and initiates the
* synchronization process.
+ *
+ * Returns true if sequences have been updated.
*/
-static void
-LogicalRepSyncSequences(void)
+static bool
+LogicalRepSyncSequences(WalReceiverConn *conn, char relstate)
{
- char *err;
- bool must_use_password;
Relation rel;
HeapTuple tup;
ScanKeyData skey[2];
SysScanDesc scan;
Oid subid = MyLogicalRepWorker->subid;
- StringInfoData app_name;
+ bool sequence_copied = false;
StartTransactionCommand();
@@ -636,9 +717,9 @@ LogicalRepSyncSequences(void)
ObjectIdGetDatum(subid));
ScanKeyInit(&skey[1],
- Anum_pg_subscription_rel_srsubstate,
- BTEqualStrategyNumber, F_CHAREQ,
- CharGetDatum(SUBREL_STATE_INIT));
+ Anum_pg_subscription_rel_srsubstate,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(relstate));
scan = systable_beginscan(rel, InvalidOid, false,
NULL, 2, skey);
@@ -693,36 +774,16 @@ LogicalRepSyncSequences(void)
* Exit early if no catalog entries found, likely due to concurrent drops.
*/
if (!seqinfos)
- return;
-
- /* Is the use of a password mandatory? */
- must_use_password = MySubscription->passwordrequired &&
- !MySubscription->ownersuperuser;
+ return false;
- initStringInfo(&app_name);
- appendStringInfo(&app_name, "pg_%u_sequence_sync_" UINT64_FORMAT,
- MySubscription->oid, GetSystemIdentifier());
+ /* Perform synchronization */
+ sequence_copied = copy_sequences(conn, relstate);
- /*
- * Establish the connection to the publisher for sequence synchronization.
- */
- LogRepWorkerWalRcvConn =
- walrcv_connect(MySubscription->conninfo, true, true,
- must_use_password,
- app_name.data, &err);
- if (LogRepWorkerWalRcvConn == NULL)
- ereport(ERROR,
- errcode(ERRCODE_CONNECTION_FAILURE),
- errmsg("sequencesync worker for subscription \"%s\" could not connect to the publisher: %s",
- MySubscription->name, err));
-
- pfree(app_name.data);
-
- copy_sequences(LogRepWorkerWalRcvConn);
+ return sequence_copied;
}
/*
- * Execute the initial sync with error handling. Disable the subscription,
+ * Execute the sequence sync with error handling. Disable the subscription,
* if required.
*
* Note that we don't handle FATAL errors which are probably because of system
@@ -735,8 +796,65 @@ start_sequence_sync(void)
PG_TRY();
{
- /* Call initial sync. */
- LogicalRepSyncSequences();
+ char *err;
+ bool must_use_password;
+ StringInfoData app_name;
+
+ /* Is the use of a password mandatory? */
+ must_use_password = MySubscription->passwordrequired &&
+ !MySubscription->ownersuperuser;
+
+ initStringInfo(&app_name);
+ appendStringInfo(&app_name, "pg_%u_sequence_sync_" UINT64_FORMAT,
+ MySubscription->oid, GetSystemIdentifier());
+
+ /*
+ * Establish the connection to the publisher for sequence synchronization.
+ */
+ LogRepWorkerWalRcvConn =
+ walrcv_connect(MySubscription->conninfo, true, true,
+ must_use_password,
+ app_name.data, &err);
+ if (LogRepWorkerWalRcvConn == NULL)
+ ereport(ERROR,
+ errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("sequencesync worker for subscription \"%s\" could not connect to the publisher: %s",
+ MySubscription->name, err));
+
+ pfree(app_name.data);
+
+ for (;;)
+ {
+ bool sequence_copied = false;
+
+ CHECK_FOR_INTERRUPTS();
+
+ /* Run sync for sequences in READY state */
+ sequence_copied |= LogicalRepSyncSequences(LogRepWorkerWalRcvConn, SUBREL_STATE_READY);
+
+ /* Call initial sync for sequences in INIT state */
+ sequence_copied |= LogicalRepSyncSequences(LogRepWorkerWalRcvConn, SUBREL_STATE_INIT);
+
+ /* Adjust sleep interval based on drift detection: */
+ if (sequence_copied)
+ {
+ sleep_ms = SEQSYNC_MIN_SLEEP_MS;
+ }
+ else
+ {
+ /* No drift detected, so double the sleep time, but not beyond
+ * the maximum allowable value.
+ */
+ sleep_ms = Min(sleep_ms * 2, SEQSYNC_MAX_SLEEP_MS);
+ }
+
+ /* Sleep for the configured interval */
+ (void) WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ sleep_ms,
+ WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+ ResetLatch(MyLatch);
+ }
}
PG_CATCH();
{
diff --git a/src/backend/replication/logical/syncutils.c b/src/backend/replication/logical/syncutils.c
index 535ffb6f09e..668325c355b 100644
--- a/src/backend/replication/logical/syncutils.c
+++ b/src/backend/replication/logical/syncutils.c
@@ -190,13 +190,13 @@ ProcessSyncingRelations(XLogRecPtr current_lsn)
*
* The pg_subscription_rel catalog is shared by tables and sequences. Changes
* to either sequences or tables can affect the validity of relation states, so
- * we identify non-READY tables and non-READY sequences together to ensure
+ * we identify non-READY tables and sequences (in any state) together to ensure
* consistency.
*
* has_pending_subtables: true if the subscription has one or more tables that
* are not in READY state, otherwise false.
* has_pending_subsequences: true if the subscription has one or more sequences
- * that are not in READY state, otherwise false.
+ * (in any state), otherwise false.
*/
void
FetchRelationStates(bool *has_pending_subtables,
@@ -204,53 +204,47 @@ FetchRelationStates(bool *has_pending_subtables,
bool *started_tx)
{
/*
- * has_subtables and has_subsequences_non_ready are declared as static,
+ * has_subtables and has_subsequences are declared as static,
* since the same value can be used until the system table is invalidated.
*/
static bool has_subtables = false;
- static bool has_subsequences_non_ready = false;
-
+ static bool has_subsequences = false;
*started_tx = false;
-
if (relation_states_validity != SYNC_RELATIONS_STATE_VALID)
{
MemoryContext oldctx;
List *rstates;
+ List *seq_states;
SubscriptionRelState *rstate;
-
relation_states_validity = SYNC_RELATIONS_STATE_REBUILD_STARTED;
- has_subsequences_non_ready = false;
-
+ has_subsequences = false;
/* Clean the old lists. */
list_free_deep(table_states_not_ready);
table_states_not_ready = NIL;
-
if (!IsTransactionState())
{
StartTransactionCommand();
*started_tx = true;
}
-
- /* Fetch tables and sequences that are in non-READY state. */
- rstates = GetSubscriptionRelations(MySubscription->oid, true, true,
+ /* Fetch tables that are in non-READY state. */
+ rstates = GetSubscriptionRelations(MySubscription->oid, true, false,
true);
-
+ /* Fetch all sequences (regardless of state). */
+ seq_states = GetSubscriptionRelations(MySubscription->oid, false, true,
+ false);
/* Allocate the tracking info in a permanent memory context. */
oldctx = MemoryContextSwitchTo(CacheMemoryContext);
foreach_ptr(SubscriptionRelState, subrel, rstates)
{
- if (get_rel_relkind(subrel->relid) == RELKIND_SEQUENCE)
- has_subsequences_non_ready = true;
- else
- {
- rstate = palloc_object(SubscriptionRelState);
- memcpy(rstate, subrel, sizeof(SubscriptionRelState));
- table_states_not_ready = lappend(table_states_not_ready,
- rstate);
- }
+ rstate = palloc_object(SubscriptionRelState);
+ memcpy(rstate, subrel, sizeof(SubscriptionRelState));
+ table_states_not_ready = lappend(table_states_not_ready,
+ rstate);
}
+ /* Check if there are any sequences. */
+ has_subsequences = (seq_states != NIL);
MemoryContextSwitchTo(oldctx);
-
+ list_free_deep(seq_states);
/*
* Does the subscription have tables?
*
@@ -260,7 +254,6 @@ FetchRelationStates(bool *has_pending_subtables,
*/
has_subtables = (table_states_not_ready != NIL) ||
HasSubscriptionTables(MySubscription->oid);
-
/*
* If the subscription relation cache has been invalidated since we
* entered this routine, we still use and return the relations we just
@@ -271,10 +264,8 @@ FetchRelationStates(bool *has_pending_subtables,
if (relation_states_validity == SYNC_RELATIONS_STATE_REBUILD_STARTED)
relation_states_validity = SYNC_RELATIONS_STATE_VALID;
}
-
if (has_pending_subtables)
*has_pending_subtables = has_subtables;
-
if (has_pending_subsequences)
- *has_pending_subsequences = has_subsequences_non_ready;
+ *has_pending_subsequences = has_subsequences;
}
diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c
index 8b91bc00062..b8bdfebb15a 100644
--- a/src/bin/psql/tab-complete.in.c
+++ b/src/bin/psql/tab-complete.in.c
@@ -2331,8 +2331,8 @@ match_previous_words(int pattern_id,
/* ALTER SUBSCRIPTION <name> */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny))
COMPLETE_WITH("CONNECTION", "ENABLE", "DISABLE", "OWNER TO",
- "RENAME TO", "REFRESH PUBLICATION", "REFRESH SEQUENCES",
- "SET", "SKIP (", "ADD PUBLICATION", "DROP PUBLICATION");
+ "RENAME TO", "REFRESH PUBLICATION", "SET", "SKIP (",
+ "ADD PUBLICATION", "DROP PUBLICATION");
/* ALTER SUBSCRIPTION <name> REFRESH */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH"))
COMPLETE_WITH("PUBLICATION", "SEQUENCES");
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 2c3c4a3f074..98559209e70 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -47,6 +47,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 GetSequence(Oid relid, int64 *last_value, bool *is_called);
extern void SetSequence(Oid relid, int64 next, bool iscalled);
extern void ResetSequenceCaches(void);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 646d6ced763..43b3292fd2c 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4391,7 +4391,6 @@ typedef enum AlterSubscriptionType
ALTER_SUBSCRIPTION_ADD_PUBLICATION,
ALTER_SUBSCRIPTION_DROP_PUBLICATION,
ALTER_SUBSCRIPTION_REFRESH_PUBLICATION,
- ALTER_SUBSCRIPTION_REFRESH_SEQUENCES,
ALTER_SUBSCRIPTION_ENABLED,
ALTER_SUBSCRIPTION_SKIP,
} AlterSubscriptionType;
diff --git a/src/test/subscription/t/026_stats.pl b/src/test/subscription/t/026_stats.pl
index 5d457060a02..2fe209f461f 100644
--- a/src/test/subscription/t/026_stats.pl
+++ b/src/test/subscription/t/026_stats.pl
@@ -16,6 +16,8 @@ $node_publisher->start;
# Create subscriber node.
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ "max_logical_replication_workers = 10");
$node_subscriber->start;
diff --git a/src/test/subscription/t/036_sequences.pl b/src/test/subscription/t/036_sequences.pl
index 471780a3585..fac16d6e964 100644
--- a/src/test/subscription/t/036_sequences.pl
+++ b/src/test/subscription/t/036_sequences.pl
@@ -75,8 +75,7 @@ is($result, '100|t',
##########
## ALTER SUBSCRIPTION ... REFRESH PUBLICATION should cause sync of new
-# sequences of the publisher, but changes to existing sequences should
-# not be synced.
+# sequences of the publisher.
##########
# Create a new sequence 'regress_s2', and update existing sequence 'regress_s1'
@@ -97,19 +96,6 @@ $result = $node_subscriber->safe_psql(
$node_subscriber->poll_query_until('postgres', $synced_query)
or die "Timed out while waiting for subscriber to synchronize data";
-$result = $node_publisher->safe_psql(
- 'postgres', qq(
- SELECT last_value, is_called FROM regress_s1;
-));
-is($result, '200|t', 'Check sequence value in the publisher');
-
-# Check - existing sequence ('regress_s1') is not synced
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
- SELECT last_value, is_called FROM regress_s1;
-));
-is($result, '100|t', 'REFRESH PUBLICATION will not sync existing sequence');
-
# Check - newly published sequence ('regress_s2') is synced
$result = $node_subscriber->safe_psql(
'postgres', qq(
@@ -119,12 +105,9 @@ is($result, '100|t',
'REFRESH PUBLICATION will sync newly published sequence');
##########
-# Test: REFRESH SEQUENCES and REFRESH PUBLICATION (copy_data = false)
-#
-# 1. ALTER SUBSCRIPTION ... REFRESH SEQUENCES should re-synchronize all
-# existing sequences, but not synchronize newly added ones.
-# 2. ALTER SUBSCRIPTION ... REFRESH PUBLICATION with (copy_data = false) should
-# also not update sequence values for newly added sequences.
+# Test:
+# 1. Automatic update of existing sequence values
+# 2. Newly added sequences are not automatically updated.
##########
# Create a new sequence 'regress_s3', and update the existing sequence
@@ -138,51 +121,25 @@ $node_publisher->safe_psql(
INSERT INTO regress_seq_test SELECT nextval('regress_s2') FROM generate_series(1,100);
));
-# 1. Do ALTER SUBSCRIPTION ... REFRESH SEQUENCES
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
- ALTER SUBSCRIPTION regress_seq_sub REFRESH SEQUENCES;
-));
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-
# Check - existing sequences ('regress_s1' and 'regress_s2') are synced
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
- SELECT last_value, is_called FROM regress_s1;
-));
-is($result, '200|t', 'REFRESH SEQUENCES will sync existing sequences');
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
- SELECT last_value, is_called FROM regress_s2;
-));
-is($result, '200|t', 'REFRESH SEQUENCES will sync existing sequences');
-# Check - newly published sequence ('regress_s3') is not synced
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
- SELECT last_value, is_called FROM regress_s3;
-));
-is($result, '1|f',
- 'REFRESH SEQUENCES will not sync newly published sequence');
+# Poll until regress_s1 reflects the updated sequence value
+$node_subscriber->poll_query_until('postgres',
+ qq(SELECT last_value = 200 AND is_called = 't' FROM regress_s1;))
+ or die "Timed out while waiting for regress_s1 sequence to sync";
-# 2. Do ALTER SUBSCRIPTION ... REFRESH PUBLICATION with copy_data as false
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
- ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION WITH (copy_data = false);
-));
-$node_subscriber->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+# Poll until regress_s2 reflects the updated sequence value
+$node_subscriber->poll_query_until('postgres',
+ qq(SELECT last_value = 200 AND is_called = 't' FROM regress_s2;))
+ or die "Timed out while waiting for regress_s2 sequence to sync";
-# Check - newly published sequence ('regress_s3') is not synced with copy_data
-# as false.
+# Check - newly published sequence ('regress_s3') is not synced
$result = $node_subscriber->safe_psql(
'postgres', qq(
SELECT last_value, is_called FROM regress_s3;
));
is($result, '1|f',
- 'REFRESH PUBLICATION will not sync newly published sequence with copy_data as false'
-);
+ 'Newly published sequence are not synced automatically');
##########
# ALTER SUBSCRIPTION ... REFRESH PUBLICATION should report an error when:
--
2.47.3
^ permalink raw reply [nested|flat] 2+ messages in thread
* Re: [PATCH] Support automatic sequence replication
@ 2026-02-06 07:38 Peter Smith <[email protected]>
parent: Ajin Cherian <[email protected]>
0 siblings, 0 replies; 2+ messages in thread
From: Peter Smith @ 2026-02-06 07:38 UTC (permalink / raw)
To: Ajin Cherian <[email protected]>; +Cc: shveta malik <[email protected]>; PostgreSQL Hackers <[email protected]>
Hi Ajin.
Some review comments for patch v2-0001.
======
.../replication/logical/sequencesync.c
copy_sequences:
1.
+ return drift_detected;
This seems a bit strange. And it is not doing quite what the function
comment says it does. I felt you should have another variable like
'sequences_copied', which is set to true only when that
'batch_succeeded_count++' is incremented. This is what you ultimately
want to return. IMO, the variable 'drift_detected' isn't needed at
all.
======
src/test/subscription/t/036_sequences.pl
2.
##########
## ALTER SUBSCRIPTION ... REFRESH PUBLICATION should cause sync of new
# sequences of the publisher.
##########
# Create a new sequence 'regress_s2', and update existing sequence 'regress_s1'
$node_publisher->safe_psql(5.
'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);
));
~
IIUC, you are no longer sync of testing "existing sequences" in this
test part, so you might also want to remove that comment and INSERT
for 'regress_s1'.
======
Kind Regards,
Peter Smith.
Fujitsu Australia
^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2026-02-06 07:38 UTC | newest]
Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2026-02-05 05:03 Re: [PATCH] Support automatic sequence replication Ajin Cherian <[email protected]>
2026-02-06 07:38 ` Peter Smith <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox