v13-0005-Divide-LogicalReplInfo-into-some-strcutures.patch

From 6ad9ce2ebc5c0fecf9412e66bf7ec2fc62d76213 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 v13 5/5] Divide LogicalReplInfo into some strcutures

---
 src/bin/pg_basebackup/pg_createsubscriber.c   | 648 ++++++++++--------
 .../t/041_pg_createsubscriber_standby.pl      |   3 +-
 src/tools/pgindent/typedefs.list              |   5 +-
 3 files changed, 351 insertions(+), 305 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 02291ba505..bd55639251 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -33,54 +33,78 @@
 
 #define	PGS_OUTPUT_DIR	"pg_createsubscriber_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);
-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 *get_primary_conninfo(const char *base_conninfo);
+static char *get_primary_conninfo(StandbyInfo *standby);
 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 check_publisher(LogicalRepInfo *dbinfo);
-static bool setup_publisher(LogicalRepInfo *dbinfo);
-static bool check_subscriber(LogicalRepInfo *dbinfo);
-static bool setup_subscriber(LogicalRepInfo *dbinfo, const char *consistent_lsn);
-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 check_publisher(PrimaryInfo *primary, LogicalRepPerdbInfoArr *dbarr);
+static bool setup_publisher(PrimaryInfo *primary, LogicalRepPerdbInfoArr *dbarr);
+static bool check_subscriber(StandbyInfo *standby, LogicalRepPerdbInfoArr *dbarr);
+static bool setup_subscriber(StandbyInfo *standby, PrimaryInfo *primary,
+							 LogicalRepPerdbInfoArr *dbarr,
+							 const char *consistent_lsn);
+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 char *server_logfile_name(const char *datadir);
-static void start_standby_server(const char *pg_ctl_path, const char *datadir, const char *logfile);
-static void stop_standby_server(const char *pg_ctl_path, const char *datadir);
+static void start_standby_server(StandbyInfo *standby);
+static void stop_standby_server(StandbyInfo *standby);
 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 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);
 
 #define	USEC_PER_SEC	1000000
 #define	WAIT_INTERVAL	1		/* 1 second */
@@ -88,21 +112,17 @@ static void enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
 /* Options */
 static const char *progname;
 
-static char *subscriber_dir = 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 LogicalRepPerdbInfoArr dbarr;
+static PrimaryInfo primary;
+static StandbyInfo standby;
 
 enum WaitPMResult
 {
@@ -112,6 +132,30 @@ enum WaitPMResult
 	POSTMASTER_FAILED
 };
 
+/*
+ * Build the replication slot name. The name must not exceed
+ * NAMEDATALEN - 1. This current schema uses a maximum of 42
+ * characters (20 + 10 + 1 + 10 + '\0'). PID 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_createsubscriber_%u_%d", oid, pid);
+}
+
+/*
+ * Build the publication name. The name must not exceed NAMEDATALEN -
+ * 1. This current schema uses a maximum of 31 characters (20 + 10 +
+ * '\0').
+ */
+static inline void
+get_publication_name(Oid oid, char *pubname, Size szpub)
+{
+	snprintf(pubname, szpub, "pg_createsubscriber_%u", oid);
+}
+
 
 /*
  * Cleanup objects that were created by pg_createsubscriber if there is an error.
@@ -129,30 +173,35 @@ 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]);
-				if (dbinfo[i].made_publication)
-					drop_publication(conn, &dbinfo[i]);
+				drop_subscription(conn, perdb);
+
+				if (perdb->made_publication)
+					drop_publication(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);
-			if (conn != NULL)
+			if (perdb->made_publication)
+				drop_publication(conn, perdb);
+			if (perdb->made_replslot)
 			{
-				if (dbinfo[i].made_publication)
-					drop_publication(conn, &dbinfo[i]);
-				if (dbinfo[i].made_replslot)
-					drop_replication_slot(conn, &dbinfo[i], dbinfo[i].subname);
-				disconnect_database(conn);
+				char replslotname[NAMEDATALEN];
+
+				get_subscription_name(perdb->oid, (int) getpid(),
+									  replslotname, NAMEDATALEN);
+				drop_replication_slot(conn, perdb, replslotname);
 			}
 		}
 	}
@@ -237,15 +286,16 @@ get_base_conninfo(char *conninfo, char *dbname)
 }
 
 /*
- * 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);
@@ -270,30 +320,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;
 }
@@ -357,45 +389,31 @@ 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;
+	char	   *conninfo = concat_conninfo_dbname(base_conninfo,
+														 dbname);
 
 	conn = PQconnectdb(conninfo);
 	if (PQstatus(conn) != CONNECTION_OK)
@@ -413,6 +431,7 @@ connect_database(const char *conninfo)
 	}
 	PQclear(res);
 
+	pfree(conninfo);
 	return conn;
 }
 
@@ -430,11 +449,10 @@ disconnect_database(PGconn *conn)
  * worker.
  */
 static char *
-get_primary_conninfo(const char *base_conninfo)
+get_primary_conninfo(StandbyInfo *standby)
 {
 	PGconn	   *conn;
 	PGresult   *res;
-	char	   *conninfo;
 	char	   *primaryconninfo;
 
 	pg_log_info("getting primary_conninfo from standby");
@@ -444,9 +462,7 @@ get_primary_conninfo(const char *base_conninfo)
 	 * not stored infomation yet, we must directly get the first element of the
 	 * database list.
 	 */
-	conninfo = concat_conninfo_dbname(base_conninfo, database_names.head->val);
-
-	conn = connect_database(conninfo);
+	conn = connect_database(standby->base_conninfo, database_names.head->val);
 	if (conn == NULL)
 		exit(1);
 
@@ -469,7 +485,6 @@ get_primary_conninfo(const char *base_conninfo)
 		exit(1);
 	}
 
-	pg_free(conninfo);
 	PQclear(res);
 	disconnect_database(conn);
 
@@ -477,19 +492,18 @@ get_primary_conninfo(const char *base_conninfo)
 }
 
 /*
- * 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);
 
@@ -512,13 +526,12 @@ get_sysid_from_conn(const char *conninfo)
 		exit(1);
 	}
 
-	sysid = strtou64(PQgetvalue(res, 0, 2), NULL, 10);
+	primary->sysid = strtou64(PQgetvalue(res, 0, 2), NULL, 10);
 
-	pg_log_info("system identifier is %llu on publisher", (unsigned long long) sysid);
+	pg_log_info("system identifier is %llu on publisher",
+				(unsigned long long) primary->sysid);
 
 	disconnect_database(conn);
-
-	return sysid;
 }
 
 /*
@@ -526,29 +539,26 @@ get_sysid_from_conn(const char *conninfo)
  * if a data directory is a clone of another one. This routine is used locally
  * and avoids a connection establishment.
  */
-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 %llu on subscriber", (unsigned long long) standby->sysid);
 
 	pfree(cf);
-
-	return sysid;
 }
 
 /*
@@ -557,7 +567,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;
@@ -592,7 +602,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);
 
@@ -613,17 +623,16 @@ modify_sysid(const char *pg_resetwal_path, const char *datadir)
  * replication.
  */
 static bool
-setup_publisher(LogicalRepInfo *dbinfo)
+setup_publisher(PrimaryInfo *primary, LogicalRepPerdbInfoArr *dbarr)
 {
 	PGconn	   *conn;
 	PGresult   *res;
 
-	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);
 
@@ -643,43 +652,20 @@ 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 31 characters (20 + 10 +
-		 * '\0').
-		 */
-		snprintf(pubname, sizeof(pubname), "pg_createsubscriber_%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 42
-		 * characters (20 + 10 + 1 + 10 + '\0'). PID is included to reduce the
-		 * probability of collision. By default, subscription name is used as
-		 * replication slot name.
-		 */
-		snprintf(replslotname, sizeof(replslotname),
-				 "pg_createsubscriber_%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);
@@ -692,7 +678,7 @@ setup_publisher(LogicalRepInfo *dbinfo)
  * Is the primary server ready for logical replication?
  */
 static bool
-check_publisher(LogicalRepInfo *dbinfo)
+check_publisher(PrimaryInfo *primary, LogicalRepPerdbInfoArr *dbarr)
 {
 	PGconn	   *conn;
 	PGresult   *res;
@@ -715,7 +701,7 @@ check_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);
 
@@ -753,10 +739,11 @@ check_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);
 
@@ -771,13 +758,14 @@ check_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);
@@ -791,17 +779,21 @@ check_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;
 	}
 
@@ -812,7 +804,7 @@ check_publisher(LogicalRepInfo *dbinfo)
  * Is the standby server ready for logical replication?
  */
 static bool
-check_subscriber(LogicalRepInfo *dbinfo)
+check_subscriber(StandbyInfo *standby, LogicalRepPerdbInfoArr *dbarr)
 {
 	PGconn	   *conn;
 	PGresult   *res;
@@ -832,7 +824,7 @@ check_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);
 
@@ -849,35 +841,41 @@ check_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;
 	}
 
@@ -889,14 +887,17 @@ check_subscriber(LogicalRepInfo *dbinfo)
  * enable the subscriptions. That's the last step for logical repliation setup.
  */
 static bool
-setup_subscriber(LogicalRepInfo *dbinfo, const char *consistent_lsn)
+setup_subscriber(StandbyInfo *standby, PrimaryInfo *primary,
+				 LogicalRepPerdbInfoArr *dbarr, const char *consistent_lsn)
 {
 	PGconn	   *conn;
 
-	for (int i = 0; i < num_dbs; i++)
+	for (int 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);
 
@@ -905,15 +906,15 @@ setup_subscriber(LogicalRepInfo *dbinfo, const char *consistent_lsn)
 		 * available on the subscriber when the physical replica is promoted.
 		 * Remove publications from the subscriber because it has no use.
 		 */
-		drop_publication(conn, &dbinfo[i]);
+		drop_publication(conn, perdb);
 
-		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]);
+		enable_subscription(conn, perdb);
 
 		disconnect_database(conn);
 	}
@@ -929,31 +930,35 @@ setup_subscriber(LogicalRepInfo *dbinfo, const char *consistent_lsn)
  * 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.
-	 */
-	if (slot_name[0] == '\0')
-	{
-		snprintf(slot_name, NAMEDATALEN, "pg_createsubscriber_%d_startpoint",
+	 * 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 (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, "SELECT * FROM pg_create_logical_replication_slot('%s', 'pgoutput', %s, false, false);",
-					  slot_name, transient_replslot ? "true" : "false");
+					  slot_name, temporary ? "true" : "false");
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -962,15 +967,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)
-		dbinfo->made_replslot = true;
+	if (temporary)
+		primary.made_transient_replslot = true;
+	else
+		perdb->made_replslot = true;
 
 	if (!dry_run)
 	{
@@ -984,14 +993,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, "SELECT * FROM pg_drop_replication_slot('%s');",
 					  slot_name);
@@ -1002,8 +1013,8 @@ drop_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo, const char *slot_nam
 	{
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
-			pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s", slot_name, dbinfo->dbname,
-						 PQerrorMessage(conn));
+			pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s",
+						 slot_name, perdb->dbname, PQerrorMessage(conn));
 
 		PQclear(res);
 	}
@@ -1039,23 +1050,25 @@ server_logfile_name(const char *datadir)
 }
 
 static void
-start_standby_server(const char *pg_ctl_path, const char *datadir, const char *logfile)
+start_standby_server(StandbyInfo *standby)
 {
 	char	   *pg_ctl_cmd;
 	int			rc;
 
-	pg_ctl_cmd = psprintf("\"%s\" start -D \"%s\" -s -l \"%s\"", pg_ctl_path, datadir, logfile);
+	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);
 }
 
 static void
-stop_standby_server(const char *pg_ctl_path, const char *datadir)
+stop_standby_server(StandbyInfo *standby)
 {
 	char	   *pg_ctl_cmd;
 	int			rc;
 
-	pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path, datadir);
+	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);
 }
@@ -1104,7 +1117,7 @@ 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;
@@ -1113,7 +1126,7 @@ wait_for_end_recovery(const char *conninfo)
 
 	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);
 
@@ -1155,7 +1168,7 @@ wait_for_end_recovery(const char *conninfo)
 		if (recovery_timeout > 0 && timer >= recovery_timeout)
 		{
 			pg_log_error("recovery timed out");
-			stop_standby_server(pg_ctl_path, subscriber_dir);
+			stop_standby_server(standby);
 			exit(1);
 		}
 
@@ -1180,17 +1193,21 @@ 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)
 	{
@@ -1210,7 +1227,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
@@ -1223,7 +1240,7 @@ create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
 			 * exact 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);
@@ -1234,9 +1251,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);
 
@@ -1246,14 +1263,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);
@@ -1265,16 +1282,20 @@ 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);
+
+	pg_log_info("dropping publication \"%s\" on database \"%s\"",
+				pubname, perdb->dbname);
 
-	appendPQExpBuffer(str, "DROP PUBLICATION %s", dbinfo->pubname);
+	appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1282,7 +1303,8 @@ drop_publication(PGconn *conn, LogicalRepInfo *dbinfo)
 	{
 		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);
 	}
@@ -1303,22 +1325,32 @@ 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	   *conninfo;
+	char		subname[NAMEDATALEN];
+	char		pubname[NAMEDATALEN];
+	char	   *escaped_conninfo;
 
 	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);
 
-	conninfo = escape_single_quotes_ascii(dbinfo->pubconninfo);
+	pg_log_info("creating subscription \"%s\" on database \"%s\"", subname,
+				perdb->dbname);
+
+	escaped_conninfo = escape_single_quotes_ascii(primary->base_conninfo);
 
 	appendPQExpBuffer(str,
 					  "CREATE SUBSCRIPTION %s CONNECTION '%s' PUBLICATION %s "
 					  "WITH (create_slot = false, copy_data = false, enabled = false)",
-					  dbinfo->subname, conninfo, dbinfo->pubname);
+					  subname,
+					  concat_conninfo_dbname(escaped_conninfo, perdb->dbname),
+					  pubname);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1328,19 +1360,19 @@ 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);
 
-	pg_free(conninfo);
+	pg_free(escaped_conninfo);
 	destroyPQExpBuffer(str);
 }
 
@@ -1348,16 +1380,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);
 
@@ -1365,7 +1401,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);
 	}
@@ -1384,18 +1421,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)
@@ -1436,7 +1478,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,
@@ -1450,7 +1492,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);
 		}
@@ -1469,16 +1511,20 @@ 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);
 
@@ -1487,7 +1533,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);
@@ -1520,16 +1566,10 @@ main(int argc, char **argv)
 	int			option_index;
 
 	char	   *base_dir;
-	char	   *server_start_log;
 	int			len;
 
-	char	   *pub_base_conninfo = NULL;
-	char	   *sub_base_conninfo = NULL;
 	char	   *dbname_conninfo = NULL;
-	char		temp_replslot[NAMEDATALEN] = {0};
 
-	uint64		pub_sysid;
-	uint64		sub_sysid;
 	struct stat statbuf;
 
 	PGconn	   *conn;
@@ -1579,7 +1619,7 @@ main(int argc, char **argv)
 		switch (c)
 		{
 			case 'D':
-				subscriber_dir = pg_strdup(optarg);
+				standby.pgdata = pg_strdup(optarg);
 				break;
 			case 'S':
 				sub_conninfo_str = pg_strdup(optarg);
@@ -1589,7 +1629,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':
@@ -1625,7 +1665,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);
@@ -1638,8 +1678,8 @@ 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, dbname_conninfo);
-	if (sub_base_conninfo == NULL)
+	standby.base_conninfo = get_base_conninfo(sub_conninfo_str, dbname_conninfo);
+	if (standby.base_conninfo == NULL)
 		exit(1);
 
 	if (database_names.head == NULL)
@@ -1654,7 +1694,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 subscriber connection string",
 						dbname_conninfo);
@@ -1668,20 +1708,20 @@ main(int argc, char **argv)
 	}
 
 	/* Obtain a connection string from the target */
-	pub_base_conninfo = get_primary_conninfo(sub_base_conninfo);
+	primary.base_conninfo = get_primary_conninfo(&standby);
 
 	/*
 	 * 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);
 
 	/* Register a function to clean up objects in case of failure. */
 	atexit(cleanup_objects_atexit);
@@ -1690,9 +1730,9 @@ main(int argc, char **argv)
 	 * 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);
@@ -1702,7 +1742,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");
@@ -1715,10 +1755,10 @@ main(int argc, char **argv)
 		exit(1);
 	}
 
-	server_start_log = server_logfile_name(subscriber_dir);
+	standby.server_log = server_logfile_name(standby.pgdata);
 
 	/* 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
@@ -1731,7 +1771,7 @@ main(int argc, char **argv)
 		/*
 		 * Check if the standby server is ready for logical replication.
 		 */
-		if (!check_subscriber(dbinfo))
+		if (!check_subscriber(&standby, &dbarr))
 			exit(1);
 
 		/*
@@ -1740,7 +1780,7 @@ main(int argc, char **argv)
 		 * relies on check_subscriber() to obtain the primary_slot_name.
 		 * That's why it is called after it.
 		 */
-		if (!check_publisher(dbinfo))
+		if (!check_publisher(&primary, &dbarr))
 			exit(1);
 
 		/*
@@ -1749,13 +1789,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);
 
 		/* Stop the standby server. */
 		pg_log_info("standby is up and running");
 		pg_log_info("stopping the server to start the transformation steps");
-		stop_standby_server(pg_ctl_path, subscriber_dir);
+		stop_standby_server(&standby);
 	}
 	else
 	{
@@ -1776,11 +1816,10 @@ main(int argc, char **argv)
 	 * consistent LSN but it should be changed after adding pg_basebackup
 	 * support.
 	 */
-	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.
@@ -1806,7 +1845,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);
 
@@ -1816,12 +1855,12 @@ main(int argc, char **argv)
 	 * Start subscriber and wait until accepting connections.
 	 */
 	pg_log_info("starting the subscriber");
-	start_standby_server(pg_ctl_path, subscriber_dir, server_start_log);
+	start_standby_server(&standby);
 
 	/*
 	 * Waiting the subscriber to be promoted.
 	 */
-	wait_for_end_recovery(dbinfo[0].subconninfo);
+	wait_for_end_recovery(&standby, dbarr.perdb[0].dbname);
 
 	/*
 	 * Create the subscription for each database on subscriber. It does not
@@ -1830,7 +1869,7 @@ main(int argc, char **argv)
 	 * set_replication_progress). It also cleans up publications created by
 	 * this tool and replication to the standby.
 	 */
-	if (!setup_subscriber(dbinfo, consistent_lsn))
+	if (!setup_subscriber(&standby, &primary, &dbarr, consistent_lsn))
 		exit(1);
 
 	/*
@@ -1839,12 +1878,15 @@ main(int argc, char **argv)
 	 * XXX we might not fail here. Instead, we provide a warning so the user
 	 * eventually drops this replication slot later.
 	 */
-	if (primary_slot_name != NULL)
+	if (standby.primary_slot_name != NULL)
 	{
-		conn = connect_database(dbinfo[0].pubconninfo);
+		char *primary_slot_name = standby.primary_slot_name;
+		LogicalRepPerdbInfo *perdb = &dbarr.perdb[0];
+
+		conn = connect_database(primary.base_conninfo, perdb->dbname);
 		if (conn != NULL)
 		{
-			drop_replication_slot(conn, &dbinfo[0], primary_slot_name);
+			drop_replication_slot(conn, perdb, primary_slot_name);
 		}
 		else
 		{
@@ -1858,19 +1900,19 @@ main(int argc, char **argv)
 	 * Stop the subscriber.
 	 */
 	pg_log_info("stopping the subscriber");
-	stop_standby_server(pg_ctl_path, subscriber_dir);
+	stop_standby_server(&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
 	 * not run successfully. Otherwise, log file is removed.
 	 */
 	if (!retain)
-		unlink(server_start_log);
+		unlink(standby.server_log);
 
 	success = true;
 
diff --git a/src/bin/pg_basebackup/t/041_pg_createsubscriber_standby.pl b/src/bin/pg_basebackup/t/041_pg_createsubscriber_standby.pl
index a9d03acc87..1544953843 100644
--- a/src/bin/pg_basebackup/t/041_pg_createsubscriber_standby.pl
+++ b/src/bin/pg_basebackup/t/041_pg_createsubscriber_standby.pl
@@ -17,6 +17,7 @@ my $result;
 # Set up node P as primary
 $node_p = PostgreSQL::Test::Cluster->new('node_p');
 $node_p->init(allows_streaming => 'logical');
+$node_p->append_conf('postgresql.conf', 'log_line_prefix = \'%m [%p] [%d] \'');
 $node_p->start;
 
 # Set up node F as about-to-fail node
@@ -88,7 +89,7 @@ command_ok(
 		'--pgdata', $node_s->data_dir,
 		'--subscriber-server', $node_s->connstr('pg1'),
 		'--database', 'pg1',
-		'--database', 'pg2'
+		'--database', 'pg2', '-r'
 	],
 	'run pg_createsubscriber on node S');
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index f51f1ff23f..3b1ec3fce1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1505,9 +1505,10 @@ LogicalRepBeginData
 LogicalRepCommitData
 LogicalRepCommitPreparedTxnData
 LogicalRepCtxStruct
-LogicalRepInfo
 LogicalRepMsgType
 LogicalRepPartMapEntry
+LogicalRepPerdbInfo
+LogicalRepPerdbInfoArr
 LogicalRepPreparedTxnData
 LogicalRepRelId
 LogicalRepRelMapEntry
@@ -1886,6 +1887,7 @@ PREDICATELOCK
 PREDICATELOCKTAG
 PREDICATELOCKTARGET
 PREDICATELOCKTARGETTAG
+PrimaryInfo
 PROCESS_INFORMATION
 PROCLOCK
 PROCLOCKTAG
@@ -2461,6 +2463,7 @@ SQLValueFunctionOp
 SSL
 SSLExtensionInfoContext
 SSL_CTX
+StandbyInfo
 STARTUPINFO
 STRLEN
 SV
-- 
2.43.0