v10-0004-Divide-LogicalReplInfo-into-some-strcutures.patch

From afd8eb318c40e4602268cf5a82b209cb7fa07af2 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Date: Mon, 29 Jan 2024 07:03:59 +0000
Subject: [PATCH v10 4/4] Divide LogicalReplInfo into some strcutures

---
 src/bin/pg_basebackup/pg_subscriber.c | 674 ++++++++++++++------------
 src/tools/pgindent/typedefs.list      |   4 +
 2 files changed, 374 insertions(+), 304 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_subscriber.c b/src/bin/pg_basebackup/pg_subscriber.c
index f535e7160f..a8df8d5135 100644
--- a/src/bin/pg_basebackup/pg_subscriber.c
+++ b/src/bin/pg_basebackup/pg_subscriber.c
@@ -32,51 +32,72 @@
 
 #define	PGS_OUTPUT_DIR	"pg_subscriber_output.d"
 
-typedef struct LogicalRepInfo
+typedef struct LogicalRepPerdbInfo
 {
-	Oid			oid;			/* database OID */
-	char	   *dbname;			/* database name */
-	char	   *pubconninfo;	/* publication connection string for logical
-								 * replication */
-	char	   *subconninfo;	/* subscription connection string for logical
-								 * replication */
-	char	   *pubname;		/* publication name */
-	char	   *subname;		/* subscription name (also replication slot
-								 * name) */
-
-	bool		made_replslot;	/* replication slot was created */
-	bool		made_publication;	/* publication was created */
-	bool		made_subscription;	/* subscription was created */
-} LogicalRepInfo;
+	Oid		oid;
+	char   *dbname;
+	bool	made_replslot;		/* replication slot was created */
+	bool	made_publication;	/* publication was created */
+	bool	made_subscription; 	/* subscription was created */
+} LogicalRepPerdbInfo;
+
+typedef struct LogicalRepPerdbInfoArr
+{
+	LogicalRepPerdbInfo    *perdb;	/* array of db infos */
+	int						ndbs;	/* number of db infos */
+} LogicalRepPerdbInfoArr;
+
+typedef struct PrimaryInfo
+{
+	char   *base_conninfo;
+	uint64	sysid;
+	bool	made_transient_replslot;
+} PrimaryInfo;
+
+typedef struct StandbyInfo
+{
+	char   *base_conninfo;
+	char   *bindir;
+	char   *pgdata;
+	char   *primary_slot_name;
+	char   *server_log;
+	uint64	sysid;
+} StandbyInfo;
 
 static void cleanup_objects_atexit(void);
 static void usage();
 static char *get_base_conninfo(char *conninfo, char *dbname,
 							   const char *noderole);
-static bool get_exec_path(const char *path);
+static bool get_exec_base_path(const char *path);
 static bool check_data_directory(const char *datadir);
 static char *concat_conninfo_dbname(const char *conninfo, const char *dbname);
-static LogicalRepInfo *store_pub_sub_info(const char *pub_base_conninfo, const char *sub_base_conninfo);
-static PGconn *connect_database(const char *conninfo);
+static void store_db_names(LogicalRepPerdbInfo **perdb, int ndbs);
+static PGconn *connect_database(const char *base_conninfo, const char*dbname);
 static void disconnect_database(PGconn *conn);
-static uint64 get_sysid_from_conn(const char *conninfo);
-static uint64 get_control_from_datadir(const char *datadir);
-static void modify_sysid(const char *pg_resetwal_path, const char *datadir);
-static bool setup_publisher(LogicalRepInfo *dbinfo);
-static char *create_logical_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo,
-											 char *slot_name);
-static void drop_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo, const char *slot_name);
+static void get_sysid_for_primary(PrimaryInfo *primary, char *dbname);
+static void get_sysid_for_standby(StandbyInfo *standby);
+static void modify_sysid(const char *bindir, const char *datadir);
+static bool setup_publisher(PrimaryInfo *primary, LogicalRepPerdbInfoArr *dbarr);
+static char *create_logical_replication_slot(PGconn *conn, bool temporary,
+											 LogicalRepPerdbInfo *perdb);
+static void drop_replication_slot(PGconn *conn, LogicalRepPerdbInfo *perdb,
+								  const char *slot_name);
 static void pg_ctl_status(const char *pg_ctl_cmd, int rc, int action);
-static void wait_for_end_recovery(const char *conninfo);
-static void create_publication(PGconn *conn, LogicalRepInfo *dbinfo);
-static void drop_publication(PGconn *conn, LogicalRepInfo *dbinfo);
-static void create_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
-static void drop_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
-static void set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn);
-static void enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
-
-static void start_standby(char *server_start_log);
-static void stop_standby(void);
+static void wait_for_end_recovery(StandbyInfo *standby,
+								  const char *dbname);
+static void create_publication(PGconn *conn, PrimaryInfo *primary,
+							   LogicalRepPerdbInfo *perdb);
+static void drop_publication(PGconn *conn, LogicalRepPerdbInfo *perdb);
+static void create_subscription(PGconn *conn, StandbyInfo *standby,
+								PrimaryInfo *primary,
+								LogicalRepPerdbInfo *perdb);
+static void drop_subscription(PGconn *conn, LogicalRepPerdbInfo *perdb);
+static void set_replication_progress(PGconn *conn, LogicalRepPerdbInfo *perdb,
+									 const char *lsn);
+static void enable_subscription(PGconn *conn, LogicalRepPerdbInfo *perdb);
+
+static void start_standby(StandbyInfo *standby);
+static void stop_standby(StandbyInfo *standby);
 
 #define	USEC_PER_SEC	1000000
 #define	WAIT_INTERVAL	1		/* 1 second */
@@ -84,25 +105,18 @@ static void stop_standby(void);
 /* Options */
 static const char *progname;
 
-static char *subscriber_dir = NULL;
 static char *pub_conninfo_str = NULL;
 static char *sub_conninfo_str = NULL;
 static SimpleStringList database_names = {NULL, NULL};
-static char *primary_slot_name = NULL;
 static bool dry_run = false;
 static bool retain = false;
 static int	recovery_timeout = 0;
 
 static bool success = false;
 
-static char *pg_ctl_path = NULL;
-static char *pg_resetwal_path = NULL;
-
-static LogicalRepInfo *dbinfo;
-static int	num_dbs = 0;
-
-static char temp_replslot[NAMEDATALEN] = {0};
-static bool made_transient_replslot = false;
+static LogicalRepPerdbInfoArr dbarr;
+static PrimaryInfo primary;
+static StandbyInfo standby;
 
 enum WaitPMResult
 {
@@ -112,6 +126,30 @@ enum WaitPMResult
 	POSTMASTER_FAILED
 };
 
+/*
+ * Build the replication slot and subscription name. The name must not exceed
+ * NAMEDATALEN - 1. This current schema uses a maximum of 36 characters
+ * (14 + 10 + 1 + 10 + '\0'). System identifier is included to reduce the
+ * probability of collision. By default, subscription name is used as
+ * replication slot name.
+ */
+static inline void
+get_subscription_name(Oid oid, int pid, char *subname, Size szsub)
+{
+	snprintf(subname, szsub, "pg_subscriber_%u_%d", oid, pid);
+}
+
+/*
+ * Build the publication name. The name must not exceed NAMEDATALEN -
+ * 1. This current schema uses a maximum of 35 characters (14 + 10 +
+ * '\0').
+ */
+static inline void
+get_publication_name(Oid oid, char *pubname, Size szpub)
+{
+	snprintf(pubname, szpub, "pg_subscriber_%u", oid);
+}
+
 
 /*
  * Cleanup objects that were created by pg_subscriber if there is an error.
@@ -129,38 +167,51 @@ cleanup_objects_atexit(void)
 	if (success)
 		return;
 
-	for (i = 0; i < num_dbs; i++)
+	for (i = 0; i < dbarr.ndbs; i++)
 	{
-		if (dbinfo[i].made_subscription)
+		LogicalRepPerdbInfo *perdb = &dbarr.perdb[i];
+
+		if (perdb->made_subscription)
 		{
-			conn = connect_database(dbinfo[i].subconninfo);
+			conn = connect_database(standby.base_conninfo, perdb->dbname);
 			if (conn != NULL)
 			{
-				drop_subscription(conn, &dbinfo[i]);
+				drop_subscription(conn, perdb);
 				disconnect_database(conn);
 			}
 		}
 
-		if (dbinfo[i].made_publication || dbinfo[i].made_replslot)
+		if (perdb->made_publication || perdb->made_replslot)
 		{
-			conn = connect_database(dbinfo[i].pubconninfo);
+			conn = connect_database(primary.base_conninfo, perdb->dbname);
 			if (conn != NULL)
 			{
-				if (dbinfo[i].made_publication)
-					drop_publication(conn, &dbinfo[i]);
-				if (dbinfo[i].made_replslot)
-					drop_replication_slot(conn, &dbinfo[i], NULL);
-				disconnect_database(conn);
+				if (perdb->made_publication)
+					drop_publication(conn, perdb);
+				if (perdb->made_replslot)
+				{
+					char replslotname[NAMEDATALEN];
+
+					get_subscription_name(perdb->oid, (int) getpid(),
+										  replslotname, NAMEDATALEN);
+					drop_replication_slot(conn, perdb, replslotname);
+				}
 			}
 		}
 	}
 
-	if (made_transient_replslot)
+	if (primary.made_transient_replslot)
 	{
-		conn = connect_database(dbinfo[0].pubconninfo);
+		char transient_replslot[NAMEDATALEN];
+
+		conn = connect_database(primary.base_conninfo, dbarr.perdb[0].dbname);
+
 		if (conn != NULL)
 		{
-			drop_replication_slot(conn, &dbinfo[0], temp_replslot);
+			snprintf(transient_replslot, NAMEDATALEN, "pg_subscriber_%d_startpoint",
+					 (int) getpid());
+
+			drop_replication_slot(conn, &dbarr.perdb[0], transient_replslot);
 			disconnect_database(conn);
 		}
 	}
@@ -246,15 +297,16 @@ get_base_conninfo(char *conninfo, char *dbname, const char *noderole)
 }
 
 /*
- * Get the absolute path from other PostgreSQL binaries (pg_ctl and
- * pg_resetwal) that is used by it.
+ * Get the absolute binary path from another PostgreSQL binary (pg_ctl) and set
+ * to StandbyInfo.
  */
 static bool
-get_exec_path(const char *path)
+get_exec_base_path(const char *path)
 {
-	int			rc;
+	int		rc;
+	char	pg_ctl_path[MAXPGPATH];
+	char   *p;
 
-	pg_ctl_path = pg_malloc(MAXPGPATH);
 	rc = find_other_exec(path, "pg_ctl",
 						 "pg_ctl (PostgreSQL) " PG_VERSION "\n",
 						 pg_ctl_path);
@@ -279,30 +331,12 @@ get_exec_path(const char *path)
 
 	pg_log_debug("pg_ctl path is: %s", pg_ctl_path);
 
-	pg_resetwal_path = pg_malloc(MAXPGPATH);
-	rc = find_other_exec(path, "pg_resetwal",
-						 "pg_resetwal (PostgreSQL) " PG_VERSION "\n",
-						 pg_resetwal_path);
-	if (rc < 0)
-	{
-		char		full_path[MAXPGPATH];
+	/* Extract the directory part from the path */
+	p = strrchr(pg_ctl_path, 'p');
+	Assert(p);
 
-		if (find_my_exec(path, full_path) < 0)
-			strlcpy(full_path, progname, sizeof(full_path));
-		if (rc == -1)
-			pg_log_error("The program \"%s\" is needed by %s but was not found in the\n"
-						 "same directory as \"%s\".\n"
-						 "Check your installation.",
-						 "pg_resetwal", progname, full_path);
-		else
-			pg_log_error("The program \"%s\" was found by \"%s\"\n"
-						 "but was not the same version as %s.\n"
-						 "Check your installation.",
-						 "pg_resetwal", full_path, progname);
-		return false;
-	}
-
-	pg_log_debug("pg_resetwal path is: %s", pg_resetwal_path);
+	*p = '\0';
+	standby.bindir = pg_strdup(pg_ctl_path);
 
 	return true;
 }
@@ -366,49 +400,35 @@ concat_conninfo_dbname(const char *conninfo, const char *dbname)
 }
 
 /*
- * Store publication and subscription information.
+ * Initialize per-db structure and store the name of databases.
  */
-static LogicalRepInfo *
-store_pub_sub_info(const char *pub_base_conninfo, const char *sub_base_conninfo)
+static void
+store_db_names(LogicalRepPerdbInfo **perdb, int ndbs)
 {
-	LogicalRepInfo *dbinfo;
-	SimpleStringListCell *cell;
-	int			i = 0;
+	SimpleStringListCell   *cell;
+	int						i = 0;
 
-	dbinfo = (LogicalRepInfo *) pg_malloc(num_dbs * sizeof(LogicalRepInfo));
+	dbarr.perdb = (LogicalRepPerdbInfo *) pg_malloc0(ndbs *
+											   sizeof(LogicalRepPerdbInfo));
 
 	for (cell = database_names.head; cell; cell = cell->next)
 	{
-		char	   *conninfo;
-
-		/* Publisher. */
-		conninfo = concat_conninfo_dbname(pub_base_conninfo, cell->val);
-		dbinfo[i].pubconninfo = conninfo;
-		dbinfo[i].dbname = cell->val;
-		dbinfo[i].made_replslot = false;
-		dbinfo[i].made_publication = false;
-		dbinfo[i].made_subscription = false;
-		/* other struct fields will be filled later. */
-
-		/* Subscriber. */
-		conninfo = concat_conninfo_dbname(sub_base_conninfo, cell->val);
-		dbinfo[i].subconninfo = conninfo;
-
+		(*perdb)[i].dbname = pg_strdup(cell->val);
 		i++;
 	}
-
-	return dbinfo;
 }
 
 static PGconn *
-connect_database(const char *conninfo)
+connect_database(const char *base_conninfo, const char*dbname)
 {
 	PGconn	   *conn;
 	PGresult   *res;
-	const char *rconninfo;
+	char	   *rconninfo;
+	char	   *concat_conninfo = concat_conninfo_dbname(base_conninfo,
+														 dbname);
 
 	/* logical replication mode */
-	rconninfo = psprintf("%s replication=database", conninfo);
+	rconninfo = psprintf("%s replication=database", concat_conninfo);
 
 	conn = PQconnectdb(rconninfo);
 	if (PQstatus(conn) != CONNECTION_OK)
@@ -426,6 +446,8 @@ connect_database(const char *conninfo)
 	}
 	PQclear(res);
 
+	pfree(rconninfo);
+	pfree(concat_conninfo);
 	return conn;
 }
 
@@ -438,19 +460,18 @@ disconnect_database(PGconn *conn)
 }
 
 /*
- * Obtain the system identifier using the provided connection. It will be used
- * to compare if a data directory is a clone of another one.
+ * Obtain the system identifier from the primary server. It will be used to
+ * compare if a data directory is a clone of another one.
  */
-static uint64
-get_sysid_from_conn(const char *conninfo)
+static void
+get_sysid_for_primary(PrimaryInfo *primary, char *dbname)
 {
 	PGconn	   *conn;
 	PGresult   *res;
-	uint64		sysid;
 
 	pg_log_info("getting system identifier from publisher");
 
-	conn = connect_database(conninfo);
+	conn = connect_database(primary->base_conninfo, dbname);
 	if (conn == NULL)
 		exit(1);
 
@@ -473,43 +494,40 @@ get_sysid_from_conn(const char *conninfo)
 		exit(1);
 	}
 
-	sysid = strtou64(PQgetvalue(res, 0, 0), NULL, 10);
+	primary->sysid = strtou64(PQgetvalue(res, 0, 0), NULL, 10);
 
-	pg_log_info("system identifier is %llu on publisher", (unsigned long long) sysid);
+	pg_log_info("system identifier is " UINT64_FORMAT " on publisher",
+				primary->sysid);
 
 	disconnect_database(conn);
-
-	return sysid;
 }
 
 /*
- * Obtain the system identifier from control file. It will be used to compare
- * if a data directory is a clone of another one. This routine is used locally
- * and avoids a replication connection.
+ * Obtain the system identifier from a standby server. It will be used to
+ * compare if a data directory is a clone of another one. This routine is used
+ * locally and avoids a replication connection.
  */
-static uint64
-get_control_from_datadir(const char *datadir)
+static void
+get_sysid_for_standby(StandbyInfo *standby)
 {
 	ControlFileData *cf;
 	bool		crc_ok;
-	uint64		sysid;
 
 	pg_log_info("getting system identifier from subscriber");
 
-	cf = get_controlfile(datadir, &crc_ok);
+	cf = get_controlfile(standby->pgdata, &crc_ok);
 	if (!crc_ok)
 	{
 		pg_log_error("control file appears to be corrupt");
 		exit(1);
 	}
 
-	sysid = cf->system_identifier;
+	standby->sysid = cf->system_identifier;
 
-	pg_log_info("system identifier is %llu on subscriber", (unsigned long long) sysid);
+	pg_log_info("system identifier is " UINT64_FORMAT " on subscriber",
+				standby->sysid);
 
 	pfree(cf);
-
-	return sysid;
 }
 
 /*
@@ -518,7 +536,7 @@ get_control_from_datadir(const char *datadir)
  * files from one of the systems might be used in the other one.
  */
 static void
-modify_sysid(const char *pg_resetwal_path, const char *datadir)
+modify_sysid(const char *bindir, const char *datadir)
 {
 	ControlFileData *cf;
 	bool		crc_ok;
@@ -553,7 +571,7 @@ modify_sysid(const char *pg_resetwal_path, const char *datadir)
 
 	pg_log_info("running pg_resetwal on the subscriber");
 
-	cmd_str = psprintf("\"%s\" -D \"%s\"", pg_resetwal_path, datadir);
+	cmd_str = psprintf("\"%s/pg_resetwal\" -D \"%s\"", bindir, datadir);
 
 	pg_log_debug("command is: %s", cmd_str);
 
@@ -574,7 +592,7 @@ modify_sysid(const char *pg_resetwal_path, const char *datadir)
  * publications and replication slots in preparation for logical replication.
  */
 static bool
-setup_publisher(LogicalRepInfo *dbinfo)
+setup_publisher(PrimaryInfo *primary, LogicalRepPerdbInfoArr *dbarr)
 {
 	PGconn	   *conn;
 	PGresult   *res;
@@ -597,7 +615,7 @@ setup_publisher(LogicalRepInfo *dbinfo)
 	 * max_replication_slots >= current + number of dbs to be converted
 	 * max_wal_senders >= current + number of dbs to be converted
 	 */
-	conn = connect_database(dbinfo[0].pubconninfo);
+	conn = connect_database(primary->base_conninfo, dbarr->perdb[0].dbname);
 	if (conn == NULL)
 		exit(1);
 
@@ -635,10 +653,11 @@ setup_publisher(LogicalRepInfo *dbinfo)
 	 * use after the transformation, hence, it will be removed at the end of
 	 * this process.
 	 */
-	if (primary_slot_name)
+	if (standby.primary_slot_name)
 	{
 		appendPQExpBuffer(str,
-						  "SELECT 1 FROM pg_replication_slots WHERE active AND slot_name = '%s'", primary_slot_name);
+						  "SELECT 1 FROM pg_replication_slots WHERE active AND slot_name = '%s'",
+						  standby.primary_slot_name);
 
 		pg_log_debug("command is: %s", str->data);
 
@@ -653,13 +672,14 @@ setup_publisher(LogicalRepInfo *dbinfo)
 		{
 			pg_log_error("could not obtain replication slot information: got %d rows, expected %d row",
 						 PQntuples(res), 1);
-			pg_free(primary_slot_name); /* it is not being used. */
-			primary_slot_name = NULL;
+			pg_free(standby.primary_slot_name); /* it is not being used. */
+			standby.primary_slot_name = NULL;
 			return false;
 		}
 		else
 		{
-			pg_log_info("primary has replication slot \"%s\"", primary_slot_name);
+			pg_log_info("primary has replication slot \"%s\"",
+						standby.primary_slot_name);
 		}
 
 		PQclear(res);
@@ -673,26 +693,29 @@ setup_publisher(LogicalRepInfo *dbinfo)
 		return false;
 	}
 
-	if (max_repslots - cur_repslots < num_dbs)
+	if (max_repslots - cur_repslots < dbarr->ndbs)
 	{
-		pg_log_error("publisher requires %d replication slots, but only %d remain", num_dbs, max_repslots - cur_repslots);
-		pg_log_error_hint("Consider increasing max_replication_slots to at least %d.", cur_repslots + num_dbs);
+		pg_log_error("publisher requires %d replication slots, but only %d remain",
+					 dbarr->ndbs, max_repslots - cur_repslots);
+		pg_log_error_hint("Consider increasing max_replication_slots to at least %d.",
+						  cur_repslots + dbarr->ndbs);
 		return false;
 	}
 
-	if (max_walsenders - cur_walsenders < num_dbs)
+	if (max_walsenders - cur_walsenders < dbarr->ndbs)
 	{
-		pg_log_error("publisher requires %d wal sender processes, but only %d remain", num_dbs, max_walsenders - cur_walsenders);
-		pg_log_error_hint("Consider increasing max_wal_senders to at least %d.", cur_walsenders + num_dbs);
+		pg_log_error("publisher requires %d wal sender processes, but only %d remain",
+					 dbarr->ndbs, max_walsenders - cur_walsenders);
+		pg_log_error_hint("Consider increasing max_wal_senders to at least %d.",
+						  cur_walsenders + dbarr->ndbs);
 		return false;
 	}
 
-	for (int i = 0; i < num_dbs; i++)
+	for (int i = 0; i < dbarr->ndbs; i++)
 	{
-		char		pubname[NAMEDATALEN];
-		char		replslotname[NAMEDATALEN];
+		LogicalRepPerdbInfo *perdb = &dbarr->perdb[i];
 
-		conn = connect_database(dbinfo[i].pubconninfo);
+		conn = connect_database(primary->base_conninfo, perdb->dbname);
 		if (conn == NULL)
 			exit(1);
 
@@ -712,43 +735,21 @@ setup_publisher(LogicalRepInfo *dbinfo)
 		}
 
 		/* Remember database OID. */
-		dbinfo[i].oid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
+		perdb->oid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
 
 		PQclear(res);
 
-		/*
-		 * Build the publication name. The name must not exceed NAMEDATALEN -
-		 * 1. This current schema uses a maximum of 35 characters (14 + 10 +
-		 * '\0').
-		 */
-		snprintf(pubname, sizeof(pubname), "pg_subscriber_%u", dbinfo[i].oid);
-		dbinfo[i].pubname = pg_strdup(pubname);
-
 		/*
 		 * Create publication on publisher. This step should be executed
 		 * *before* promoting the subscriber to avoid any transactions between
 		 * consistent LSN and the new publication rows (such transactions
 		 * wouldn't see the new publication rows resulting in an error).
 		 */
-		create_publication(conn, &dbinfo[i]);
-
-		/*
-		 * Build the replication slot name. The name must not exceed
-		 * NAMEDATALEN - 1. This current schema uses a maximum of 36
-		 * characters (14 + 10 + 1 + 10 + '\0'). System identifier is included
-		 * to reduce the probability of collision. By default, subscription
-		 * name is used as replication slot name.
-		 */
-		snprintf(replslotname, sizeof(replslotname),
-				 "pg_subscriber_%u_%d",
-				 dbinfo[i].oid,
-				 (int) getpid());
-		dbinfo[i].subname = pg_strdup(replslotname);
+		create_publication(conn, primary, perdb);
 
 		/* Create replication slot on publisher. */
-		if (create_logical_replication_slot(conn, &dbinfo[i], replslotname) != NULL || dry_run)
-			pg_log_info("create replication slot \"%s\" on publisher", replslotname);
-		else
+		if (create_logical_replication_slot(conn, false, perdb) == NULL &&
+			!dry_run)
 			return false;
 
 		disconnect_database(conn);
@@ -761,7 +762,7 @@ setup_publisher(LogicalRepInfo *dbinfo)
  * Is the target server ready for logical replication?
  */
 static bool
-setup_subscriber(LogicalRepInfo *dbinfo)
+setup_subscriber(StandbyInfo *standby, LogicalRepPerdbInfoArr *dbarr)
 {
 	PGconn	   *conn;
 	PGresult   *res;
@@ -781,7 +782,7 @@ setup_subscriber(LogicalRepInfo *dbinfo)
 	 * max_logical_replication_workers >= number of dbs to be converted
 	 * max_worker_processes >= 1 + number of dbs to be converted
 	 */
-	conn = connect_database(dbinfo[0].subconninfo);
+	conn = connect_database(standby->base_conninfo, dbarr->perdb[0].dbname);
 	if (conn == NULL)
 		exit(1);
 
@@ -798,35 +799,41 @@ setup_subscriber(LogicalRepInfo *dbinfo)
 	max_repslots = atoi(PQgetvalue(res, 1, 0));
 	max_wprocs = atoi(PQgetvalue(res, 2, 0));
 	if (strcmp(PQgetvalue(res, 3, 0), "") != 0)
-		primary_slot_name = pg_strdup(PQgetvalue(res, 3, 0));
+		standby->primary_slot_name = pg_strdup(PQgetvalue(res, 3, 0));
 
 	pg_log_debug("subscriber: max_logical_replication_workers: %d", max_lrworkers);
 	pg_log_debug("subscriber: max_replication_slots: %d", max_repslots);
 	pg_log_debug("subscriber: max_worker_processes: %d", max_wprocs);
-	pg_log_debug("subscriber: primary_slot_name: %s", primary_slot_name);
+	pg_log_debug("subscriber: primary_slot_name: %s", standby->primary_slot_name);
 
 	PQclear(res);
 
 	disconnect_database(conn);
 
-	if (max_repslots < num_dbs)
+	if (max_repslots < dbarr->ndbs)
 	{
-		pg_log_error("subscriber requires %d replication slots, but only %d remain", num_dbs, max_repslots);
-		pg_log_error_hint("Consider increasing max_replication_slots to at least %d.", num_dbs);
+		pg_log_error("subscriber requires %d replication slots, but only %d remain",
+					 dbarr->ndbs, max_repslots);
+		pg_log_error_hint("Consider increasing max_replication_slots to at least %d.",
+						  dbarr->ndbs);
 		return false;
 	}
 
-	if (max_lrworkers < num_dbs)
+	if (max_lrworkers < dbarr->ndbs)
 	{
-		pg_log_error("subscriber requires %d logical replication workers, but only %d remain", num_dbs, max_lrworkers);
-		pg_log_error_hint("Consider increasing max_logical_replication_workers to at least %d.", num_dbs);
+		pg_log_error("subscriber requires %d logical replication workers, but only %d remain",
+					 dbarr->ndbs, max_lrworkers);
+		pg_log_error_hint("Consider increasing max_logical_replication_workers to at least %d.",
+						  dbarr->ndbs);
 		return false;
 	}
 
-	if (max_wprocs < num_dbs + 1)
+	if (max_wprocs < dbarr->ndbs + 1)
 	{
-		pg_log_error("subscriber requires %d worker processes, but only %d remain", num_dbs + 1, max_wprocs);
-		pg_log_error_hint("Consider increasing max_worker_processes to at least %d.", num_dbs + 1);
+		pg_log_error("subscriber requires %d worker processes, but only %d remain",
+					 dbarr->ndbs + 1, max_wprocs);
+		pg_log_error_hint("Consider increasing max_worker_processes to at least %d.",
+						  dbarr->ndbs + 1);
 		return false;
 	}
 
@@ -841,28 +848,32 @@ setup_subscriber(LogicalRepInfo *dbinfo)
  * result set that contains the consistent LSN.
  */
 static char *
-create_logical_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo,
-								char *slot_name)
+create_logical_replication_slot(PGconn *conn, bool temporary,
+								LogicalRepPerdbInfo *perdb)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res = NULL;
 	char	   *lsn = NULL;
-	bool		transient_replslot = false;
+	char		slot_name[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
 	/*
-	 * If no slot name is informed, it is a transient replication slot used
-	 * only for catch up purposes.
+	 * Construct a name of logical replication slot. The formatting is
+	 * different depends on its persistency.
+	 *
+	 * For persistent slots: the name must be same as the subscription.
+	 * For temporary slots: OID is not needed, but another string is added.
 	 */
-	if (slot_name[0] == '\0')
-	{
+	if (temporary)
 		snprintf(slot_name, NAMEDATALEN, "pg_subscriber_%d_startpoint",
 				 (int) getpid());
-		transient_replslot = true;
-	}
+	else
+		get_subscription_name(perdb->oid, (int) getpid(), slot_name,
+							  NAMEDATALEN);
 
-	pg_log_info("creating the replication slot \"%s\" on database \"%s\"", slot_name, dbinfo->dbname);
+	pg_log_info("creating the replication slot \"%s\" on database \"%s\"",
+				slot_name, perdb->dbname);
 
 	appendPQExpBuffer(str, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	appendPQExpBufferStr(str, " LOGICAL \"pgoutput\" NOEXPORT_SNAPSHOT");
@@ -874,17 +885,19 @@ create_logical_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo,
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 		{
-			pg_log_error("could not create replication slot \"%s\" on database \"%s\": %s", slot_name, dbinfo->dbname,
-						 PQresultErrorMessage(res));
+			pg_log_error("could not create replication slot \"%s\" on database \"%s\": %s",
+						 slot_name, perdb->dbname, PQresultErrorMessage(res));
 			return lsn;
 		}
+
+		pg_log_info("create replication slot \"%s\" on publisher", slot_name);
 	}
 
 	/* for cleanup purposes */
-	if (transient_replslot)
-		made_transient_replslot = true;
+	if (temporary)
+		primary.made_transient_replslot = true;
 	else
-		dbinfo->made_replslot = true;
+		perdb->made_replslot = true;
 
 	if (!dry_run)
 	{
@@ -898,14 +911,16 @@ create_logical_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo,
 }
 
 static void
-drop_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo, const char *slot_name)
+drop_replication_slot(PGconn *conn, LogicalRepPerdbInfo *perdb,
+					  const char *slot_name)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
 
 	Assert(conn != NULL);
 
-	pg_log_info("dropping the replication slot \"%s\" on database \"%s\"", slot_name, dbinfo->dbname);
+	pg_log_info("dropping the replication slot \"%s\" on database \"%s\"",
+				slot_name, perdb->dbname);
 
 	appendPQExpBuffer(str, "DROP_REPLICATION_SLOT \"%s\"", slot_name);
 
@@ -915,7 +930,8 @@ drop_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo, const char *slot_nam
 	{
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
-			pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s", slot_name, dbinfo->dbname,
+			pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s",
+						 slot_name, perdb->dbname,
 						 PQerrorMessage(conn));
 
 		PQclear(res);
@@ -968,19 +984,17 @@ pg_ctl_status(const char *pg_ctl_cmd, int rc, int action)
  * the recovery process. By default, it waits forever.
  */
 static void
-wait_for_end_recovery(const char *conninfo)
+wait_for_end_recovery(StandbyInfo *standby,
+					  const char *dbname)
 {
 	PGconn	   *conn;
 	PGresult   *res;
 	int			status = POSTMASTER_STILL_STARTING;
 	int			timer = 0;
 
-	char	   *pg_ctl_cmd;
-	int			rc;
-
 	pg_log_info("waiting the postmaster to reach the consistent state");
 
-	conn = connect_database(conninfo);
+	conn = connect_database(standby->base_conninfo, dbname);
 	if (conn == NULL)
 		exit(1);
 
@@ -1021,9 +1035,13 @@ wait_for_end_recovery(const char *conninfo)
 		 */
 		if (recovery_timeout > 0 && timer >= recovery_timeout)
 		{
+			char *pg_ctl_cmd;
+			int	 rc;
+
 			pg_log_error("recovery timed out");
 
-			pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path, subscriber_dir);
+			pg_ctl_cmd = psprintf("\"%s/pg_ctl\" stop -D \"%s\" -s",
+								  standby->bindir, standby->pgdata);
 			rc = system(pg_ctl_cmd);
 			pg_ctl_status(pg_ctl_cmd, rc, 0);
 
@@ -1051,17 +1069,22 @@ wait_for_end_recovery(const char *conninfo)
  * Create a publication that includes all tables in the database.
  */
 static void
-create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
+create_publication(PGconn *conn, PrimaryInfo *primary,
+				   LogicalRepPerdbInfo *perdb)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
+	char		pubname[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
+	get_publication_name(perdb->oid, pubname, NAMEDATALEN);
+
+
 	/* Check if the publication needs to be created. */
 	appendPQExpBuffer(str,
 					  "SELECT puballtables FROM pg_catalog.pg_publication WHERE pubname = '%s'",
-					  dbinfo->pubname);
+					  pubname);
 	res = PQexec(conn, str->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
@@ -1081,7 +1104,7 @@ create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
 		 */
 		if (strcmp(PQgetvalue(res, 0, 0), "t") == 0)
 		{
-			pg_log_info("publication \"%s\" already exists", dbinfo->pubname);
+			pg_log_info("publication \"%s\" already exists", pubname);
 			return;
 		}
 		else
@@ -1094,7 +1117,7 @@ create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
 			 * database oid in which puballtables is false.
 			 */
 			pg_log_error("publication \"%s\" does not replicate changes for all tables",
-						 dbinfo->pubname);
+						 pubname);
 			pg_log_error_hint("Consider renaming this publication.");
 			PQclear(res);
 			PQfinish(conn);
@@ -1105,9 +1128,9 @@ create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
 	PQclear(res);
 	resetPQExpBuffer(str);
 
-	pg_log_info("creating publication \"%s\" on database \"%s\"", dbinfo->pubname, dbinfo->dbname);
+	pg_log_info("creating publication \"%s\" on database \"%s\"", pubname, perdb->dbname);
 
-	appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES", dbinfo->pubname);
+	appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES", pubname);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1117,14 +1140,14 @@ create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		{
 			pg_log_error("could not create publication \"%s\" on database \"%s\": %s",
-						 dbinfo->pubname, dbinfo->dbname, PQerrorMessage(conn));
+						 pubname, perdb->dbname, PQerrorMessage(conn));
 			PQfinish(conn);
 			exit(1);
 		}
 	}
 
 	/* for cleanup purposes */
-	dbinfo->made_publication = true;
+	perdb->made_publication = true;
 
 	if (!dry_run)
 		PQclear(res);
@@ -1136,24 +1159,28 @@ create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
  * Remove publication if it couldn't finish all steps.
  */
 static void
-drop_publication(PGconn *conn, LogicalRepInfo *dbinfo)
+drop_publication(PGconn *conn, LogicalRepPerdbInfo *perdb)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
+	char		pubname[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
-	pg_log_info("dropping publication \"%s\" on database \"%s\"", dbinfo->pubname, dbinfo->dbname);
+	get_publication_name(perdb->oid, pubname, NAMEDATALEN);
 
-	appendPQExpBuffer(str, "DROP PUBLICATION %s", dbinfo->pubname);
+	pg_log_info("dropping publication \"%s\" on database \"%s\"",
+				pubname, perdb->dbname);
+
+	appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname);
 
-	pg_log_debug("command is: %s", str->data);
 
 	if (!dry_run)
 	{
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
-			pg_log_error("could not drop publication \"%s\" on database \"%s\": %s", dbinfo->pubname, dbinfo->dbname, PQerrorMessage(conn));
+			pg_log_error("could not drop publication \"%s\" on database \"%s\": %s",
+						 pubname, perdb->dbname, PQerrorMessage(conn));
 
 		PQclear(res);
 	}
@@ -1174,19 +1201,30 @@ drop_publication(PGconn *conn, LogicalRepInfo *dbinfo)
  * initial location.
  */
 static void
-create_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
+create_subscription(PGconn *conn, StandbyInfo *standby,
+					PrimaryInfo *primary,
+					LogicalRepPerdbInfo *perdb)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
+	char		subname[NAMEDATALEN];
+	char		pubname[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
-	pg_log_info("creating subscription \"%s\" on database \"%s\"", dbinfo->subname, dbinfo->dbname);
+	get_subscription_name(perdb->oid, (int) getpid(), subname, NAMEDATALEN);
+	get_publication_name(perdb->oid, pubname, NAMEDATALEN);
+
+	pg_log_info("creating subscription \"%s\" on database \"%s\"", subname,
+				perdb->dbname);
 
 	appendPQExpBuffer(str,
 					  "CREATE SUBSCRIPTION %s CONNECTION '%s' PUBLICATION %s "
 					  "WITH (create_slot = false, copy_data = false, enabled = false)",
-					  dbinfo->subname, dbinfo->pubconninfo, dbinfo->pubname);
+					  subname,
+					  concat_conninfo_dbname(primary->base_conninfo,
+											 perdb->dbname),
+					  pubname);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1196,14 +1234,14 @@ create_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		{
 			pg_log_error("could not create subscription \"%s\" on database \"%s\": %s",
-						 dbinfo->subname, dbinfo->dbname, PQerrorMessage(conn));
+						 subname, perdb->dbname, PQerrorMessage(conn));
 			PQfinish(conn);
 			exit(1);
 		}
 	}
 
 	/* for cleanup purposes */
-	dbinfo->made_subscription = true;
+	perdb->made_subscription = true;
 
 	if (!dry_run)
 		PQclear(res);
@@ -1215,16 +1253,20 @@ create_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
  * Remove subscription if it couldn't finish all steps.
  */
 static void
-drop_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
+drop_subscription(PGconn *conn, LogicalRepPerdbInfo *perdb)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
+	char		subname[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
-	pg_log_info("dropping subscription \"%s\" on database \"%s\"", dbinfo->subname, dbinfo->dbname);
+	get_subscription_name(perdb->oid, (int) getpid(), subname, NAMEDATALEN);
+
+	pg_log_info("dropping subscription \"%s\" on database \"%s\"",
+				subname, perdb->dbname);
 
-	appendPQExpBuffer(str, "DROP SUBSCRIPTION %s", dbinfo->subname);
+	appendPQExpBuffer(str, "DROP SUBSCRIPTION %s", subname);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1232,7 +1274,8 @@ drop_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
 	{
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
-			pg_log_error("could not drop subscription \"%s\" on database \"%s\": %s", dbinfo->subname, dbinfo->dbname, PQerrorMessage(conn));
+			pg_log_error("could not drop subscription \"%s\" on database \"%s\": %s",
+						 subname, perdb->dbname, PQerrorMessage(conn));
 
 		PQclear(res);
 	}
@@ -1251,18 +1294,23 @@ drop_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
  * printing purposes.
  */
 static void
-set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn)
+set_replication_progress(PGconn *conn, LogicalRepPerdbInfo *perdb,
+						 const char *lsn)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
 	Oid			suboid;
 	char		originname[NAMEDATALEN];
 	char		lsnstr[17 + 1]; /* MAXPG_LSNLEN = 17 */
+	char		subname[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
+	get_subscription_name(perdb->oid, (int) getpid(), subname, NAMEDATALEN);
+
 	appendPQExpBuffer(str,
-					  "SELECT oid FROM pg_catalog.pg_subscription WHERE subname = '%s'", dbinfo->subname);
+					  "SELECT oid FROM pg_catalog.pg_subscription WHERE subname = '%s'",
+					  subname);
 
 	res = PQexec(conn, str->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -1303,7 +1351,7 @@ set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn)
 	PQclear(res);
 
 	pg_log_info("setting the replication progress (node name \"%s\" ; LSN %s) on database \"%s\"",
-				originname, lsnstr, dbinfo->dbname);
+				originname, lsnstr, perdb->dbname);
 
 	resetPQExpBuffer(str);
 	appendPQExpBuffer(str,
@@ -1317,7 +1365,7 @@ set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn)
 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 		{
 			pg_log_error("could not set replication progress for the subscription \"%s\": %s",
-						 dbinfo->subname, PQresultErrorMessage(res));
+						 subname, PQresultErrorMessage(res));
 			PQfinish(conn);
 			exit(1);
 		}
@@ -1336,16 +1384,19 @@ set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn)
  * of this setup.
  */
 static void
-enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
+enable_subscription(PGconn *conn, LogicalRepPerdbInfo *perdb)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
+	char		subname[NAMEDATALEN];
 
 	Assert(conn != NULL);
 
-	pg_log_info("enabling subscription \"%s\" on database \"%s\"", dbinfo->subname, dbinfo->dbname);
+	get_subscription_name(perdb->oid, (int) getpid(), subname, NAMEDATALEN);
+	pg_log_info("enabling subscription \"%s\" on database \"%s\"", subname,
+				perdb->dbname);
 
-	appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", dbinfo->subname);
+	appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", subname);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1354,7 +1405,7 @@ enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		{
-			pg_log_error("could not enable subscription \"%s\": %s", dbinfo->subname,
+			pg_log_error("could not enable subscription \"%s\": %s", subname,
 						 PQerrorMessage(conn));
 			PQfinish(conn);
 			exit(1);
@@ -1372,20 +1423,20 @@ enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
  * messages.
  */
 static void
-start_standby(char *server_start_log)
+start_standby(StandbyInfo *standby)
 {
 	int			rc;
 	char	   *pg_ctl_cmd;
 
-	Assert(server_start_log != NULL);
-
-	if (server_start_log[0] == '\0')
+	if (standby->server_log == NULL)
 	{
 		char		timebuf[128];
 		struct timeval time;
 		time_t		tt;
 		int			len;
 
+		standby->server_log = (char *) pg_malloc0(MAXPGPATH);
+
 		/* append timestamp with ISO 8601 format. */
 		gettimeofday(&time, NULL);
 		tt = (time_t) time.tv_sec;
@@ -1393,8 +1444,8 @@ start_standby(char *server_start_log)
 		snprintf(timebuf + strlen(timebuf), sizeof(timebuf) - strlen(timebuf),
 				 ".%03d", (int) (time.tv_usec / 1000));
 
-		len = snprintf(server_start_log, MAXPGPATH,
-					   "%s/%s/server_start_%s.log", subscriber_dir,
+		len = snprintf(standby->server_log, MAXPGPATH,
+					   "%s/%s/server_start_%s.log", standby->pgdata,
 					   PGS_OUTPUT_DIR, timebuf);
 		if (len >= MAXPGPATH)
 		{
@@ -1404,8 +1455,8 @@ start_standby(char *server_start_log)
 	}
 
 	pg_log_info("starting the standby server");
-	pg_ctl_cmd = psprintf("\"%s\" start -D \"%s\" -s -l \"%s\"", pg_ctl_path,
-						  subscriber_dir, server_start_log);
+	pg_ctl_cmd = psprintf("\"%s/pg_ctl\" start -D \"%s\" -s -l \"%s\"",
+						  standby->bindir, standby->pgdata, standby->server_log);
 	rc = system(pg_ctl_cmd);
 	pg_ctl_status(pg_ctl_cmd, rc, 1);
 }
@@ -1415,15 +1466,15 @@ start_standby(char *server_start_log)
  * messages.
  */
 static void
-stop_standby(void)
+stop_standby(StandbyInfo *standby)
 {
 	int			rc;
 	char	   *pg_ctl_cmd;
 
 	pg_log_info("stopping the standby server");
 
-	pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path,
-						  subscriber_dir);
+	pg_ctl_cmd = psprintf("\"%s/pg_ctl\" stop -D \"%s\" -s", standby->bindir,
+						  standby->pgdata);
 	rc = system(pg_ctl_cmd);
 	pg_ctl_status(pg_ctl_cmd, rc, 0);
 }
@@ -1454,12 +1505,8 @@ main(int argc, char **argv)
 
 	int			len;
 
-	char	   *pub_base_conninfo = NULL;
-	char	   *sub_base_conninfo = NULL;
 	char	   *dbname_conninfo = NULL;
 
-	uint64		pub_sysid;
-	uint64		sub_sysid;
 	struct stat statbuf;
 
 	PGconn	   *conn;
@@ -1513,7 +1560,7 @@ main(int argc, char **argv)
 		switch (c)
 		{
 			case 'D':
-				subscriber_dir = pg_strdup(optarg);
+				standby.pgdata = pg_strdup(optarg);
 				break;
 			case 'P':
 				pub_conninfo_str = pg_strdup(optarg);
@@ -1526,7 +1573,7 @@ main(int argc, char **argv)
 				if (!simple_string_list_member(&database_names, optarg))
 				{
 					simple_string_list_append(&database_names, optarg);
-					num_dbs++;
+					dbarr.ndbs++;
 				}
 				break;
 			case 'n':
@@ -1562,7 +1609,7 @@ main(int argc, char **argv)
 	/*
 	 * Required arguments
 	 */
-	if (subscriber_dir == NULL)
+	if (standby.pgdata == NULL)
 	{
 		pg_log_error("no subscriber data directory specified");
 		pg_log_error_hint("Try \"%s --help\" for more information.", progname);
@@ -1585,9 +1632,10 @@ main(int argc, char **argv)
 		pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 		exit(1);
 	}
-	pub_base_conninfo = get_base_conninfo(pub_conninfo_str, dbname_conninfo,
-										  "publisher");
-	if (pub_base_conninfo == NULL)
+	primary.base_conninfo = get_base_conninfo(pub_conninfo_str,
+											  dbname_conninfo,
+											  "publisher");
+	if (primary.base_conninfo == NULL)
 		exit(1);
 
 	if (sub_conninfo_str == NULL)
@@ -1596,8 +1644,12 @@ main(int argc, char **argv)
 		pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 		exit(1);
 	}
-	sub_base_conninfo = get_base_conninfo(sub_conninfo_str, NULL, "subscriber");
-	if (sub_base_conninfo == NULL)
+
+	standby.base_conninfo = get_base_conninfo(sub_conninfo_str, NULL,
+											  "subscriber");
+
+
+	if (standby.base_conninfo == NULL)
 		exit(1);
 
 	if (database_names.head == NULL)
@@ -1612,7 +1664,7 @@ main(int argc, char **argv)
 		if (dbname_conninfo)
 		{
 			simple_string_list_append(&database_names, dbname_conninfo);
-			num_dbs++;
+			dbarr.ndbs++;
 
 			pg_log_info("database \"%s\" was extracted from the publisher connection string",
 						dbname_conninfo);
@@ -1628,23 +1680,23 @@ main(int argc, char **argv)
 	/*
 	 * Get the absolute path of pg_ctl and pg_resetwal on the subscriber.
 	 */
-	if (!get_exec_path(argv[0]))
+	if (!get_exec_base_path(argv[0]))
 		exit(1);
 
 	/* rudimentary check for a data directory. */
-	if (!check_data_directory(subscriber_dir))
+	if (!check_data_directory(standby.pgdata))
 		exit(1);
 
-	/* Store database information for publisher and subscriber. */
-	dbinfo = store_pub_sub_info(pub_base_conninfo, sub_base_conninfo);
+	/* Store database information to dbarr */
+	store_db_names(&dbarr.perdb, dbarr.ndbs);
 
 	/*
 	 * Check if the subscriber data directory has the same system identifier
 	 * than the publisher data directory.
 	 */
-	pub_sysid = get_sysid_from_conn(dbinfo[0].pubconninfo);
-	sub_sysid = get_control_from_datadir(subscriber_dir);
-	if (pub_sysid != sub_sysid)
+	get_sysid_for_primary(&primary, dbarr.perdb[0].dbname);
+	get_sysid_for_standby(&standby);
+	if (primary.sysid != standby.sysid)
 	{
 		pg_log_error("subscriber data directory is not a copy of the source database cluster");
 		exit(1);
@@ -1654,7 +1706,7 @@ main(int argc, char **argv)
 	 * Create the output directory to store any data generated by this tool.
 	 */
 	base_dir = (char *) pg_malloc0(MAXPGPATH);
-	len = snprintf(base_dir, MAXPGPATH, "%s/%s", subscriber_dir, PGS_OUTPUT_DIR);
+	len = snprintf(base_dir, MAXPGPATH, "%s/%s", standby.pgdata, PGS_OUTPUT_DIR);
 	if (len >= MAXPGPATH)
 	{
 		pg_log_error("directory path for subscriber is too long");
@@ -1668,7 +1720,7 @@ main(int argc, char **argv)
 	}
 
 	/* subscriber PID file. */
-	snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);
+	snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", standby.pgdata);
 
 	/*
 	 * The standby server must be running. That's because some checks will be
@@ -1681,7 +1733,7 @@ main(int argc, char **argv)
 		/*
 		 * Check if the standby server is ready for logical replication.
 		 */
-		if (!setup_subscriber(dbinfo))
+		if (!setup_subscriber(&standby, &dbarr))
 			exit(1);
 
 		/*
@@ -1691,13 +1743,13 @@ main(int argc, char **argv)
 		 * if the primary slot is in use. We could use an extra connection for
 		 * it but it doesn't seem worth.
 		 */
-		if (!setup_publisher(dbinfo))
+		if (!setup_publisher(&primary, &dbarr))
 			exit(1);
 
 		pg_log_info("standby is up and running");
 		pg_log_info("stopping the server to start the transformation steps");
 
-		stop_standby();
+		stop_standby(&standby);
 	}
 	else
 	{
@@ -1723,11 +1775,11 @@ main(int argc, char **argv)
 	 * replication connection open (depending when base backup was taken, the
 	 * connection should be open for a few hours).
 	 */
-	conn = connect_database(dbinfo[0].pubconninfo);
+	conn = connect_database(primary.base_conninfo, dbarr.perdb[0].dbname);
 	if (conn == NULL)
 		exit(1);
-	consistent_lsn = create_logical_replication_slot(conn, &dbinfo[0],
-													 temp_replslot);
+	consistent_lsn = create_logical_replication_slot(conn, true, &dbarr.perdb[0]);
+
 
 	/*
 	 * Write recovery parameters.
@@ -1753,7 +1805,7 @@ main(int argc, char **argv)
 	{
 		appendPQExpBuffer(recoveryconfcontents, "recovery_target_lsn = '%s'\n",
 						  consistent_lsn);
-		WriteRecoveryConfig(conn, subscriber_dir, recoveryconfcontents);
+		WriteRecoveryConfig(conn, standby.pgdata, recoveryconfcontents);
 	}
 	disconnect_database(conn);
 
@@ -1762,32 +1814,32 @@ main(int argc, char **argv)
 	/*
 	 * Start subscriber and wait until accepting connections.
 	 */
-	start_standby(server_start_log);
+	start_standby(&standby);
 
 	/*
 	 * Waiting the subscriber to be promoted.
 	 */
-	wait_for_end_recovery(dbinfo[0].subconninfo);
+	wait_for_end_recovery(&standby, dbarr.perdb[0].dbname);
 
 	/*
 	 * Create a subscription for each database.
 	 */
-	for (i = 0; i < num_dbs; i++)
+	for (i = 0; i < dbarr.ndbs; i++)
 	{
+		LogicalRepPerdbInfo *perdb = &dbarr.perdb[i];
+
 		/* Connect to subscriber. */
-		conn = connect_database(dbinfo[i].subconninfo);
+		conn = connect_database(standby.base_conninfo, perdb->dbname);
 		if (conn == NULL)
 			exit(1);
 
-		create_subscription(conn, &dbinfo[i]);
+		create_subscription(conn, &standby, &primary, perdb);
 
 		/* Set the replication progress to the correct LSN. */
-		set_replication_progress(conn, &dbinfo[i], consistent_lsn);
+		set_replication_progress(conn, perdb, consistent_lsn);
 
 		/* Enable subscription. */
-		enable_subscription(conn, &dbinfo[i]);
-
-		disconnect_database(conn);
+		enable_subscription(conn, perdb);
 	}
 
 	/*
@@ -1798,31 +1850,45 @@ main(int argc, char **argv)
 	 * XXX we might not fail here. Instead, we provide a warning so the user
 	 * eventually drops the replication slot later.
 	 */
-	conn = connect_database(dbinfo[0].pubconninfo);
+	conn = connect_database(primary.base_conninfo, dbarr.perdb[0].dbname);
 	if (conn == NULL)
 	{
-		if (primary_slot_name != NULL)
-			pg_log_warning("could not drop replication slot \"%s\" on primary", primary_slot_name);
-		pg_log_warning("could not drop transient replication slot \"%s\" on publisher", temp_replslot);
+		char transient_replslot[NAMEDATALEN];
+
+		snprintf(transient_replslot, NAMEDATALEN, "pg_subscriber_%d_startpoint",
+				 (int) getpid());
+
+		if (standby.primary_slot_name != NULL)
+			pg_log_warning("could not drop replication slot \"%s\" on primary",
+						   standby.primary_slot_name);
+		pg_log_warning("could not drop transient replication slot \"%s\" on publisher",
+					   transient_replslot);
 		pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
 	}
 	else
 	{
-		drop_replication_slot(conn, &dbinfo[0], temp_replslot);
+		LogicalRepPerdbInfo *perdb = &dbarr.perdb[0];
+		char *primary_slot_name = standby.primary_slot_name;
+		char transient_replslot[NAMEDATALEN];
+
 		if (primary_slot_name != NULL)
-			drop_replication_slot(conn, &dbinfo[0], primary_slot_name);
+			drop_replication_slot(conn, perdb, primary_slot_name);
+
+		snprintf(transient_replslot, NAMEDATALEN, "pg_subscriber_%d_startpoint",
+				 (int) getpid());
+		drop_replication_slot(conn, perdb, transient_replslot);
 		disconnect_database(conn);
 	}
 
 	/*
 	 * Stop the subscriber.
 	 */
-	stop_standby();
+	stop_standby(&standby);
 
 	/*
 	 * Change system identifier.
 	 */
-	modify_sysid(pg_resetwal_path, subscriber_dir);
+	modify_sysid(standby.bindir, standby.pgdata);
 
 	/*
 	 * The log file is kept if retain option is specified or this tool does
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 7e866e3c3d..e9e3db9ffc 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1506,6 +1506,8 @@ LogicalRepCommitPreparedTxnData
 LogicalRepCtxStruct
 LogicalRepMsgType
 LogicalRepPartMapEntry
+LogicalRepPerdbInfo
+LogicalRepPerdbInfoArr
 LogicalRepPreparedTxnData
 LogicalRepRelId
 LogicalRepRelMapEntry
@@ -1884,6 +1886,7 @@ PREDICATELOCK
 PREDICATELOCKTAG
 PREDICATELOCKTARGET
 PREDICATELOCKTARGETTAG
+PrimaryInfo
 PROCESS_INFORMATION
 PROCLOCK
 PROCLOCKTAG
@@ -2460,6 +2463,7 @@ SQLValueFunctionOp
 SSL
 SSLExtensionInfoContext
 SSL_CTX
+StandbyInfo
 STARTUPINFO
 STRLEN
 SV
-- 
2.43.0