From dd7be7f34e1ac68c2b6fd193dec2aa6b2832e1b9 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Mon, 29 Jan 2024 07:03:59 +0000 Subject: [PATCH v12 5/5] Divide LogicalReplInfo into some strcutures --- src/bin/pg_basebackup/pg_createsubscriber.c | 660 ++++++++++-------- .../t/041_pg_createsubscriber_standby.pl | 2 +- src/tools/pgindent/typedefs.list | 5 +- 3 files changed, 365 insertions(+), 302 deletions(-) diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c index 6f832f7551..898fa3f114 100644 --- a/src/bin/pg_basebackup/pg_createsubscriber.c +++ b/src/bin/pg_basebackup/pg_createsubscriber.c @@ -32,54 +32,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 */ @@ -87,21 +111,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 { @@ -111,6 +131,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. @@ -128,33 +172,54 @@ 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); 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 (primary.made_transient_replslot) + { + char transient_replslot[NAMEDATALEN]; + + conn = connect_database(primary.base_conninfo, dbarr.perdb[0].dbname); + + if (conn != NULL) + { + snprintf(transient_replslot, NAMEDATALEN, "pg_subscriber_%d_startpoint", + (int) getpid()); + + drop_replication_slot(conn, &dbarr.perdb[0], transient_replslot); + disconnect_database(conn); + } + } } static void @@ -236,15 +301,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); @@ -269,30 +335,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]; - - 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; - } + /* Extract the directory part from the path */ + p = strrchr(pg_ctl_path, 'p'); + Assert(p); - pg_log_debug("pg_resetwal path is: %s", pg_resetwal_path); + *p = '\0'; + standby.bindir = pg_strdup(pg_ctl_path); return true; } @@ -356,45 +404,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) @@ -412,6 +446,7 @@ connect_database(const char *conninfo) } PQclear(res); + pfree(conninfo); return conn; } @@ -429,11 +464,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"); @@ -443,9 +477,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); @@ -468,7 +500,6 @@ get_primary_conninfo(const char *base_conninfo) exit(1); } - pg_free(conninfo); PQclear(res); disconnect_database(conn); @@ -476,19 +507,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); @@ -511,13 +541,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; } /* @@ -525,29 +554,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; } /* @@ -556,7 +582,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; @@ -590,7 +616,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); @@ -609,17 +635,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); @@ -639,43 +664,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) - pg_log_info("create replication slot \"%s\" on publisher", replslotname); - else + if (create_logical_replication_slot(conn, false, perdb) == NULL) return false; disconnect_database(conn); @@ -688,7 +690,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; @@ -711,7 +713,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); @@ -749,10 +751,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); @@ -767,13 +770,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); @@ -787,17 +791,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; } @@ -808,7 +816,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; @@ -828,7 +836,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); @@ -845,35 +853,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; } @@ -885,14 +899,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); @@ -901,15 +918,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); } @@ -925,45 +942,55 @@ 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. + * 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') - { - snprintf(slot_name, NAMEDATALEN, "pg_createsubscriber_%d_startpoint", + 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); 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; + lsn = pg_strdup(PQgetvalue(res, 0, 1)); PQclear(res); @@ -974,14 +1001,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); @@ -990,8 +1019,9 @@ 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, - PQerrorMessage(conn)); + pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s", + slot_name, perdb->dbname, + PQerrorMessage(conn)); PQclear(res); @@ -1026,23 +1056,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); } @@ -1091,7 +1123,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; @@ -1100,7 +1132,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); @@ -1139,7 +1171,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); } @@ -1164,17 +1196,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) { @@ -1194,7 +1230,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 @@ -1207,7 +1243,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); @@ -1218,9 +1254,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); @@ -1228,13 +1264,13 @@ 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; PQclear(res); destroyPQExpBuffer(str); @@ -1244,25 +1280,29 @@ 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); 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); - destroyPQExpBuffer(str); } @@ -1279,19 +1319,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); @@ -1299,13 +1350,13 @@ 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; PQclear(res); destroyPQExpBuffer(str); @@ -1315,22 +1366,27 @@ 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); 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); @@ -1348,18 +1404,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) @@ -1392,7 +1453,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, @@ -1404,7 +1465,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); } @@ -1421,24 +1482,27 @@ 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); res = PQexec(conn, str->data); if (PQresultStatus(res) != PGRES_COMMAND_OK) { - pg_log_error("could not enable subscription \"%s\": %s", dbinfo->subname, - PQerrorMessage(conn)); + pg_log_error("could not enable subscription \"%s\": %s", subname, + PQerrorMessage(conn)); PQfinish(conn); exit(1); } @@ -1468,16 +1532,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; @@ -1527,7 +1585,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); @@ -1537,7 +1595,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': @@ -1573,7 +1631,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); @@ -1586,8 +1644,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) @@ -1602,7 +1660,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); @@ -1616,20 +1674,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); @@ -1638,9 +1696,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); @@ -1650,7 +1708,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"); @@ -1663,10 +1721,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 @@ -1679,7 +1737,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); /* @@ -1688,7 +1746,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); /* @@ -1697,13 +1755,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 { @@ -1732,11 +1790,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. @@ -1752,7 +1809,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,12 +1819,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 @@ -1776,7 +1833,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); /* @@ -1785,12 +1842,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], temp_replslot); + drop_replication_slot(conn, perdb, primary_slot_name); } else { @@ -1804,12 +1864,12 @@ 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); cleanup: /* @@ -1817,7 +1877,7 @@ cleanup: * 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..856bf0de3c 100644 --- a/src/bin/pg_basebackup/t/041_pg_createsubscriber_standby.pl +++ b/src/bin/pg_basebackup/t/041_pg_createsubscriber_standby.pl @@ -88,7 +88,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