On Mon, Oct 25, 2021 at 02:40:05PM +0530, Bharath Rupireddy wrote:
> 2) I think we should tweak the below error message
> to
> pg_log_error("could not read replication slot \"%s\": %s",
> "IDENTIFY_SYSTEM", PQerrorMessage(conn));
> Having slot name in the error message helps to isolate the error
> message from tons of server logs that gets generated.
Yes, this suggestion makes sense.
> 3) Also for the same reasons stated as above, change the below error message
> pg_log_error("could not read replication slot: got %d rows and %d
> fields, expected %d rows and %d or more fields",
> to
> pg_log_error("could not read replication slot \"%s\": got %d rows and
> %d fields, expected %d rows and %d or more fields", slot_name,....
We can even get rid of "or more" to match the condition used.
> 4) Also for the same reasons, change below
> + pg_log_error("could not parse slot's restart_lsn \"%s\"",
> to
> pg_log_error("could not parse replicaton slot \"%s\" restart_lsn \"%s\"",
> slot_name, PQgetvalue(res, 0, 1));
Appending the slot name makes sense.
> 5) I think we should also have assertion for the timeline id:
> Assert(stream.startpos != InvalidXLogRecPtr);
> Assert(stream.timeline!= 0);
Okay.
> 6) Why do we need these two assignements?
> I think we can just get rid of lsn_loc and tli_loc, initlaize
> *restart_lsn = InvalidXLogRecPtr and *restart_tli = 0 at the start of
> the function and directly assign the requrested values to *restart_lsn
> and *restart_tli, also see comment (8).
FWIW, I find the style of the patch easier to follow.
> 9) 80char limit crossed:
> +GetSlotInformation(PGconn *conn, const char *slot_name, XLogRecPtr
> *restart_lsn, TimeLineID *restart_tli)
pgindent says nothing.
> 10) Missing word "command", and use "issued to the server", so change the
> below:
> + <command>READ_REPLICATION_SLOT</command> is issued to retrieve the
> to
> + <command>READ_REPLICATION_SLOT</command> command is issued to
> the server to retrieve the
Okay.
> 11) Will replication_slot ever be NULL? If it ever be null, then we
> don't reach this far right? We see the pg_log_error("%s needs a slot
> to be specified using --slot". Please revmove below if condition:
> + * server may not support this option.
Did you notice that this applies when creating or dropping a slot, for
code paths entirely different than what we are dealing with here?
--
Michael
From bc38918b44e3e41836bb075e7f915dcc40fddcb9 Mon Sep 17 00:00:00 2001
From: Michael Paquier <mich...@paquier.xyz>
Date: Mon, 25 Oct 2021 19:48:11 +0900
Subject: [PATCH v12] Allow pg_receivewal to stream from a slot's restart LSN
Prior to this patch, when running pg_receivewal, the streaming start
point would be the current location of the archives if anything is
found, and pg_receivewal would fall back to the current WAL flush
location if there are no archives, as of the result of an
IDENTIFY_SYSTEM command.
If for some reason the WAL files from pg_receivewal were moved, we want
to restart where we left at, which is the replication slot's restart_lsn
instead of skipping right to the current flush location. So this makes
pg_receivewal use the following sequence of methods to determine the
starting streaming LSN:
- Scan the local archives.
- Use the slot's restart_lsn, if supported.
- Fallback to the current flush LSN.
To keep compatibility with prior server versions, we only attempt to use
READ_REPLICATION_SLOT if the backend version is at least 15, and
fallback to the older behavior of streaming from the current flush
LSN if the command is not supported.
Some new TAP tests are added to cover this feature.
---
src/bin/pg_basebackup/pg_receivewal.c | 31 ++++++-
src/bin/pg_basebackup/streamutil.c | 98 ++++++++++++++++++++
src/bin/pg_basebackup/streamutil.h | 3 +
src/bin/pg_basebackup/t/020_pg_receivewal.pl | 53 ++++++++++-
doc/src/sgml/ref/pg_receivewal.sgml | 11 +++
5 files changed, 192 insertions(+), 4 deletions(-)
diff --git a/src/bin/pg_basebackup/pg_receivewal.c b/src/bin/pg_basebackup/pg_receivewal.c
index d5140a79fe..04ba20b197 100644
--- a/src/bin/pg_basebackup/pg_receivewal.c
+++ b/src/bin/pg_basebackup/pg_receivewal.c
@@ -404,15 +404,40 @@ StreamLog(void)
exit(1);
/*
- * Figure out where to start streaming.
+ * Figure out where to start streaming. First scan the local directory.
*/
stream.startpos = FindStreamingStart(&stream.timeline);
if (stream.startpos == InvalidXLogRecPtr)
{
- stream.startpos = serverpos;
- stream.timeline = servertli;
+ /*
+ * Try to get the starting point from the slot if any. This is
+ * supported in PostgreSQL 15 and newer.
+ */
+ if (replication_slot != NULL &&
+ PQserverVersion(conn) >= 150000)
+ {
+ if (!GetSlotInformation(conn, replication_slot, &stream.startpos,
+ &stream.timeline))
+ {
+ /* Error is logged by GetSlotInformation() */
+ return;
+ }
+ }
+
+ /*
+ * If it the starting point is still not known, use the current WAL
+ * flush value as last resort.
+ */
+ if (stream.startpos == InvalidXLogRecPtr)
+ {
+ stream.startpos = serverpos;
+ stream.timeline = servertli;
+ }
}
+ Assert(stream.startpos != InvalidXLogRecPtr &&
+ stream.timeline != 0);
+
/*
* Always start streaming at the beginning of a segment
*/
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index a9bc1ce214..6f14ed7748 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -479,6 +479,104 @@ RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli,
return true;
}
+
+/*
+ * Run READ_REPLICATION_SLOT through a given connection and give back to
+ * caller some result information if requested for this slot:
+ * - Start LSN position, InvalidXLogRecPtr if unknown.
+ * - Current timeline ID, 0 if unknown.
+ * Returns false on failure, and true otherwise.
+ */
+bool
+GetSlotInformation(PGconn *conn, const char *slot_name,
+ XLogRecPtr *restart_lsn, TimeLineID *restart_tli)
+{
+ PGresult *res;
+ PQExpBuffer query;
+ XLogRecPtr lsn_loc = InvalidXLogRecPtr;
+ TimeLineID tli_loc = 0;
+
+ if (*restart_lsn)
+ *restart_lsn = lsn_loc;
+ if (restart_tli != NULL)
+ *restart_tli = tli_loc;
+
+ query = createPQExpBuffer();
+ appendPQExpBuffer(query, "READ_REPLICATION_SLOT %s", slot_name);
+ res = PQexec(conn, query->data);
+ destroyPQExpBuffer(query);
+
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not send replication command \"%s\": %s",
+ "READ_REPLICATION_SLOT", PQerrorMessage(conn));
+ PQclear(res);
+ return false;
+ }
+
+ /* The command should always return precisely one tuple and three fields */
+ if (PQntuples(res) != 1 || PQnfields(res) != 3)
+ {
+ pg_log_error("could not read replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
+ slot_name, PQntuples(res), PQnfields(res), 1, 3);
+ PQclear(res);
+ return false;
+ }
+
+ /*
+ * When the slot doesn't exist, the command returns a tuple with NULL
+ * values. This checks only the slot type field.
+ */
+ if (PQgetisnull(res, 0, 0))
+ {
+ pg_log_error("could not find replication slot \"%s\"", slot_name);
+ PQclear(res);
+ return false;
+ }
+
+ /*
+ * Note that this cannot happen as READ_REPLICATION_SLOT supports only
+ * physical slots, but play it safe.
+ */
+ if (strcmp(PQgetvalue(res, 0, 0), "physical") != 0)
+ {
+ pg_log_error("expected a physical replication slot, got type \"%s\" instead",
+ PQgetvalue(res, 0, 0));
+ PQclear(res);
+ return false;
+ }
+
+ /* restart LSN */
+ if (!PQgetisnull(res, 0, 1))
+ {
+ uint32 hi,
+ lo;
+
+ if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &hi, &lo) != 2)
+ {
+ pg_log_error("could not parse restart_lsn \"%s\" for replication slot \"%s\"",
+ PQgetvalue(res, 0, 1), slot_name);
+ PQclear(res);
+ return false;
+ }
+ lsn_loc = ((uint64) hi) << 32 | lo;
+ }
+
+ /* current TLI */
+ if (!PQgetisnull(res, 0, 2))
+ tli_loc = (TimeLineID) atol(PQgetvalue(res, 0, 2));
+
+ PQclear(res);
+
+ /* Assign results if requested */
+ if (restart_lsn)
+ *restart_lsn = lsn_loc;
+ if (restart_tli)
+ *restart_tli = tli_loc;
+
+ return true;
+}
+
/*
* Create a replication slot for the given connection. This function
* returns true in case of success.
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 65135c79e0..7918935cb3 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -52,6 +52,9 @@ extern void AppendIntegerCommandOption(PQExpBuffer buf,
bool use_new_option_syntax,
char *option_name, int32 option_value);
+extern bool GetSlotInformation(PGconn *conn, const char *slot_name,
+ XLogRecPtr *restart_lsn,
+ TimeLineID *restart_tli);
extern bool RetrieveWalSegSize(PGconn *conn);
extern TimestampTz feGetCurrentTimestamp(void);
extern void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
diff --git a/src/bin/pg_basebackup/t/020_pg_receivewal.pl b/src/bin/pg_basebackup/t/020_pg_receivewal.pl
index b93493b5e9..092c9b6f25 100644
--- a/src/bin/pg_basebackup/t/020_pg_receivewal.pl
+++ b/src/bin/pg_basebackup/t/020_pg_receivewal.pl
@@ -5,7 +5,7 @@ use strict;
use warnings;
use PostgreSQL::Test::Utils;
use PostgreSQL::Test::Cluster;
-use Test::More tests => 27;
+use Test::More tests => 31;
program_help_ok('pg_receivewal');
program_version_ok('pg_receivewal');
@@ -72,6 +72,8 @@ $primary->command_ok(
my @partial_wals = glob "$stream_dir/*\.partial";
is(scalar(@partial_wals), 1, "one partial WAL segment was created");
+note "Testing pg_receivewal with compression methods";
+
# Check ZLIB compression if available.
SKIP:
{
@@ -155,3 +157,52 @@ SKIP:
ok(check_mode_recursive($stream_dir, 0700, 0600),
"check stream dir permissions");
}
+
+note "Testing pg_receivewal with slot as starting streaming point";
+
+# When using a replication slot, archiving should be resumed from the slot's
+# restart LSN. Use a new archive location and new slot for this test.
+my $slot_dir = $primary->basedir . '/slot_wal';
+mkdir($slot_dir);
+$slot_name = 'archive_slot';
+
+# Setup the slot, reserving WAL at creation (corresponding to the
+# last redo LSN here, actually).
+$primary->psql('postgres',
+ "SELECT pg_create_physical_replication_slot('$slot_name', true);");
+
+# Get the segment name associated with the slot's restart LSN, that should
+# be archived.
+my $walfile_streamed = $primary->safe_psql(
+ 'postgres',
+ "SELECT pg_walfile_name(restart_lsn)
+ FROM pg_replication_slots
+ WHERE slot_name = '$slot_name';");
+
+# Switch to a new segment, to make sure that the segment retained by the
+# slot is still streamed. This may not be necessary, but play it safe.
+$primary->psql('postgres',
+ 'INSERT INTO test_table VALUES (generate_series(1,100));');
+$primary->psql('postgres', 'SELECT pg_switch_wal();');
+$nextlsn =
+ $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn();');
+chomp($nextlsn);
+
+# Check case where the slot does not exist.
+$primary->command_fails_like(
+ [
+ 'pg_receivewal', '-D', $slot_dir, '--slot',
+ 'nonexistentslot', '-n', '--no-sync', '--verbose',
+ '--endpos', $nextlsn
+ ],
+ qr/pg_receivewal: error: could not find replication slot "nonexistentslot"/,
+ 'pg_receivewal fails with non-existing slot');
+$primary->command_ok(
+ [
+ 'pg_receivewal', '-D', $slot_dir, '--slot',
+ $slot_name, '-n', '--no-sync', '--verbose',
+ '--endpos', $nextlsn
+ ],
+ "WAL streamed from the slot's restart_lsn");
+ok(-e "$slot_dir/$walfile_streamed",
+ "WAL from the slot's restart_lsn has been archived");
diff --git a/doc/src/sgml/ref/pg_receivewal.sgml b/doc/src/sgml/ref/pg_receivewal.sgml
index 6da8b2be8c..d3c7488293 100644
--- a/doc/src/sgml/ref/pg_receivewal.sgml
+++ b/doc/src/sgml/ref/pg_receivewal.sgml
@@ -88,6 +88,17 @@ PostgreSQL documentation
</para>
</listitem>
+ <listitem>
+ <para>
+ If a starting point cannot not be calculated with the previous method,
+ and if a replication slot is used, an extra
+ <command>READ_REPLICATION_SLOT</command> command is issued to retrieve
+ the slot's <literal>restart_lsn</literal> to use as starting point.
+ This option is only available when streaming write-ahead logs from
+ <productname>PostgreSQL</productname> 15 and up.
+ </para>
+ </listitem>
+
<listitem>
<para>
If a starting point cannot be calculated with the previous method,
--
2.33.0
signature.asc
Description: PGP signature
