From 75c5e59ef08e1408c2130bda77c39773f659f760 Mon Sep 17 00:00:00 2001
From: Michael Banck <michael.banck@credativ.de>
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

