public inbox for [email protected]
help / color / mirror / Atom feedFrom: Zhijie Hou (Fujitsu) <[email protected]>
To: Zhijie Hou (Fujitsu) <[email protected]>
To: Amit Kapila <[email protected]>
To: Ajin Cherian <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: shveta malik <[email protected]>
Cc: Ashutosh Sharma <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: [PATCH] Support automatic sequence replication
Date: Mon, 2 Mar 2026 07:58:54 +0000
Message-ID: <TY4PR01MB1690715895CDE6FEFA13C2A2C947EA@TY4PR01MB16907.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OS9PR01MB1691377CDB1468CDC9820BBEB9470A@OS9PR01MB16913.jpnprd01.prod.outlook.com>
References: <CAFPTHDZXX9WQ_X1ZfEvS248T+pKuk6SmCnXcvgPM059N1xPUfA@mail.gmail.com>
<CAJpy0uDLUEjHHME8om1vAf6qkXCeRR6cBvkpK8yWBAC=T0ZFLA@mail.gmail.com>
<CAFPTHDZg1JrunGgOj332hr+gUuH_Jm7skqPpYSvd-QE3yEdRDQ@mail.gmail.com>
<CAJpy0uBz7MCSUkvFJD9ij65vBahNmY+bfCgdGKRqXovYs+K_TA@mail.gmail.com>
<CAJpy0uDsuNqjWd-TmGBxqSS1rnVCJ3B8=SYrtxQ=Vs8kb71QFA@mail.gmail.com>
<CAJpy0uAMWg3KcXtVBS7B0rnchLNrCCVYBByJCzAp=u5LERgtfA@mail.gmail.com>
<CAFPTHDZwEhxhDAeqcPi0GuYN6xBs8gFXHOMUnbg3u2Xigcz4Zg@mail.gmail.com>
<CAE9k0PmTyCU1A9YEf+MRgfeZ8yK1bAYJu=o0bH8DNUTzXejQyQ@mail.gmail.com>
<CAA4eK1L6czEzG4mLNZSyjYC5nX0FMSjjk3csKuxPD3Ph5-7Yvw@mail.gmail.com>
<CAJpy0uAhGQJ=msVsn2GsqWXr+YESJK6x9NBvrUtKvtvp1OVuKQ@mail.gmail.com>
<CAJpy0uAOuu-M6wobH2wHOdTymm-cX9+MqwPyRNoOt=sPKBdCew@mail.gmail.com>
<CAFPTHDZiWYXoKoo4VcBYNH9a=gxDZhfkcBeXt5w6cLw4_ysyKw@mail.gmail.com>
<OS9PR01MB12149D9054CC7F2DC3F0D26A1F577A@OS9PR01MB12149.jpnprd01.prod.outlook.com>
<CAA4eK1KYxQALt46k5uWOO6SUtNjvjOaXwfNjH0AU656YrcGZEw@mail.gmail.com>
<CAFPTHDZYonM+SXG19VVjgWduXQJSuDhcOUWq0NCiiuQubCSt6g@mail.gmail.com>
<CAFPTHDYud1zr0VyizhyhEQXfHMgXVcHrPzE56WUKGCFNskQq2A@mail.gmail.com>
<CAA4eK1JTau3fV7br6xwAV+LXXwM65RuGCuM2J3PQpCONtL1KXA@mail.gmail.com>
<OS9PR01MB1691377CDB1468CDC9820BBEB9470A@OS9PR01MB16913.jpnprd01.prod.outlook.com>
On Saturday, February 28, 2026 4:41 PM Zhijie Hou (Fujitsu) <[email protected]> wrote:
> Here is the V8 patch set addressing the previous review comments:
>
> - For 0001, I noticed that the GetSequence() function added in the patch
> fetches the local sequence value without any privilege check. This
> allows the worker to read sequence data even without proper SELECT
> privilege, which seems unsafe. I've added a SELECT privilege check
> before fetching the sequence value. Additionally, I've updated several
> comments, made cosmetic changes, commit message update, and run
> pgindent on all patches.
>
> - 0002 includes the changes to synchronize sequences directly in the
> REFRESH SEQUENCES command
Rebased the patch to silence compile warning due to a recent commit
a2c89835.
Best Regards,
Hou zj
Attachments:
[application/octet-stream] v9-0002-Synchronize-sequences-directly-in-REFRESH-SEQUENC.patch (17.2K, 2-v9-0002-Synchronize-sequences-directly-in-REFRESH-SEQUENC.patch)
download | inline diff:
From 25d17db1e61f9078dab37417ebb8e06d2a4ba766 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <[email protected]>
Date: Sat, 28 Feb 2026 16:14:14 +0800
Subject: [PATCH v9 2/2] Synchronize sequences directly in REFRESH SEQUENCES
command
The ALTER SUBSCRIPTION ... REFRESH SEQUENCES command currently sets all
sequence states in pg_subscription_rel to INIT and relies on the sequence sync
worker to perform the actual synchronization and update states to READY.
With the recent change making the sequence sync worker long-lived, most
sequences are now synchronized in the background, reducing the need for
REFRESH SEQUENCES. However, the command remains necessary for sequences that
haven't been synchronized.
This commit enhances REFRESH SEQUENCES to synchronize sequences directly within
the command itself, eliminating the overhead of launching a worker and updating
catalog entries unnecessarily.
---
src/backend/commands/subscriptioncmds.c | 17 +-
.../replication/logical/sequencesync.c | 165 ++++++++++++++----
src/include/replication/logicalworker.h | 5 +
src/test/subscription/t/036_sequences.pl | 49 ++++++
4 files changed, 190 insertions(+), 46 deletions(-)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 5e3c0964d38..0a5acfda0ff 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1245,25 +1245,10 @@ AlterSubscription_refresh_seq(Subscription *sub)
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));
- }
+ AlterSubSyncSequences(wrconn, sub->oid, sub->name, sub->runasowner);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
index fad1bb548b2..b4081216477 100644
--- a/src/backend/replication/logical/sequencesync.c
+++ b/src/backend/replication/logical/sequencesync.c
@@ -209,7 +209,7 @@ get_sequences_string(List *seqindexes, List *seqinfos, StringInfo buf)
*/
static void
report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
- List *missing_seqs_idx, List *seqinfos)
+ List *missing_seqs_idx, List *seqinfos, char *subname)
{
StringInfo seqstr;
@@ -255,7 +255,7 @@ report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
ereport(ERROR,
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("logical replication sequence synchronization failed for subscription \"%s\"",
- MySubscription->name));
+ subname));
}
/*
@@ -283,6 +283,7 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
HeapTuple tup;
Form_pg_sequence local_seq;
LogicalRepSequenceInfo *seqinfo_local;
+ LOCKMODE lockmode;
*seqidx = DatumGetInt32(slot_getattr(slot, ++col, &isnull));
Assert(!isnull);
@@ -329,7 +330,20 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
seqinfo_local->found_on_pub = true;
- *sequence_rel = try_table_open(seqinfo_local->localrelid, RowExclusiveLock);
+ /*
+ * We take a stronger lock during DDL commands (currently only ALTER
+ * SUBSCRIPTION ... REFRESH SEQUENCES) to prevent concurrent sequencesync
+ * workers from updating the page_lsn while the DDL is also updating the
+ * same sequence. This ensures we can always fetch the latest page_lsn to
+ * determine whether the remote sequence value should be synchronized (see
+ * validate_seqsync_state).
+ */
+ if (IsLogicalWorker())
+ lockmode = RowExclusiveLock;
+ else
+ lockmode = ShareRowExclusiveLock;
+
+ *sequence_rel = try_table_open(seqinfo_local->localrelid, lockmode);
/* Sequence was concurrently dropped? */
if (!*sequence_rel)
@@ -367,7 +381,8 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
* Preliminary check to determine if copying the sequence is allowed.
*/
static CopySeqResult
-validate_seqsync_state(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
+validate_seqsync_state(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel,
+ XLogRecPtr local_page_lsn)
{
AclResult aclresult;
Oid seqoid = seqinfo->localrelid;
@@ -377,6 +392,16 @@ validate_seqsync_state(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
/* Perform drift check if it's not the initial sync */
if (seqinfo->relstate == SUBREL_STATE_READY)
{
+ /*
+ * Skip synchronization if we are processing outdated sequence info
+ * based on the LSN. This occurs when the sequence has been updated to
+ * more recent data concurrently (via either ALTER SUBSCRIPTION ...
+ * REFRESH SEQUENCES or the sequencesync worker).
+ */
+ if (XLogRecPtrIsValid(local_page_lsn) &&
+ local_page_lsn > seqinfo->page_lsn)
+ return COPYSEQ_NO_DRIFT;
+
/*
* Verify that the current user has SELECT privilege on the sequence.
* This is required to read the sequence state below.
@@ -390,9 +415,32 @@ validate_seqsync_state(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
GetSequence(sequence_rel, &local_last_value, &local_is_called);
/*
- * Skip synchronization if the sequence is already in READY state and
- * has not drifted from the publisher's value.
+ * Skip synchronization if the local sequence value is already ahead of
+ * the publisher's value.
+ *
+ * XXX This occurs not only when the local sequence has been
+ * synchronized to a newer value from the publisher (where skipping is
+ * necessary to avoid backward movement), but also when the local
+ * sequence has been manually updated by the user on the subscriber. The
+ * latter could be considered a replication conflict, and overwriting
+ * the user's update might be acceptable. However, since we cannot
+ * easily distinguish between these two scenarios, we choose to skip
+ * synchronization in all cases and emit a WARNING to notify the user to
+ * manually resolve the conflict.
*/
+ if (local_last_value > seqinfo->last_value)
+ {
+ ereport(WARNING,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("skipped synchronizing the sequence \"%s.%s\"",
+ seqinfo->nspname, seqinfo->seqname),
+ errdetail("The local last_value %lld is ahead of the one on publisher",
+ (long long int) local_last_value));
+
+ return COPYSEQ_NO_DRIFT;
+ }
+
+ /* Skip synchronization if the sequence hasn't drifted */
if (local_last_value == seqinfo->last_value &&
local_is_called == seqinfo->is_called)
return COPYSEQ_NO_DRIFT;
@@ -413,12 +461,16 @@ validate_seqsync_state(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
* synchronized.
*/
static CopySeqResult
-copy_sequence(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
+copy_sequence(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel,
+ Oid subid, bool run_as_owner)
{
UserContext ucxt;
- bool run_as_owner = MySubscription->runasowner;
Oid seqoid = seqinfo->localrelid;
CopySeqResult result;
+ XLogRecPtr local_page_lsn;
+
+ (void) GetSubscriptionRelState(subid, RelationGetRelid(sequence_rel),
+ &local_page_lsn);
/*
* If the user did not opt to run as the owner of the subscription
@@ -427,7 +479,7 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
if (!run_as_owner)
SwitchToUntrustedUser(sequence_rel->rd_rel->relowner, &ucxt);
- result = validate_seqsync_state(seqinfo, sequence_rel);
+ result = validate_seqsync_state(seqinfo, sequence_rel, local_page_lsn);
if (result != COPYSEQ_SUCCESS)
{
@@ -454,8 +506,9 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
* Record the remote sequence's LSN in pg_subscription_rel and mark the
* sequence as READY if updating a sequence that is in INIT state.
*/
- if (seqinfo->relstate == SUBREL_STATE_INIT)
- UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY,
+ if (seqinfo->relstate == SUBREL_STATE_INIT ||
+ seqinfo->page_lsn != local_page_lsn)
+ UpdateSubscriptionRelState(subid, seqoid, SUBREL_STATE_READY,
seqinfo->page_lsn, false);
return COPYSEQ_SUCCESS;
@@ -471,7 +524,8 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
* Returns true/false if any sequences were actually copied.
*/
static bool
-copy_sequences(WalReceiverConn *conn, List *seqinfos)
+copy_sequences(WalReceiverConn *conn, List *seqinfos, Oid subid, char *subname,
+ bool runasowner)
{
int cur_batch_base_index = 0;
int n_seqinfos = list_length(seqinfos);
@@ -497,11 +551,16 @@ copy_sequences(WalReceiverConn *conn, List *seqinfos)
int batch_no_drift = 0;
int batch_missing_count;
Relation sequence_rel = NULL;
+ bool started_tx = false;
WalRcvExecResult *res;
TupleTableSlot *slot;
- StartTransactionCommand();
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
for (int idx = cur_batch_base_index; idx < n_seqinfos; idx++)
{
@@ -597,14 +656,15 @@ copy_sequences(WalReceiverConn *conn, List *seqinfos)
* sequences in READY state, only sync if there's drift.
*/
if (sync_status == COPYSEQ_SUCCESS)
- sync_status = copy_sequence(seqinfo, sequence_rel);
+ sync_status = copy_sequence(seqinfo, sequence_rel,
+ subid, runasowner);
switch (sync_status)
{
case COPYSEQ_SUCCESS:
elog(DEBUG1,
"logical replication synchronization has updated sequence \"%s.%s\" in subscription \"%s\"",
- seqinfo->nspname, seqinfo->seqname, MySubscription->name);
+ seqinfo->nspname, seqinfo->seqname, subname);
batch_succeeded_count++;
sequence_copied = true;
break;
@@ -612,9 +672,8 @@ copy_sequences(WalReceiverConn *conn, List *seqinfos)
case COPYSEQ_MISMATCH:
/*
- * Remember mismatched sequences in SequenceSyncContext
- * since these will be used after the transaction is
- * committed.
+ * Remember mismatched sequences in SequenceSyncContext since
+ * these will be used after the transaction is committed.
*/
oldctx = MemoryContextSwitchTo(SequenceSyncContext);
mismatched_seqs_idx = lappend_int(mismatched_seqs_idx,
@@ -680,13 +739,17 @@ copy_sequences(WalReceiverConn *conn, List *seqinfos)
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, %d no drift",
- MySubscription->name,
+ subname,
(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();
+ /*
+ * Commit this batch if started a transaction, and prepare for next
+ * batch.
+ */
+ if (started_tx)
+ CommitTransactionCommand();
if (batch_missing_count)
{
@@ -711,7 +774,7 @@ copy_sequences(WalReceiverConn *conn, List *seqinfos)
/* Report mismatches, permission issues, or missing sequences */
report_sequence_errors(mismatched_seqs_idx, insuffperm_seqs_idx,
- missing_seqs_idx, seqinfos);
+ missing_seqs_idx, seqinfos, subname);
return sequence_copied;
}
@@ -723,20 +786,23 @@ copy_sequences(WalReceiverConn *conn, List *seqinfos)
* Returns true if sequences have been updated.
*/
static bool
-LogicalRepSyncSequences(WalReceiverConn *conn)
+LogicalRepSyncSequences(WalReceiverConn *conn, Oid subid, char *subname,
+ bool runasowner)
{
Relation rel;
HeapTuple tup;
ScanKeyData skey[1];
SysScanDesc scan;
- Oid subid = MyLogicalRepWorker->subid;
bool sequence_copied = false;
List *seqinfos = NIL;
MemoryContext oldctx;
+ bool started_tx = false;
- Assert(SequenceSyncContext);
-
- StartTransactionCommand();
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
rel = table_open(SubscriptionRelRelationId, AccessShareLock);
@@ -796,7 +862,8 @@ LogicalRepSyncSequences(WalReceiverConn *conn)
systable_endscan(scan);
table_close(rel, AccessShareLock);
- CommitTransactionCommand();
+ if (started_tx)
+ CommitTransactionCommand();
/*
* Exit early if no catalog entries found, likely due to concurrent drops.
@@ -805,7 +872,8 @@ LogicalRepSyncSequences(WalReceiverConn *conn)
return false;
/* Process sequences */
- sequence_copied = copy_sequences(conn, seqinfos);
+ sequence_copied = copy_sequences(conn, seqinfos, subid, subname,
+ runasowner);
return sequence_copied;
}
@@ -880,7 +948,10 @@ start_sequence_sync(void)
/*
* Synchronize all sequences (both READY and INIT states).
*/
- sequence_copied = LogicalRepSyncSequences(LogRepWorkerWalRcvConn);
+ sequence_copied = LogicalRepSyncSequences(LogRepWorkerWalRcvConn,
+ MySubscription->oid,
+ MySubscription->name,
+ MySubscription->runasowner);
MemoryContextReset(SequenceSyncContext);
MemoryContextSwitchTo(oldctx);
@@ -942,3 +1013,37 @@ SequenceSyncWorkerMain(Datum main_arg)
FinishSyncWorker();
}
+
+/*
+ * Wrapper for LogicalRepSyncSequences to synchronize all sequences of a
+ * subscription from outside the sequencesync worker
+ */
+void
+AlterSubSyncSequences(WalReceiverConn *conn, Oid subid, char *subname,
+ bool runasowner)
+{
+ /*
+ * Init the SequenceSyncContext which we clean up after the sequence
+ * synchronization.
+ */
+ SequenceSyncContext = AllocSetContextCreate(CurrentMemoryContext,
+ "SequenceSyncContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ PG_TRY();
+ {
+ MemoryContext oldctx;
+
+ oldctx = MemoryContextSwitchTo(SequenceSyncContext);
+
+ LogicalRepSyncSequences(conn, subid, subname, runasowner);
+
+ MemoryContextSwitchTo(oldctx);
+ }
+ PG_FINALLY();
+ {
+ MemoryContextDelete(SequenceSyncContext);
+ SequenceSyncContext = NULL;
+ }
+ PG_END_TRY();
+}
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 7d748a28da8..73afd7853d0 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -14,6 +14,8 @@
#include <signal.h>
+#include "replication/walreceiver.h"
+
extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending;
extern void ApplyWorkerMain(Datum main_arg);
@@ -31,4 +33,7 @@ extern void LogicalRepWorkersWakeupAtCommit(Oid subid);
extern void AtEOXact_LogicalRepWorkers(bool isCommit);
+extern void AlterSubSyncSequences(WalReceiverConn *conn, Oid subid,
+ char *subname, bool runasowner);
+
#endif /* LOGICALWORKER_H */
diff --git a/src/test/subscription/t/036_sequences.pl b/src/test/subscription/t/036_sequences.pl
index 1d81518fe22..9a61b7bd0c8 100644
--- a/src/test/subscription/t/036_sequences.pl
+++ b/src/test/subscription/t/036_sequences.pl
@@ -176,4 +176,53 @@ $node_subscriber->wait_for_log(
qr/WARNING: ( [A-Z0-9]+:)? missing sequence on publisher \("public.regress_s4"\)/,
$log_offset);
+##########
+# ALTER SUBSCRIPTION ... REFRESH SEQUENCES synchronizes sequences online,
+# eliminating the need to launch a sequencesync worker.
+##########
+
+# Reduce max_logical_replication_workers to disallow sequence worker from running
+$node_subscriber->append_conf('postgresql.conf',
+ qq(max_logical_replication_workers = 0));
+$node_subscriber->restart;
+
+# Verify there is no logical replication apply worker running
+$result = $node_subscriber->safe_psql(
+ 'postgres',
+ "SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'logical replication apply worker'");
+
+is($result, '0', 'no logical replication worker is running');
+
+# Increment sequence on publisher
+$node_publisher->safe_psql('postgres',
+ qq(SELECT nextval('regress_s1');));
+
+# The command should fail due to missing sequence ('regress_s4')
+my ($cmdret, $stdout, $stderr) = $node_subscriber->psql('postgres',
+ "ALTER SUBSCRIPTION regress_seq_sub REFRESH SEQUENCES;");
+
+like(
+ $stderr,
+ qr/WARNING: missing sequence on publisher \("public.regress_s4"\)/,
+ "output the wanring for the missing sequence regress_s4");
+
+like(
+ $stderr,
+ qr/ERROR: logical replication sequence synchronization failed for subscription \"regress_seq_sub\"/,
+ "the command failed due to the missing sequence regress_s4");
+
+# Refresh the publication to remove the missing sequence
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_seq_sub REFRESH PUBLICATION;");
+
+# Sync the sequence regress_s1
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_seq_sub REFRESH SEQUENCES;");
+
+# Get the current sequence value on subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ qq(SELECT last_value FROM regress_s1;));
+
+is($result, '201', 'sequence regress_s1 is synced now');
+
done_testing();
--
2.51.1.windows.1
[application/octet-stream] v9-0001-Support-automatic-sequence-replication.patch (41.6K, 3-v9-0001-Support-automatic-sequence-replication.patch)
download | inline diff:
From 641532c6b3034c345968d74215576f7020ee537f Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Tue, 24 Feb 2026 21:37:01 +1100
Subject: [PATCH v9 1/2] Support automatic sequence replication.
Currently, sequence values are synchronized from publisher to subscriber only
when the user manually runs ALTER SUBSCRIPTION ... REFRESH PUBLICATION (which
affects only newly subscribed sequences) or REFRESH SEQUENCES. The sequence sync
worker exits immediately after completing each synchronization round.
The primary use case for sequence replication is during upgrades, where it's
recommended that users ensure sequences are in sync by running REFRESH SEQUENCES
before finishing the upgrade. However, this command can be slow when
synchronizing a large number of sequences, potentially increasing downtime.
To address this, this commit makes the sequence sync worker long-lived,
continuously monitoring sequences and resynchronizing 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.
With this change, most sequences are silently synchronized in the background,
eliminating the need to run REFRESH SEQUENCES for the majority of cases.
However, frequently updated sequences may still lag behind, requiring a final
REFRESH SEQUENCES before upgrade completion. Users can monitor progress by
checking whether sequence states transition from INIT to READY in
pg_subscription_rel.
The REFRESH SEQUENCES command is retained for this final synchronization step,
though it currently updates all sequence states to INIT, which has room for
improvement. A future patch will enhance this command to synchronize sequences
directly without launching a worker, reducing catalog overhead.
Author: Ajin Cherian <[email protected]>
Author: Zhijie Hou <[email protected]>
Reviewed-by: Shveta Malik <[email protected]>
Reviewed-by: Peter Smith <[email protected]>
Reviewed-by: Ashutosh Sharma <[email protected]>
Reviewed-by: Amit Kapila <[email protected]>
---
doc/src/sgml/logical-replication.sgml | 23 +-
doc/src/sgml/ref/alter_subscription.sgml | 5 -
src/backend/commands/sequence.c | 29 ++
.../replication/logical/sequencesync.c | 374 +++++++++++++-----
src/backend/replication/logical/syncutils.c | 47 +--
src/backend/replication/logical/worker.c | 11 +
src/include/catalog/pg_subscription_rel.h | 1 +
src/include/commands/sequence.h | 1 +
src/include/replication/worker_internal.h | 2 +-
src/test/subscription/t/026_stats.pl | 2 +
src/test/subscription/t/036_sequences.pl | 79 +---
11 files changed, 370 insertions(+), 204 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 5028fe9af09..bb523af5d37 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1787,8 +1787,9 @@ Publications:
<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. The worker will
+ remain running for the life of the subscription, periodically
+ synchronizing all published sequences.
</para>
<para>
The ability to launch a sequence synchronization worker is limited by the
@@ -1817,7 +1818,7 @@ Publications:
<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
+ Subscriber sequence values can become out of sync as the publisher
advances them.
</para>
<para>
@@ -2335,15 +2336,13 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<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">
+ Incremental sequence changes are continuously replicated. If, however,
+ some kind of switchover or failover to the subscriber database is
+ intended, then the sequences replication could be lagging behind and
+ the sequences on the subscriber should be compared with that of the
+ publisher to make sure that they are up to date, if not they
+ 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
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 5318998e80c..5a7b4f3c2c2 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -190,11 +190,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
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index e1b808bbb60..ac5a0197e1c 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -929,6 +929,35 @@ lastval(PG_FUNCTION_ARGS)
PG_RETURN_INT64(result);
}
+/*
+ * Read the current sequence values (last_value and is_called)
+ *
+ * This is a read-only operation used by logical replication sequence
+ * synchronization to detect drift. The caller must hold a lock on the sequence
+ * and have sufficient privileges to access it.
+ */
+void
+GetSequence(Relation seqrel, int64 *last_value, bool *is_called)
+{
+ Buffer buf;
+ HeapTupleData seqtuple;
+ Form_pg_sequence_data seq;
+
+ /* Confirm that the relation is a sequence and is locked */
+ Assert(seqrel->rd_rel->relkind == RELKIND_SEQUENCE);
+ Assert(CheckRelationLockedByMe(seqrel, AccessShareLock, true));
+
+ /* 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 */
+ UnlockReleaseBuffer(buf);
+}
+
/*
* Main internal procedure that handles 2 & 3 arg forms of SETVAL.
*
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
index 9c92fddd624..fad1bb548b2 100644
--- 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 as 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/synchronize)
+ * | |
+ * +--<---+
*
* To avoid creating too many transactions, up to MAX_SEQUENCES_SYNC_PER_BATCH
* sequences are synchronized per transaction. The locks on the sequence
@@ -60,6 +72,7 @@
#include "postmaster/interrupt.h"
#include "replication/logicalworker.h"
#include "replication/worker_internal.h"
+#include "storage/latch.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
@@ -78,23 +91,35 @@ typedef enum CopySeqResult
COPYSEQ_SUCCESS,
COPYSEQ_MISMATCH,
COPYSEQ_INSUFFICIENT_PERM,
- COPYSEQ_SKIPPED
+ COPYSEQ_SKIPPED,
+ COPYSEQ_NO_DRIFT,
} CopySeqResult;
-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;
+
+static MemoryContext SequenceSyncContext = NULL;
/*
- * Apply worker determines if sequence synchronization is needed.
+ * Apply worker determines whether a sequence sync worker is needed.
+ *
+ * Check if the subscription includes sequences and start a sequencesync
+ * worker if one is not already running. The active sequencesync worker will
+ * handle all pending sequence synchronization. If any sequences remain
+ * unsynchronized after it exits, a new worker can be started in the next
+ * iteration.
*
- * Start a sequencesync worker if one is not already running. The active
- * sequencesync worker will handle all pending sequence synchronization. If any
- * sequences remain unsynchronized after it exits, a new worker can be started
- * in the next iteration.
+ * The pointer to the sequencesync worker is cached to avoid scanning the
+ * workers array each time via logicalrep_worker_find().
*/
void
-ProcessSequencesForSync(void)
+MaybeLaunchSequenceSyncWorker(void)
{
- LogicalRepWorker *sequencesync_worker;
+ static LogicalRepWorker *sequencesync_worker = NULL;
+
int nsyncworkers;
bool has_pending_sequences;
bool started_tx;
@@ -112,6 +137,19 @@ ProcessSequencesForSync(void)
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ /*
+ * Quick exit if the sequence sync worker for the current subscription is
+ * already alive.
+ */
+ if (sequencesync_worker &&
+ sequencesync_worker->proc &&
+ isSequenceSyncWorker(sequencesync_worker) &&
+ sequencesync_worker->subid == MyLogicalRepWorker->subid)
+ {
+ LWLockRelease(LogicalRepWorkerLock);
+ return;
+ }
+
/* Check if there is a sequencesync worker already running? */
sequencesync_worker = logicalrep_worker_find(WORKERTYPE_SEQUENCESYNC,
MyLogicalRepWorker->subid,
@@ -144,7 +182,7 @@ ProcessSequencesForSync(void)
* for the given list of sequence indexes.
*/
static void
-get_sequences_string(List *seqindexes, StringInfo buf)
+get_sequences_string(List *seqindexes, List *seqinfos, StringInfo buf)
{
resetStringInfo(buf);
foreach_int(seqidx, seqindexes)
@@ -171,7 +209,7 @@ get_sequences_string(List *seqindexes, StringInfo buf)
*/
static void
report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
- List *missing_seqs_idx)
+ List *missing_seqs_idx, List *seqinfos)
{
StringInfo seqstr;
@@ -183,7 +221,7 @@ report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
if (mismatched_seqs_idx)
{
- get_sequences_string(mismatched_seqs_idx, seqstr);
+ get_sequences_string(mismatched_seqs_idx, seqinfos, seqstr);
ereport(WARNING,
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg_plural("mismatched or renamed sequence on subscriber (%s)",
@@ -194,7 +232,7 @@ report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
if (insuffperm_seqs_idx)
{
- get_sequences_string(insuffperm_seqs_idx, seqstr);
+ get_sequences_string(insuffperm_seqs_idx, seqinfos, seqstr);
ereport(WARNING,
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg_plural("insufficient privileges on sequence (%s)",
@@ -205,7 +243,7 @@ report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
if (missing_seqs_idx)
{
- get_sequences_string(missing_seqs_idx, seqstr);
+ get_sequences_string(missing_seqs_idx, seqinfos, seqstr);
ereport(WARNING,
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg_plural("missing sequence on publisher (%s)",
@@ -229,7 +267,8 @@ report_sequence_errors(List *mismatched_seqs_idx, List *insuffperm_seqs_idx,
*/
static CopySeqResult
get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
- LogicalRepSequenceInfo **seqinfo, int *seqidx)
+ LogicalRepSequenceInfo **seqinfo, int *seqidx,
+ List *seqinfos)
{
bool isnull;
int col = 0;
@@ -325,32 +364,77 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
}
/*
- * Apply remote sequence state to local sequence and mark it as
- * synchronized (READY).
+ * Preliminary check to determine if copying the sequence is allowed.
*/
static CopySeqResult
-copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner)
+validate_seqsync_state(LogicalRepSequenceInfo *seqinfo, Relation sequence_rel)
{
- UserContext ucxt;
AclResult aclresult;
+ Oid seqoid = seqinfo->localrelid;
+ int64 local_last_value;
+ bool local_is_called;
+
+ /* Perform drift check if it's not the initial sync */
+ if (seqinfo->relstate == SUBREL_STATE_READY)
+ {
+ /*
+ * Verify that the current user has SELECT privilege on the sequence.
+ * This is required to read the sequence state below.
+ */
+ aclresult = pg_class_aclcheck(seqoid, GetUserId(), ACL_SELECT);
+
+ if (aclresult != ACLCHECK_OK)
+ return COPYSEQ_INSUFFICIENT_PERM;
+
+ /* Get current local sequence state */
+ GetSequence(sequence_rel, &local_last_value, &local_is_called);
+
+ /*
+ * Skip synchronization if the sequence is already in READY state and
+ * has not drifted from the publisher's value.
+ */
+ if (local_last_value == seqinfo->last_value &&
+ local_is_called == seqinfo->is_called)
+ return COPYSEQ_NO_DRIFT;
+ }
+
+ /* Verify that the current user can update the sequence */
+ aclresult = pg_class_aclcheck(seqoid, GetUserId(), ACL_UPDATE);
+
+ if (aclresult != ACLCHECK_OK)
+ return COPYSEQ_INSUFFICIENT_PERM;
+
+ return COPYSEQ_SUCCESS;
+}
+
+/*
+ * 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, Relation sequence_rel)
+{
+ UserContext ucxt;
bool run_as_owner = MySubscription->runasowner;
Oid seqoid = seqinfo->localrelid;
+ CopySeqResult result;
/*
* If the user did not opt to run as the owner of the subscription
* ('run_as_owner'), then copy the sequence as the owner of the sequence.
*/
if (!run_as_owner)
- SwitchToUntrustedUser(seqowner, &ucxt);
+ SwitchToUntrustedUser(sequence_rel->rd_rel->relowner, &ucxt);
- aclresult = pg_class_aclcheck(seqoid, GetUserId(), ACL_UPDATE);
+ result = validate_seqsync_state(seqinfo, sequence_rel);
- if (aclresult != ACLCHECK_OK)
+ if (result != COPYSEQ_SUCCESS)
{
if (!run_as_owner)
RestoreUserContext(&ucxt);
- return COPYSEQ_INSUFFICIENT_PERM;
+ return result;
}
/*
@@ -368,19 +452,26 @@ 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 (seqinfo->relstate == SUBREL_STATE_INIT)
+ UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY,
+ seqinfo->page_lsn, false);
return COPYSEQ_SUCCESS;
}
/*
* 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, List *seqinfos)
{
int cur_batch_base_index = 0;
int n_seqinfos = list_length(seqinfos);
@@ -390,13 +481,10 @@ copy_sequences(WalReceiverConn *conn)
StringInfo seqstr = makeStringInfo();
StringInfo cmd = makeStringInfo();
MemoryContext oldctx;
+ bool sequence_copied = 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 +494,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;
@@ -501,46 +590,53 @@ copy_sequences(WalReceiverConn *conn)
}
sync_status = get_and_validate_seq_info(slot, &sequence_rel,
- &seqinfo, &seqidx);
+ &seqinfo, &seqidx, seqinfos);
+
+ /*
+ * For sequences in INIT state, always sync. Otherwise, for
+ * sequences in READY state, only sync if there's drift.
+ */
if (sync_status == COPYSEQ_SUCCESS)
- sync_status = copy_sequence(seqinfo,
- sequence_rel->rd_rel->relowner);
+ sync_status = copy_sequence(seqinfo, sequence_rel);
switch (sync_status)
{
case COPYSEQ_SUCCESS:
elog(DEBUG1,
- "logical replication synchronization for subscription \"%s\", sequence \"%s.%s\" has finished",
- MySubscription->name, seqinfo->nspname,
- seqinfo->seqname);
+ "logical replication synchronization has updated sequence \"%s.%s\" in subscription \"%s\"",
+ seqinfo->nspname, seqinfo->seqname, MySubscription->name);
batch_succeeded_count++;
+ sequence_copied = true;
break;
+
case COPYSEQ_MISMATCH:
/*
- * Remember mismatched sequences in a long-lived memory
- * context since these will be used after the transaction
- * is committed.
+ * Remember mismatched sequences in SequenceSyncContext
+ * since these will be used after the transaction is
+ * committed.
*/
- oldctx = MemoryContextSwitchTo(ApplyContext);
+ oldctx = MemoryContextSwitchTo(SequenceSyncContext);
mismatched_seqs_idx = lappend_int(mismatched_seqs_idx,
seqidx);
MemoryContextSwitchTo(oldctx);
batch_mismatched_count++;
break;
+
case COPYSEQ_INSUFFICIENT_PERM:
/*
- * Remember sequences with insufficient privileges in a
- * long-lived memory context since these will be used
- * after the transaction is committed.
+ * Remember sequences with insufficient privileges in
+ * SequenceSyncContext since these will be used after the
+ * transaction is committed.
*/
- oldctx = MemoryContextSwitchTo(ApplyContext);
+ oldctx = MemoryContextSwitchTo(SequenceSyncContext);
insuffperm_seqs_idx = lappend_int(insuffperm_seqs_idx,
seqidx);
MemoryContextSwitchTo(oldctx);
batch_insuffperm_count++;
break;
+
case COPYSEQ_SKIPPED:
/*
@@ -558,6 +654,15 @@ copy_sequences(WalReceiverConn *conn)
batch_skipped_count++;
}
break;
+
+ case COPYSEQ_NO_DRIFT:
+ /* Nothing to do */
+ batch_no_drift++;
+ break;
+
+ default:
+ elog(ERROR, "unrecognized Sequence replication result: %d", (int) sync_status);
+
}
if (sequence_rel)
@@ -566,20 +671,19 @@ copy_sequences(WalReceiverConn *conn)
ExecDropSingleTupleTableSlot(slot);
walrcv_clear_result(res);
- resetStringInfo(seqstr);
- resetStringInfo(cmd);
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",
+ "logical replication sequence synchronization for subscription \"%s\" - batch #%d = %d attempted, %d succeeded, %d mismatched, %d insufficient permission, %d missing from publisher, %d skipped, %d no drift",
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);
+ batch_insuffperm_count, batch_missing_count, batch_skipped_count, batch_no_drift);
/* Commit this batch, and prepare for next batch */
CommitTransactionCommand();
@@ -607,51 +711,55 @@ copy_sequences(WalReceiverConn *conn)
/* Report mismatches, permission issues, or missing sequences */
report_sequence_errors(mismatched_seqs_idx, insuffperm_seqs_idx,
- missing_seqs_idx);
+ missing_seqs_idx, seqinfos);
+
+ return sequence_copied;
}
/*
* 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 *err;
- bool must_use_password;
Relation rel;
HeapTuple tup;
- ScanKeyData skey[2];
+ ScanKeyData skey[1];
SysScanDesc scan;
Oid subid = MyLogicalRepWorker->subid;
- StringInfoData app_name;
+ bool sequence_copied = false;
+ List *seqinfos = NIL;
+ MemoryContext oldctx;
+
+ Assert(SequenceSyncContext);
StartTransactionCommand();
rel = table_open(SubscriptionRelRelationId, AccessShareLock);
+ /* Scan for all sequences belonging to this subscription */
ScanKeyInit(&skey[0],
Anum_pg_subscription_rel_srsubid,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(subid));
- ScanKeyInit(&skey[1],
- Anum_pg_subscription_rel_srsubstate,
- BTEqualStrategyNumber, F_CHAREQ,
- CharGetDatum(SUBREL_STATE_INIT));
-
scan = systable_beginscan(rel, InvalidOid, false,
- NULL, 2, skey);
+ NULL, 1, skey);
+
while (HeapTupleIsValid(tup = systable_getnext(scan)))
{
Form_pg_subscription_rel subrel;
LogicalRepSequenceInfo *seq;
Relation sequence_rel;
- MemoryContext oldctx;
+ char relstate;
CHECK_FOR_INTERRUPTS();
subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
+ relstate = subrel->srsubstate;
sequence_rel = try_table_open(subrel->srrelid, RowExclusiveLock);
@@ -666,18 +774,19 @@ LogicalRepSyncSequences(void)
continue;
}
+ Assert(relstate == SUBREL_STATE_INIT || relstate == SUBREL_STATE_READY);
+
/*
* Worker needs to process sequences across transaction boundary, so
- * allocate them under long-lived context.
+ * allocate them under SequenceSyncContext.
*/
- oldctx = MemoryContextSwitchTo(ApplyContext);
-
+ oldctx = MemoryContextSwitchTo(SequenceSyncContext);
seq = palloc0_object(LogicalRepSequenceInfo);
seq->localrelid = subrel->srrelid;
seq->nspname = get_namespace_name(RelationGetNamespace(sequence_rel));
seq->seqname = pstrdup(RelationGetRelationName(sequence_rel));
+ seq->relstate = relstate;
seqinfos = lappend(seqinfos, seq);
-
MemoryContextSwitchTo(oldctx);
table_close(sequence_rel, NoLock);
@@ -693,36 +802,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());
+ /* Process sequences */
+ sequence_copied = copy_sequences(conn, seqinfos);
- /*
- * 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 +824,91 @@ 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);
+
+ /*
+ * Init the SequenceSyncContext which we clean up after each sequence
+ * synchronization.
+ */
+ SequenceSyncContext = AllocSetContextCreate(ApplyContext,
+ "SequenceSyncContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ for (;;)
+ {
+ bool sequence_copied = false;
+ MemoryContext oldctx;
+
+ CHECK_FOR_INTERRUPTS();
+
+ /* Process any invalidation messages that might have accumulated */
+ AcceptInvalidationMessages();
+ maybe_reread_subscription();
+
+ /*
+ * Perform sequence synchronization under SequenceSyncContext and
+ * reset it each cycle to avoid manual memory management.
+ */
+ oldctx = MemoryContextSwitchTo(SequenceSyncContext);
+
+ /*
+ * Synchronize all sequences (both READY and INIT states).
+ */
+ sequence_copied = LogicalRepSyncSequences(LogRepWorkerWalRcvConn);
+
+ MemoryContextReset(SequenceSyncContext);
+ MemoryContextSwitchTo(oldctx);
+
+ /*
+ * Adjust sleep interval based on whether sequences were copied
+ * over
+ */
+ if (sequence_copied)
+ {
+ sleep_ms = SEQSYNC_MIN_SLEEP_MS;
+ }
+ else
+ {
+ /*
+ * 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 ef61ca0437d..233ac7ae873 100644
--- a/src/backend/replication/logical/syncutils.c
+++ b/src/backend/replication/logical/syncutils.c
@@ -172,7 +172,9 @@ ProcessSyncingRelations(XLogRecPtr current_lsn)
case WORKERTYPE_APPLY:
ProcessSyncingTablesForApply(current_lsn);
- ProcessSequencesForSync();
+
+ /* Check if sequence worker needs to be started */
+ MaybeLaunchSequenceSyncWorker();
break;
case WORKERTYPE_SEQUENCESYNC:
@@ -191,13 +193,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,
@@ -205,23 +207,22 @@ FetchRelationStates(bool *has_pending_subtables,
bool *started_tx)
{
/*
- * has_subtables and has_subsequences_non_ready are declared as static,
- * since the same value can be used until the system table is invalidated.
+ * 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;
@@ -231,27 +232,27 @@ FetchRelationStates(bool *has_pending_subtables,
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?
*
@@ -277,5 +278,5 @@ FetchRelationStates(bool *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/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f9c4b484754..f91c8f9ecde 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -5099,6 +5099,9 @@ maybe_reread_subscription(void)
* worker won't restart if the streaming option's value is changed from
* 'parallel' to any other value or the server decides not to stream the
* in-progress transaction.
+ *
+ * Note: some parameters may not be relevant to the sequence sync worker,
+ * but exit anyway.
*/
if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
strcmp(newsub->name, MySubscription->name) != 0 ||
@@ -5114,6 +5117,10 @@ maybe_reread_subscription(void)
ereport(LOG,
(errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
MySubscription->name)));
+ else if (am_sequencesync_worker())
+ ereport(LOG,
+ (errmsg("logical replication sequence synchronization worker for subscription \"%s\" will stop because of a parameter change",
+ MySubscription->name)));
else
ereport(LOG,
(errmsg("logical replication worker for subscription \"%s\" will restart because of a parameter change",
@@ -5132,6 +5139,10 @@ maybe_reread_subscription(void)
ereport(LOG,
errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
MySubscription->name));
+ else if (am_sequencesync_worker())
+ ereport(LOG,
+ errmsg("logical replication sequence synchronization worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
+ MySubscription->name));
else
ereport(LOG,
errmsg("logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 502640d3018..86574b69169 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -96,6 +96,7 @@ typedef struct LogicalRepSequenceInfo
char *seqname;
char *nspname;
Oid localrelid;
+ char relstate;
/* Sequence information retrieved from the publisher node */
XLogRecPtr page_lsn;
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 2c3c4a3f074..f00eea9fbd1 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(Relation seqrel, int64 *last_value, bool *is_called);
extern void SetSequence(Oid relid, int64 next, bool iscalled);
extern void ResetSequenceCaches(void);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 4ecbdcfadac..a41cb045f19 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -286,7 +286,7 @@ extern void UpdateTwoPhaseState(Oid suboid, char new_state);
extern void ProcessSyncingTablesForSync(XLogRecPtr current_lsn);
extern void ProcessSyncingTablesForApply(XLogRecPtr current_lsn);
-extern void ProcessSequencesForSync(void);
+extern void MaybeLaunchSequenceSyncWorker(void);
pg_noreturn extern void FinishSyncWorker(void);
extern void InvalidateSyncingRelStates(Datum arg, SysCacheIdentifier cacheid,
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..1d81518fe22 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'
@@ -84,9 +83,6 @@ $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
@@ -97,19 +93,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,16 +102,13 @@ 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
-# 'regress_s2'.
+# Create a new sequence 'regress_s3', and update the existing sequences
+# 'regress_s2' and 'regress_s1'.
$node_publisher->safe_psql(
'postgres', qq(
CREATE SEQUENCE regress_s3;
@@ -136,53 +116,28 @@ $node_publisher->safe_psql(
-- Existing sequence
INSERT INTO regress_seq_test SELECT nextval('regress_s2') FROM generate_series(1,100);
+ INSERT INTO regress_seq_test SELECT nextval('regress_s1') 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 sequences are not synced automatically');
##########
# ALTER SUBSCRIPTION ... REFRESH PUBLICATION should report an error when:
--
2.51.1.windows.1
view thread (58+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: [PATCH] Support automatic sequence replication
In-Reply-To: <TY4PR01MB1690715895CDE6FEFA13C2A2C947EA@TY4PR01MB16907.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox