Add two-phase option in pg_createsubscriber.

This patch introduces the '--enable-two-phase' option to the
'pg_createsubscriber' utility, allowing users to enable two-phase commit
for all subscriptions during their creation.

Note that even without this option users can enable the two_phase option
for the subscriptions created by pg_createsubscriber. However, it requires
the subscription to be disabled first which could be inconvenient for
users.

When two-phase commit is enabled, prepared transactions are sent to the
subscriber at the time of 'PREPARE TRANSACTION', and they are processed as
two-phase transactions on the subscriber as well. If disabled, prepared
transactions are sent only when committed and are processed immediately by
the subscriber.

Author: Shubham Khanna <[email protected]>
Reviewed-by: vignesh C <[email protected]>
Reviewed-by: Hayato Kuroda <[email protected]>
Reviewed-by: Peter Smith <[email protected]>
Reviewed-by: Ajin Cherian <[email protected]>
Reviewed-by: Amit Kapila <[email protected]>
Discussion: https://postgr.es/m/CAHv8RjLPdFP=kA5LNSmWZ=+GMXmO+LczvV6p9HJjsXxZz10KGA@mail.gmail.com

diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml
index d56487fe2ca8158a6c3bf8bdb5e93b08dbe8da58..65caa3f50959eb1b74063edfca89cd4880461a35 100644 (file)
--- a/doc/src/sgml/ref/pg_createsubscriber.sgml
+++ b/doc/src/sgml/ref/pg_createsubscriber.sgml
@@ -165,6 +165,19 @@ PostgreSQL documentation
      </listitem>
     </varlistentry>
 
+    <varlistentry>
+     <term><option>-T</option></term>
+     <term><option>--enable-two-phase</option></term>
+     <listitem>
+      <para>
+       Enables <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>
+       commit for the subscription. When multiple databases are specified, this
+       option applies uniformly to all subscriptions created on those databases.
+       The default is <literal>false</literal>.
+      </para>
+     </listitem>
+    </varlistentry>
+
     <varlistentry>
      <term><option>-U <replaceable class="parameter">username</replaceable></option></term>
      <term><option>--subscriber-username=<replaceable class="parameter">username</replaceable></option></term>
@@ -300,7 +313,9 @@ PostgreSQL documentation
     greater than or equal to the number of specified databases.  The target
     server must have <xref linkend="guc-max-worker-processes"/> configured to a
     value greater than the number of specified databases.  The target server
-    must accept local connections.
+    must accept local connections. If you are planning to use the
+    <option>--enable-two-phase</option> switch then you will also need to set
+    the <xref linkend="guc-max-prepared-transactions"/> appropriately.
    </para>
 
    <para>
@@ -360,6 +375,7 @@ PostgreSQL documentation
    </para>
 
    <para>
+    Unless the <option>--enable-two-phase</option> switch is specified,
     <application>pg_createsubscriber</application> sets up logical
     replication with two-phase commit disabled.  This means that any
     prepared transactions will be replicated at the time

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 9fdf15e5ac03bdd0e4ca350e9bd3de6b75030976..a5a2d61165d2993cdd94679d4c0a1faa790efaa0 100644 (file)
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -38,6 +38,7 @@ struct CreateSubscriberOptions
    char       *socket_dir;     /* directory for Unix-domain socket, if any */
    char       *sub_port;       /* subscriber port number */
    const char *sub_username;   /* subscriber username */
+   bool        two_phase;      /* enable-two-phase option */
    SimpleStringList database_names;    /* list of database names */
    SimpleStringList pub_names; /* list of publication names */
    SimpleStringList sub_names; /* list of subscription names */
@@ -45,6 +46,7 @@ struct CreateSubscriberOptions
    int         recovery_timeout;   /* stop recovery after this time */
 };
 
+/* per-database publication/subscription info */
 struct LogicalRepInfo
 {
    char       *dbname;         /* database name */
@@ -58,6 +60,16 @@ struct LogicalRepInfo
    bool        made_publication;   /* publication was created */
 };
 
+/*
+ * Information shared across all the databases (or publications and
+ * subscriptions).
+ */
+struct LogicalRepInfos
+{
+   struct LogicalRepInfo *dbinfo;
+   bool        two_phase;      /* enable-two-phase option */
+};
+
 static void cleanup_objects_atexit(void);
 static void usage();
 static char *get_base_conninfo(const char *conninfo, char **dbname);
@@ -117,7 +129,7 @@ static bool dry_run = false;
 
 static bool success = false;
 
-static struct LogicalRepInfo *dbinfo;
+static struct LogicalRepInfos dbinfos;
 static int num_dbs = 0;        /* number of specified databases */
 static int num_pubs = 0;       /* number of specified publications */
 static int num_subs = 0;       /* number of specified subscriptions */
@@ -172,17 +184,17 @@ cleanup_objects_atexit(void)
 
    for (int i = 0; i < num_dbs; i++)
    {
-       if (dbinfo[i].made_publication || dbinfo[i].made_replslot)
+       if (dbinfos.dbinfo[i].made_publication || dbinfos.dbinfo[i].made_replslot)
        {
            PGconn     *conn;
 
-           conn = connect_database(dbinfo[i].pubconninfo, false);
+           conn = connect_database(dbinfos.dbinfo[i].pubconninfo, false);
            if (conn != NULL)
            {
-               if (dbinfo[i].made_publication)
-                   drop_publication(conn, &dbinfo[i]);
-               if (dbinfo[i].made_replslot)
-                   drop_replication_slot(conn, &dbinfo[i], dbinfo[i].replslotname);
+               if (dbinfos.dbinfo[i].made_publication)
+                   drop_publication(conn, &dbinfos.dbinfo[i]);
+               if (dbinfos.dbinfo[i].made_replslot)
+                   drop_replication_slot(conn, &dbinfos.dbinfo[i], dbinfos.dbinfo[i].replslotname);
                disconnect_database(conn, false);
            }
            else
@@ -192,16 +204,18 @@ cleanup_objects_atexit(void)
                 * that some objects were left on primary and should be
                 * removed before trying again.
                 */
-               if (dbinfo[i].made_publication)
+               if (dbinfos.dbinfo[i].made_publication)
                {
                    pg_log_warning("publication \"%s\" created in database \"%s\" on primary was left behind",
-                                  dbinfo[i].pubname, dbinfo[i].dbname);
+                                  dbinfos.dbinfo[i].pubname,
+                                  dbinfos.dbinfo[i].dbname);
                    pg_log_warning_hint("Drop this publication before trying again.");
                }
-               if (dbinfo[i].made_replslot)
+               if (dbinfos.dbinfo[i].made_replslot)
                {
                    pg_log_warning("replication slot \"%s\" created in database \"%s\" on primary was left behind",
-                                  dbinfo[i].replslotname, dbinfo[i].dbname);
+                                  dbinfos.dbinfo[i].replslotname,
+                                  dbinfos.dbinfo[i].dbname);
                    pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
                }
            }
@@ -227,6 +241,7 @@ usage(void)
    printf(_("  -P, --publisher-server=CONNSTR  publisher connection string\n"));
    printf(_("  -s, --socketdir=DIR             socket directory to use (default current dir.)\n"));
    printf(_("  -t, --recovery-timeout=SECS     seconds to wait for recovery to end\n"));
+   printf(_("  -T, --enable-two-phase          enable two-phase commit for all subscriptions\n"));
    printf(_("  -U, --subscriber-username=NAME  user name for subscriber connection\n"));
    printf(_("  -v, --verbose                   output verbose messages\n"));
    printf(_("      --config-file=FILENAME      use specified main server configuration\n"
@@ -479,9 +494,10 @@ store_pub_sub_info(const struct CreateSubscriberOptions *opt,
                     dbinfo[i].pubname ? dbinfo[i].pubname : "(auto)",
                     dbinfo[i].replslotname ? dbinfo[i].replslotname : "(auto)",
                     dbinfo[i].pubconninfo);
-       pg_log_debug("subscriber(%d): subscription: %s ; connection string: %s", i,
+       pg_log_debug("subscriber(%d): subscription: %s ; connection string: %s, two_phase: %s", i,
                     dbinfo[i].subname ? dbinfo[i].subname : "(auto)",
-                    dbinfo[i].subconninfo);
+                    dbinfo[i].subconninfo,
+                    dbinfos.two_phase ? "true" : "false");
 
        if (num_pubs > 0)
            pubcell = pubcell->next;
@@ -938,11 +954,12 @@ check_publisher(const struct LogicalRepInfo *dbinfo)
        failed = true;
    }
 
-   if (max_prepared_transactions != 0)
+   if (max_prepared_transactions != 0 && !dbinfos.two_phase)
    {
        pg_log_warning("two_phase option will not be enabled for replication slots");
        pg_log_warning_detail("Subscriptions will be created with the two_phase option disabled.  "
                              "Prepared transactions will be replicated at COMMIT PREPARED.");
+       pg_log_warning_hint("You can use --enable-two-phase switch to enable two_phase.");
    }
 
    /*
@@ -1345,8 +1362,9 @@ create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
    slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
 
    appendPQExpBuffer(str,
-                     "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",
-                     slot_name_esc);
+                     "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, %s, false)",
+                     slot_name_esc,
+                     dbinfos.two_phase ? "true" : "false");
 
    PQfreemem(slot_name_esc);
 
@@ -1722,8 +1740,9 @@ create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
    appendPQExpBuffer(str,
                      "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
                      "WITH (create_slot = false, enabled = false, "
-                     "slot_name = %s, copy_data = false)",
-                     subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc);
+                     "slot_name = %s, copy_data = false, two_phase = %s)",
+                     subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc,
+                     dbinfos.two_phase ? "true" : "false");
 
    PQfreemem(pubname_esc);
    PQfreemem(subname_esc);
@@ -1895,6 +1914,7 @@ main(int argc, char **argv)
        {"publisher-server", required_argument, NULL, 'P'},
        {"socketdir", required_argument, NULL, 's'},
        {"recovery-timeout", required_argument, NULL, 't'},
+       {"enable-two-phase", no_argument, NULL, 'T'},
        {"subscriber-username", required_argument, NULL, 'U'},
        {"verbose", no_argument, NULL, 'v'},
        {"version", no_argument, NULL, 'V'},
@@ -1950,6 +1970,7 @@ main(int argc, char **argv)
    opt.socket_dir = NULL;
    opt.sub_port = DEFAULT_SUB_PORT;
    opt.sub_username = NULL;
+   opt.two_phase = false;
    opt.database_names = (SimpleStringList)
    {
        0
@@ -1972,7 +1993,7 @@ main(int argc, char **argv)
 
    get_restricted_token();
 
-   while ((c = getopt_long(argc, argv, "d:D:np:P:s:t:U:v",
+   while ((c = getopt_long(argc, argv, "d:D:np:P:s:t:TU:v",
                            long_options, &option_index)) != -1)
    {
        switch (c)
@@ -2009,6 +2030,9 @@ main(int argc, char **argv)
            case 't':
                opt.recovery_timeout = atoi(optarg);
                break;
+           case 'T':
+               opt.two_phase = true;
+               break;
            case 'U':
                opt.sub_username = pg_strdup(optarg);
                break;
@@ -2170,12 +2194,14 @@ main(int argc, char **argv)
    /* Rudimentary check for a data directory */
    check_data_directory(subscriber_dir);
 
+   dbinfos.two_phase = opt.two_phase;
+
    /*
     * Store database information for publisher and subscriber. It should be
     * called before atexit() because its return is used in the
     * cleanup_objects_atexit().
     */
-   dbinfo = store_pub_sub_info(&opt, pub_base_conninfo, sub_base_conninfo);
+   dbinfos.dbinfo = store_pub_sub_info(&opt, pub_base_conninfo, sub_base_conninfo);
 
    /* Register a function to clean up objects in case of failure */
    atexit(cleanup_objects_atexit);
@@ -2184,7 +2210,7 @@ main(int argc, char **argv)
     * Check if the subscriber data directory has the same system identifier
     * than the publisher data directory.
     */
-   pub_sysid = get_primary_sysid(dbinfo[0].pubconninfo);
+   pub_sysid = get_primary_sysid(dbinfos.dbinfo[0].pubconninfo);
    sub_sysid = get_standby_sysid(subscriber_dir);
    if (pub_sysid != sub_sysid)
        pg_fatal("subscriber data directory is not a copy of the source database cluster");
@@ -2214,10 +2240,10 @@ main(int argc, char **argv)
    start_standby_server(&opt, true, false);
 
    /* Check if the standby server is ready for logical replication */
-   check_subscriber(dbinfo);
+   check_subscriber(dbinfos.dbinfo);
 
    /* Check if the primary server is ready for logical replication */
-   check_publisher(dbinfo);
+   check_publisher(dbinfos.dbinfo);
 
    /*
     * Stop the target server. The recovery process requires that the server
@@ -2230,10 +2256,10 @@ main(int argc, char **argv)
    stop_standby_server(subscriber_dir);
 
    /* Create the required objects for each database on publisher */
-   consistent_lsn = setup_publisher(dbinfo);
+   consistent_lsn = setup_publisher(dbinfos.dbinfo);
 
    /* Write the required recovery parameters */
-   setup_recovery(dbinfo, subscriber_dir, consistent_lsn);
+   setup_recovery(dbinfos.dbinfo, subscriber_dir, consistent_lsn);
 
    /*
     * Start subscriber so the recovery parameters will take effect. Wait
@@ -2244,7 +2270,7 @@ main(int argc, char **argv)
    start_standby_server(&opt, true, true);
 
    /* Waiting the subscriber to be promoted */
-   wait_for_end_recovery(dbinfo[0].subconninfo, &opt);
+   wait_for_end_recovery(dbinfos.dbinfo[0].subconninfo, &opt);
 
    /*
     * Create the subscription for each database on subscriber. It does not
@@ -2252,13 +2278,13 @@ main(int argc, char **argv)
     * point to the LSN reported by setup_publisher().  It also cleans up
     * publications created by this tool and replication to the standby.
     */
-   setup_subscriber(dbinfo, consistent_lsn);
+   setup_subscriber(dbinfos.dbinfo, consistent_lsn);
 
    /* Remove primary_slot_name if it exists on primary */
-   drop_primary_replication_slot(dbinfo, primary_slot_name);
+   drop_primary_replication_slot(dbinfos.dbinfo, primary_slot_name);
 
    /* Remove failover replication slots if they exist on subscriber */
-   drop_failover_replication_slots(dbinfo);
+   drop_failover_replication_slots(dbinfos.dbinfo);
 
    /* Stop the subscriber */
    pg_log_info("stopping the subscriber");

diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
index c8dbdb7e9b79c74a2c6708b4ef3eefb563be670f..c35fa108ce3459afd09d793f6217e390f78fae7a 100644 (file)
--- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -373,6 +373,7 @@ command_ok(
 
 # Run pg_createsubscriber on node S.  --verbose is used twice
 # to show more information.
+# In passing, also test the --enable-two-phase option
 command_ok(
    [
        'pg_createsubscriber',
@@ -388,6 +389,7 @@ command_ok(
        '--replication-slot' => 'replslot2',
        '--database' => $db1,
        '--database' => $db2,
+       '--enable-two-phase'
    ],
    'run pg_createsubscriber on node S');
 
@@ -406,6 +408,15 @@ $node_p->safe_psql($db2, "INSERT INTO tbl2 VALUES('row 1')");
 # Start subscriber
 $node_s->start;
 
+# Verify that all subtwophase states are pending or enabled,
+# e.g. there are no subscriptions where subtwophase is disabled ('d')
+is( $node_s->safe_psql(
+       'postgres',
+       "SELECT count(1) = 0 FROM pg_subscription WHERE subtwophasestate = 'd'"
+   ),
+   't',
+   'subscriptions are created with the two-phase option enabled');
+
 # Confirm the pre-existing subscription has been removed
 $result = $node_s->safe_psql(
    'postgres', qq(