Re: Logical Replication of sequences

On Tue, Oct 28, 2025 at 6:00 PM vignesh C <vignesh21(at)gmail(dot)com> wrote:

> On Mon, 27 Oct 2025 at 14:42, Chao Li <li(dot)evan(dot)chao(at)gmail(dot)com> wrote:
> >
> >
> > > On Oct 24, 2025, at 23:22, vignesh C <vignesh21(at)gmail(dot)com> wrote:
> > >
> > > Regards,
> > > Vignesh
> > >
> <v20251024-0001-Rename-sync_error_count-to-tbl_sync_error_.patch><v20251024-0002-Add-worker-type-argument-to-logicalrep_wor.patch><v20251024-0003-New-worker-for-sequence-synchronization-du.patch><v20251024-0004-Documentation-for-sequence-synchronization.patch>
> >
> >
> > The changes in 0001 are straightforward, looks good. I haven’t reviewed
> 0004 yet. Got a few comments for 0002 and 0003.
>
> > 5 - 0003
> > ```
> > +/*
> > + * Reset the last_seqsync_start_time of the sequencesync worker in the
> > + * subscription's apply worker.
> > + */
> > +void
> > +logicalrep_reset_seqsync_start_time(void)
> > +{
> > + LogicalRepWorker *worker;
> > +
> > + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
> > +
> > + /*
> > + * Set the last_seqsync_start_time for the sequence worker in
> the apply
> > + * worker instead of the sequence sync worker, as the sequence
> sync worker
> > + * has finished and is about to exit.
> > + */
> > + worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
> InvalidOid,
> > +
> WORKERTYPE_APPLY, true);
> > + if (worker)
> > + worker->last_seqsync_start_time = 0;
> > +
> > + LWLockRelease(LogicalRepWorkerLock);
> > +}
> > ```
> >
> > Two comments for this new function:
> >
> > * The function comment and in-code comment are redundant. Suggesting
> move the in-code comment to function comment.
> > * Why LW_SHARED is used? We are writing worker->last_seqsync_start_time,
> shouldn’t LW_EXCLUSIVE be used?
>
> There will be only one sequence sync worker and only this process is
> going to update this, so LW_SHARED is enough to find the apply worker.
>
> > 6 - 0003
> > ```
> > + /*
> > + * Count running sync workers for this subscription, while we
> have the
> > + * lock.
> > + */
> > + nsyncworkers =
> logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
> > + LWLockRelease(LogicalRepWorkerLock);
> > +
> > + launch_sync_worker(nsyncworkers, InvalidOid,
> > +
> &MyLogicalRepWorker->last_seqsync_start_time);
> > ```
> >
> > I think here could be a race condition. Because the lock is acquired in
> LW_SHARED, meaning multiple caller may get the same nsyncworkers. Then it
> launches sync worker based on nsyncworkers, which would use inaccurate
> nsyncworkers, because between LWLockRelease() and launch_sync_worker(),
> another worker might be started.
> >
> > But if that is not the case, only one caller should call
> ProcessSyncingSequencesForApply(), then why the lock is needed?
>
> Sequence sync worker will be started only by the apply worker, another
> worker cannot be started for this subscription between LWLockRelease()
> and launch_sync_worker() as this apply worker is responsible for it
> and the apply worker is active with current work. Same logic is used
> for table sync workers too.
>
> > 7 - 0003
> > ```
> > + if (insuffperm_seqs->len)
> > + {
> > + appendStringInfo(combined_error_detail, "Insufficient
> permission for sequence(s): (%s)",
> > + insuffperm_seqs->data);
> > + appendStringInfoString(combined_error_hint, "Grant
> permissions for the sequence(s).");
> > + }
> > ```
> >
> > “Grant permissions” is unclear. Should it be “Grant UPDATE privilege”?
>
> Modified
>
> > 8 - 0003
> > ```
> > + appendStringInfoString(combined_error_hint, "
> For mismatched sequences, alter or re-create local sequences to have
> matching parameters as publishers.");
> > ```
> >
> > “To have matching parameters as publishers” grammatically sound not
> good. Maybe revision to “to match the publisher’s parameters”.
>
> Modified
>
> > 9 - 0003
> > ```
> > + /*
> > + * current_indexes is not incremented sequentially
> because some
> > + * sequences may be missing, and the number of fetched
> rows may not
> > + * match the batch size. The `hash_search` with
> HASH_REMOVE takes care
> > + * of the count.
> > + */
> > ```
> >
> > Typo: current_indexes => current_index
>
> Modified
>
> > 10 - 0003
> > ```
> > - /* Find the leader apply worker and signal it. */
> > - logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
> > + /*
> > + * This is a clean exit of the sequencesync worker; reset the
> > + * last_seqsync_start_time.
> > + */
> > + if (wtype == WORKERTYPE_SEQUENCESYNC)
> > + logicalrep_reset_seqsync_start_time();
> > + else
> > + /* Find the leader apply worker and signal it. */
> > + logicalrep_worker_wakeup(MyLogicalRepWorker->subid,
> InvalidOid);
> > ```
> >
> > The comment “this is a clean exist of sequencsync worker” is specific to
> “if”, so suggesting moving into “if”. And “this is a clean exis of the
> sequencesyc worker” is not needed, keep consistent with the comment in
> “else”.
>
> Modified
>
> > 11 - 0003
> > ```
> > +void
> > +launch_sync_worker(int nsyncworkers, Oid relid, TimestampTz
> *last_start_time)
> > +{
> > + /* If there is a free sync worker slot, start a new sync worker
> */
> > + if (nsyncworkers < max_sync_workers_per_subscription)
> > + {
> > ```
> >
> > The entire function is under an “if”, so we can do “if (!…) return”, so
> saves a level of indent.
>
> Modified
>
> Peter's comments from [1] have also been addressed. The attached
> v20251029 version patch has the changes for the same.
>
> [1] -
> https://www.postgresql.org/message-id/CAHut%2BPtMc1fr6cQvUAnxRE%2Bbuim5m-d9M2dM0YAeEHNkS9KzBw%40mail.gmail.com

Are you planning to merge 0001 and 0002, I don't think first we want to
commit the solution with a hash and then commit with the list. So for
review also we better merge these 2 so that we don't need to review the
changes with hash as we are not going to commit that change if we agree
that we want to go ahead with the list.

--
Regards,
Dilip Kumar
Google