Subject: [PATCH] Add option to create a replication slot in pg_basebackup if not yet present.

From: Michael Banck <[email-masked]>
Date: Sun, 10 Sep 2017 18:07:33 +0200

When requesting a particular replication slot, the new pg_basebackup option
--create-slot tries to create it before starting to replicate from it unless
it exists already.

In order to allow reporting of temporary replication slot creation in --verbose
mode, further refactor the slot creation logic. 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 and mention it in --verbose mode. At the same time, also create
the temporary slot via CreateReplicationSlot() instead of creating the
temporary slot with an explicit SQL command.
---
 doc/src/sgml/ref/pg_basebackup.sgml          | 12 ++++++
 src/bin/pg_basebackup/pg_basebackup.c        | 56 ++++++++++++++++++++++++++--
 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           | 15 +++++---
 src/bin/pg_basebackup/streamutil.h           |  2 +-
 src/bin/pg_basebackup/t/010_pg_basebackup.pl | 34 +++++++++++++++--
 8 files changed, 107 insertions(+), 34 deletions(-)

diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml
index 2454d35af3..57086295d3 100644
--- a/doc/src/sgml/ref/pg_basebackup.sgml
+++ b/doc/src/sgml/ref/pg_basebackup.sgml
@@ -269,6 +269,18 @@ PostgreSQL documentation
      </varlistentry>
 
      <varlistentry>
+      <term><option>-C</option></term>
+      <term><option>--create-slot</option></term>
+      <listitem>
+       <para>
+        This option can only be used together with the <literal>--slot</literal>
+        option.  It causes the specified slot name to be created unless it
+        exists already.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
       <term><option>-T <replaceable class="parameter">olddir</replaceable>=<replaceable class="parameter">newdir</replaceable></option></term>
       <term><option>--tablespace-mapping=<replaceable class="parameter">olddir</replaceable>=<replaceable class="parameter">newdir</replaceable></option></term>
       <listitem>
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 51509d150e..35ea707384 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -92,6 +92,8 @@ static pg_time_t last_progress_report = 0;
 static int32 maxrate = 0;		/* no limit by default */
 static char *replication_slot = NULL;
 static bool temp_replication_slot = true;
+static bool create_slot = false;
+static bool no_slot = false;
 
 static bool success = false;
 static bool made_new_pgdata = false;
@@ -337,6 +339,7 @@ usage(void)
 			 "                         write recovery.conf for replication\n"));
 	printf(_("  -S, --slot=SLOTNAME    replication slot to use\n"));
 	printf(_("      --no-slot          prevent creation of temporary replication slot\n"));
+	printf(_("  -C, --create-slot      create replication slot if not present already\n"));
 	printf(_("  -T, --tablespace-mapping=OLDDIR=NEWDIR\n"
 			 "                         relocate tablespace in OLDDIR to NEWDIR\n"));
 	printf(_("  -X, --wal-method=none|fetch|stream\n"
@@ -492,8 +495,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) PQbackendPID(param->bgconn));
 
 	if (format == 'p')
 		stream.walmethod = CreateWalDirectoryMethod(param->xlog, 0, do_sync);
@@ -586,6 +587,27 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier)
 	else
 		param->temp_slot = temp_replication_slot;
 
+	/*
+	 * Create replication slot if one is needed.
+	 */
+	if (!no_slot && (temp_replication_slot || create_slot))
+	{
+		if (!replication_slot)
+			replication_slot = psprintf("pg_basebackup_%d", (int) PQbackendPID(param->bgconn));
+
+		if (!CreateReplicationSlot(param->bgconn, replication_slot, NULL, true,
+								   temp_replication_slot, false))
+			disconnect_and_exit(1);
+
+		if (verbose)
+			if (temp_replication_slot)
+				fprintf(stderr, _("%s: temporary replication slot \"%s\" created\n"),
+						progname, replication_slot);
+			else
+				fprintf(stderr, _("%s: replication slot \"%s\" created\n"),
+						progname, replication_slot);
+	}
+
 	if (format == 'p')
 	{
 		/*
@@ -2099,12 +2121,12 @@ main(int argc, char **argv)
 		{"progress", no_argument, NULL, 'P'},
 		{"waldir", required_argument, NULL, 1},
 		{"no-slot", no_argument, NULL, 2},
+		{"create-slot", no_argument, NULL, 'C'},
 		{NULL, 0, NULL, 0}
 	};
 	int			c;
 
 	int			option_index;
-	bool		no_slot = false;
 
 	progname = get_progname(argv[0]);
 	set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
@@ -2126,7 +2148,7 @@ main(int argc, char **argv)
 
 	atexit(cleanup_directories_atexit);
 
-	while ((c = getopt_long(argc, argv, "D:F:r:RT:X:l:nNzZ:d:c:h:p:U:s:S:wWvP",
+	while ((c = getopt_long(argc, argv, "D:F:r:RS:CT:X:l:nNzZ:d:c:h:p:U:s:wWvP",
 							long_options, &option_index)) != -1)
 	{
 		switch (c)
@@ -2162,6 +2184,9 @@ main(int argc, char **argv)
 				replication_slot = pg_strdup(optarg);
 				temp_replication_slot = false;
 				break;
+			case 'C':
+				create_slot = true;
+				break;
 			case 2:
 				no_slot = true;
 				break;
@@ -2347,6 +2372,29 @@ main(int argc, char **argv)
 		temp_replication_slot = false;
 	}
 
+	if (create_slot)
+	{
+		if (!replication_slot)
+		{
+			fprintf(stderr,
+					_("%s: creation of replication slots requires a slot name\n"),
+					progname);
+			fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+					progname);
+			exit(1);
+		}
+
+		if (no_slot)
+		{
+			fprintf(stderr,
+					_("%s: creation of replication slots requires a slot, but --no-slot was requested\n"),
+					progname);
+			fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+					progname);
+			exit(1);
+		}
+	}
+
 	if (xlog_dir)
 	{
 		if (format != 'p')
diff --git a/src/bin/pg_basebackup/pg_receivewal.c b/src/bin/pg_basebackup/pg_receivewal.c
index 4a1a5658fb..8dfe7fd549 100644
--- a/src/bin/pg_basebackup/pg_receivewal.c
+++ b/src/bin/pg_basebackup/pg_receivewal.c
@@ -700,7 +700,7 @@ main(int argc, char **argv)
 					_("%s: creating replication slot \"%s\"\n"),
 					progname, replication_slot);
 
-		if (!CreateReplicationSlot(conn, replication_slot, NULL, true,
+		if (!CreateReplicationSlot(conn, replication_slot, NULL, true, 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 6811a55e76..6499f58e3c 100644
--- a/src/bin/pg_basebackup/pg_recvlogical.c
+++ b/src/bin/pg_basebackup/pg_recvlogical.c
@@ -979,7 +979,7 @@ main(int argc, char **argv)
 					_("%s: creating replication slot \"%s\"\n"),
 					progname, replication_slot);
 
-		if (!CreateReplicationSlot(conn, replication_slot, plugin,
+		if (!CreateReplicationSlot(conn, replication_slot, plugin, 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 888458f4a9..1f1e5e1268 100644
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -522,24 +522,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 9d40744a34..be3a9786fd 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;
@@ -334,13 +334,18 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	Assert(slot_name != NULL);
 
 	/* Build query */
+	appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	if (is_physical)
-		appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" PHYSICAL",
-						  slot_name);
+	{
+		if (is_temporary)
+			appendPQExpBuffer(query, " TEMPORARY");
+		appendPQExpBuffer(query, " PHYSICAL RESERVE_WAL");
+	}
 	else
 	{
-		appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"",
-						  slot_name, plugin);
+		if (is_temporary)
+			appendPQExpBuffer(query, " TEMPORARY");
+		appendPQExpBuffer(query, " LOGICAL \"%s\"", 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 6f6878679f..1dffb55b91 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,
diff --git a/src/bin/pg_basebackup/t/010_pg_basebackup.pl b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
index a00f7b0e1a..e287645745 100644
--- a/src/bin/pg_basebackup/t/010_pg_basebackup.pl
+++ b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
@@ -4,7 +4,7 @@ use Cwd;
 use Config;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 72;
+use Test::More tests => 78;
 
 program_help_ok('pg_basebackup');
 program_version_ok('pg_basebackup');
@@ -256,18 +256,44 @@ $node->command_ok(
 	'pg_basebackup -X stream runs with --no-slot');
 
 $node->command_fails(
-	[ 'pg_basebackup', '-D', "$tempdir/fail", '-S', 'slot1' ],
+	[ 'pg_basebackup', '-D', "$tempdir/fail", '-X', 'none', '-S', 'slot0' ],
 	'pg_basebackup with replication slot fails without -X stream');
 $node->command_fails(
 	[   'pg_basebackup',             '-D',
 		"$tempdir/backupxs_sl_fail", '-X',
 		'stream',                    '-S',
-		'slot1' ],
+		'slot0' ],
 	'pg_basebackup fails with nonexistent replication slot');
 
+$node->command_fails(
+	[   'pg_basebackup', '-D', "$tempdir/backupxs_slot", '-C' ],
+	'pg_basebackup -C fails without slot name');
+
+$node->command_fails(
+	[   'pg_basebackup', '-D', "$tempdir/backupxs_slot", '-C', '-S', 'slot0', '--no-slot' ],
+	'pg_basebackup fails with -C -S --no-slot');
+
+$node->command_ok(
+	[   'pg_basebackup', '-D', "$tempdir/backupxs_slot", '-C', '-S', 'slot0' ],
+	'pg_basebackup -C -S creates previously nonexistent replication slot');
+
+my $lsn = $node->safe_psql('postgres',
+	q{SELECT restart_lsn FROM pg_replication_slots WHERE slot_name = 'slot0'}
+);
+like($lsn, qr!^0/[0-9A-F]{7,8}$!, 'replication slot is present');
+
+my $checkpoint_lsn = $node->safe_psql('postgres',
+	q{SELECT checkpoint_lsn FROM pg_control_checkpoint(), pg_replication_slots WHERE slot_name = 'slot0' AND restart_lsn <= pg_control_checkpoint.checkpoint_lsn}
+);
+like($lsn, qr!^0/[0-9A-F]{7,8}$!, "replication slot LSN $lsn is lower than latest checkpoint LSN $checkpoint_lsn");
+
+$node->command_fails(
+	[   'pg_basebackup', '-D', "$tempdir/backupxs_slot1", '-C', '-S', 'slot0' ],
+	'pg_basebackup fails with -C -S and a previously existing slot');
+
 $node->safe_psql('postgres',
 	q{SELECT * FROM pg_create_physical_replication_slot('slot1')});
-my $lsn = $node->safe_psql('postgres',
+$lsn = $node->safe_psql('postgres',
 	q{SELECT restart_lsn FROM pg_replication_slots WHERE slot_name = 'slot1'}
 );
 is($lsn, '', 'restart LSN of new slot is null');
-- 
2.11.0

--Y7xTucakfITjPcLV
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0

-- 
Sent via pgsql-hackers mailing list ([email-masked])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

--Y7xTucakfITjPcLV--