Implement NOWAIT option for BASE_BACKUP command

Specifying this option makes the server not wait for the
xlog to be archived, or emit a warning that it can't,
instead leaving the responsibility with the client.

This is useful when the log is being streamed using
the streaming protocol in parallel with the backup,
without having log archiving enabled.

doc/src/sgml/protocol.sgml

@@ -1473,7 +1473,7 @@ The commands accepted in walsender mode are:
   </varlistentry>
 
   <varlistentry>
-    <term>BASE_BACKUP [<literal>LABEL</literal> <replaceable>'label'</replaceable>] [<literal>PROGRESS</literal>] [<literal>FAST</literal>] [<literal>WAL</literal>]</term>
+    <term>BASE_BACKUP [<literal>LABEL</literal> <replaceable>'label'</replaceable>] [<literal>PROGRESS</literal>] [<literal>FAST</literal>] [<literal>WAL</literal>] [<literal>NOWAIT</literal>]</term>
     <listitem>
      <para>
       Instructs the server to start streaming a base backup.
@@ -1530,6 +1530,19 @@ The commands accepted in walsender mode are:
          </para>
         </listitem>
        </varlistentry>
+
+       <varlistentry>
+        <term><literal>NOWAIT</literal></term>
+        <listitem>
+         <para>
+          By default, the backup will wait until the last required xlog
+          segment has been archived, or emit a warning if log archiving is
+          not enabled. Specifying <literal>NOWAIT</literal> disables both
+          the waiting and the warning, leaving the client responsible for
+          ensuring the required log is available.
+         </para>
+         </listitem>
+       </varlistentry>
       </variablelist>
      </para>
      <para>

src/backend/access/transam/xlog.c

@@ -8865,7 +8865,7 @@ pg_stop_backup(PG_FUNCTION_ARGS)
 	XLogRecPtr	stoppoint;
 	char		stopxlogstr[MAXFNAMELEN];
 
-	stoppoint = do_pg_stop_backup(NULL);
+	stoppoint = do_pg_stop_backup(NULL, true);
 
 	snprintf(stopxlogstr, sizeof(stopxlogstr), "%X/%X",
 			 stoppoint.xlogid, stoppoint.xrecoff);
@@ -8880,7 +8880,7 @@ pg_stop_backup(PG_FUNCTION_ARGS)
  * the non-exclusive backup specified by 'labelfile'.
  */
 XLogRecPtr
-do_pg_stop_backup(char *labelfile)
+do_pg_stop_backup(char *labelfile, bool waitforarchive)
 {
 	bool		exclusive = (labelfile == NULL);
 	XLogRecPtr	startpoint;
@@ -9079,7 +9079,7 @@ do_pg_stop_backup(char *labelfile)
 	 * wish to wait, you can set statement_timeout.  Also, some notices are
 	 * issued to clue in anyone who might be doing this interactively.
 	 */
-	if (XLogArchivingActive())
+	if (waitforarchive && XLogArchivingActive())
 	{
 		XLByteToPrevSeg(stoppoint, _logId, _logSeg);
 		XLogFileName(lastxlogfilename, ThisTimeLineID, _logId, _logSeg);
@@ -9120,7 +9120,7 @@ do_pg_stop_backup(char *labelfile)
 		ereport(NOTICE,
 				(errmsg("pg_stop_backup complete, all required WAL segments have been archived")));
 	}
-	else
+	else if (waitforarchive)
 		ereport(NOTICE,
 				(errmsg("WAL archiving is not enabled; you must ensure that all required WAL segments are copied through other means to complete the backup")));
 

src/backend/replication/basebackup.c

@@ -37,6 +37,7 @@ typedef struct
 	const char *label;
 	bool		progress;
 	bool		fastcheckpoint;
+	bool		nowait;
 	bool		includewal;
 }	basebackup_options;
 
@@ -173,7 +174,7 @@ perform_base_backup(basebackup_options *opt, DIR *tblspcdir)
 	}
 	PG_END_ENSURE_ERROR_CLEANUP(base_backup_cleanup, (Datum) 0);
 
-	endptr = do_pg_stop_backup(labelfile);
+	endptr = do_pg_stop_backup(labelfile, !opt->nowait);
 
 	if (opt->includewal)
 	{
@@ -260,6 +261,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 	bool		o_label = false;
 	bool		o_progress = false;
 	bool		o_fast = false;
+	bool		o_nowait = false;
 	bool		o_wal = false;
 
 	MemSet(opt, 0, sizeof(*opt));
@@ -294,6 +296,15 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			opt->fastcheckpoint = true;
 			o_fast = true;
 		}
+		else if (strcmp(defel->defname, "nowait") == 0)
+		{
+			if (o_nowait)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("duplicate option \"%s\"", defel->defname)));
+			opt->nowait = true;
+			o_nowait = true;
+		}
 		else if (strcmp(defel->defname, "wal") == 0)
 		{
 			if (o_wal)

src/backend/replication/repl_gram.y

@@ -71,6 +71,7 @@ Node *replication_parse_result;
 %token K_LABEL
 %token K_PROGRESS
 %token K_FAST
+%token K_NOWAIT
 %token K_WAL
 %token K_START_REPLICATION
 
@@ -107,7 +108,7 @@ identify_system:
 			;
 
 /*
- * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL]
+ * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  */
 base_backup:
 			K_BASE_BACKUP base_backup_opt_list
@@ -142,6 +143,11 @@ base_backup_opt:
 				  $$ = makeDefElem("wal",
 						   (Node *)makeInteger(TRUE));
 				}
+			| K_NOWAIT
+				{
+				  $$ = makeDefElem("nowait",
+						   (Node *)makeInteger(TRUE));
+				}
 			;
 
 /*

src/backend/replication/repl_scanner.l

@@ -60,6 +60,7 @@ BASE_BACKUP			{ return K_BASE_BACKUP; }
 FAST			{ return K_FAST; }
 IDENTIFY_SYSTEM		{ return K_IDENTIFY_SYSTEM; }
 LABEL			{ return K_LABEL; }
+NOWAIT			{ return K_NOWAIT; }
 PROGRESS			{ return K_PROGRESS; }
 WAL			{ return K_WAL; }
 START_REPLICATION	{ return K_START_REPLICATION; }

src/include/access/xlog.h

@@ -318,7 +318,7 @@ extern void WakeupRecovery(void);
  * Starting/stopping a base backup
  */
 extern XLogRecPtr do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile);
-extern XLogRecPtr do_pg_stop_backup(char *labelfile);
+extern XLogRecPtr do_pg_stop_backup(char *labelfile, bool waitforarchive);
 extern void do_pg_abort_backup(void);
 
 /* File path names (all relative to $PGDATA) */