On Sun, Aug 31, 2014 at 10:45 PM, Magnus Hagander <mag...@hagander.net> wrote:
> As this is a number of patches rolled into one - do you happen to keep
> them separate in your local repo? If so can you send them as separate
> ones (refactor identify_system should be quite unrelated to supporting
> replication slots, right?), for easier review? (if not, I'll just
> split them apart mentally, but it's easier to review separately)
Thanks for your review!
OK, here are 2 patches, the 2nd needing the 1st one:
1) Refactor IDENTIFY_SYSTEM and replslot create/drop APIs
2) Support for --create and --drop in pg_receivexlog
> On the identify_system part - my understanding of the code is that
> what you pass in as num_cols is the number of columns required for it
> to work, right?
The argument is I would say cross-version compatibility and
consistency with the existing 9.4 code, but... (see below for the rest
of the story).
> We probably need to adjust the error message as well
> in that case, because it's no longer what's "expected", it's what's
> "required"?
OK, changed this way.
> And we might want to include a hint about the reason (wrong version)?
I am not sure about that, a simple error message looks fine IMO, and
there is no notion of error hinting in the other client utilities as
well.
> There's also a note "get LSN start position if necessary", but it
> tries to do it unconditionally. What is the "if necessary" supposed to
> refer to?
That's remnant of some old code, so I removed it. Thanks for spotting that.
> Actually - why do we even care about the 3 vs 4 in RunIdentifySystem,
> as it never actually looks at the 4th column anyway? If we do
> specifically want it to fail in the case of pg_recvlogical, we really
> need to think up a better error message for it, and perhaps a
> different way of specifying it?
Hm. I'd vote to simplify the code a bit based on the argument that the
current API only looks at the 3 first columns and does not care about
the 4th which is the plugin name.
> Do we really want those Asserts? There is not a single Assert in
> bin/pg_basebackup today - as is the case for most things in bin/. We
> typically use regular if statements for things that "can happen", and
> just ignore the others I think - since the callers are fairly simple
> to trace.
OK, removed.
Regards,
--
Michael
From fdca8988480cac602157c3ae24ae61311bdaf960 Mon Sep 17 00:00:00 2001
From: Michael Paquier <mich...@otacoo.com>
Date: Mon, 1 Sep 2014 20:48:43 +0900
Subject: [PATCH 1/2] Refactoring of pg_basebackup utilities
Code duplication is reduced with the introduction of new APIs for each
individual replication command:
- IDENTIFY_SYSTEM
- CREATE_REPLICATION_SLOT
- DROP_REPLICATION_SLOT
A couple of variables used to identify a timeline ID are changed as well
to be more consistent with core code.
---
src/bin/pg_basebackup/pg_basebackup.c | 21 +----
src/bin/pg_basebackup/pg_receivexlog.c | 49 +++---------
src/bin/pg_basebackup/pg_recvlogical.c | 119 +++++----------------------
src/bin/pg_basebackup/streamutil.c | 142 +++++++++++++++++++++++++++++++++
src/bin/pg_basebackup/streamutil.h | 9 +++
5 files changed, 183 insertions(+), 157 deletions(-)
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 8b9acea..49675cf 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1569,8 +1569,8 @@ BaseBackup(void)
{
PGresult *res;
char *sysidentifier;
- uint32 latesttli;
- uint32 starttli;
+ TimeLineID latesttli;
+ TimeLineID starttli;
char *basebkp;
char escaped_label[MAXPGPATH];
char *maxrate_clause = NULL;
@@ -1624,23 +1624,8 @@ BaseBackup(void)
/*
* Run IDENTIFY_SYSTEM so we can get the timeline
*/
- res = PQexec(conn, "IDENTIFY_SYSTEM");
- if (PQresultStatus(res) != PGRES_TUPLES_OK)
- {
- fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
- progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
+ if (!RunIdentifySystem(conn, &sysidentifier, &latesttli, NULL))
disconnect_and_exit(1);
- }
- if (PQntuples(res) != 1 || PQnfields(res) < 3)
- {
- fprintf(stderr,
- _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"),
- progname, PQntuples(res), PQnfields(res), 1, 3);
- disconnect_and_exit(1);
- }
- sysidentifier = pg_strdup(PQgetvalue(res, 0, 0));
- latesttli = atoi(PQgetvalue(res, 0, 1));
- PQclear(res);
/*
* Start the actual backup
diff --git a/src/bin/pg_basebackup/pg_receivexlog.c b/src/bin/pg_basebackup/pg_receivexlog.c
index a8b9ad3..f722374 100644
--- a/src/bin/pg_basebackup/pg_receivexlog.c
+++ b/src/bin/pg_basebackup/pg_receivexlog.c
@@ -253,21 +253,10 @@ FindStreamingStart(uint32 *tli)
static void
StreamLog(void)
{
- PGresult *res;
XLogRecPtr startpos;
- uint32 starttli;
+ TimeLineID starttli;
XLogRecPtr serverpos;
- uint32 servertli;
- uint32 hi,
- lo;
-
- /*
- * Connect in replication mode to the server
- */
- conn = GetConnection();
- if (!conn)
- /* Error message already written in GetConnection() */
- return;
+ TimeLineID servertli;
if (!CheckServerVersionForStreaming(conn))
{
@@ -280,33 +269,11 @@ StreamLog(void)
}
/*
- * Run IDENTIFY_SYSTEM so we can get the timeline and current xlog
- * position.
+ * Identify server, obtaining start LSN position and current timeline ID
+ * at the same time.
*/
- res = PQexec(conn, "IDENTIFY_SYSTEM");
- if (PQresultStatus(res) != PGRES_TUPLES_OK)
- {
- fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
- progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
- disconnect_and_exit(1);
- }
- if (PQntuples(res) != 1 || PQnfields(res) < 3)
- {
- fprintf(stderr,
- _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"),
- progname, PQntuples(res), PQnfields(res), 1, 3);
+ if (!RunIdentifySystem(conn, NULL, &servertli, &serverpos))
disconnect_and_exit(1);
- }
- servertli = atoi(PQgetvalue(res, 0, 1));
- if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2)
- {
- fprintf(stderr,
- _("%s: could not parse transaction log location \"%s\"\n"),
- progname, PQgetvalue(res, 0, 2));
- disconnect_and_exit(1);
- }
- serverpos = ((uint64) hi) << 32 | lo;
- PQclear(res);
/*
* Figure out where to start streaming.
@@ -492,6 +459,12 @@ main(int argc, char **argv)
pqsignal(SIGINT, sigint_handler);
#endif
+ /* Obtain a connection before doing anything */
+ conn = GetConnection();
+ if (!conn)
+ /* Error message already written in GetConnection() */
+ exit(1);
+
while (true)
{
StreamLog();
diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c
index f3b0f34..1e53c82 100644
--- a/src/bin/pg_basebackup/pg_recvlogical.c
+++ b/src/bin/pg_basebackup/pg_recvlogical.c
@@ -208,15 +208,6 @@ StreamLog(void)
query = createPQExpBuffer();
/*
- * Connect in replication mode to the server
- */
- if (!conn)
- conn = GetConnection();
- if (!conn)
- /* Error message already written in GetConnection() */
- return;
-
- /*
* Start the replication
*/
if (verbose)
@@ -596,7 +587,6 @@ sighup_handler(int signum)
int
main(int argc, char **argv)
{
- PGresult *res;
static struct option long_options[] = {
/* general options */
{"file", required_argument, NULL, 'f'},
@@ -834,121 +824,48 @@ main(int argc, char **argv)
#endif
/*
- * don't really need this but it actually helps to get more precise error
- * messages about authentication, required GUCs and such without starting
- * to loop around connection attempts lateron.
+ * Obtain a connection to server. This is not really necessary but it
+ * helps to get more precise error messages about authentification,
+ * required GUC parameters and such.
*/
- {
- conn = GetConnection();
- if (!conn)
- /* Error message already written in GetConnection() */
- exit(1);
-
- /*
- * Run IDENTIFY_SYSTEM so we can get the timeline and current xlog
- * position.
- */
- res = PQexec(conn, "IDENTIFY_SYSTEM");
- if (PQresultStatus(res) != PGRES_TUPLES_OK)
- {
- fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
- progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
- disconnect_and_exit(1);
- }
-
- if (PQntuples(res) != 1 || PQnfields(res) < 4)
- {
- fprintf(stderr,
- _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"),
- progname, PQntuples(res), PQnfields(res), 1, 4);
- disconnect_and_exit(1);
- }
- PQclear(res);
- }
-
+ conn = GetConnection();
+ if (!conn)
+ /* Error message already written in GetConnection() */
+ exit(1);
- /*
- * stop a replication slot
- */
+ /* Drop a replication slot */
if (do_drop_slot)
{
- char query[256];
-
if (verbose)
fprintf(stderr,
- _("%s: freeing replication slot \"%s\"\n"),
+ _("%s: dropping replication slot \"%s\"\n"),
progname, replication_slot);
- snprintf(query, sizeof(query), "DROP_REPLICATION_SLOT \"%s\"",
- replication_slot);
- res = PQexec(conn, query);
- if (PQresultStatus(res) != PGRES_COMMAND_OK)
- {
- fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
- progname, query, PQerrorMessage(conn));
- disconnect_and_exit(1);
- }
-
- if (PQntuples(res) != 0 || PQnfields(res) != 0)
- {
- fprintf(stderr,
- _("%s: could not stop logical replication: got %d rows and %d fields, expected %d rows and %d fields\n"),
- progname, PQntuples(res), PQnfields(res), 0, 0);
+ if (!DropReplicationSlot(conn, false))
disconnect_and_exit(1);
- }
-
- PQclear(res);
disconnect_and_exit(0);
}
- /*
- * init a replication slot
- */
+ /* Create a replication slot */
if (do_create_slot)
{
- char query[256];
-
if (verbose)
fprintf(stderr,
- _("%s: initializing replication slot \"%s\"\n"),
+ _("%s: creating replication slot \"%s\"\n"),
progname, replication_slot);
- snprintf(query, sizeof(query), "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"",
- replication_slot, plugin);
-
- res = PQexec(conn, query);
- if (PQresultStatus(res) != PGRES_TUPLES_OK)
- {
- fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
- progname, query, PQerrorMessage(conn));
+ if (!CreateReplicationSlot(conn, plugin, false))
disconnect_and_exit(1);
- }
-
- if (PQntuples(res) != 1 || PQnfields(res) != 4)
- {
- fprintf(stderr,
- _("%s: could not init logical replication: got %d rows and %d fields, expected %d rows and %d fields\n"),
- progname, PQntuples(res), PQnfields(res), 1, 4);
- disconnect_and_exit(1);
- }
-
- if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &hi, &lo) != 2)
- {
- fprintf(stderr,
- _("%s: could not parse transaction log location \"%s\"\n"),
- progname, PQgetvalue(res, 0, 1));
- disconnect_and_exit(1);
- }
- startpos = ((uint64) hi) << 32 | lo;
-
- replication_slot = strdup(PQgetvalue(res, 0, 0));
- PQclear(res);
}
-
if (!do_start_slot)
disconnect_and_exit(0);
+ /* Identify system, obtaining start LSN position at the same time */
+ if (!RunIdentifySystem(conn, NULL, NULL, &startpos))
+ disconnect_and_exit(1);
+
+ /* Stream loop */
while (true)
{
StreamLog();
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index 1100260..173dda4 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -227,6 +227,148 @@ GetConnection(void)
return tmpconn;
}
+/*
+ * Run IDENTIFY_SYSTEM through a given connection and give back to caller
+ * some result information if requested:
+ * - Start LSN position
+ * - Current timeline ID
+ * - system identifier
+ */
+bool
+RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli,
+ XLogRecPtr *startpos)
+{
+ PGresult *res;
+ uint32 hi, lo;
+
+ /* Leave if no connection present */
+ if (conn == NULL)
+ return false;
+
+ res = PQexec(conn, "IDENTIFY_SYSTEM");
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
+ progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
+ return false;
+ }
+ if (PQntuples(res) != 1 || PQnfields(res) < 3)
+ {
+ fprintf(stderr,
+ _("%s: could not identify system: got %d rows and %d fields, required %d rows and %d or more fields\n"),
+ progname, PQntuples(res), PQnfields(res), 1, 3);
+ return false;
+ }
+
+ /* Get system identifier */
+ if (sysid != NULL)
+ *sysid = pg_strdup(PQgetvalue(res, 0, 0));
+
+ /* Get timeline ID to start streaming from */
+ if (starttli != NULL)
+ *starttli = atoi(PQgetvalue(res, 0, 1));
+
+ /* Get LSN start position if necessary */
+ if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2)
+ {
+ fprintf(stderr,
+ _("%s: could not parse transaction log location \"%s\"\n"),
+ progname, PQgetvalue(res, 0, 2));
+ return false;
+ }
+ if (startpos != NULL)
+ *startpos = ((uint64) hi) << 32 | lo;
+
+ PQclear(res);
+ return true;
+}
+
+/*
+ * Create a replication slot for the given connection. This function
+ * returns true in case of success as well as the start position
+ * obtained after the slot creation.
+ */
+bool
+CreateReplicationSlot(PGconn *conn, const char *plugin,
+ bool is_physical)
+{
+ char query[256];
+ PGresult *res;
+ uint32 hi, lo;
+
+ /* Build query */
+ if (is_physical)
+ snprintf(query, sizeof(query), "CREATE_REPLICATION_SLOT \"%s\" PHYSICAL",
+ replication_slot);
+ else
+ snprintf(query, sizeof(query), "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"",
+ replication_slot, plugin);
+
+ res = PQexec(conn, query);
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
+ progname, query, PQerrorMessage(conn));
+ return false;
+ }
+
+ if (PQntuples(res) != 1 || PQnfields(res) != 4)
+ {
+ fprintf(stderr,
+ _("%s: could not init %s replication: got %d rows and %d fields, expected %d rows and %d fields\n"),
+ progname, is_physical ? "physical" : "logical",
+ PQntuples(res), PQnfields(res), 1, 4);
+ return false;
+ }
+
+ /* Check LSN format obtained as consistent point */
+ if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &hi, &lo) != 2)
+ {
+ fprintf(stderr,
+ _("%s: could not parse transaction log location \"%s\"\n"),
+ progname, PQgetvalue(res, 0, 1));
+ return false;
+ }
+
+ replication_slot = strdup(PQgetvalue(res, 0, 0));
+ PQclear(res);
+ return true;
+}
+
+/*
+ * Drop a replication slot for the given connection. This function
+ * returns true in case of success.
+ */
+bool
+DropReplicationSlot(PGconn *conn, bool is_physical)
+{
+ char query[256];
+ PGresult *res;
+
+ /* Build query */
+ snprintf(query, sizeof(query), "DROP_REPLICATION_SLOT \"%s\"",
+ replication_slot);
+ res = PQexec(conn, query);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ {
+ fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
+ progname, query, PQerrorMessage(conn));
+ return false;
+ }
+
+ if (PQntuples(res) != 0 || PQnfields(res) != 0)
+ {
+ fprintf(stderr,
+ _("%s: could not stop %s replication: got %d rows and %d fields, expected %d rows and %d fields\n"),
+ progname, is_physical ? "physical" : "logical",
+ PQntuples(res), PQnfields(res), 0, 0);
+ return false;
+ }
+
+ PQclear(res);
+ return true;
+}
+
/*
* Frontend version of GetCurrentTimestamp(), since we are not linked with
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 8c6691f..e03687c 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -14,6 +14,8 @@
#include "libpq-fe.h"
+#include "access/xlogdefs.h"
+
extern const char *progname;
extern char *connection_string;
extern char *dbhost;
@@ -28,6 +30,13 @@ extern PGconn *conn;
extern PGconn *GetConnection(void);
+/* Replication commands */
+extern bool CreateReplicationSlot(PGconn *conn, const char *plugin,
+ bool is_physical);
+extern bool DropReplicationSlot(PGconn *conn, bool is_physical);
+extern bool RunIdentifySystem(PGconn *conn, char **sysid,
+ TimeLineID *starttli,
+ XLogRecPtr *startpos);
extern int64 feGetCurrentTimestamp(void);
extern void feTimestampDifference(int64 start_time, int64 stop_time,
long *secs, int *microsecs);
--
2.1.0
From 8ff1dfbaaab139f1796d3b604d7b8c5bdbf0d163 Mon Sep 17 00:00:00 2001
From: Michael Paquier <mich...@otacoo.com>
Date: Mon, 1 Sep 2014 20:53:45 +0900
Subject: [PATCH 2/2] Support for replslot creation and drop in pg_receivexlog
Using the new actions --create and --drop that are similarly present
in pg_recvlogical, a user can respectively create and drop a replication
slot that can be used afterwards when fetching WALs.
---
doc/src/sgml/ref/pg_receivexlog.sgml | 29 +++++++++++++++++
src/bin/pg_basebackup/pg_receivexlog.c | 59 +++++++++++++++++++++++++++++++++-
2 files changed, 87 insertions(+), 1 deletion(-)
diff --git a/doc/src/sgml/ref/pg_receivexlog.sgml b/doc/src/sgml/ref/pg_receivexlog.sgml
index 5916b8f..51d93ea 100644
--- a/doc/src/sgml/ref/pg_receivexlog.sgml
+++ b/doc/src/sgml/ref/pg_receivexlog.sgml
@@ -72,6 +72,35 @@ PostgreSQL documentation
<title>Options</title>
<para>
+ <application>pg_receivexlog</application> can run in one of two following
+ modes, which control physical replication slot:
+
+ <variablelist>
+
+ <varlistentry>
+ <term><option>--create</option></term>
+ <listitem>
+ <para>
+ Create a new physical replication slot with the name specified in
+ <option>--slot</option>, then exit.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term><option>--drop</option></term>
+ <listitem>
+ <para>
+ Drop the replication slot with the name specified in
+ <option>--slot</option>, then exit.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+
+ </para>
+
+ <para>
The following command-line options control the location and format of the
output.
diff --git a/src/bin/pg_basebackup/pg_receivexlog.c b/src/bin/pg_basebackup/pg_receivexlog.c
index f722374..e87839a 100644
--- a/src/bin/pg_basebackup/pg_receivexlog.c
+++ b/src/bin/pg_basebackup/pg_receivexlog.c
@@ -38,6 +38,8 @@ static int noloop = 0;
static int standby_message_timeout = 10 * 1000; /* 10 sec = default */
static int fsync_interval = 0; /* 0 = default */
static volatile bool time_to_abort = false;
+static bool do_create_slot = false;
+static bool do_drop_slot = false;
static void usage(void);
@@ -78,6 +80,9 @@ usage(void)
printf(_(" -w, --no-password never prompt for password\n"));
printf(_(" -W, --password force password prompt (should happen automatically)\n"));
printf(_(" -S, --slot=SLOTNAME replication slot to use\n"));
+ printf(_("\nOptional actions:\n"));
+ printf(_(" --create create a new replication slot (for the slot's name see --slot)\n"));
+ printf(_(" --drop drop the replication slot (for the slot's name see --slot)\n"));
printf(_("\nReport bugs to <pgsql-b...@postgresql.org>.\n"));
}
@@ -337,6 +342,9 @@ main(int argc, char **argv)
{"status-interval", required_argument, NULL, 's'},
{"slot", required_argument, NULL, 'S'},
{"verbose", no_argument, NULL, 'v'},
+/* action */
+ {"create", no_argument, NULL, 1},
+ {"drop", no_argument, NULL, 2},
{NULL, 0, NULL, 0}
};
@@ -420,6 +428,13 @@ main(int argc, char **argv)
case 'v':
verbose++;
break;
+/* action */
+ case 1:
+ do_create_slot = true;
+ break;
+ case 2:
+ do_drop_slot = true;
+ break;
default:
/*
@@ -444,10 +459,26 @@ main(int argc, char **argv)
exit(1);
}
+ if (replication_slot == NULL && (do_drop_slot || do_create_slot))
+ {
+ fprintf(stderr, _("%s: replication slot needed with action --create or --drop\n"), progname);
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+
+ if (do_drop_slot && do_create_slot)
+ {
+ fprintf(stderr, _("%s: cannot use --create together with --drop\n"), progname);
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+
/*
* Required arguments
*/
- if (basedir == NULL)
+ if (basedir == NULL && !do_create_slot && !do_drop_slot)
{
fprintf(stderr, _("%s: no target directory specified\n"), progname);
fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
@@ -465,6 +496,32 @@ main(int argc, char **argv)
/* Error message already written in GetConnection() */
exit(1);
+ /* Drop a replication slot */
+ if (do_drop_slot)
+ {
+ if (verbose)
+ fprintf(stderr,
+ _("%s: dropping replication slot \"%s\"\n"),
+ progname, replication_slot);
+
+ if (!DropReplicationSlot(conn, true))
+ disconnect_and_exit(1);
+ disconnect_and_exit(0);
+ }
+
+ /* Create a replication slot */
+ if (do_create_slot)
+ {
+ if (verbose)
+ fprintf(stderr,
+ _("%s: creating replication slot \"%s\"\n"),
+ progname, replication_slot);
+
+ if (!CreateReplicationSlot(conn, NULL, true))
+ disconnect_and_exit(1);
+ disconnect_and_exit(0);
+ }
+
while (true)
{
StreamLog();
--
2.1.0
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
