Re: Logical Replication of sequences

On Mon, Nov 3, 2025 at 12:35 AM vignesh C <vignesh21(at)gmail(dot)com> wrote:
>

Some inor comments on 0001.
1.
+ /*
+ * Acquire LogicalRepWorkerLock in LW_EXCLUSIVE mode to block the apply
+ * worker (holding LW_SHARED) from reading or updating
+ * last_seqsync_start_time. See ProcessSyncingSequencesForApply().
+ */
+ LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);

Is it required to have LW_EXCLUSIVE lock here? In the function
ProcessSyncingSequencesForApply(), apply_worker access/update
last_seqsync_start_time only once it ensures that sequence sync worker
has exited. I have made changes related to this in the attached to
show you what I have in mind.

2.
+ /*
+ * Worker needs to process sequences across transaction boundary, so
+ * allocate them under long-lived context.
+ */
+ oldctx = MemoryContextSwitchTo(TopMemoryContext);
+
+ seq = palloc0_object(LogicalRepSequenceInfo);
…
...
+ /*
+ * Allocate in a long-lived memory context, since these
+ * errors will be reported after the transaction commits.
+ */
+ oldctx = MemoryContextSwitchTo(TopMemoryContext);
+ mismatched_seqs = lappend_int(mismatched_seqs, seqidx);

At the above and other places in syncworker, we don't need to use
TopMemoryContext; rather, we can use ApplyContext allocated via
SequenceSyncWorkerMain()->SetupApplyOrSyncWorker()->InitializeLogRepWorker().

3.
ProcessSyncingTablesForApply(current_lsn);
+ ProcessSyncingSequencesForApply();

I am not sure if the function name ProcessSyncingSequencesForApply is
appropriate. For tables, we do some work for concurrently running
tablesync workers and launch new as well but for sequences, we don't
do any work for sequences that are already being synced. How about
ProcessSequencesForSync()?

4.
+      /* Should never happen. */
+      elog(ERROR, "Sequence synchronization worker not expected to
process relations");

The first letter of the ERROR message should be small. How about:
"sequence synchronization worker is not expected to process
relations"? I have made this change in the attached.

5.
@@ -5580,7 +5606,8 @@ start_apply(XLogRecPtr origin_startpos)
* idle state.
*/
AbortOutOfAnyTransaction();
- pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+ pgstat_report_subscription_error(MySubscription->oid,
+ !am_tablesync_worker());

Why this change?

6.
@@ -264,6 +267,8 @@ extern bool
logicalrep_worker_launch(LogicalRepWorkerType wtype,
Oid userid, Oid relid,
dsm_handle subworker_dsm,
bool retain_dead_tuples);
+extern void launch_sync_worker(LogicalRepWorkerType wtype, int nsyncworkers,
+ Oid relid, TimestampTz *last_start_time);
extern void logicalrep_worker_stop(LogicalRepWorkerType wtype, Oid subid,
Oid relid);
All the other functions except the newly added one are from
launcher.c. So, this one should be after those, no? It should be after
the InvalidateSyncingRelStates() declaration.

Apart from above, please find attached top-up patch to improve
comments and some other cosmetic stuff. The 0001 patch looks good to
me apart from the above minor points.

--
With Regards,
Amit Kapila.