From 75c5e59ef08e1408c2130bda77c39773f659f760 Mon Sep 17 00:00:00 2001 From: Michael Banck Date: Sun, 19 Mar 2017 19:44:25 +0100 Subject: [PATCH 2/2] Refactor replication slot creation in pg_basebackup et al. Add new argument `is_temporary' to CreateReplicationSlot() in streamutil.c which specifies whether a physical replication slot is temporary or not. Update all callers. Move the creation of the temporary replication slot for pg_basebackup from receivelog.c to pg_basebackup.c. At the same time, call CreateReplicationSlot() instead of creating the temporary slot with an explicit SQL command. --- src/bin/pg_basebackup/pg_basebackup.c | 22 ++++++++++++++++++---- src/bin/pg_basebackup/pg_receivewal.c | 2 +- src/bin/pg_basebackup/pg_recvlogical.c | 2 +- src/bin/pg_basebackup/receivelog.c | 18 ------------------ src/bin/pg_basebackup/streamutil.c | 18 +++++++++++++----- src/bin/pg_basebackup/streamutil.h | 2 +- 6 files changed, 34 insertions(+), 30 deletions(-) diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index 69ca4be..622099c 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -487,8 +487,6 @@ LogStreamerMain(logstreamer_param *param) stream.partial_suffix = NULL; stream.replication_slot = replication_slot; stream.temp_slot = param->temp_slot; - if (stream.temp_slot && !stream.replication_slot) - stream.replication_slot = psprintf("pg_basebackup_%d", (int) getpid()); if (format == 'p') stream.walmethod = CreateWalDirectoryMethod(param->xlog, 0, do_sync); @@ -582,6 +580,22 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier) param->temp_slot = temp_replication_slot; /* + * Create temporary replication slot if one is needed + */ + if (temp_replication_slot) + { + if (!replication_slot) + replication_slot = psprintf("pg_basebackup_%d", (int) getpid()); + + if (verbose) + fprintf(stderr, _("%s: creating temporary replication slot \"%s\"\n"), + progname, replication_slot); + + if (!CreateReplicationSlot(param->bgconn, replication_slot, NULL, true, true, false)) + disconnect_and_exit(1); + } + + /* * Try to create a permanent replication slot if one is specified. Do * not error out if the slot already exists, other errors are already * reported by CreateReplicationSlot(). @@ -592,8 +606,8 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier) fprintf(stderr, _("%s: creating replication slot \"%s\"\n"), progname, replication_slot); - if (!CreateReplicationSlot(param->bgconn, replication_slot, NULL, true, true)) - return 1; + if (!CreateReplicationSlot(param->bgconn, replication_slot, NULL, true, false, true)) + disconnect_and_exit(1); } if (format == 'p') diff --git a/src/bin/pg_basebackup/pg_receivewal.c b/src/bin/pg_basebackup/pg_receivewal.c index 15348ad..d221633 100644 --- a/src/bin/pg_basebackup/pg_receivewal.c +++ b/src/bin/pg_basebackup/pg_receivewal.c @@ -699,7 +699,7 @@ main(int argc, char **argv) progname, replication_slot); if (!CreateReplicationSlot(conn, replication_slot, NULL, true, - slot_exists_ok)) + false, slot_exists_ok)) disconnect_and_exit(1); disconnect_and_exit(0); } diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index 6b081bd..14ce537 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -980,7 +980,7 @@ main(int argc, char **argv) progname, replication_slot); if (!CreateReplicationSlot(conn, replication_slot, plugin, - false, slot_exists_ok)) + false, false, slot_exists_ok)) disconnect_and_exit(1); startpos = InvalidXLogRecPtr; } diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c index f415135..d884793 100644 --- a/src/bin/pg_basebackup/receivelog.c +++ b/src/bin/pg_basebackup/receivelog.c @@ -509,24 +509,6 @@ ReceiveXlogStream(PGconn *conn, StreamCtl *stream) } /* - * Create temporary replication slot if one is needed - */ - if (stream->temp_slot) - { - snprintf(query, sizeof(query), - "CREATE_REPLICATION_SLOT \"%s\" TEMPORARY PHYSICAL RESERVE_WAL", - stream->replication_slot); - res = PQexec(conn, query); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - fprintf(stderr, _("%s: could not create temporary replication slot \"%s\": %s"), - progname, stream->replication_slot, PQerrorMessage(conn)); - PQclear(res); - return false; - } - } - - /* * initialize flush position to starting point, it's the caller's * responsibility that that's sane. */ diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c index 507da5e..67d801a 100644 --- a/src/bin/pg_basebackup/streamutil.c +++ b/src/bin/pg_basebackup/streamutil.c @@ -322,7 +322,7 @@ RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, */ bool CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin, - bool is_physical, bool slot_exists_ok) + bool is_physical, bool is_temporary, bool slot_exists_ok) { PQExpBuffer query; PGresult *res; @@ -335,12 +335,20 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin, /* Build query */ if (is_physical) - appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" PHYSICAL", - slot_name); + if (is_temporary) + appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" TEMPORARY PHYSICAL", + slot_name); + else + appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" PHYSICAL", + slot_name); else { - appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", - slot_name, plugin); + if (is_temporary) + appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" TEMPORARY LOGICAL \"%s\"", + slot_name, plugin); + else + appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", + slot_name, plugin); if (PQserverVersion(conn) >= 100000) /* pg_recvlogical doesn't use an exported snapshot, so suppress */ appendPQExpBuffer(query, " NOEXPORT_SNAPSHOT"); diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h index 460dcb5..1fe4168 100644 --- a/src/bin/pg_basebackup/streamutil.h +++ b/src/bin/pg_basebackup/streamutil.h @@ -32,7 +32,7 @@ extern PGconn *GetConnection(void); /* Replication commands */ extern bool CreateReplicationSlot(PGconn *conn, const char *slot_name, - const char *plugin, bool is_physical, + const char *plugin, bool is_physical, bool is_temporary, bool slot_exists_ok); extern bool DropReplicationSlot(PGconn *conn, const char *slot_name); extern bool RunIdentifySystem(PGconn *conn, char **sysid, -- 2.1.4