Re: Logical Replication of sequences

> On Oct 24, 2025, at 23:22, vignesh C <vignesh21@gmail.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.

1 - 0002
```
  * We are only interested in the leader apply worker or table sync worker.
+ * For apply workers, the relid should be set to InvalidOid, as they manage
+ * changes across all tables and sequences. For table sync workers, the relid
+ * should be set to the OID of the relation being synchronized.
  */
 LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid subid, Oid relid, LogicalRepWorkerType wtype,
+					   bool only_running)
 {
 	int			i;
 	LogicalRepWorker *res = NULL;
 
 	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
```

The comment says that “for apply workers, the relid should be set to InvalidOid”, so is it worthy adding an assert for that?

2 - 0002
```
-	/* Search for attached worker for a given subscription id. */
+	/* Search for the attached worker matching the specified criteria. */
 	for (i = 0; i < max_logical_replication_workers; i++)
 	{
```

Minor issue with the comment:

* we are not search for a specific work, so “the” should be “a”
* “attached” is confusing. In the old comment, ‘attached” tied to “a given subscription id”, but now, attach to what?

So suggested revision:

“/* Search for a logical replication worker matching the specified criteria */”

3 - 0002
```
 /*
  * Stop the logical replication worker for subid/relid, if any.
+ *
+ * Similar to logicalrep_worker_find, relid should be set to a valid OID only
+ * for table sync workers.
  */
 void
-logicalrep_worker_stop(Oid subid, Oid relid)
+logicalrep_worker_stop(Oid subid, Oid relid, LogicalRepWorkerType wtype)
```

The comment should be updated: subid/relid => subid/relid/wtype.

4 - 0002
```
@@ -477,7 +477,8 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-												rstate->relid, false);
+												rstate->relid,
+												WORKERTYPE_TABLESYNC, true);
```

Why changed only_running from false to true? This commit adds a new worker type, but don’t tend to change the existing logic.

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?

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?

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”?

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”.

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

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”.

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.

Best regards,
—
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/