From c314ead5be0c627a6f654a74f18099466c566c47 Mon Sep 17 00:00:00 2001
From: Andres Freund <andres@anarazel.de>
Date: Sun, 6 Sep 2015 13:17:23 +0200
Subject: [PATCH] Add ability to reserve WAL upon slot creation via replication
 protocol.

Since 6fcd885 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             | 13 ++++++++++++-
 src/backend/replication/repl_gram.y    | 12 ++++++++++--
 src/backend/replication/repl_scanner.l |  1 +
 src/backend/replication/walsender.c    |  8 ++++++++
 src/include/nodes/replnodes.h          |  1 +
 5 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 42e94971741..2caa84a1253 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/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>
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index e9177ca0db6..1b753644223 100644
--- a/src/backend/replication/repl_gram.y
+++ b/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; }
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index 91bac21b06b..126e7d90640 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/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; }
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 27de156b1ff..c95fe75a721 100644
--- a/src/backend/replication/walsender.c
+++ b/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",
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index cac64198508..4e35be6a286 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -55,6 +55,7 @@ typedef struct CreateReplicationSlotCmd
 	char	   *slotname;
 	ReplicationKind kind;
 	char	   *plugin;
+	bool		reserve_wal;
 } CreateReplicationSlotCmd;