Re: speed up a logical replica setup

On Mon, 19 Feb 2024 at 11:15, Hayato Kuroda (Fujitsu)
<kuroda.hayato@fujitsu.com> wrote:
>
> Dear hackers,
>
> > Since it may be useful, I will post top-up patch on Monday, if there are no
> > updating.
>
> And here are top-up patches. Feel free to check and include.
>
> v22-0001: Same as v21-0001.
> === rebased patches ===
> v22-0002: Update docs per recent changes. Same as v20-0002.
> v22-0003: Add check versions of the target. Extracted from v20-0003.
> v22-0004: Remove -S option. Mostly same as v20-0009, but commit massage was
>           slightly changed.
> === Newbie ===
> V22-0005: Addressed my comments which seems to be trivial[1].
>           Comments #1, 3, 4, 8, 10, 14, 17 were addressed here.
> v22-0006: Consider the scenario when commands are failed after the recovery.
>           drop_subscription() is removed and some messages are added per [2].
> V22-0007: Revise server_is_in_recovery() per [1]. Comments #5, 6, 7, were addressed here.
> V22-0008: Fix a strange report when physical_primary_slot is null. Per comment #9 [1].
> V22-0009: Prohibit reuse publications when it has already existed. Per comments #11 and 12 [1].
> V22-0010: Avoid to call PQclear()/PQfinish()/pg_free() if the process exits soon. Per comment #15 [1].
> V22-0011: Update testcode. Per comments #17- [1].
>
> I did not handle below points because I have unclear points.
>
> a.
> This patch set cannot detect the disconnection between the target (standby) and
> source (primary) during the catch up. Because the connection status must be gotten
> at the same time (=in the same query) with the recovery status, but now it is now an
> independed function (server_is_in_recovery()).
>
> b.
> This patch set cannot detect the inconsistency reported by Shubham [3]. I could not
> come up with solutions without removing -P...

Few comments:
1) The below code can lead to assertion failure if the publisher is
stopped while dropping the replication slot:
+       if (primary_slot_name != NULL)
+       {
+               conn = connect_database(dbinfo[0].pubconninfo);
+               if (conn != NULL)
+               {
+                       drop_replication_slot(conn, &dbinfo[0],
primary_slot_name);
+               }
+               else
+               {
+                       pg_log_warning("could not drop replication
slot \"%s\" on primary",
+                                                  primary_slot_name);
+                       pg_log_warning_hint("Drop this replication
slot soon to avoid retention of WAL files.");
+               }
+               disconnect_database(conn);
+       }

pg_createsubscriber: error: connection to database failed: connection
to server on socket "/tmp/.s.PGSQL.5432" failed: No such file or
directory
Is the server running locally and accepting connections on that socket?
pg_createsubscriber: warning: could not drop replication slot
"standby_1" on primary
pg_createsubscriber: hint: Drop this replication slot soon to avoid
retention of WAL files.
pg_createsubscriber: pg_createsubscriber.c:432: disconnect_database:
Assertion `conn != ((void *)0)' failed.
Aborted (core dumped)

This is happening because we are calling disconnect_database in case
of connection failure case too which has the following assert:
+static void
+disconnect_database(PGconn *conn)
+{
+       Assert(conn != NULL);
+
+       PQfinish(conn);
+}

2) There is a CheckDataVersion function which does exactly this, will
we be able to use this:
+       /* Check standby server version */
+       if ((ver_fd = fopen(versionfile, "r")) == NULL)
+               pg_fatal("could not open file \"%s\" for reading: %m",
versionfile);
+
+       /* Version number has to be the first line read */
+       if (!fgets(rawline, sizeof(rawline), ver_fd))
+       {
+               if (!ferror(ver_fd))
+                       pg_fatal("unexpected empty file \"%s\"", versionfile);
+               else
+                       pg_fatal("could not read file \"%s\": %m", versionfile);
+       }
+
+       /* Strip trailing newline and carriage return */
+       (void) pg_strip_crlf(rawline);
+
+       if (strcmp(rawline, PG_MAJORVERSION) != 0)
+       {
+               pg_log_error("standby server is of wrong version");
+               pg_log_error_detail("File \"%s\" contains \"%s\",
which is not compatible with this program's version \"%s\".",
+                                                       versionfile,
rawline, PG_MAJORVERSION);
+               exit(1);
+       }
+
+       fclose(ver_fd);

3) Should this be added to typedefs.list:
+enum WaitPMResult
+{
+       POSTMASTER_READY,
+       POSTMASTER_STILL_STARTING
+};

4) pgCreateSubscriber should be mentioned after pg_controldata to keep
the ordering consistency:
diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml
index aa94f6adf6..c5edd244ef 100644
--- a/doc/src/sgml/reference.sgml
+++ b/doc/src/sgml/reference.sgml
@@ -285,6 +285,7 @@
    &pgCtl;
    &pgResetwal;
    &pgRewind;
+   &pgCreateSubscriber;
    &pgtestfsync;

5) Here pg_replication_slots should be pg_catalog.pg_replication_slots:
+       if (primary_slot_name)
+       {
+               appendPQExpBuffer(str,
+                                                 "SELECT 1 FROM
pg_replication_slots "
+                                                 "WHERE active AND
slot_name = '%s'",
+                                                 primary_slot_name);

6) Here pg_settings should be pg_catalog.pg_settings:
+        * - max_worker_processes >= 1 + number of dbs to be converted
+        *------------------------------------------------------------------------
+        */
+       res = PQexec(conn,
+                                "SELECT setting FROM pg_settings
WHERE name IN ("
+                                "'max_logical_replication_workers', "
+                                "'max_replication_slots', "
+                                "'max_worker_processes', "
+                                "'primary_slot_name') "
+                                "ORDER BY name");

Regards,
Vignesh