Skip to content

Commit 5a991ef

Browse files
robertmhaasrobertmhaas
committed
Allow logical decoding via the walsender interface.
In order for this to work, walsenders need the optional ability to connect to a database, so the "replication" keyword now allows true or false, for backward-compatibility, and the new value "database" (which causes the "dbname" parameter to be respected). walsender needs to loop not only when idle but also when sending decoded data to the user and when waiting for more xlog data to decode. This means that there are now three separate loops inside walsender.c; although some refactoring has been done here, this is still a bit ugly. Andres Freund, with contributions from Álvaro Herrera, and further review by me.
1 parent cb9a0c7 commit 5a991ef

File tree

12 files changed

+915
-167
lines changed

12 files changed

+915
-167
lines changed

‎doc/src/sgml/protocol.sgml

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1302,10 +1302,13 @@
13021302

13031303
<para>
13041304
To initiate streaming replication, the frontend sends the
1305-
<literal>replication</> parameter in the startup message. This tells the
1306-
backend to go into walsender mode, wherein a small set of replication commands
1307-
can be issued instead of SQL statements. Only the simple query protocol can be
1308-
used in walsender mode.
1305+
<literal>replication</> parameter in the startup message. A boolean value
1306+
of <literal>true</> tells the backend to go into walsender mode, wherein a
1307+
small set of replication commands can be issued instead of SQL statements. Only
1308+
the simple query protocol can be used in walsender mode.
1309+
Passing <literal>database</> as the value instructs walsender to connect to
1310+
the database specified in the <literal>dbname</> parameter, which will allow
1311+
the connection to be used for logical replication from that database.
13091312

13101313
The commands accepted in walsender mode are:
13111314

@@ -1315,7 +1318,7 @@ The commands accepted in walsender mode are:
13151318
<listitem>
13161319
<para>
13171320
Requests the server to identify itself. Server replies with a result
1318-
set of a single row, containing three fields:
1321+
set of a single row, containing four fields:
13191322
</para>
13201323

13211324
<para>
@@ -1357,6 +1360,17 @@ The commands accepted in walsender mode are:
13571360
</listitem>
13581361
</varlistentry>
13591362

1363+
<varlistentry>
1364+
<term>
1365+
dbname
1366+
</term>
1367+
<listitem>
1368+
<para>
1369+
Database connected to or NULL.
1370+
</para>
1371+
</listitem>
1372+
</varlistentry>
1373+
13601374
</variablelist>
13611375
</para>
13621376
</listitem>

‎src/backend/postmaster/postmaster.c

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1884,10 +1884,23 @@ ProcessStartupPacket(Port *port, bool SSLdone)
18841884
port->cmdline_options = pstrdup(valptr);
18851885
else if (strcmp(nameptr, "replication") == 0)
18861886
{
1887-
if (!parse_bool(valptr, &am_walsender))
1887+
/*
1888+
* Due to backward compatibility concerns the replication
1889+
* parameter is a hybrid beast which allows the value to be
1890+
* either boolean or the string 'database'. The latter
1891+
* connects to a specific database which is e.g. required for
1892+
* logical decoding while.
1893+
*/
1894+
if (strcmp(valptr, "database") == 0)
1895+
{
1896+
am_walsender = true;
1897+
am_db_walsender = true;
1898+
}
1899+
else if (!parse_bool(valptr, &am_walsender))
18881900
ereport(FATAL,
18891901
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1890-
errmsg("invalid value for boolean option \"replication\"")));
1902+
errmsg("invalid value for parameter \"replication\""),
1903+
errhint("Valid values are: false, 0, true, 1, database.")));
18911904
}
18921905
else
18931906
{
@@ -1968,8 +1981,15 @@ ProcessStartupPacket(Port *port, bool SSLdone)
19681981
if (strlen(port->user_name) >= NAMEDATALEN)
19691982
port->user_name[NAMEDATALEN - 1] = '\0';
19701983

1971-
/* Walsender is not related to a particular database */
1972-
if (am_walsender)
1984+
/*
1985+
* Normal walsender backends, e.g. for streaming replication, are not
1986+
* connected to a particular database. But walsenders used for logical
1987+
* replication need to connect to a specific database. We allow streaming
1988+
* replication commands to be issued even if connected to a database as it
1989+
* can make sense to first make a basebackup and then stream changes
1990+
* starting from that.
1991+
*/
1992+
if (am_walsender && !am_db_walsender)
19731993
port->database_name[0] = '\0';
19741994

19751995
/*

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -131,16 +131,16 @@ libpqrcv_identify_system(TimeLineID *primary_tli)
131131
"the primary server: %s",
132132
PQerrorMessage(streamConn))));
133133
}
134-
if (PQnfields(res) != 3 || PQntuples(res) != 1)
134+
if (PQnfields(res) < 3 || PQntuples(res) != 1)
135135
{
136136
int ntuples = PQntuples(res);
137137
int nfields = PQnfields(res);
138138

139139
PQclear(res);
140140
ereport(ERROR,
141141
(errmsg("invalid response from primary server"),
142-
errdetail("Expected 1 tuple with 3 fields, got %d tuples with %d fields.",
143-
ntuples, nfields)));
142+
errdetail("Could not identify system: Got %d rows and %d fields, expected %d rows and %d or more fields.",
143+
ntuples, nfields, 3, 1)));
144144
}
145145
primary_sysid = PQgetvalue(res, 0, 0);
146146
*primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0);

‎src/backend/replication/repl_gram.y

Lines changed: 68 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -73,13 +73,17 @@ Node *replication_parse_result;
7373
%token K_WAL
7474
%token K_TIMELINE
7575
%token K_PHYSICAL
76+
%token K_LOGICAL
7677
%token K_SLOT
7778

7879
%type <node> command
79-
%type <node> base_backup start_replication create_replication_slot drop_replication_slot identify_system timeline_history
80+
%type <node> base_backup start_replication start_logical_replication create_replication_slot drop_replication_slot identify_system timeline_history
8081
%type <list> base_backup_opt_list
8182
%type <defelt> base_backup_opt
8283
%type <uintval> opt_timeline
84+
%type <list> plugin_options plugin_opt_list
85+
%type <defelt> plugin_opt_elem
86+
%type <node> plugin_opt_arg
8387
%type <str> opt_slot
8488

8589
%%
@@ -98,6 +102,7 @@ command:
98102
identify_system
99103
| base_backup
100104
| start_replication
105+
| start_logical_replication
101106
| create_replication_slot
102107
| drop_replication_slot
103108
| timeline_history
@@ -165,8 +170,8 @@ base_backup_opt:
165170
}
166171
;
167172

168-
/* CREATE_REPLICATION_SLOT SLOT slot PHYSICAL */
169173
create_replication_slot:
174+
/* CREATE_REPLICATION_SLOT slot PHYSICAL */
170175
K_CREATE_REPLICATION_SLOT IDENT K_PHYSICAL
171176
{
172177
CreateReplicationSlotCmd *cmd;
@@ -175,9 +180,19 @@ create_replication_slot:
175180
cmd->slotname = $2;
176181
$$ = (Node *) cmd;
177182
}
183+
/* CREATE_REPLICATION_SLOT slot LOGICAL plugin */
184+
| K_CREATE_REPLICATION_SLOT IDENT K_LOGICAL IDENT
185+
{
186+
CreateReplicationSlotCmd *cmd;
187+
cmd = makeNode(CreateReplicationSlotCmd);
188+
cmd->kind = REPLICATION_KIND_LOGICAL;
189+
cmd->slotname = $2;
190+
cmd->plugin = $4;
191+
$$ = (Node *) cmd;
192+
}
178193
;
179194

180-
/* DROP_REPLICATION_SLOT SLOT slot */
195+
/* DROP_REPLICATION_SLOT slot */
181196
drop_replication_slot:
182197
K_DROP_REPLICATION_SLOT IDENT
183198
{
@@ -205,19 +220,19 @@ start_replication:
205220
}
206221
;
207222

208-
opt_timeline:
209-
K_TIMELINE UCONST
223+
/* START_REPLICATION SLOT slot LOGICAL %X/%X options */
224+
start_logical_replication:
225+
K_START_REPLICATION K_SLOT IDENT K_LOGICAL RECPTR plugin_options
210226
{
211-
if ($2 <= 0)
212-
ereport(ERROR,
213-
(errcode(ERRCODE_SYNTAX_ERROR),
214-
(errmsg("invalid timeline %u", $2))));
215-
$$ = $2;
227+
StartReplicationCmd *cmd;
228+
cmd = makeNode(StartReplicationCmd);
229+
cmd->kind = REPLICATION_KIND_LOGICAL;;
230+
cmd->slotname = $3;
231+
cmd->startpoint = $5;
232+
cmd->options = $6;
233+
$$ = (Node *) cmd;
216234
}
217-
| /* EMPTY */
218-
{ $$ = 0; }
219235
;
220-
221236
/*
222237
* TIMELINE_HISTORY %d
223238
*/
@@ -250,6 +265,46 @@ opt_slot:
250265
{ $$ = NULL; }
251266
;
252267

268+
opt_timeline:
269+
K_TIMELINE UCONST
270+
{
271+
if ($2 <= 0)
272+
ereport(ERROR,
273+
(errcode(ERRCODE_SYNTAX_ERROR),
274+
(errmsg("invalid timeline %u", $2))));
275+
$$ = $2;
276+
}
277+
| /* EMPTY */ { $$ = 0; }
278+
;
279+
280+
281+
plugin_options:
282+
'(' plugin_opt_list ')' { $$ = $2; }
283+
| /* EMPTY */ { $$ = NIL; }
284+
;
285+
286+
plugin_opt_list:
287+
plugin_opt_elem
288+
{
289+
$$ = list_make1($1);
290+
}
291+
| plugin_opt_list ',' plugin_opt_elem
292+
{
293+
$$ = lappend($1, $3);
294+
}
295+
;
296+
297+
plugin_opt_elem:
298+
IDENT plugin_opt_arg
299+
{
300+
$$ = makeDefElem($1, $2);
301+
}
302+
;
303+
304+
plugin_opt_arg:
305+
SCONST { $$ = (Node *) makeString($1); }
306+
| /* EMPTY */ { $$ = NULL; }
307+
;
253308
%%
254309

255310
#include "repl_scanner.c"

‎src/backend/replication/repl_scanner.l

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@ CREATE_REPLICATION_SLOT { return K_CREATE_REPLICATION_SLOT; }
9494
DROP_REPLICATION_SLOT { return K_DROP_REPLICATION_SLOT; }
9595
TIMELINE_HISTORY { return K_TIMELINE_HISTORY; }
9696
PHYSICAL { return K_PHYSICAL; }
97+
LOGICAL { return K_LOGICAL; }
9798
SLOT { return K_SLOT; }
9899

99100
"," { return ','; }

0 commit comments

Comments
 (0)