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