Skip to content

Commit eb4da3e

Browse files
peterepetere
committed
Add option to control snapshot export to CREATE_REPLICATION_SLOT
We used to export snapshots unconditionally in CREATE_REPLICATION_SLOT in the replication protocol, but several upcoming patches want more control over what happens. Suppress snapshot export in pg_recvlogical, which neither needs nor can use the exported snapshot. Since snapshot exporting can fail this improves reliability. This also paves the way for allowing the creation of replication slots on standbys, which cannot export snapshots because they cannot allocate new XIDs. Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>
1 parent 7150402 commit eb4da3e

File tree

10 files changed

+141
-26
lines changed

10 files changed

+141
-26
lines changed

‎doc/src/sgml/logicaldecoding.sgml

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -271,8 +271,9 @@ $ pg_recvlogical -d postgres --slot test --drop-slot
271271
<sect2>
272272
<title>Exported Snapshots</title>
273273
<para>
274-
When a new replication slot is created using the streaming replication interface,
275-
a snapshot is exported
274+
When a new replication slot is created using the streaming replication
275+
interface (see <xref linkend="protocol-replication-create-slot">), a
276+
snapshot is exported
276277
(see <xref linkend="functions-snapshot-synchronization">), which will show
277278
exactly the state of the database after which all changes will be
278279
included in the change stream. This can be used to create a new replica by
@@ -282,6 +283,12 @@ $ pg_recvlogical -d postgres --slot test --drop-slot
282283
database's state at that point in time, which afterwards can be updated
283284
using the slot's contents without losing any changes.
284285
</para>
286+
<para>
287+
Creation of a snapshot is not always possible. In particular, it will
288+
fail when connected to a hot standby. Applications that do not require
289+
snapshot export may suppress it with the <literal>NOEXPORT_SNAPSHOT</>
290+
option.
291+
</para>
285292
</sect2>
286293
</sect1>
287294

‎doc/src/sgml/protocol.sgml

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1486,8 +1486,8 @@ The commands accepted in walsender mode are:
14861486
</listitem>
14871487
</varlistentry>
14881488

1489-
<varlistentry>
1490-
<term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</> [ <literal>TEMPORARY</> ] { <literal>PHYSICAL</> [ <literal>RESERVE_WAL</> ] | <literal>LOGICAL</> <replaceable class="parameter">output_plugin</> }
1489+
<varlistentry id="protocol-replication-create-slot" xreflabel="CREATE_REPLICATION_SLOT">
1490+
<term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</> [ <literal>TEMPORARY</> ] { <literal>PHYSICAL</> [ <literal>RESERVE_WAL</> ] | <literal>LOGICAL</> <replaceable class="parameter">output_plugin</> [ <literal>EXPORT_SNAPSHOT</> | <literal>NOEXPORT_SNAPSHOT</> ] }
14911491
<indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
14921492
</term>
14931493
<listitem>
@@ -1538,6 +1538,21 @@ The commands accepted in walsender mode are:
15381538
</para>
15391539
</listitem>
15401540
</varlistentry>
1541+
1542+
<varlistentry>
1543+
<term><literal>EXPORT_SNAPSHOT</></term>
1544+
<term><literal>NOEXPORT_SNAPSHOT</></term>
1545+
<listitem>
1546+
<para>
1547+
Decides what to do with the snapshot created during logical slot
1548+
initialization. <literal>EXPORT_SNAPSHOT</>, which is the default,
1549+
will export the snapshot for use in other sessions. This option can't
1550+
be used inside a transaction. <literal>NOEXPORT_SNAPSHOT</> will
1551+
just use the snapshot for logical decoding as normal but won't do
1552+
anything else with it.
1553+
</para>
1554+
</listitem>
1555+
</varlistentry>
15411556
</variablelist>
15421557

15431558
<para>

‎src/backend/commands/subscriptioncmds.c

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -314,7 +314,11 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
314314

315315
PG_TRY();
316316
{
317-
walrcv_create_slot(wrconn, slotname, false, &lsn);
317+
/*
318+
* Create permanent slot for the subscription. We won't use the
319+
* initial snapshot for anything, so no need to export it.
320+
*/
321+
walrcv_create_slot(wrconn, slotname, false, false, &lsn);
318322
ereport(NOTICE,
319323
(errmsg("created replication slot \"%s\" on publisher",
320324
slotname)));

‎src/backend/replication/libpqwalreceiver/libpqwalreceiver.c

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ static void libpqrcv_send(WalReceiverConn *conn, const char *buffer,
6868
static char *libpqrcv_create_slot(WalReceiverConn *conn,
6969
const char *slotname,
7070
bool temporary,
71+
bool export_snapshot,
7172
XLogRecPtr *lsn);
7273
static bool libpqrcv_command(WalReceiverConn *conn,
7374
const char *cmd, char **err);
@@ -720,21 +721,27 @@ libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
720721
*/
721722
static char *
722723
libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
723-
bool temporary, XLogRecPtr *lsn)
724+
bool temporary, bool export_snapshot, XLogRecPtr *lsn)
724725
{
725726
PGresult *res;
726727
StringInfoData cmd;
727728
char *snapshot;
728729

729730
initStringInfo(&cmd);
730731

731-
appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\" ", slotname);
732+
appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
732733

733734
if (temporary)
734-
appendStringInfo(&cmd, "TEMPORARY ");
735+
appendStringInfo(&cmd, " TEMPORARY");
735736

736737
if (conn->logical)
737-
appendStringInfo(&cmd, "LOGICAL pgoutput");
738+
{
739+
appendStringInfo(&cmd, " LOGICAL pgoutput");
740+
if (export_snapshot)
741+
appendStringInfo(&cmd, " EXPORT_SNAPSHOT");
742+
else
743+
appendStringInfo(&cmd, " NOEXPORT_SNAPSHOT");
744+
}
738745

739746
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
740747
pfree(cmd.data);

‎src/backend/replication/repl_gram.y

Lines changed: 34 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,8 @@ Node *replication_parse_result;
7979
%token K_SLOT
8080
%token K_RESERVE_WAL
8181
%token K_TEMPORARY
82+
%token K_EXPORT_SNAPSHOT
83+
%token K_NOEXPORT_SNAPSHOT
8284

8385
%type <node> command
8486
%type <node> base_backup start_replication start_logical_replication
@@ -91,7 +93,9 @@ Node *replication_parse_result;
9193
%type <defelt> plugin_opt_elem
9294
%type <node> plugin_opt_arg
9395
%type <str> opt_slot var_name
94-
%type <boolval> opt_reserve_wal opt_temporary
96+
%type <boolval> opt_temporary
97+
%type <list> create_slot_opt_list
98+
%type <defelt> create_slot_opt
9599

96100
%%
97101

@@ -202,29 +206,55 @@ base_backup_opt:
202206

203207
create_replication_slot:
204208
/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
205-
K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL opt_reserve_wal
209+
K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
206210
{
207211
CreateReplicationSlotCmd *cmd;
208212
cmd = makeNode(CreateReplicationSlotCmd);
209213
cmd->kind = REPLICATION_KIND_PHYSICAL;
210214
cmd->slotname = $2;
211215
cmd->temporary = $3;
212-
cmd->reserve_wal = $5;
216+
cmd->options = $5;
213217
$$ = (Node *) cmd;
214218
}
215219
/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
216-
| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT
220+
| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
217221
{
218222
CreateReplicationSlotCmd *cmd;
219223
cmd = makeNode(CreateReplicationSlotCmd);
220224
cmd->kind = REPLICATION_KIND_LOGICAL;
221225
cmd->slotname = $2;
222226
cmd->temporary = $3;
223227
cmd->plugin = $5;
228+
cmd->options = $6;
224229
$$ = (Node *) cmd;
225230
}
226231
;
227232

233+
create_slot_opt_list:
234+
create_slot_opt_list create_slot_opt
235+
{ $$ = lappend($1, $2); }
236+
| /* EMPTY */
237+
{ $$ = NIL; }
238+
;
239+
240+
create_slot_opt:
241+
K_EXPORT_SNAPSHOT
242+
{
243+
$$ = makeDefElem("export_snapshot",
244+
(Node *)makeInteger(TRUE), -1);
245+
}
246+
| K_NOEXPORT_SNAPSHOT
247+
{
248+
$$ = makeDefElem("export_snapshot",
249+
(Node *)makeInteger(FALSE), -1);
250+
}
251+
| K_RESERVE_WAL
252+
{
253+
$$ = makeDefElem("reserve_wal",
254+
(Node *)makeInteger(TRUE), -1);
255+
}
256+
;
257+
228258
/* DROP_REPLICATION_SLOT slot */
229259
drop_replication_slot:
230260
K_DROP_REPLICATION_SLOT IDENT
@@ -291,11 +321,6 @@ opt_physical:
291321
| /* EMPTY */
292322
;
293323

294-
opt_reserve_wal:
295-
K_RESERVE_WAL { $$ = true; }
296-
| /* EMPTY */ { $$ = false; }
297-
;
298-
299324
opt_temporary:
300325
K_TEMPORARY { $$ = true; }
301326
| /* EMPTY */ { $$ = false; }

‎src/backend/replication/repl_scanner.l

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,8 @@ RESERVE_WAL { return K_RESERVE_WAL; }
100100
LOGICAL { return K_LOGICAL; }
101101
SLOT { return K_SLOT; }
102102
TEMPORARY { return K_TEMPORARY; }
103+
EXPORT_SNAPSHOT { return K_EXPORT_SNAPSHOT; }
104+
NOEXPORT_SNAPSHOT { return K_NOEXPORT_SNAPSHOT; }
103105

104106
"," { return ','; }
105107
";" { return ';'; }

‎src/backend/replication/walsender.c

Lines changed: 54 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151

5252
#include "catalog/pg_type.h"
5353
#include "commands/dbcommands.h"
54+
#include "commands/defrem.h"
5455
#include "funcapi.h"
5556
#include "libpq/libpq.h"
5657
#include "libpq/pqformat.h"
@@ -737,6 +738,48 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
737738
return count;
738739
}
739740

741+
/*
742+
* Process extra options given to CREATE_REPLICATION_SLOT.
743+
*/
744+
static void
745+
parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
746+
bool *reserve_wal,
747+
bool *export_snapshot)
748+
{
749+
ListCell *lc;
750+
bool snapshot_action_given = false;
751+
bool reserve_wal_given = false;
752+
753+
/* Parse options */
754+
foreach (lc, cmd->options)
755+
{
756+
DefElem *defel = (DefElem *) lfirst(lc);
757+
758+
if (strcmp(defel->defname, "export_snapshot") == 0)
759+
{
760+
if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
761+
ereport(ERROR,
762+
(errcode(ERRCODE_SYNTAX_ERROR),
763+
errmsg("conflicting or redundant options")));
764+
765+
snapshot_action_given = true;
766+
*export_snapshot = defGetBoolean(defel);
767+
}
768+
else if (strcmp(defel->defname, "reserve_wal") == 0)
769+
{
770+
if (reserve_wal_given || cmd->kind != REPLICATION_KIND_PHYSICAL)
771+
ereport(ERROR,
772+
(errcode(ERRCODE_SYNTAX_ERROR),
773+
errmsg("conflicting or redundant options")));
774+
775+
reserve_wal_given = true;
776+
*reserve_wal = true;
777+
}
778+
else
779+
elog(ERROR, "unrecognized option: %s", defel->defname);
780+
}
781+
}
782+
740783
/*
741784
* Create a new replication slot.
742785
*/
@@ -746,6 +789,8 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
746789
const char *snapshot_name = NULL;
747790
char xpos[MAXFNAMELEN];
748791
char *slot_name;
792+
bool reserve_wal = false;
793+
bool export_snapshot = true;
749794
DestReceiver *dest;
750795
TupOutputState *tstate;
751796
TupleDesc tupdesc;
@@ -754,6 +799,8 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
754799

755800
Assert(!MyReplicationSlot);
756801

802+
parseCreateReplSlotOptions(cmd, &reserve_wal, &export_snapshot);
803+
757804
/* setup state for XLogReadPage */
758805
sendTimeLineIsHistoric = false;
759806
sendTimeLine = ThisTimeLineID;
@@ -799,18 +846,21 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
799846
DecodingContextFindStartpoint(ctx);
800847

801848
/*
802-
* Export a plain (not of the snapbuild.c type) snapshot to the user
803-
* that can be imported into another session.
849+
* Export the snapshot if we've been asked to do so.
850+
*
851+
* NB. We will convert the snapbuild.c kind of snapshot to normal
852+
* snapshot when doing this.
804853
*/
805-
snapshot_name = SnapBuildExportSnapshot(ctx->snapshot_builder);
854+
if (export_snapshot)
855+
snapshot_name = SnapBuildExportSnapshot(ctx->snapshot_builder);
806856

807857
/* don't need the decoding context anymore */
808858
FreeDecodingContext(ctx);
809859

810860
if (!cmd->temporary)
811861
ReplicationSlotPersist();
812862
}
813-
else if (cmd->kind == REPLICATION_KIND_PHYSICAL && cmd->reserve_wal)
863+
else if (cmd->kind == REPLICATION_KIND_PHYSICAL && reserve_wal)
814864
{
815865
ReplicationSlotReserveWal();
816866

‎src/bin/pg_basebackup/streamutil.c

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -338,8 +338,13 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
338338
appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" PHYSICAL",
339339
slot_name);
340340
else
341+
{
341342
appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"",
342343
slot_name, plugin);
344+
if (PQserverVersion(conn) >= 100000)
345+
/* pg_recvlogical doesn't use an exported snapshot, so suppress */
346+
appendPQExpBuffer(query, " NOEXPORT_SNAPSHOT");
347+
}
343348

344349
res = PQexec(conn, query->data);
345350
if (PQresultStatus(res) != PGRES_TUPLES_OK)

‎src/include/nodes/replnodes.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ typedef struct CreateReplicationSlotCmd
5656
ReplicationKind kind;
5757
char *plugin;
5858
bool temporary;
59-
bool reserve_wal;
59+
List *options;
6060
} CreateReplicationSlotCmd;
6161

6262

‎src/include/replication/walreceiver.h

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -183,7 +183,7 @@ typedef void (*walrcv_send_fn) (WalReceiverConn *conn, const char *buffer,
183183
int nbytes);
184184
typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
185185
const char *slotname, bool temporary,
186-
XLogRecPtr *lsn);
186+
bool export_snapshot, XLogRecPtr *lsn);
187187
typedef bool (*walrcv_command_fn) (WalReceiverConn *conn, const char *cmd,
188188
char **err);
189189
typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);
@@ -224,8 +224,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
224224
WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
225225
#define walrcv_send(conn, buffer, nbytes) \
226226
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
227-
#define walrcv_create_slot(conn, slotname, temporary, lsn) \
228-
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, lsn)
227+
#define walrcv_create_slot(conn, slotname, temporary, export_snapshot, lsn) \
228+
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, export_snapshot, lsn)
229229
#define walrcv_command(conn, cmd, err) \
230230
WalReceiverFunctions->walrcv_command(conn, cmd, err)
231231
#define walrcv_disconnect(conn) \

0 commit comments

Comments
 (0)