From 24d719f882709b02f9cef6e76ac4b74f80cc2034 Mon Sep 17 00:00:00 2001 From: Ajin Cherian 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 itsajin@gmail.com --- 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$<iteration count>:&l This catalog contains tables and sequences known to the subscription after running: - CREATE SUBSCRIPTION, + CREATE SUBSCRIPTION, or - ALTER SUBSCRIPTION ... REFRESH PUBLICATION, or - - ALTER SUBSCRIPTION ... REFRESH SEQUENCES. + ALTER SUBSCRIPTION ... REFRESH PUBLICATION. 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. - - - use - ALTER SUBSCRIPTION ... REFRESH SEQUENCES - to re-synchronize all sequences currently known to the subscription. - - A sequence synchronization worker 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. The ability to launch a sequence synchronization worker is limited by the @@ -1815,21 +1809,7 @@ Publications: - Refreshing Out-of-Sync Sequences - - Subscriber sequence values will become out of sync as the publisher - advances them. - - - To detect this, compare the - pg_subscription_rel.srsublsn - on the subscriber with the page_lsn obtained - from the pg_get_sequence_data - function for the sequence on the publisher. Then run - - ALTER SUBSCRIPTION ... REFRESH SEQUENCES to - re-synchronize if necessary. - + Out-of-Sync Sequences Each sequence caches a block of values (typically 32) in memory before @@ -1961,16 +1941,6 @@ Publications: ------------+-----------+------------ 610 | t | 0/017CEDF8 (1 row) - - - - 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 - - ALTER SUBSCRIPTION ... REFRESH SEQUENCES. - -/* sub # */ ALTER SUBSCRIPTION sub1 REFRESH SEQUENCES; @@ -2333,24 +2303,6 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER - - - 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 - ALTER SUBSCRIPTION ... REFRESH SEQUENCES - or by copying the current data from the publisher (perhaps using - pg_dump) or by determining a sufficiently high value - from the tables themselves. - - - Replication of TRUNCATE 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 name SET PUBLICA ALTER SUBSCRIPTION name ADD PUBLICATION publication_name [, ...] [ WITH ( publication_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name DROP PUBLICATION publication_name [, ...] [ WITH ( publication_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name REFRESH PUBLICATION [ WITH ( refresh_option [= value] [, ... ] ) ] -ALTER SUBSCRIPTION name REFRESH SEQUENCES ALTER SUBSCRIPTION name ENABLE ALTER SUBSCRIPTION name DISABLE ALTER SUBSCRIPTION name SET ( subscription_parameter [= value] [, ... ] ) @@ -190,11 +189,6 @@ ALTER SUBSCRIPTION name RENAME TO < Previously subscribed tables are not copied, even if a table's row filter WHERE clause has since been modified. - - Previously subscribed sequences are not re-synchronized. To do that, - use - ALTER SUBSCRIPTION ... REFRESH SEQUENCES. - See for recommendations on how to handle any warnings about sequence definition differences between @@ -219,30 +213,6 @@ ALTER SUBSCRIPTION name RENAME TO < - - REFRESH SEQUENCES - - - Re-synchronize sequence data with the publisher. Unlike - - ALTER SUBSCRIPTION ... REFRESH PUBLICATION which - only has the ability to synchronize newly added sequences, - REFRESH SEQUENCES 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. - - - See for - recommendations on how to handle any warnings about sequence definition - differences between the publisher and the subscriber. - - - See for recommendations on how to - identify and handle out-of-sync sequences. - - - - ENABLE 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 */ 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 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