Add ability to reserve WAL upon slot creation via replication protocol.

Since 6fcd8851 it is possible to immediately reserve WAL when creating a
slot via pg_create_physical_replication_slot(). Extend the replication
protocol to allow that as well.

Although, in contrast to the SQL interface, it is possible to update the
reserved location via the replication interface, it is still useful
being able to reserve upon creation there. Otherwise the logic in
ReplicationSlotReserveWal() has to be repeated in slot employing
clients.

Author: Michael Paquier
Discussion: CAB7nPqT0Wc1W5mdYGeJ_wbutbwNN+3qgrFR64avXaQCiJMGaYA@mail.gmail.com

doc/src/sgml/protocol.sgml

@@ -1434,7 +1434,7 @@ The commands accepted in walsender mode are:
   </varlistentry>
 
   <varlistentry>
-    <term>CREATE_REPLICATION_SLOT <replaceable class="parameter">slot_name</> { <literal>PHYSICAL</> | <literal>LOGICAL</> <replaceable class="parameter">output_plugin</> }
+    <term>CREATE_REPLICATION_SLOT <replaceable class="parameter">slot_name</> { <literal>PHYSICAL</> [ RESERVE_WAL ] | <literal>LOGICAL</> <replaceable class="parameter">output_plugin</> }
      <indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
@@ -1463,6 +1463,17 @@ The commands accepted in walsender mode are:
          </para>
        </listitem>
       </varlistentry>
+
+      <varlistentry>
+       <term><literal>RESERVE_WAL</></term>
+       <listitem>
+        <para>
+         Specify that this physical replication reserves <acronym>WAL</>
+         immediately; otherwise <acronym>WAL</> is only reserved upon
+         connection from a streaming replication client.
+        </para>
+       </listitem>
+      </varlistentry>
      </variablelist>
     </listitem>
   </varlistentry>

src/backend/replication/repl_gram.y

@@ -76,6 +76,7 @@ Node *replication_parse_result;
 %token K_PHYSICAL
 %token K_LOGICAL
 %token K_SLOT
+%token K_RESERVE_WAL
 
 %type <node>	command
 %type <node>	base_backup start_replication start_logical_replication
@@ -88,6 +89,7 @@ Node *replication_parse_result;
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
 %type <str>		opt_slot
+%type <boolval>	opt_reserve_wal
 
 %%
 
@@ -181,13 +183,14 @@ base_backup_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot PHYSICAL */
-			K_CREATE_REPLICATION_SLOT IDENT K_PHYSICAL
+			/* CREATE_REPLICATION_SLOT slot PHYSICAL RESERVE_WAL */
+			K_CREATE_REPLICATION_SLOT IDENT K_PHYSICAL opt_reserve_wal
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
 					cmd->kind = REPLICATION_KIND_PHYSICAL;
 					cmd->slotname = $2;
+					cmd->reserve_wal = $4;
 					$$ = (Node *) cmd;
 				}
 			/* CREATE_REPLICATION_SLOT slot LOGICAL plugin */
@@ -268,6 +271,11 @@ opt_physical:
 			| /* EMPTY */
 			;
 
+opt_reserve_wal:
+			K_RESERVE_WAL					{ $$ = true; }
+			| /* EMPTY */					{ $$ = false; }
+			;
+
 opt_slot:
 			K_SLOT IDENT
 				{ $$ = $2; }

src/backend/replication/repl_scanner.l

@@ -95,6 +95,7 @@ CREATE_REPLICATION_SLOT		{ return K_CREATE_REPLICATION_SLOT; }
 DROP_REPLICATION_SLOT		{ return K_DROP_REPLICATION_SLOT; }
 TIMELINE_HISTORY	{ return K_TIMELINE_HISTORY; }
 PHYSICAL			{ return K_PHYSICAL; }
+RESERVE_WAL			{ return K_RESERVE_WAL; }
 LOGICAL				{ return K_LOGICAL; }
 SLOT				{ return K_SLOT; }
 

src/backend/replication/walsender.c

@@ -826,6 +826,14 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 
 		ReplicationSlotPersist();
 	}
+	else if (cmd->kind == REPLICATION_KIND_PHYSICAL && cmd->reserve_wal)
+	{
+		ReplicationSlotReserveWal();
+
+		/* Write this slot to disk */
+		ReplicationSlotMarkDirty();
+		ReplicationSlotSave();
+	}
 
 	slot_name = NameStr(MyReplicationSlot->data.name);
 	snprintf(xpos, sizeof(xpos), "%X/%X",

src/include/nodes/replnodes.h

@@ -55,6 +55,7 @@ typedef struct CreateReplicationSlotCmd
 	char	   *slotname;
 	ReplicationKind kind;
 	char	   *plugin;
+	bool		reserve_wal;
 } CreateReplicationSlotCmd;