Re: Adding a '--two-phase' option to 'pg_createsubscriber' utility.

[vignesh C]

On Wed, 11 Dec 2024 at 11:21, Shubham Khanna
<khannashubham1...@gmail.com> wrote:
>
> On Wed, Dec 11, 2024 at 4:21 AM Peter Smith <smithpb2...@gmail.com> wrote:
>
> I have fixed the given comments. The attached patch contains the
> suggested changes.

Since all the subscriptions are created based on the two_phase option
provided, there is no need to store this for each database:
@@ -53,6 +54,7 @@ struct LogicalRepInfo
        char       *pubname;            /* publication name */
        char       *subname;            /* subscription name */
        char       *replslotname;       /* replication slot name */
+       bool            two_phase;              /* two-phase enabled
for the subscription */

                dbinfo[i].dbname = cell->val;
+               dbinfo[i].two_phase = opt->two_phase;
                if (num_pubs > 0)

How about we handle something like in the attached changes.

Regards,
Vignesh

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 01a203983d..d141b13bd6 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -54,7 +54,6 @@ struct LogicalRepInfo
 	char	   *pubname;		/* publication name */
 	char	   *subname;		/* subscription name */
 	char	   *replslotname;	/* replication slot name */
-	bool		two_phase;		/* two-phase enabled for the subscription */
 
 	bool		made_replslot;	/* replication slot was created */
 	bool		made_publication;	/* publication was created */
@@ -81,7 +80,7 @@ static void check_publisher(const struct LogicalRepInfo *dbinfo);
 static char *setup_publisher(struct LogicalRepInfo *dbinfo);
 static void check_subscriber(const struct LogicalRepInfo *dbinfo);
 static void setup_subscriber(struct LogicalRepInfo *dbinfo,
-							 const char *consistent_lsn);
+							 const char *consistent_lsn, bool two_phase);
 static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir,
 						   const char *lsn);
 static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
@@ -100,7 +99,9 @@ static void wait_for_end_recovery(const char *conninfo,
 								  const struct CreateSubscriberOptions *opt);
 static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
 static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
-static void create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
+static void create_subscription(PGconn *conn,
+								const struct LogicalRepInfo *dbinfo,
+								bool two_phase);
 static void set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo,
 									 const char *lsn);
 static void enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
@@ -459,7 +460,6 @@ store_pub_sub_info(const struct CreateSubscriberOptions *opt,
 		conninfo = concat_conninfo_dbname(pub_base_conninfo, cell->val);
 		dbinfo[i].pubconninfo = conninfo;
 		dbinfo[i].dbname = cell->val;
-		dbinfo[i].two_phase = opt->two_phase;
 		if (num_pubs > 0)
 			dbinfo[i].pubname = pubcell->val;
 		else
@@ -486,7 +486,7 @@ store_pub_sub_info(const struct CreateSubscriberOptions *opt,
 		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].two_phase ? "true" : "false");
+					 opt->two_phase ? "true" : "false");
 
 		if (num_pubs > 0)
 			pubcell = pubcell->next;
@@ -1143,7 +1143,8 @@ check_and_drop_existing_subscriptions(PGconn *conn,
  * replication setup.
  */
 static void
-setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
+setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn,
+				 bool two_phase)
 {
 	for (int i = 0; i < num_dbs; i++)
 	{
@@ -1167,7 +1168,7 @@ setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
 		 */
 		drop_publication(conn, &dbinfo[i]);
 
-		create_subscription(conn, &dbinfo[i]);
+		create_subscription(conn, &dbinfo[i], two_phase);
 
 		/* Set the replication progress to the correct LSN */
 		set_replication_progress(conn, &dbinfo[i], consistent_lsn);
@@ -1682,7 +1683,8 @@ drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
  * initial location.
  */
 static void
-create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
+create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo,
+					bool two_phase)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
@@ -1706,7 +1708,7 @@ create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
 					  "WITH (create_slot = false, enabled = false, "
 					  "slot_name = %s, copy_data = false, two_phase = %s)",
 					  subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc,
-					  dbinfo->two_phase ? "true" : "false");
+					  two_phase ? "true" : "false");
 
 	pg_free(pubname_esc);
 	pg_free(subname_esc);
@@ -2240,7 +2242,7 @@ 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(dbinfo, consistent_lsn, opt.two_phase);
 
 	/* Remove primary_slot_name if it exists on primary */
 	drop_primary_replication_slot(dbinfo, primary_slot_name);