Re: Logical Replication of sequences

On Fri, 27 Dec 2024 at 15:18, Hayato Kuroda (Fujitsu)
<kuroda.hayato@fujitsu.com> wrote:
>
> Dear Vignesh,
>
> Thanks for updating the patch! Here are my comments.
>
> 01. SyncingRelationsState
>
> ```
>  * SYNC_RELATIONS_STATE_NEEDS_REBUILD The subscription relations state is no
>  * longer valid and the subscription relations should be rebuilt.
> ```
>
> Can we follow the style like other lines? Like:
>
> SYNC_RELATIONS_STATE_NEEDS_REBUILD indicates that the subscription relations
> state is no longer valid and the subscription relations should be rebuilt.

Modfied

> 02. FetchRelationStates()
>
> ```
>                 /* Fetch tables and sequences that are in non-ready state. */
>                 rstates = GetSubscriptionRelations(MySubscription->oid, true, true,
>                                                                                    false);
> ```
>
> I think rstates should be list_free_deep()'d after the foreach().

Since rstates is allocated in TopTransactionContext and there is a
CommitTransactionCommand which will take care of releasing the memory
from AtCommit_Memory:
        /*
         * Release all transaction-local memory.  TopTransactionContext survives
         * but becomes empty; any sub-contexts go away.
         */
        Assert(TopTransactionContext != NULL);
        MemoryContextReset(TopTransactionContext);

So I felt deep free is not required in this case.

> 03. LogicalRepSyncSequences
>
> ```
>         char            slotname[NAMEDATALEN];
> ...
>         snprintf(slotname, NAMEDATALEN, "pg_%u_sync_sequences_" UINT64_FORMAT,
>                          subid, GetSystemIdentifier());
>
>         /*
>          * Here we use the slot name instead of the subscription name as the
>          * application_name, so that it is different from the leader apply worker,
>          * so that synchronous replication can distinguish them.
>          */
>         LogRepWorkerWalRcvConn =
>                 walrcv_connect(MySubscription->conninfo, true, true,
>                                            must_use_password,
>                                            slotname, &err);
> ```
>
> Hmm, IIUC the sequence sync worker does not require any replication slots.
> I feel the variable name should be "application_name" and the comment can be updated.

Modified

> 04. LogicalRepSyncSequences
>
> ```
>         /* Get the sequences that should be synchronized. */
>         sequences = GetSubscriptionRelations(subid, false, true, false);
> ```
>
> I think rstates should be list_free_deep()'d after the foreach_ptr().

Since rstates is allocated in TopTransactionContext and there is a
CommitTransactionCommand immediately which will take care of releasing
the memory from AtCommit_Memory:
        /*
         * Release all transaction-local memory.  TopTransactionContext survives
         * but becomes empty; any sub-contexts go away.
         */
        Assert(TopTransactionContext != NULL);
        MemoryContextReset(TopTransactionContext);

So I felt deep free is not required in this case.

> 05. LogicalRepSyncSequences
>
> ```
>                 /*
>                  * COPY FROM does not honor RLS policies.  That is not a problem for
>                  * subscriptions owned by roles with BYPASSRLS privilege (or
>                  * superuser, who has it implicitly), but other roles should not be
>                  * able to circumvent RLS.  Disallow logical replication into RLS
>                  * enabled relations for such roles.
>                  */
>                 if (check_enable_rls(RelationGetRelid(sequence_rel), InvalidOid, false) == RLS_ENABLED)
>                         ereport(ERROR,
>                                         errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
>                                         errmsg("user \"%s\" cannot replicate into sequence with row-level security enabled: \"%s\"",
>                                                    GetUserNameFromId(GetUserId(), true),
>                                                    RelationGetRelationName(sequence_rel)));
> ```
>
> Can we really set a row level security policy to sequences? I've tested but
> I couldn't.
>
> ```
> postgres=# CREATE SEQUENCE s;
> CREATE SEQUENCE
> postgres=# ALTER TABLE s ENABLE ROW LEVEL SECURITY ;
> ERROR:  ALTER action ENABLE ROW SECURITY cannot be performed on relation "s"
> DETAIL:  This operation is not supported for sequences.
> ```

Removed this check

> 06. copy_sequence
>
> ```
>         appendStringInfo(&cmd, "SELECT c.oid, c.relkind\n"
>                                          "FROM pg_catalog.pg_class c\n"
>                                          "INNER JOIN pg_catalog.pg_namespace n\n"
>                                          "  ON (c.relnamespace = n.oid)\n"
>                                          "WHERE n.nspname = %s AND c.relname = %s",
>                                          quote_literal_cstr(nspname),
>                                          quote_literal_cstr(relname));
> ```
>
> I feel the function is not so efficient because it can obtain only a tuple, i.e.,
> information for one sequence at once. I think you basically copied from fetch_remote_table_info(),
> but it was OK for it because the tablesync worker handles only a table.
>
> Can you obtain all sequences at once and check whether each sequences match or not?

I still could not come up with a simple way to do this, I will think
more and handle this in the next version.

> 07. LogicalRepSyncSequences
>
> ```
>                         /* LOG all the sequences synchronized during current batch. */
>                         for (int i = 0; i < curr_batch_seq; i++)
>                         {
>                                 SubscriptionRelState *done_seq;
>
>                                 done_seq = (SubscriptionRelState *) lfirst(list_nth_cell(sequences_not_synced,
>                                                                                                                                                  (curr_seq - curr_batch_seq) + i));
>
>                                 ereport(DEBUG1,
>                                                 errmsg_internal("logical replication synchronization for subscription \"%s\", sequence \"%s\" has finished",
>                                                                                 get_subscription_name(subid, false), get_rel_name(done_seq->relid)));
>                         }
> ```
>
> The loop is needed only when the debug messages should be output the system.
> Can we use message_level_is_interesting() to skip the loop for some cases?

Modified

> 08. pg_dump.c
>
> ```
> /*
>  * getSubscriptionTables
>  *        Get information about subscription membership for dumpable tables. This
>  *    will be used only in binary-upgrade mode for PG17 or later versions.
>  */
> void
> getSubscriptionTables(Archive *fout)
> ```
>
> I was bit confused of the pg_dump codes. I doubt that the pg_upgrade might not
> be able to transfer pg_subscripion_rel entries of sequences, but it seemed to work
> well because sequences are handled mostly same as normal tables on the pg_dump layer.
> But based on other codes, the function should be "getSubscriptionRelations" and
> comments should be also updated.

Modified

> 09. logical-replication.sgml
>
> I feel we can add descriptions in "Publication" section. E.g.:
>
> Publications can also include sequences, but the behavior differs from a table
> or a group of tables: users can synchronize current states of sequences at an
> arbitrary timing. For more details, please see "Replicating Sequences".

Added this

> 10. pg_proc.dat
>
> ```
> +{ oid => '6313',
> +  descr => 'current on-disk sequence state',
> +  proname => 'pg_sequence_state', provolatile => 'v',
> +  prorettype => 'record', proargtypes => 'regclass',
> +  proallargtypes => '{regclass,pg_lsn,int8,int8,bool}',
> +  proargmodes => '{i,o,o,o,o}',
> +  proargnames => '{seq_oid,page_lsn,last_value,log_cnt,is_called}',
> +  prosrc => 'pg_sequence_state' },
> ```
>
> I think this is not wrong, but according to the src/include/catalog/unused_oids,
> the oid should be at 8000-9999 while developing:
>
> ```
> $ ./unused_oids
> ...
> Patches should use a more-or-less consecutive range of OIDs.
> Best practice is to start with a random choice in the range 8000-9999.
> Suggested random unused OID: 9293 (415 consecutive OID(s) available starting here)
> ```

Modified

Thanks for the comments, the attached patch has the changes for the same.

Regards,
Vignesh