Re: Logical Replication of sequences - Mailing list pgsql-hackers

Hi Vignesh,

WIP - Some comments for patch v20251027-0003

======
General.

1.
When referring to synchronization workers AFAIK the convention has always been:

- code/comments refer to "tablesync workers" and "sequencesync workers"
- but errors/docs mostly use the long form like "table synchronization
workers" and "sequence synchronization workers"

This patch still has places saying "sequence sync worker" instead of
"sequencesync worker" etc. IMO these should be changed, otherwise
there are too many variations of saying the same thing.

======
src/backend/commands/sequence.c

pg_get_sequence_data:

2.
/*
* See the comment in copy_sequence() above
* UpdateSubscriptionRelState() for details on recording the LSN.
*/

Consider rewording that more like:
For details about recording the LSN, see the
UpdateSubscriptionRelState() call in copy_sequence().

======
src/backend/replication/logical/launcher.c

logicalrep_worker_find:

3.
/sequence sync workers/sequencesync workers/
/table sync workers/tablesync workers/

~~~

logicalrep_reset_seqsync_start_time:

4.
+ /*
+ * 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.
+ */

Half of this comment was already described in the function comment.
Maybe better to remove this comment, and instead just add more to the
function comment.

SUGGESTION (function comment)
Reset the last_seqsync_start_time of the sequencesync worker in the
subscription's apply worker. Note -- this value is not stored in the
sequencesync worker, because that has finished already and is about to
exit.

======
src/backend/replication/logical/syncutils.c

FinishSyncWorker:

5.
+FinishSyncWorker()
 {
+ LogicalRepWorkerType wtype = MyLogicalRepWorker->type;
+
+ Assert(wtype == WORKERTYPE_TABLESYNC || wtype == WORKERTYPE_SEQUENCESYNC);
+

We already have some macros so I think it's better to make use of them

SUGGESTION

Assert(am_tablesync_worker() || am_sequencesync_worker())

~

Similarly, the subsequent code like:

+ if (wtype == WORKERTYPE_TABLESYNC)
and
+ if (wtype == WORKERTYPE_SEQUENCESYNC)

becomes
if (am_tablesync_worker()) ...
if (am_sequencesync_worker()) ...

~~~

6.
+ /*
+ * 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(WORKERTYPE_APPLY, MyLogicalRepWorker->subid,
+ InvalidOid);

I think those comments belong within the if blocks, so add some {}

SUGGESTION

if (is_sequencesync)
{
/* comment ... */
logicalrep_reset_seqsync_start_time(...)
}
else
{
/* comment ... */
logicalrep_worker_wakeup(...)
}

~~~

launch_sync_worker:

7.
 /*
- * Process possible state change(s) of relations that are being synchronized.
+ * Attempt to launch a sync worker (sequence or table) if there is a sync
+ * worker slot available and the retry interval has elapsed.
+ *
+ * nsyncworkers: Number of currently running sync workers for the subscription.
+ * relid:  InvalidOid for sequence sync worker, actual relid for table sync
+ * worker.
+ * last_start_time: Pointer to the last start time of the worker.
+ */

/sequence sync worker/sequencesync worker/
/table sync worker/tablesync worker/

~~~

8.
+ (void) logicalrep_worker_launch((relid == InvalidOid) ?
WORKERTYPE_SEQUENCESYNC : WORKERTYPE_TABLESYNC,

Tidier to use macro:
OidIsValid(relid) ? WORKERTYPE_TABLESYNC : WORKERTYPE_SEQUENCESYNC

OTOH, I think the caller already knows the WORKERTYPE_xxx it is
launching, perhaps it is best to pass 'wtype' as another  parameter to
launch_sync_worker()?

~~~

FetchRelationStates

9.
+ /*
+ * has_subtables and has_subsequences_non_ready is declared as static,
+ * since the same value can be used until the system table is invalidated.
+ */

typo: /is declared/are declared/

======
src/backend/replication/logical/worker.c

DisableSubscriptionAndExit:

10.
+ LogicalRepWorkerType wtype = am_tablesync_worker() ? WORKERTYPE_TABLESYNC :
+ (am_sequencesync_worker()) ? WORKERTYPE_SEQUENCESYNC :
+ WORKERTYPE_APPLY;
+

Why is this code needed at all?

I think we don't need wtype, because later code that says:
+ if (wtype != WORKERTYPE_SEQUENCESYNC)

Can instead just say:
+ if (am_apply_worker() || am_tablesync_worker())

======
src/include/catalog/pg_subscription_rel.h

11.
+typedef struct LogicalRepSeqHashKey
+{
+ const char *seqname;
+ const char *nspname;
+} LogicalRepSeqHashKey;
+
+typedef struct LogicalRepSequenceInfo
+{
+ char    *seqname;
+ char    *nspname;
+ Oid localrelid;
+ bool remote_seq_queried;
+ Oid seqowner;
+ bool entry_valid;
+} LogicalRepSequenceInfo;

No comments?

======
src/include/commands/sequence.h

12.
+#define SEQ_LOG_CNT_INVALID 0
+

Unused?

======
Kind Regards,
Peter Smith.
Fujitsu Australia