Add a WAIT option to DROP_REPLICATION_SLOT

Commit 9915de6c1cb2 changed the default behavior of
DROP_REPLICATION_SLOT so that it would wait until any session holding
the slot active would release it, instead of raising an error.  But
users are already depending on the original behavior, so revert to it by
default and add a WAIT option to invoke the new behavior.

Per complaint from Simone Gotti, in
Discussion: https://postgr.es/m/CAEvsy6Wgdf90O6pUvg2wSVXL2omH5OPC-38OD4Zzgk-FXavj3Q@mail.gmail.com

doc/src/sgml/logicaldecoding.sgml

@@ -303,7 +303,7 @@ $ pg_recvlogical -d postgres --slot test --drop-slot
      </listitem>
 
      <listitem>
-      <para><literal>DROP_REPLICATION_SLOT <replaceable>slot_name</replaceable></literal></para>
+      <para><literal>DROP_REPLICATION_SLOT <replaceable>slot_name</replaceable></literal> <optional> <literal>WAIT</> </></para>
      </listitem>
 
      <listitem>

doc/src/sgml/protocol.sgml

@@ -2173,13 +2173,13 @@ The commands accepted in walsender mode are:
   </varlistentry>
 
   <varlistentry>
-    <term><literal>DROP_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</>
+    <term>
+     <literal>DROP_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</> <optional> <literal>WAIT</> </optional>
      <indexterm><primary>DROP_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
      <para>
-      Drops a replication slot, freeing any reserved server-side resources. If
-      the slot is currently in use by an active connection, this command fails.
+      Drops a replication slot, freeing any reserved server-side resources.
       If the slot is a logical slot that was created in a database other than
       the database the walsender is connected to, this command fails.
      </para>
@@ -2192,6 +2192,17 @@ The commands accepted in walsender mode are:
          </para>
        </listitem>
       </varlistentry>
+
+      <varlistentry>
+       <term><literal>WAIT</literal></term>
+       <listitem>
+        <para>
+         This option causes the command to wait if the slot is active until
+         it becomes inactive, instead of the default behavior of raising an
+         error.
+        </para>
+       </listitem>
+      </varlistentry>
      </variablelist>
     </listitem>
   </varlistentry>

src/backend/commands/subscriptioncmds.c

@@ -959,7 +959,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	load_file("libpqwalreceiver", false);
 
 	initStringInfo(&cmd);
-	appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s", quote_identifier(slotname));
+	appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname));
 
 	wrconn = walrcv_connect(conninfo, true, subname, &err);
 	if (wrconn == NULL)

src/backend/replication/repl_gram.y

@@ -72,6 +72,7 @@ static SQLCmd *make_sqlcmd(void);
 %token K_LABEL
 %token K_PROGRESS
 %token K_FAST
+%token K_WAIT
 %token K_NOWAIT
 %token K_MAX_RATE
 %token K_WAL
@@ -272,6 +273,15 @@ drop_replication_slot:
 					DropReplicationSlotCmd *cmd;
 					cmd = makeNode(DropReplicationSlotCmd);
 					cmd->slotname = $2;
+					cmd->wait = false;
+					$$ = (Node *) cmd;
+				}
+			| K_DROP_REPLICATION_SLOT IDENT K_WAIT
+				{
+					DropReplicationSlotCmd *cmd;
+					cmd = makeNode(DropReplicationSlotCmd);
+					cmd->slotname = $2;
+					cmd->wait = true;
 					$$ = (Node *) cmd;
 				}
 			;

src/backend/replication/repl_scanner.l

@@ -103,6 +103,7 @@ TEMPORARY			{ return K_TEMPORARY; }
 EXPORT_SNAPSHOT		{ return K_EXPORT_SNAPSHOT; }
 NOEXPORT_SNAPSHOT	{ return K_NOEXPORT_SNAPSHOT; }
 USE_SNAPSHOT		{ return K_USE_SNAPSHOT; }
+WAIT				{ return K_WAIT; }
 
 ","				{ return ','; }
 ";"				{ return ';'; }

src/backend/replication/slotfuncs.c

@@ -171,7 +171,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 
 	CheckSlotRequirements();
 
-	ReplicationSlotDrop(NameStr(*name), false);
+	ReplicationSlotDrop(NameStr(*name), true);
 
 	PG_RETURN_VOID();
 }

src/backend/replication/walsender.c

@@ -1028,7 +1028,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 static void
 DropReplicationSlot(DropReplicationSlotCmd *cmd)
 {
-	ReplicationSlotDrop(cmd->slotname, false);
+	ReplicationSlotDrop(cmd->slotname, !cmd->wait);
 	EndCommand("DROP_REPLICATION_SLOT", DestRemote);
 }
 

src/include/nodes/replnodes.h

@@ -68,6 +68,7 @@ typedef struct DropReplicationSlotCmd
 {
 	NodeTag		type;
 	char	   *slotname;
+	bool		wait;
 } DropReplicationSlotCmd;