extensible options syntax for replication parser?

From 0eba2ff07ccf77c903b9ee0331ede9f31bdc1560 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Wed, 24 Jun 2020 11:49:00 -0400
Subject: [PATCH v2] Flexible options for BASE_BACKUP and
 CREATE_REPLICATION_SLOT.

---
 src/backend/replication/basebackup.c          |  33 ++---
 .../libpqwalreceiver/libpqwalreceiver.c       |   8 +-
 src/backend/replication/repl_gram.y           | 115 +++++++++++++++---
 src/backend/replication/walsender.c           |  15 +--
 src/bin/pg_basebackup/pg_basebackup.c         | 108 ++++++++++++----
 src/bin/pg_basebackup/streamutil.c            |  14 ++-
 6 files changed, 220 insertions(+), 73 deletions(-)

diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index 096b0fcef0..7d992e81e0 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -777,7 +778,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -786,7 +787,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
 		else if (strcmp(defel->defname, "fast") == 0)
@@ -795,16 +796,16 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
+			opt->fastcheckpoint = defGetBoolean(defel);
 			o_fast = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -813,19 +814,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -841,21 +842,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -880,7 +881,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -895,8 +896,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index e4fd1f9bb6..d93cdaba1b 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -824,19 +824,19 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 		switch (snapshot_action)
 		{
 			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, " (EXPORT_SNAPSHOT TRUE)");
 				break;
 			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, " (EXPORT_SNAPSHOT FALSE)");
 				break;
 			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
+				appendStringInfoString(&cmd, " (USE_SNAPSHOT)");
 				break;
 		}
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index f93a0de218..e86abf7a20 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -94,16 +94,16 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -156,12 +156,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -169,14 +181,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -199,8 +211,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -214,8 +226,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -230,8 +242,8 @@ base_backup_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -241,8 +253,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -255,14 +267,19 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
 				  $$ = makeDefElem("export_snapshot",
@@ -416,6 +433,64 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index e2477c47e0..12941c8a49 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -898,7 +898,8 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
+			if (defGetBoolean(defel))
+				*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -908,7 +909,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -975,7 +976,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (EXPORT_SNAPSHOT)")));
 
 			need_full_snapshot = true;
 		}
@@ -985,25 +986,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 4f29671d0c..0645e983c6 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1787,6 +1787,49 @@ ReceiveBackupManifestInMemoryChunk(size_t r, char *copybuf,
 	appendPQExpBuffer(buf, copybuf, r);
 }
 
+static void
+AppendBaseBackupPlainOption(StringInfo buf, bool use_new_option_syntax,
+							char *option_name)
+{
+	if (buf->len > 0)
+	{
+		if (use_new_option_syntax)
+			appendStringInfoString(buf, ", ");
+		else
+			appendStringInfoChar(buf, ' ');
+	}
+
+	appendStringInfoString(buf, option_name);
+}
+
+static void
+AppendBaseBackupStringOption(StringInfo buf, bool use_new_option_syntax,
+							 char *option_name, char *option_value)
+{
+	AppendBaseBackupPlainOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t length = strlen(option_value);
+		char *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendStringInfoString(buf, " '");
+		appendStringInfoString(buf, escaped_value);
+		appendStringInfoChar(buf, '\'');
+		pfree(escaped_value);
+	}
+}
+
+static void
+AppendBaseBackupIntegerOption(StringInfo buf, bool use_new_option_syntax,
+							  char *option_name, int32 option_value)
+{
+	AppendBaseBackupPlainOption(buf, use_new_option_syntax, option_name);
+
+	appendStringInfo(buf, " %d", option_value);
+}
+
 static void
 BaseBackup(void)
 {
@@ -1795,10 +1838,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1807,8 +1846,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	StringInfoData	buf;
 
 	Assert(conn != NULL);
+	initStringInfo(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1826,6 +1868,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1400)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1856,20 +1900,42 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendBaseBackupStringOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "FAST");
+	if (includewal == NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendBaseBackupIntegerOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendBaseBackupIntegerOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendBaseBackupIntegerOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendBaseBackupPlainOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendBaseBackupStringOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendBaseBackupStringOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1884,18 +1950,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index 410116492e..8ed628d8fa 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -505,14 +505,24 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	{
 		appendPQExpBufferStr(query, " PHYSICAL");
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+		{
+			if (PQserverVersion(conn) >= 140000)
+				appendPQExpBufferStr(query, " (RESERVE_WAL)");
+			else
+				appendPQExpBufferStr(query, " RESERVE_WAL");
+		}
 	}
 	else
 	{
 		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (PQserverVersion(conn) >= 100000)
+		{
 			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+			if (PQserverVersion(conn) >= 140000)
+				appendPQExpBufferStr(query, " (EXPORT_SNAPSHOT 0)");
+			else
+				appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+		}
 	}
 
 	res = PQexec(conn, query->data);
-- 
2.24.3 (Apple Git-128)

From 956f54a87d3e847a9cc6052285d4d3cdb5de6f10 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Thu, 8 Oct 2020 10:27:25 -0400
Subject: [PATCH v3] Flexible options for BASE_BACKUP and
 CREATE_REPLICATION_SLOT.

v2: Fix compile error.
v3: Fix inverted test, as reported by Tushar Ahuja.
---
 src/backend/replication/basebackup.c          |  33 ++---
 .../libpqwalreceiver/libpqwalreceiver.c       |   8 +-
 src/backend/replication/repl_gram.y           | 115 +++++++++++++++---
 src/backend/replication/walsender.c           |  15 +--
 src/bin/pg_basebackup/pg_basebackup.c         | 108 ++++++++++++----
 src/bin/pg_basebackup/streamutil.c            |  14 ++-
 6 files changed, 220 insertions(+), 73 deletions(-)

diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index b89df01fa7..e0c59f4adb 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -780,7 +781,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -789,7 +790,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
 		else if (strcmp(defel->defname, "fast") == 0)
@@ -798,16 +799,16 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
+			opt->fastcheckpoint = defGetBoolean(defel);
 			o_fast = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -816,19 +817,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -844,21 +845,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -883,7 +884,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -898,8 +899,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index ad574099ff..abf15874f4 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -849,19 +849,19 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 		switch (snapshot_action)
 		{
 			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, " (EXPORT_SNAPSHOT TRUE)");
 				break;
 			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, " (EXPORT_SNAPSHOT FALSE)");
 				break;
 			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
+				appendStringInfoString(&cmd, " (USE_SNAPSHOT)");
 				break;
 		}
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index f93a0de218..e86abf7a20 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -94,16 +94,16 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -156,12 +156,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -169,14 +181,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -199,8 +211,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -214,8 +226,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -230,8 +242,8 @@ base_backup_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -241,8 +253,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -255,14 +267,19 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
 				  $$ = makeDefElem("export_snapshot",
@@ -416,6 +433,64 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 7c9d1b67df..302039b6f6 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -897,7 +897,8 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
+			if (defGetBoolean(defel))
+				*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -907,7 +908,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -974,7 +975,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (EXPORT_SNAPSHOT)")));
 
 			need_full_snapshot = true;
 		}
@@ -984,25 +985,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (USE_SNAPSHOT)")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 7a5d4562f9..e12c4d2cb8 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1793,6 +1793,49 @@ ReceiveBackupManifestInMemoryChunk(size_t r, char *copybuf,
 	appendPQExpBuffer(buf, copybuf, r);
 }
 
+static void
+AppendBaseBackupPlainOption(StringInfo buf, bool use_new_option_syntax,
+							char *option_name)
+{
+	if (buf->len > 0)
+	{
+		if (use_new_option_syntax)
+			appendStringInfoString(buf, ", ");
+		else
+			appendStringInfoChar(buf, ' ');
+	}
+
+	appendStringInfoString(buf, option_name);
+}
+
+static void
+AppendBaseBackupStringOption(StringInfo buf, bool use_new_option_syntax,
+							 char *option_name, char *option_value)
+{
+	AppendBaseBackupPlainOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t length = strlen(option_value);
+		char *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendStringInfoString(buf, " '");
+		appendStringInfoString(buf, escaped_value);
+		appendStringInfoChar(buf, '\'');
+		pfree(escaped_value);
+	}
+}
+
+static void
+AppendBaseBackupIntegerOption(StringInfo buf, bool use_new_option_syntax,
+							  char *option_name, int32 option_value)
+{
+	AppendBaseBackupPlainOption(buf, use_new_option_syntax, option_name);
+
+	appendStringInfo(buf, " %d", option_value);
+}
+
 static void
 BaseBackup(void)
 {
@@ -1801,10 +1844,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1813,8 +1852,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	StringInfoData	buf;
 
 	Assert(conn != NULL);
+	initStringInfo(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1832,6 +1874,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1400)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1862,20 +1906,42 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendBaseBackupStringOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "FAST");
+	if (includewal != NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendBaseBackupIntegerOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendBaseBackupIntegerOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendBaseBackupPlainOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendBaseBackupIntegerOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendBaseBackupPlainOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendBaseBackupStringOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendBaseBackupStringOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1890,18 +1956,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index be653ebb2d..07b6fe7c8b 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -506,14 +506,24 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	{
 		appendPQExpBufferStr(query, " PHYSICAL");
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+		{
+			if (PQserverVersion(conn) >= 140000)
+				appendPQExpBufferStr(query, " (RESERVE_WAL)");
+			else
+				appendPQExpBufferStr(query, " RESERVE_WAL");
+		}
 	}
 	else
 	{
 		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (PQserverVersion(conn) >= 100000)
+		{
 			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+			if (PQserverVersion(conn) >= 140000)
+				appendPQExpBufferStr(query, " (EXPORT_SNAPSHOT 0)");
+			else
+				appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+		}
 	}
 
 	res = PQexec(conn, query->data);
-- 
2.24.3 (Apple Git-128)

From bd36c10d4413d0e2be4f3bab3a1de1a9736886c6 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Fri, 10 Sep 2021 11:50:05 -0400
Subject: [PATCH v5 1/2] Flexible options for BASE_BACKUP.

Previously, BASE_BACKUP used an entirely hard-coded syntax, but that's
hard to extend. Instead, adopt the same kind of syntax we've used for
SQL commands such as VACUUM, ANALYZE, COPY, and EXPLAIN, where it's
not necessary for all of the option names to be parser keywords.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_basebackup prefer
the new syntax when the server is new enough to support it.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml            | 68 ++++++++++++--------
 src/backend/replication/basebackup.c  | 33 +++++-----
 src/backend/replication/repl_gram.y   | 93 +++++++++++++++++++++++----
 src/bin/pg_basebackup/pg_basebackup.c | 65 ++++++++++++-------
 src/bin/pg_basebackup/streamutil.c    | 61 ++++++++++++++++++
 src/bin/pg_basebackup/streamutil.h    | 12 ++++
 6 files changed, 254 insertions(+), 78 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a232546b1d..32d1eeabdc 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2517,8 +2517,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-base-backup" xreflabel="BASE_BACKUP">
-    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
-     <indexterm><primary>BASE_BACKUP</primary></indexterm>
+    <term><literal>BASE_BACKUP</literal> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
     </term>
     <listitem>
      <para>
@@ -2540,52 +2539,55 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>PROGRESS</literal></term>
+        <term><literal>PROGRESS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Request information required to generate a progress report. This will
-          send back an approximate size in the header of each tablespace, which
-          can be used to calculate how far along the stream is done. This is
-          calculated by enumerating all the file sizes once before the transfer
-          is even started, and might as such have a negative impact on the
-          performance.  In particular, it might take longer before the first data
+          If set to true, request information required to generate a progress
+          report. This will send back an approximate size in the header of each
+          tablespace, which can be used to calculate how far along the stream
+          is done. This is calculated by enumerating all the file sizes once
+          before the transfer is even started, and might as such have a
+          negative impact on the performance.  In particular, it might take
+          longer before the first data
           is streamed. Since the database files can change during the backup,
           the size is only approximate and might both grow and shrink between
           the time of approximation and the sending of the actual files.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>FAST</literal></term>
+        <term><literal>FAST [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Request a fast checkpoint.
+          If set to true, a fast checkpoint is requested.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>WAL</literal></term>
+        <term><literal>WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include the necessary WAL segments in the backup. This will include
-          all the files between start and stop backup in the
+          If set to true, include the necessary WAL segments in the backup.
+          This will include all the files between start and stop backup in the
           <filename>pg_wal</filename> directory of the base directory tar
-          file.
+          file. The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOWAIT</literal></term>
+        <term><literal>WAIT [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, the backup will wait until the last required WAL
+          If set to true, the backup will wait until the last required WAL
           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.
+          not enabled. If false, the backup will neither wait nor warn,
+          leaving the client responsible for ensuring the required log is
+          available. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2605,25 +2607,25 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>TABLESPACE_MAP</literal></term>
+        <term><literal>TABLESPACE_MAP [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include information about symbolic links present in the directory
-          <filename>pg_tblspc</filename> in a file named
+          If true, include information about symbolic links present in the
+          directory <filename>pg_tblspc</filename> in a file named
           <filename>tablespace_map</filename>. The tablespace map file includes
           each symbolic link name as it exists in the directory
           <filename>pg_tblspc/</filename> and the full path of that symbolic link.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOVERIFY_CHECKSUMS</literal></term>
+        <term><literal>VERIFY_CHECKSUMS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, checksums are verified during a base backup if they are
-          enabled. Specifying <literal>NOVERIFY_CHECKSUMS</literal> disables
-          this verification.
+          If true, checksums are verified during a base backup if they are
+          enabled. If false, this is skipped. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2708,6 +2710,7 @@ The commands accepted in replication mode are:
        </varlistentry>
       </variablelist>
      </para>
+
      <para>
       After the second regular result set, one or more CopyOutResponse results
       will be sent, one for the main data directory and one for each additional tablespace other
@@ -2788,6 +2791,17 @@ The commands accepted in replication mode are:
      </para>
     </listitem>
   </varlistentry>
+
+  <varlistentry>
+    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>BASE_BACKUP</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
 </variablelist>
 
 </para>
diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index e09108d0ec..b0b52d3b1a 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -787,7 +788,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -796,7 +797,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
 		else if (strcmp(defel->defname, "fast") == 0)
@@ -805,16 +806,16 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
+			opt->fastcheckpoint = defGetBoolean(defel);
 			o_fast = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -823,19 +824,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -851,21 +852,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -890,7 +891,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -905,8 +906,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index e1e8ec29cc..ce51a5e322 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -95,13 +95,13 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
 %type <list>	create_slot_opt_list
 %type <defelt>	create_slot_opt
@@ -157,12 +157,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -170,14 +182,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -200,8 +212,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -215,8 +227,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -422,6 +434,65 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_TWO_PHASE					{ $$ = "two_phase"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 7296eb97d0..0c8be22558 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1809,10 +1809,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1821,8 +1817,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	PQExpBufferData buf;
 
 	Assert(conn != NULL);
+	initPQExpBuffer(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1840,6 +1839,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1500)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1870,20 +1871,42 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendStringCommandOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "FAST");
+	if (includewal != NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendIntegerCommandOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendStringCommandOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1898,18 +1921,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index f5b3b476e5..d782b81adc 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -603,6 +603,67 @@ DropReplicationSlot(PGconn *conn, const char *slot_name)
 	return true;
 }
 
+/*
+ * Append a "plain" option - one with no value - to a server command that
+ * is being constructed.
+ *
+ * In the old syntax, all options were parser keywords, so you could just
+ * write things like SOME_COMMAND OPTION1 OPTION2 'opt2value' OPTION3 42. The
+ * new syntax uses a comma-separated list surrounded by parentheses, so the
+ * equivalent is SOME_COMMAND (OPTION1, OPTION2 'optvalue', OPTION3 42).
+ */
+void
+AppendPlainCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						 char *option_name)
+{
+	if (buf->len > 0 && buf->data[buf->len - 1] != '(')
+	{
+		if (use_new_option_syntax)
+			appendPQExpBufferStr(buf, ", ");
+		else
+			appendPQExpBufferChar(buf, ' ');
+	}
+
+	appendPQExpBuffer(buf, " %s", option_name);
+}
+
+/*
+ * Append an option with an associated string value to a server command that
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendStringCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						  char *option_name, char *option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t		length = strlen(option_value);
+		char	   *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendPQExpBuffer(buf, " '%s'", escaped_value);
+		pfree(escaped_value);
+	}
+}
+
+/*
+ * Append an option with an associated integer value to a server command
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendIntegerCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						   char *option_name, int32 option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	appendPQExpBuffer(buf, " %d", option_value);
+}
 
 /*
  * Frontend version of GetCurrentTimestamp(), since we are not linked with
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 504803b976..65135c79e0 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -15,6 +15,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "libpq-fe.h"
+#include "pqexpbuffer.h"
 
 extern const char *progname;
 extern char *connection_string;
@@ -40,6 +41,17 @@ extern bool RunIdentifySystem(PGconn *conn, char **sysid,
 							  TimeLineID *starttli,
 							  XLogRecPtr *startpos,
 							  char **db_name);
+
+extern void AppendPlainCommandOption(PQExpBuffer buf,
+									 bool use_new_option_syntax,
+									 char *option_value);
+extern void AppendStringCommandOption(PQExpBuffer buf,
+									  bool use_new_option_syntax,
+									  char *option_name, char *option_value);
+extern void AppendIntegerCommandOption(PQExpBuffer buf,
+									   bool use_new_option_syntax,
+									   char *option_name, int32 option_value);
+
 extern bool RetrieveWalSegSize(PGconn *conn);
 extern TimestampTz feGetCurrentTimestamp(void);
 extern void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
-- 
2.24.3 (Apple Git-128)

From f2b38f680683bc5317c7ba7140f77e8efcb9aa43 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Fri, 10 Sep 2021 11:50:05 -0400
Subject: [PATCH v5 2/2] Flexible options for CREATE_REPLICATION_SLOT.

Like BASE_BACKUP, CREATE_REPLICATION_SLOT has historically used a
hard-coded syntax.  To improve future extensibility, adopt a flexible
options syntax here, too.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_receivewal and
pg_recvlogical use it.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml                    | 37 ++++++++++++-----
 .../libpqwalreceiver/libpqwalreceiver.c       | 16 ++++----
 src/backend/replication/repl_gram.y           | 35 +++++++++-------
 src/backend/replication/walsender.c           | 40 ++++++++++--------
 src/bin/pg_basebackup/streamutil.c            | 41 +++++++++++++++----
 5 files changed, 110 insertions(+), 59 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 32d1eeabdc..31bf5a7ffd 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1914,7 +1914,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-create-slot" xreflabel="CREATE_REPLICATION_SLOT">
-   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> | <literal>LOGICAL</literal> } [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
      <indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
@@ -1954,46 +1954,50 @@ The commands accepted in replication mode are:
         </para>
        </listitem>
       </varlistentry>
+     </variablelist>
+
+     <para>The following options are supported:</para>
 
+     <variablelist>
       <varlistentry>
-       <term><literal>TWO_PHASE</literal></term>
+       <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this logical replication slot supports decoding of two-phase
+         If true, this logical replication slot supports decoding of two-phase
          transactions. With this option, two-phase commands like
          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
          The transaction will be decoded and transmitted at
          <literal>PREPARE TRANSACTION</literal> time.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>RESERVE_WAL</literal></term>
+       <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this physical replication slot reserves <acronym>WAL</acronym>
+         If true, this physical replication slot reserves <acronym>WAL</acronym>
          immediately.  Otherwise, <acronym>WAL</acronym> is only reserved upon
          connection from a streaming replication client.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>EXPORT_SNAPSHOT</literal></term>
-       <term><literal>NOEXPORT_SNAPSHOT</literal></term>
-       <term><literal>USE_SNAPSHOT</literal></term>
+       <term><literal>SNAPSHOT { 'export' | 'use' | 'nothing' }</literal></term>
        <listitem>
         <para>
          Decides what to do with the snapshot created during logical slot
-         initialization. <literal>EXPORT_SNAPSHOT</literal>, which is the default,
+         initialization. <literal>'export'</literal>, which is the default,
          will export the snapshot for use in other sessions. This option can't
-         be used inside a transaction.  <literal>USE_SNAPSHOT</literal> will use the
+         be used inside a transaction.  <literal>'use'</literal> will use the
          snapshot for the current transaction executing the command. This
          option must be used in a transaction, and
          <literal>CREATE_REPLICATION_SLOT</literal> must be the first command
-         run in that transaction.  Finally, <literal>NOEXPORT_SNAPSHOT</literal> will
+         run in that transaction.  Finally, <literal>'nothing'</literal> will
          just use the snapshot for logical decoding as normal but won't do
          anything else with it.
         </para>
@@ -2052,6 +2056,17 @@ The commands accepted in replication mode are:
     </listitem>
   </varlistentry>
 
+  <varlistentry>
+    <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>CREATE_REPLICATION_SLOT</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
+
   <varlistentry>
     <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ]
      <indexterm><primary>START_REPLICATION</primary></indexterm>
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 19ea159af4..e3a783ebec 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -872,26 +872,28 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 
 	if (conn->logical)
 	{
-		appendStringInfoString(&cmd, " LOGICAL pgoutput");
-		if (two_phase)
-			appendStringInfoString(&cmd, " TWO_PHASE");
+		appendStringInfoString(&cmd, " LOGICAL pgoutput (");
 
 		switch (snapshot_action)
 		{
 			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, "SNAPSHOT 'export'");
 				break;
 			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
 				break;
 			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
+				appendStringInfoString(&cmd, "SNAPSHOT 'use'");
 				break;
 		}
+
+		if (two_phase)
+			appendStringInfoString(&cmd, ", TWO_PHASE");
+		appendStringInfoChar(&cmd, ')');
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index ce51a5e322..e5f66610c3 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -103,8 +103,8 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	plugin_opt_arg
 %type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -243,8 +243,8 @@ base_backup_legacy_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -254,8 +254,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -268,28 +268,33 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("export"), -1);
 				}
 			| K_NOEXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(false), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("nothing"), -1);
 				}
 			| K_USE_SNAPSHOT
 				{
-				  $$ = makeDefElem("use_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("use"), -1);
 				}
 			| K_RESERVE_WAL
 				{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3ca2a11389..b811a5c0ef 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -872,26 +872,30 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 	{
 		DefElem    *defel = (DefElem *) lfirst(lc);
 
-		if (strcmp(defel->defname, "export_snapshot") == 0)
+		if (strcmp(defel->defname, "snapshot") == 0)
 		{
+			char	   *action;
+
 			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 
+			action = defGetString(defel);
 			snapshot_action_given = true;
-			*snapshot_action = defGetBoolean(defel) ? CRS_EXPORT_SNAPSHOT :
-				CRS_NOEXPORT_SNAPSHOT;
-		}
-		else if (strcmp(defel->defname, "use_snapshot") == 0)
-		{
-			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
+
+			if (strcmp(action, "export") == 0)
+				*snapshot_action = CRS_EXPORT_SNAPSHOT;
+			else if (strcmp(action, "nothing") == 0)
+				*snapshot_action = CRS_NOEXPORT_SNAPSHOT;
+			else if (strcmp(action, "use") == 0)
+				*snapshot_action = CRS_USE_SNAPSHOT;
+			else
 				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						 errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
+								defel->defname, action)));
 
-			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -901,7 +905,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
@@ -910,7 +914,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 			two_phase_given = true;
-			*two_phase = true;
+			*two_phase = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -980,7 +984,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));
 
 			need_full_snapshot = true;
 		}
@@ -990,25 +994,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index d782b81adc..72fda9f1d0 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -490,6 +490,7 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 {
 	PQExpBuffer query;
 	PGresult   *res;
+	bool		use_new_option_syntax = (PQserverVersion(conn) >= 150000);
 
 	query = createPQExpBuffer();
 
@@ -498,27 +499,51 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	Assert(!(two_phase && is_physical));
 	Assert(slot_name != NULL);
 
-	/* Build query */
+	/* Build base portion of query */
 	appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	if (is_temporary)
 		appendPQExpBufferStr(query, " TEMPORARY");
 	if (is_physical)
-	{
 		appendPQExpBufferStr(query, " PHYSICAL");
+	else
+		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
+
+	/* Add any requested options */
+	if (use_new_option_syntax)
+		appendPQExpBufferStr(query, " (");
+	if (is_physical)
+	{
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "RESERVE_WAL");
 	}
 	else
 	{
-		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (two_phase && PQserverVersion(conn) >= 150000)
-			appendPQExpBufferStr(query, " TWO_PHASE");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "TWO_PHASE");
 
-		if (PQserverVersion(conn) >= 100000)
-			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+		/* pg_recvlogical doesn't use an exported snapshot, so suppress */
+		if (use_new_option_syntax)
+			AppendStringCommandOption(query, use_new_option_syntax,
+									   "SNAPSHOT", "nothing");
+		else
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "NOEXPORT_SNAPSHOT");
+	}
+	if (use_new_option_syntax)
+	{
+		/* Suppress option list if it would be empty, otherwise terminate */
+		if (query->data[query->len - 1] == '(')
+		{
+			query->len -= 2;
+			query->data[query->len] = '\0';
+		}
+		else
+			appendPQExpBufferChar(query, ')');
 	}
 
+	/* Now run the query */
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-- 
2.24.3 (Apple Git-128)

From 3172200ac7ff934f8add6e46fb2a5fe2e741d957 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 21 Sep 2021 12:22:07 -0400
Subject: [PATCH v6 1/2] Flexible options for BASE_BACKUP.

Previously, BASE_BACKUP used an entirely hard-coded syntax, but that's
hard to extend. Instead, adopt the same kind of syntax we've used for
SQL commands such as VACUUM, ANALYZE, COPY, and EXPLAIN, where it's
not necessary for all of the option names to be parser keywords.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_basebackup prefer
the new syntax when the server is new enough to support it.

Patch by me, reviewed by Fabien Coelho and Sergei Kornilov.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml            | 68 +++++++++++--------
 src/backend/replication/basebackup.c  | 51 ++++++++------
 src/backend/replication/repl_gram.y   | 97 +++++++++++++++++++++++----
 src/bin/pg_basebackup/pg_basebackup.c | 71 +++++++++++++-------
 src/bin/pg_basebackup/streamutil.c    | 61 +++++++++++++++++
 src/bin/pg_basebackup/streamutil.h    | 12 ++++
 6 files changed, 276 insertions(+), 84 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a232546b1d..a5c07bfefd 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2517,8 +2517,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-base-backup" xreflabel="BASE_BACKUP">
-    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
-     <indexterm><primary>BASE_BACKUP</primary></indexterm>
+    <term><literal>BASE_BACKUP</literal> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
     </term>
     <listitem>
      <para>
@@ -2540,52 +2539,55 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>PROGRESS</literal></term>
+        <term><literal>PROGRESS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Request information required to generate a progress report. This will
-          send back an approximate size in the header of each tablespace, which
-          can be used to calculate how far along the stream is done. This is
-          calculated by enumerating all the file sizes once before the transfer
-          is even started, and might as such have a negative impact on the
-          performance.  In particular, it might take longer before the first data
+          If set to true, request information required to generate a progress
+          report. This will send back an approximate size in the header of each
+          tablespace, which can be used to calculate how far along the stream
+          is done. This is calculated by enumerating all the file sizes once
+          before the transfer is even started, and might as such have a
+          negative impact on the performance.  In particular, it might take
+          longer before the first data
           is streamed. Since the database files can change during the backup,
           the size is only approximate and might both grow and shrink between
           the time of approximation and the sending of the actual files.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>FAST</literal></term>
+        <term><literal>CHECKPOINT { 'fast' | 'spread' }</replaceable></literal></term>
         <listitem>
          <para>
-          Request a fast checkpoint.
+          Sets the type of checkpoint to be performed at the beginning of the
+          base backup. The default is <literal>spread</literal>.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>WAL</literal></term>
+        <term><literal>WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include the necessary WAL segments in the backup. This will include
-          all the files between start and stop backup in the
+          If set to true, include the necessary WAL segments in the backup.
+          This will include all the files between start and stop backup in the
           <filename>pg_wal</filename> directory of the base directory tar
-          file.
+          file. The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOWAIT</literal></term>
+        <term><literal>WAIT [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, the backup will wait until the last required WAL
+          If set to true, the backup will wait until the last required WAL
           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.
+          not enabled. If false, the backup will neither wait nor warn,
+          leaving the client responsible for ensuring the required log is
+          available. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2605,25 +2607,25 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>TABLESPACE_MAP</literal></term>
+        <term><literal>TABLESPACE_MAP [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include information about symbolic links present in the directory
-          <filename>pg_tblspc</filename> in a file named
+          If true, include information about symbolic links present in the
+          directory <filename>pg_tblspc</filename> in a file named
           <filename>tablespace_map</filename>. The tablespace map file includes
           each symbolic link name as it exists in the directory
           <filename>pg_tblspc/</filename> and the full path of that symbolic link.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOVERIFY_CHECKSUMS</literal></term>
+        <term><literal>VERIFY_CHECKSUMS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, checksums are verified during a base backup if they are
-          enabled. Specifying <literal>NOVERIFY_CHECKSUMS</literal> disables
-          this verification.
+          If true, checksums are verified during a base backup if they are
+          enabled. If false, this is skipped. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2708,6 +2710,7 @@ The commands accepted in replication mode are:
        </varlistentry>
       </variablelist>
      </para>
+
      <para>
       After the second regular result set, one or more CopyOutResponse results
       will be sent, one for the main data directory and one for each additional tablespace other
@@ -2788,6 +2791,17 @@ The commands accepted in replication mode are:
      </para>
     </listitem>
   </varlistentry>
+
+  <varlistentry>
+    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>BASE_BACKUP</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
 </variablelist>
 
 </para>
diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index e09108d0ec..4c97ab7b5a 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -764,7 +765,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 	ListCell   *lopt;
 	bool		o_label = false;
 	bool		o_progress = false;
-	bool		o_fast = false;
+	bool		o_checkpoint = false;
 	bool		o_nowait = false;
 	bool		o_wal = false;
 	bool		o_maxrate = false;
@@ -787,7 +788,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -796,25 +797,35 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
-		else if (strcmp(defel->defname, "fast") == 0)
+		else if (strcmp(defel->defname, "checkpoint") == 0)
 		{
-			if (o_fast)
+			char	   *optval = defGetString(defel);
+
+			if (o_checkpoint)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
-			o_fast = true;
+			if (pg_strcasecmp(optval, "fast") == 0)
+				opt->fastcheckpoint = true;
+			else if (pg_strcasecmp(optval, "spread") == 0)
+				opt->fastcheckpoint = false;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("unrecognized checkpoint type: \"%s\"",
+								optval)));
+			o_checkpoint = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -823,19 +834,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -851,21 +862,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -890,7 +901,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -905,8 +916,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index e1e8ec29cc..3b59d62ed8 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -95,13 +95,13 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
 %type <list>	create_slot_opt_list
 %type <defelt>	create_slot_opt
@@ -157,12 +157,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -170,14 +182,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -190,8 +202,8 @@ base_backup_opt:
 				}
 			| K_FAST
 				{
-				  $$ = makeDefElem("fast",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("checkpoint",
+								   (Node *)makeString("fast"), -1);
 				}
 			| K_WAL
 				{
@@ -200,8 +212,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -215,8 +227,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -422,6 +434,65 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_TWO_PHASE					{ $$ = "two_phase"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 669aa207a3..27ee6394cf 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1804,10 +1804,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1816,8 +1812,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	PQExpBufferData buf;
 
 	Assert(conn != NULL);
+	initPQExpBuffer(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1835,6 +1834,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1500)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1865,20 +1866,48 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendStringCommandOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+	{
+		if (use_new_option_syntax)
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+									  "CHECKPOINT", "fast");
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "FAST");
+	}
+	if (includewal != NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendIntegerCommandOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendStringCommandOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1893,18 +1922,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index f5b3b476e5..d782b81adc 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -603,6 +603,67 @@ DropReplicationSlot(PGconn *conn, const char *slot_name)
 	return true;
 }
 
+/*
+ * Append a "plain" option - one with no value - to a server command that
+ * is being constructed.
+ *
+ * In the old syntax, all options were parser keywords, so you could just
+ * write things like SOME_COMMAND OPTION1 OPTION2 'opt2value' OPTION3 42. The
+ * new syntax uses a comma-separated list surrounded by parentheses, so the
+ * equivalent is SOME_COMMAND (OPTION1, OPTION2 'optvalue', OPTION3 42).
+ */
+void
+AppendPlainCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						 char *option_name)
+{
+	if (buf->len > 0 && buf->data[buf->len - 1] != '(')
+	{
+		if (use_new_option_syntax)
+			appendPQExpBufferStr(buf, ", ");
+		else
+			appendPQExpBufferChar(buf, ' ');
+	}
+
+	appendPQExpBuffer(buf, " %s", option_name);
+}
+
+/*
+ * Append an option with an associated string value to a server command that
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendStringCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						  char *option_name, char *option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t		length = strlen(option_value);
+		char	   *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendPQExpBuffer(buf, " '%s'", escaped_value);
+		pfree(escaped_value);
+	}
+}
+
+/*
+ * Append an option with an associated integer value to a server command
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendIntegerCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						   char *option_name, int32 option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	appendPQExpBuffer(buf, " %d", option_value);
+}
 
 /*
  * Frontend version of GetCurrentTimestamp(), since we are not linked with
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 504803b976..65135c79e0 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -15,6 +15,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "libpq-fe.h"
+#include "pqexpbuffer.h"
 
 extern const char *progname;
 extern char *connection_string;
@@ -40,6 +41,17 @@ extern bool RunIdentifySystem(PGconn *conn, char **sysid,
 							  TimeLineID *starttli,
 							  XLogRecPtr *startpos,
 							  char **db_name);
+
+extern void AppendPlainCommandOption(PQExpBuffer buf,
+									 bool use_new_option_syntax,
+									 char *option_value);
+extern void AppendStringCommandOption(PQExpBuffer buf,
+									  bool use_new_option_syntax,
+									  char *option_name, char *option_value);
+extern void AppendIntegerCommandOption(PQExpBuffer buf,
+									   bool use_new_option_syntax,
+									   char *option_name, int32 option_value);
+
 extern bool RetrieveWalSegSize(PGconn *conn);
 extern TimestampTz feGetCurrentTimestamp(void);
 extern void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
-- 
2.24.3 (Apple Git-128)

From e2f3eb25bf15e3f4de677cd1bbcb98b0cd220ce4 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 21 Sep 2021 12:22:15 -0400
Subject: [PATCH v6 2/2] Flexible options for CREATE_REPLICATION_SLOT.

Like BASE_BACKUP, CREATE_REPLICATION_SLOT has historically used a
hard-coded syntax.  To improve future extensibility, adopt a flexible
options syntax here, too.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_receivewal and
pg_recvlogical use it.

Patch by me, reviewed by Fabien Coelho and Sergei Kornilov.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml                    | 37 ++++++++++++-----
 .../libpqwalreceiver/libpqwalreceiver.c       | 16 ++++----
 src/backend/replication/repl_gram.y           | 35 +++++++++-------
 src/backend/replication/walsender.c           | 40 ++++++++++---------
 src/bin/pg_basebackup/streamutil.c            | 40 ++++++++++++++++---
 5 files changed, 111 insertions(+), 57 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a5c07bfefd..aed06b968e 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1914,7 +1914,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-create-slot" xreflabel="CREATE_REPLICATION_SLOT">
-   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> | <literal>LOGICAL</literal> } [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
      <indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
@@ -1954,46 +1954,50 @@ The commands accepted in replication mode are:
         </para>
        </listitem>
       </varlistentry>
+     </variablelist>
+
+     <para>The following options are supported:</para>
 
+     <variablelist>
       <varlistentry>
-       <term><literal>TWO_PHASE</literal></term>
+       <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this logical replication slot supports decoding of two-phase
+         If true, this logical replication slot supports decoding of two-phase
          transactions. With this option, two-phase commands like
          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
          The transaction will be decoded and transmitted at
          <literal>PREPARE TRANSACTION</literal> time.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>RESERVE_WAL</literal></term>
+       <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this physical replication slot reserves <acronym>WAL</acronym>
+         If true, this physical replication slot reserves <acronym>WAL</acronym>
          immediately.  Otherwise, <acronym>WAL</acronym> is only reserved upon
          connection from a streaming replication client.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>EXPORT_SNAPSHOT</literal></term>
-       <term><literal>NOEXPORT_SNAPSHOT</literal></term>
-       <term><literal>USE_SNAPSHOT</literal></term>
+       <term><literal>SNAPSHOT { 'export' | 'use' | 'nothing' }</literal></term>
        <listitem>
         <para>
          Decides what to do with the snapshot created during logical slot
-         initialization. <literal>EXPORT_SNAPSHOT</literal>, which is the default,
+         initialization. <literal>'export'</literal>, which is the default,
          will export the snapshot for use in other sessions. This option can't
-         be used inside a transaction.  <literal>USE_SNAPSHOT</literal> will use the
+         be used inside a transaction.  <literal>'use'</literal> will use the
          snapshot for the current transaction executing the command. This
          option must be used in a transaction, and
          <literal>CREATE_REPLICATION_SLOT</literal> must be the first command
-         run in that transaction.  Finally, <literal>NOEXPORT_SNAPSHOT</literal> will
+         run in that transaction.  Finally, <literal>'nothing'</literal> will
          just use the snapshot for logical decoding as normal but won't do
          anything else with it.
         </para>
@@ -2052,6 +2056,17 @@ The commands accepted in replication mode are:
     </listitem>
   </varlistentry>
 
+  <varlistentry>
+    <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>CREATE_REPLICATION_SLOT</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
+
   <varlistentry>
     <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ]
      <indexterm><primary>START_REPLICATION</primary></indexterm>
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 19ea159af4..e3a783ebec 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -872,26 +872,28 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 
 	if (conn->logical)
 	{
-		appendStringInfoString(&cmd, " LOGICAL pgoutput");
-		if (two_phase)
-			appendStringInfoString(&cmd, " TWO_PHASE");
+		appendStringInfoString(&cmd, " LOGICAL pgoutput (");
 
 		switch (snapshot_action)
 		{
 			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, "SNAPSHOT 'export'");
 				break;
 			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
+				appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
 				break;
 			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
+				appendStringInfoString(&cmd, "SNAPSHOT 'use'");
 				break;
 		}
+
+		if (two_phase)
+			appendStringInfoString(&cmd, ", TWO_PHASE");
+		appendStringInfoChar(&cmd, ')');
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 3b59d62ed8..126380e2df 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -103,8 +103,8 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	plugin_opt_arg
 %type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -243,8 +243,8 @@ base_backup_legacy_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -254,8 +254,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -268,28 +268,33 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("export"), -1);
 				}
 			| K_NOEXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(false), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("nothing"), -1);
 				}
 			| K_USE_SNAPSHOT
 				{
-				  $$ = makeDefElem("use_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("use"), -1);
 				}
 			| K_RESERVE_WAL
 				{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3ca2a11389..b811a5c0ef 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -872,26 +872,30 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 	{
 		DefElem    *defel = (DefElem *) lfirst(lc);
 
-		if (strcmp(defel->defname, "export_snapshot") == 0)
+		if (strcmp(defel->defname, "snapshot") == 0)
 		{
+			char	   *action;
+
 			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 
+			action = defGetString(defel);
 			snapshot_action_given = true;
-			*snapshot_action = defGetBoolean(defel) ? CRS_EXPORT_SNAPSHOT :
-				CRS_NOEXPORT_SNAPSHOT;
-		}
-		else if (strcmp(defel->defname, "use_snapshot") == 0)
-		{
-			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
+
+			if (strcmp(action, "export") == 0)
+				*snapshot_action = CRS_EXPORT_SNAPSHOT;
+			else if (strcmp(action, "nothing") == 0)
+				*snapshot_action = CRS_NOEXPORT_SNAPSHOT;
+			else if (strcmp(action, "use") == 0)
+				*snapshot_action = CRS_USE_SNAPSHOT;
+			else
 				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						 errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
+								defel->defname, action)));
 
-			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -901,7 +905,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
@@ -910,7 +914,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 			two_phase_given = true;
-			*two_phase = true;
+			*two_phase = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -980,7 +984,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));
 
 			need_full_snapshot = true;
 		}
@@ -990,25 +994,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index d782b81adc..37237cd5d9 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -490,6 +490,7 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 {
 	PQExpBuffer query;
 	PGresult   *res;
+	bool		use_new_option_syntax = (PQserverVersion(conn) >= 150000);
 
 	query = createPQExpBuffer();
 
@@ -498,27 +499,54 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	Assert(!(two_phase && is_physical));
 	Assert(slot_name != NULL);
 
-	/* Build query */
+	/* Build base portion of query */
 	appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	if (is_temporary)
 		appendPQExpBufferStr(query, " TEMPORARY");
 	if (is_physical)
-	{
 		appendPQExpBufferStr(query, " PHYSICAL");
+	else
+		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
+
+	/* Add any requested options */
+	if (use_new_option_syntax)
+		appendPQExpBufferStr(query, " (");
+	if (is_physical)
+	{
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "RESERVE_WAL");
 	}
 	else
 	{
-		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (two_phase && PQserverVersion(conn) >= 150000)
-			appendPQExpBufferStr(query, " TWO_PHASE");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "TWO_PHASE");
 
 		if (PQserverVersion(conn) >= 100000)
+		{
 			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+			if (use_new_option_syntax)
+				AppendStringCommandOption(query, use_new_option_syntax,
+										   "SNAPSHOT", "nothing");
+			else
+				AppendPlainCommandOption(query, use_new_option_syntax,
+										 "NOEXPORT_SNAPSHOT");
+		}
+	}
+	if (use_new_option_syntax)
+	{
+		/* Suppress option list if it would be empty, otherwise terminate */
+		if (query->data[query->len - 1] == '(')
+		{
+			query->len -= 2;
+			query->data[query->len] = '\0';
+		}
+		else
+			appendPQExpBufferChar(query, ')');
 	}
 
+	/* Now run the query */
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-- 
2.24.3 (Apple Git-128)

From 2a729aa211e608943eb8ec331777123768362f7e Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 21 Sep 2021 12:22:07 -0400
Subject: [PATCH v7 1/2] Flexible options for BASE_BACKUP.

Previously, BASE_BACKUP used an entirely hard-coded syntax, but that's
hard to extend. Instead, adopt the same kind of syntax we've used for
SQL commands such as VACUUM, ANALYZE, COPY, and EXPLAIN, where it's
not necessary for all of the option names to be parser keywords.

In the new syntax, most of the options now take an optional Boolean
argument. To match our practice in other in places, the options which
the old syntax called NOWAIT and NOVERIFY_CHECKSUMS options are in the
new syntax called WAIT and VERIFY_CHECKUMS, and the default value is
false. The old FAST option in the new syntax has been replaced by a
CHECKSUM option whose value may be 'fast' or 'spread'.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_basebackup prefer
the new syntax when the server is new enough to support it.

Patch by me, reviewed by Fabien Coelho and Sergei Kornilov.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml            | 68 +++++++++++--------
 src/backend/replication/basebackup.c  | 51 ++++++++------
 src/backend/replication/repl_gram.y   | 97 +++++++++++++++++++++++----
 src/bin/pg_basebackup/pg_basebackup.c | 71 +++++++++++++-------
 src/bin/pg_basebackup/streamutil.c    | 61 +++++++++++++++++
 src/bin/pg_basebackup/streamutil.h    | 12 ++++
 6 files changed, 276 insertions(+), 84 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a232546b1d..e88998cd27 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2517,8 +2517,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-base-backup" xreflabel="BASE_BACKUP">
-    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
-     <indexterm><primary>BASE_BACKUP</primary></indexterm>
+    <term><literal>BASE_BACKUP</literal> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
     </term>
     <listitem>
      <para>
@@ -2540,52 +2539,55 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>PROGRESS</literal></term>
+        <term><literal>PROGRESS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Request information required to generate a progress report. This will
-          send back an approximate size in the header of each tablespace, which
-          can be used to calculate how far along the stream is done. This is
-          calculated by enumerating all the file sizes once before the transfer
-          is even started, and might as such have a negative impact on the
-          performance.  In particular, it might take longer before the first data
+          If set to true, request information required to generate a progress
+          report. This will send back an approximate size in the header of each
+          tablespace, which can be used to calculate how far along the stream
+          is done. This is calculated by enumerating all the file sizes once
+          before the transfer is even started, and might as such have a
+          negative impact on the performance.  In particular, it might take
+          longer before the first data
           is streamed. Since the database files can change during the backup,
           the size is only approximate and might both grow and shrink between
           the time of approximation and the sending of the actual files.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>FAST</literal></term>
+        <term><literal>CHECKPOINT { 'fast' | 'spread' }</literal></term>
         <listitem>
          <para>
-          Request a fast checkpoint.
+          Sets the type of checkpoint to be performed at the beginning of the
+          base backup. The default is <literal>spread</literal>.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>WAL</literal></term>
+        <term><literal>WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include the necessary WAL segments in the backup. This will include
-          all the files between start and stop backup in the
+          If set to true, include the necessary WAL segments in the backup.
+          This will include all the files between start and stop backup in the
           <filename>pg_wal</filename> directory of the base directory tar
-          file.
+          file. The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOWAIT</literal></term>
+        <term><literal>WAIT [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, the backup will wait until the last required WAL
+          If set to true, the backup will wait until the last required WAL
           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.
+          not enabled. If false, the backup will neither wait nor warn,
+          leaving the client responsible for ensuring the required log is
+          available. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2605,25 +2607,25 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>TABLESPACE_MAP</literal></term>
+        <term><literal>TABLESPACE_MAP [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include information about symbolic links present in the directory
-          <filename>pg_tblspc</filename> in a file named
+          If true, include information about symbolic links present in the
+          directory <filename>pg_tblspc</filename> in a file named
           <filename>tablespace_map</filename>. The tablespace map file includes
           each symbolic link name as it exists in the directory
           <filename>pg_tblspc/</filename> and the full path of that symbolic link.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOVERIFY_CHECKSUMS</literal></term>
+        <term><literal>VERIFY_CHECKSUMS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, checksums are verified during a base backup if they are
-          enabled. Specifying <literal>NOVERIFY_CHECKSUMS</literal> disables
-          this verification.
+          If true, checksums are verified during a base backup if they are
+          enabled. If false, this is skipped. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2708,6 +2710,7 @@ The commands accepted in replication mode are:
        </varlistentry>
       </variablelist>
      </para>
+
      <para>
       After the second regular result set, one or more CopyOutResponse results
       will be sent, one for the main data directory and one for each additional tablespace other
@@ -2788,6 +2791,17 @@ The commands accepted in replication mode are:
      </para>
     </listitem>
   </varlistentry>
+
+  <varlistentry>
+    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>BASE_BACKUP</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
 </variablelist>
 
 </para>
diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index e09108d0ec..4c97ab7b5a 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -764,7 +765,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 	ListCell   *lopt;
 	bool		o_label = false;
 	bool		o_progress = false;
-	bool		o_fast = false;
+	bool		o_checkpoint = false;
 	bool		o_nowait = false;
 	bool		o_wal = false;
 	bool		o_maxrate = false;
@@ -787,7 +788,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -796,25 +797,35 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
-		else if (strcmp(defel->defname, "fast") == 0)
+		else if (strcmp(defel->defname, "checkpoint") == 0)
 		{
-			if (o_fast)
+			char	   *optval = defGetString(defel);
+
+			if (o_checkpoint)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
-			o_fast = true;
+			if (pg_strcasecmp(optval, "fast") == 0)
+				opt->fastcheckpoint = true;
+			else if (pg_strcasecmp(optval, "spread") == 0)
+				opt->fastcheckpoint = false;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("unrecognized checkpoint type: \"%s\"",
+								optval)));
+			o_checkpoint = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -823,19 +834,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -851,21 +862,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -890,7 +901,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -905,8 +916,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index e1e8ec29cc..3b59d62ed8 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -95,13 +95,13 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
 %type <list>	create_slot_opt_list
 %type <defelt>	create_slot_opt
@@ -157,12 +157,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -170,14 +182,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -190,8 +202,8 @@ base_backup_opt:
 				}
 			| K_FAST
 				{
-				  $$ = makeDefElem("fast",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("checkpoint",
+								   (Node *)makeString("fast"), -1);
 				}
 			| K_WAL
 				{
@@ -200,8 +212,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -215,8 +227,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -422,6 +434,65 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_TWO_PHASE					{ $$ = "two_phase"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 669aa207a3..27ee6394cf 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1804,10 +1804,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1816,8 +1812,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	PQExpBufferData buf;
 
 	Assert(conn != NULL);
+	initPQExpBuffer(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1835,6 +1834,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1500)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1865,20 +1866,48 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendStringCommandOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+	{
+		if (use_new_option_syntax)
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+									  "CHECKPOINT", "fast");
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "FAST");
+	}
+	if (includewal != NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendIntegerCommandOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendStringCommandOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1893,18 +1922,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index f5b3b476e5..d782b81adc 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -603,6 +603,67 @@ DropReplicationSlot(PGconn *conn, const char *slot_name)
 	return true;
 }
 
+/*
+ * Append a "plain" option - one with no value - to a server command that
+ * is being constructed.
+ *
+ * In the old syntax, all options were parser keywords, so you could just
+ * write things like SOME_COMMAND OPTION1 OPTION2 'opt2value' OPTION3 42. The
+ * new syntax uses a comma-separated list surrounded by parentheses, so the
+ * equivalent is SOME_COMMAND (OPTION1, OPTION2 'optvalue', OPTION3 42).
+ */
+void
+AppendPlainCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						 char *option_name)
+{
+	if (buf->len > 0 && buf->data[buf->len - 1] != '(')
+	{
+		if (use_new_option_syntax)
+			appendPQExpBufferStr(buf, ", ");
+		else
+			appendPQExpBufferChar(buf, ' ');
+	}
+
+	appendPQExpBuffer(buf, " %s", option_name);
+}
+
+/*
+ * Append an option with an associated string value to a server command that
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendStringCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						  char *option_name, char *option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t		length = strlen(option_value);
+		char	   *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendPQExpBuffer(buf, " '%s'", escaped_value);
+		pfree(escaped_value);
+	}
+}
+
+/*
+ * Append an option with an associated integer value to a server command
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendIntegerCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						   char *option_name, int32 option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	appendPQExpBuffer(buf, " %d", option_value);
+}
 
 /*
  * Frontend version of GetCurrentTimestamp(), since we are not linked with
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 504803b976..65135c79e0 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -15,6 +15,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "libpq-fe.h"
+#include "pqexpbuffer.h"
 
 extern const char *progname;
 extern char *connection_string;
@@ -40,6 +41,17 @@ extern bool RunIdentifySystem(PGconn *conn, char **sysid,
 							  TimeLineID *starttli,
 							  XLogRecPtr *startpos,
 							  char **db_name);
+
+extern void AppendPlainCommandOption(PQExpBuffer buf,
+									 bool use_new_option_syntax,
+									 char *option_value);
+extern void AppendStringCommandOption(PQExpBuffer buf,
+									  bool use_new_option_syntax,
+									  char *option_name, char *option_value);
+extern void AppendIntegerCommandOption(PQExpBuffer buf,
+									   bool use_new_option_syntax,
+									   char *option_name, int32 option_value);
+
 extern bool RetrieveWalSegSize(PGconn *conn);
 extern TimestampTz feGetCurrentTimestamp(void);
 extern void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
-- 
2.24.3 (Apple Git-128)

From 8824b607b67458cc389b87c5e6c341f8471fb5b4 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Wed, 22 Sep 2021 15:52:23 -0400
Subject: [PATCH v7 2/2] Flexible options for CREATE_REPLICATION_SLOT.

Like BASE_BACKUP, CREATE_REPLICATION_SLOT has historically used a
hard-coded syntax.  To improve future extensibility, adopt a flexible
options syntax here, too.

In the new syntax, instead of three mutually exclusive options
EXPORT_SNAPSHOT, USE_SNAPSHOT, and NOEXPORT_SNAPSHOT, there is now a single
SNAPSHOT option with three possible values: 'export', 'use', and 'nothing'.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, makes pg_receivewal,
pg_recvlogical, and walreceiver processes use it.

Patch by me, reviewed by Fabien Coelho and Sergei Kornilov.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml                    | 37 +++++++++----
 .../libpqwalreceiver/libpqwalreceiver.c       | 55 ++++++++++++++-----
 src/backend/replication/repl_gram.y           | 35 +++++++-----
 src/backend/replication/walsender.c           | 40 ++++++++------
 src/bin/pg_basebackup/streamutil.c            | 40 ++++++++++++--
 5 files changed, 144 insertions(+), 63 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index e88998cd27..9eec9eb4e5 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1914,7 +1914,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-create-slot" xreflabel="CREATE_REPLICATION_SLOT">
-   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> | <literal>LOGICAL</literal> } [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
      <indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
@@ -1954,46 +1954,50 @@ The commands accepted in replication mode are:
         </para>
        </listitem>
       </varlistentry>
+     </variablelist>
+
+     <para>The following options are supported:</para>
 
+     <variablelist>
       <varlistentry>
-       <term><literal>TWO_PHASE</literal></term>
+       <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this logical replication slot supports decoding of two-phase
+         If true, this logical replication slot supports decoding of two-phase
          transactions. With this option, two-phase commands like
          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
          The transaction will be decoded and transmitted at
          <literal>PREPARE TRANSACTION</literal> time.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>RESERVE_WAL</literal></term>
+       <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this physical replication slot reserves <acronym>WAL</acronym>
+         If true, this physical replication slot reserves <acronym>WAL</acronym>
          immediately.  Otherwise, <acronym>WAL</acronym> is only reserved upon
          connection from a streaming replication client.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>EXPORT_SNAPSHOT</literal></term>
-       <term><literal>NOEXPORT_SNAPSHOT</literal></term>
-       <term><literal>USE_SNAPSHOT</literal></term>
+       <term><literal>SNAPSHOT { 'export' | 'use' | 'nothing' }</literal></term>
        <listitem>
         <para>
          Decides what to do with the snapshot created during logical slot
-         initialization. <literal>EXPORT_SNAPSHOT</literal>, which is the default,
+         initialization. <literal>'export'</literal>, which is the default,
          will export the snapshot for use in other sessions. This option can't
-         be used inside a transaction.  <literal>USE_SNAPSHOT</literal> will use the
+         be used inside a transaction.  <literal>'use'</literal> will use the
          snapshot for the current transaction executing the command. This
          option must be used in a transaction, and
          <literal>CREATE_REPLICATION_SLOT</literal> must be the first command
-         run in that transaction.  Finally, <literal>NOEXPORT_SNAPSHOT</literal> will
+         run in that transaction.  Finally, <literal>'nothing'</literal> will
          just use the snapshot for logical decoding as normal but won't do
          anything else with it.
         </para>
@@ -2052,6 +2056,17 @@ The commands accepted in replication mode are:
     </listitem>
   </varlistentry>
 
+  <varlistentry>
+    <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>CREATE_REPLICATION_SLOT</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
+
   <varlistentry>
     <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ]
      <indexterm><primary>START_REPLICATION</primary></indexterm>
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 19ea159af4..b97cd1df99 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -862,6 +862,9 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 	PGresult   *res;
 	StringInfoData cmd;
 	char	   *snapshot;
+	int			use_new_options_syntax;
+
+	use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 15000);
 
 	initStringInfo(&cmd);
 
@@ -872,26 +875,52 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 
 	if (conn->logical)
 	{
-		appendStringInfoString(&cmd, " LOGICAL pgoutput");
+		appendStringInfoString(&cmd, " LOGICAL pgoutput ");
+		if (use_new_options_syntax)
+			appendStringInfoChar(&cmd, '(');
 		if (two_phase)
-			appendStringInfoString(&cmd, " TWO_PHASE");
+			appendStringInfoString(&cmd, "TWO_PHASE");
 
-		switch (snapshot_action)
+		if (use_new_options_syntax)
+		{
+			switch (snapshot_action)
+			{
+				case CRS_EXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, ", SNAPSHOT 'export'");
+					break;
+				case CRS_NOEXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, ", SNAPSHOT 'nothing'");
+					break;
+				case CRS_USE_SNAPSHOT:
+					appendStringInfoString(&cmd, ", SNAPSHOT 'use'");
+					break;
+			}
+		}
+		else
 		{
-			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
-				break;
-			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
-				break;
-			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
-				break;
+			switch (snapshot_action)
+			{
+				case CRS_EXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
+					break;
+				case CRS_NOEXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
+					break;
+				case CRS_USE_SNAPSHOT:
+					appendStringInfoString(&cmd, " USE_SNAPSHOT");
+					break;
+			}
 		}
+
+		if (use_new_options_syntax)
+			appendStringInfoChar(&cmd, ')');
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		if (use_new_options_syntax)
+			appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
+		else
+			appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 3b59d62ed8..126380e2df 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -103,8 +103,8 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	plugin_opt_arg
 %type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -243,8 +243,8 @@ base_backup_legacy_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -254,8 +254,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -268,28 +268,33 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("export"), -1);
 				}
 			| K_NOEXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(false), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("nothing"), -1);
 				}
 			| K_USE_SNAPSHOT
 				{
-				  $$ = makeDefElem("use_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("use"), -1);
 				}
 			| K_RESERVE_WAL
 				{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3ca2a11389..b811a5c0ef 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -872,26 +872,30 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 	{
 		DefElem    *defel = (DefElem *) lfirst(lc);
 
-		if (strcmp(defel->defname, "export_snapshot") == 0)
+		if (strcmp(defel->defname, "snapshot") == 0)
 		{
+			char	   *action;
+
 			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 
+			action = defGetString(defel);
 			snapshot_action_given = true;
-			*snapshot_action = defGetBoolean(defel) ? CRS_EXPORT_SNAPSHOT :
-				CRS_NOEXPORT_SNAPSHOT;
-		}
-		else if (strcmp(defel->defname, "use_snapshot") == 0)
-		{
-			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
+
+			if (strcmp(action, "export") == 0)
+				*snapshot_action = CRS_EXPORT_SNAPSHOT;
+			else if (strcmp(action, "nothing") == 0)
+				*snapshot_action = CRS_NOEXPORT_SNAPSHOT;
+			else if (strcmp(action, "use") == 0)
+				*snapshot_action = CRS_USE_SNAPSHOT;
+			else
 				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						 errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
+								defel->defname, action)));
 
-			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -901,7 +905,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
@@ -910,7 +914,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 			two_phase_given = true;
-			*two_phase = true;
+			*two_phase = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -980,7 +984,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));
 
 			need_full_snapshot = true;
 		}
@@ -990,25 +994,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index d782b81adc..37237cd5d9 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -490,6 +490,7 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 {
 	PQExpBuffer query;
 	PGresult   *res;
+	bool		use_new_option_syntax = (PQserverVersion(conn) >= 150000);
 
 	query = createPQExpBuffer();
 
@@ -498,27 +499,54 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	Assert(!(two_phase && is_physical));
 	Assert(slot_name != NULL);
 
-	/* Build query */
+	/* Build base portion of query */
 	appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	if (is_temporary)
 		appendPQExpBufferStr(query, " TEMPORARY");
 	if (is_physical)
-	{
 		appendPQExpBufferStr(query, " PHYSICAL");
+	else
+		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
+
+	/* Add any requested options */
+	if (use_new_option_syntax)
+		appendPQExpBufferStr(query, " (");
+	if (is_physical)
+	{
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "RESERVE_WAL");
 	}
 	else
 	{
-		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (two_phase && PQserverVersion(conn) >= 150000)
-			appendPQExpBufferStr(query, " TWO_PHASE");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "TWO_PHASE");
 
 		if (PQserverVersion(conn) >= 100000)
+		{
 			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+			if (use_new_option_syntax)
+				AppendStringCommandOption(query, use_new_option_syntax,
+										   "SNAPSHOT", "nothing");
+			else
+				AppendPlainCommandOption(query, use_new_option_syntax,
+										 "NOEXPORT_SNAPSHOT");
+		}
+	}
+	if (use_new_option_syntax)
+	{
+		/* Suppress option list if it would be empty, otherwise terminate */
+		if (query->data[query->len - 1] == '(')
+		{
+			query->len -= 2;
+			query->data[query->len] = '\0';
+		}
+		else
+			appendPQExpBufferChar(query, ')');
 	}
 
+	/* Now run the query */
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-- 
2.24.3 (Apple Git-128)

From 2a729aa211e608943eb8ec331777123768362f7e Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 21 Sep 2021 12:22:07 -0400
Subject: [PATCH v8 1/2] Flexible options for BASE_BACKUP.

Previously, BASE_BACKUP used an entirely hard-coded syntax, but that's
hard to extend. Instead, adopt the same kind of syntax we've used for
SQL commands such as VACUUM, ANALYZE, COPY, and EXPLAIN, where it's
not necessary for all of the option names to be parser keywords.

In the new syntax, most of the options now take an optional Boolean
argument. To match our practice in other in places, the options which
the old syntax called NOWAIT and NOVERIFY_CHECKSUMS options are in the
new syntax called WAIT and VERIFY_CHECKUMS, and the default value is
false. The old FAST option in the new syntax has been replaced by a
CHECKSUM option whose value may be 'fast' or 'spread'.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_basebackup prefer
the new syntax when the server is new enough to support it.

Patch by me, reviewed by Fabien Coelho and Sergei Kornilov.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml            | 68 +++++++++++--------
 src/backend/replication/basebackup.c  | 51 ++++++++------
 src/backend/replication/repl_gram.y   | 97 +++++++++++++++++++++++----
 src/bin/pg_basebackup/pg_basebackup.c | 71 +++++++++++++-------
 src/bin/pg_basebackup/streamutil.c    | 61 +++++++++++++++++
 src/bin/pg_basebackup/streamutil.h    | 12 ++++
 6 files changed, 276 insertions(+), 84 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a232546b1d..e88998cd27 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2517,8 +2517,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-base-backup" xreflabel="BASE_BACKUP">
-    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
-     <indexterm><primary>BASE_BACKUP</primary></indexterm>
+    <term><literal>BASE_BACKUP</literal> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
     </term>
     <listitem>
      <para>
@@ -2540,52 +2539,55 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>PROGRESS</literal></term>
+        <term><literal>PROGRESS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Request information required to generate a progress report. This will
-          send back an approximate size in the header of each tablespace, which
-          can be used to calculate how far along the stream is done. This is
-          calculated by enumerating all the file sizes once before the transfer
-          is even started, and might as such have a negative impact on the
-          performance.  In particular, it might take longer before the first data
+          If set to true, request information required to generate a progress
+          report. This will send back an approximate size in the header of each
+          tablespace, which can be used to calculate how far along the stream
+          is done. This is calculated by enumerating all the file sizes once
+          before the transfer is even started, and might as such have a
+          negative impact on the performance.  In particular, it might take
+          longer before the first data
           is streamed. Since the database files can change during the backup,
           the size is only approximate and might both grow and shrink between
           the time of approximation and the sending of the actual files.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>FAST</literal></term>
+        <term><literal>CHECKPOINT { 'fast' | 'spread' }</literal></term>
         <listitem>
          <para>
-          Request a fast checkpoint.
+          Sets the type of checkpoint to be performed at the beginning of the
+          base backup. The default is <literal>spread</literal>.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>WAL</literal></term>
+        <term><literal>WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include the necessary WAL segments in the backup. This will include
-          all the files between start and stop backup in the
+          If set to true, include the necessary WAL segments in the backup.
+          This will include all the files between start and stop backup in the
           <filename>pg_wal</filename> directory of the base directory tar
-          file.
+          file. The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOWAIT</literal></term>
+        <term><literal>WAIT [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, the backup will wait until the last required WAL
+          If set to true, the backup will wait until the last required WAL
           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.
+          not enabled. If false, the backup will neither wait nor warn,
+          leaving the client responsible for ensuring the required log is
+          available. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2605,25 +2607,25 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>TABLESPACE_MAP</literal></term>
+        <term><literal>TABLESPACE_MAP [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include information about symbolic links present in the directory
-          <filename>pg_tblspc</filename> in a file named
+          If true, include information about symbolic links present in the
+          directory <filename>pg_tblspc</filename> in a file named
           <filename>tablespace_map</filename>. The tablespace map file includes
           each symbolic link name as it exists in the directory
           <filename>pg_tblspc/</filename> and the full path of that symbolic link.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOVERIFY_CHECKSUMS</literal></term>
+        <term><literal>VERIFY_CHECKSUMS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, checksums are verified during a base backup if they are
-          enabled. Specifying <literal>NOVERIFY_CHECKSUMS</literal> disables
-          this verification.
+          If true, checksums are verified during a base backup if they are
+          enabled. If false, this is skipped. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2708,6 +2710,7 @@ The commands accepted in replication mode are:
        </varlistentry>
       </variablelist>
      </para>
+
      <para>
       After the second regular result set, one or more CopyOutResponse results
       will be sent, one for the main data directory and one for each additional tablespace other
@@ -2788,6 +2791,17 @@ The commands accepted in replication mode are:
      </para>
     </listitem>
   </varlistentry>
+
+  <varlistentry>
+    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>BASE_BACKUP</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
 </variablelist>
 
 </para>
diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index e09108d0ec..4c97ab7b5a 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -764,7 +765,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 	ListCell   *lopt;
 	bool		o_label = false;
 	bool		o_progress = false;
-	bool		o_fast = false;
+	bool		o_checkpoint = false;
 	bool		o_nowait = false;
 	bool		o_wal = false;
 	bool		o_maxrate = false;
@@ -787,7 +788,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -796,25 +797,35 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
-		else if (strcmp(defel->defname, "fast") == 0)
+		else if (strcmp(defel->defname, "checkpoint") == 0)
 		{
-			if (o_fast)
+			char	   *optval = defGetString(defel);
+
+			if (o_checkpoint)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
-			o_fast = true;
+			if (pg_strcasecmp(optval, "fast") == 0)
+				opt->fastcheckpoint = true;
+			else if (pg_strcasecmp(optval, "spread") == 0)
+				opt->fastcheckpoint = false;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("unrecognized checkpoint type: \"%s\"",
+								optval)));
+			o_checkpoint = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -823,19 +834,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -851,21 +862,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -890,7 +901,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -905,8 +916,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index e1e8ec29cc..3b59d62ed8 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -95,13 +95,13 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
 %type <list>	create_slot_opt_list
 %type <defelt>	create_slot_opt
@@ -157,12 +157,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -170,14 +182,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -190,8 +202,8 @@ base_backup_opt:
 				}
 			| K_FAST
 				{
-				  $$ = makeDefElem("fast",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("checkpoint",
+								   (Node *)makeString("fast"), -1);
 				}
 			| K_WAL
 				{
@@ -200,8 +212,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -215,8 +227,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -422,6 +434,65 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_TWO_PHASE					{ $$ = "two_phase"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 669aa207a3..27ee6394cf 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1804,10 +1804,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1816,8 +1812,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	PQExpBufferData buf;
 
 	Assert(conn != NULL);
+	initPQExpBuffer(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1835,6 +1834,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1500)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1865,20 +1866,48 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendStringCommandOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+	{
+		if (use_new_option_syntax)
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+									  "CHECKPOINT", "fast");
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "FAST");
+	}
+	if (includewal != NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendIntegerCommandOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendStringCommandOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1893,18 +1922,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index f5b3b476e5..d782b81adc 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -603,6 +603,67 @@ DropReplicationSlot(PGconn *conn, const char *slot_name)
 	return true;
 }
 
+/*
+ * Append a "plain" option - one with no value - to a server command that
+ * is being constructed.
+ *
+ * In the old syntax, all options were parser keywords, so you could just
+ * write things like SOME_COMMAND OPTION1 OPTION2 'opt2value' OPTION3 42. The
+ * new syntax uses a comma-separated list surrounded by parentheses, so the
+ * equivalent is SOME_COMMAND (OPTION1, OPTION2 'optvalue', OPTION3 42).
+ */
+void
+AppendPlainCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						 char *option_name)
+{
+	if (buf->len > 0 && buf->data[buf->len - 1] != '(')
+	{
+		if (use_new_option_syntax)
+			appendPQExpBufferStr(buf, ", ");
+		else
+			appendPQExpBufferChar(buf, ' ');
+	}
+
+	appendPQExpBuffer(buf, " %s", option_name);
+}
+
+/*
+ * Append an option with an associated string value to a server command that
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendStringCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						  char *option_name, char *option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t		length = strlen(option_value);
+		char	   *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendPQExpBuffer(buf, " '%s'", escaped_value);
+		pfree(escaped_value);
+	}
+}
+
+/*
+ * Append an option with an associated integer value to a server command
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendIntegerCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						   char *option_name, int32 option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	appendPQExpBuffer(buf, " %d", option_value);
+}
 
 /*
  * Frontend version of GetCurrentTimestamp(), since we are not linked with
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 504803b976..65135c79e0 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -15,6 +15,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "libpq-fe.h"
+#include "pqexpbuffer.h"
 
 extern const char *progname;
 extern char *connection_string;
@@ -40,6 +41,17 @@ extern bool RunIdentifySystem(PGconn *conn, char **sysid,
 							  TimeLineID *starttli,
 							  XLogRecPtr *startpos,
 							  char **db_name);
+
+extern void AppendPlainCommandOption(PQExpBuffer buf,
+									 bool use_new_option_syntax,
+									 char *option_value);
+extern void AppendStringCommandOption(PQExpBuffer buf,
+									  bool use_new_option_syntax,
+									  char *option_name, char *option_value);
+extern void AppendIntegerCommandOption(PQExpBuffer buf,
+									   bool use_new_option_syntax,
+									   char *option_name, int32 option_value);
+
 extern bool RetrieveWalSegSize(PGconn *conn);
 extern TimestampTz feGetCurrentTimestamp(void);
 extern void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
-- 
2.24.3 (Apple Git-128)

From 456219649f71057755a40c8fb662648dbcea0140 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Thu, 23 Sep 2021 11:03:15 -0400
Subject: [PATCH v8 2/2] Flexible options for CREATE_REPLICATION_SLOT.

Like BASE_BACKUP, CREATE_REPLICATION_SLOT has historically used a
hard-coded syntax.  To improve future extensibility, adopt a flexible
options syntax here, too.

In the new syntax, instead of three mutually exclusive options
EXPORT_SNAPSHOT, USE_SNAPSHOT, and NOEXPORT_SNAPSHOT, there is now a single
SNAPSHOT option with three possible values: 'export', 'use', and 'nothing'.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, makes pg_receivewal,
pg_recvlogical, and walreceiver processes use it.

Patch by me, reviewed by Fabien Coelho and Sergei Kornilov.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml                    | 37 +++++++----
 .../libpqwalreceiver/libpqwalreceiver.c       | 61 +++++++++++++++----
 src/backend/replication/repl_gram.y           | 35 ++++++-----
 src/backend/replication/walsender.c           | 40 ++++++------
 src/bin/pg_basebackup/streamutil.c            | 40 ++++++++++--
 5 files changed, 150 insertions(+), 63 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index e88998cd27..9eec9eb4e5 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1914,7 +1914,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-create-slot" xreflabel="CREATE_REPLICATION_SLOT">
-   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> | <literal>LOGICAL</literal> } [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
      <indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
@@ -1954,46 +1954,50 @@ The commands accepted in replication mode are:
         </para>
        </listitem>
       </varlistentry>
+     </variablelist>
+
+     <para>The following options are supported:</para>
 
+     <variablelist>
       <varlistentry>
-       <term><literal>TWO_PHASE</literal></term>
+       <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this logical replication slot supports decoding of two-phase
+         If true, this logical replication slot supports decoding of two-phase
          transactions. With this option, two-phase commands like
          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
          The transaction will be decoded and transmitted at
          <literal>PREPARE TRANSACTION</literal> time.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>RESERVE_WAL</literal></term>
+       <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this physical replication slot reserves <acronym>WAL</acronym>
+         If true, this physical replication slot reserves <acronym>WAL</acronym>
          immediately.  Otherwise, <acronym>WAL</acronym> is only reserved upon
          connection from a streaming replication client.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>EXPORT_SNAPSHOT</literal></term>
-       <term><literal>NOEXPORT_SNAPSHOT</literal></term>
-       <term><literal>USE_SNAPSHOT</literal></term>
+       <term><literal>SNAPSHOT { 'export' | 'use' | 'nothing' }</literal></term>
        <listitem>
         <para>
          Decides what to do with the snapshot created during logical slot
-         initialization. <literal>EXPORT_SNAPSHOT</literal>, which is the default,
+         initialization. <literal>'export'</literal>, which is the default,
          will export the snapshot for use in other sessions. This option can't
-         be used inside a transaction.  <literal>USE_SNAPSHOT</literal> will use the
+         be used inside a transaction.  <literal>'use'</literal> will use the
          snapshot for the current transaction executing the command. This
          option must be used in a transaction, and
          <literal>CREATE_REPLICATION_SLOT</literal> must be the first command
-         run in that transaction.  Finally, <literal>NOEXPORT_SNAPSHOT</literal> will
+         run in that transaction.  Finally, <literal>'nothing'</literal> will
          just use the snapshot for logical decoding as normal but won't do
          anything else with it.
         </para>
@@ -2052,6 +2056,17 @@ The commands accepted in replication mode are:
     </listitem>
   </varlistentry>
 
+  <varlistentry>
+    <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>CREATE_REPLICATION_SLOT</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
+
   <varlistentry>
     <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ]
      <indexterm><primary>START_REPLICATION</primary></indexterm>
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 19ea159af4..5c6e56a5b2 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -862,6 +862,9 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 	PGresult   *res;
 	StringInfoData cmd;
 	char	   *snapshot;
+	int			use_new_options_syntax;
+
+	use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
 
 	initStringInfo(&cmd);
 
@@ -872,26 +875,58 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 
 	if (conn->logical)
 	{
-		appendStringInfoString(&cmd, " LOGICAL pgoutput");
+		appendStringInfoString(&cmd, " LOGICAL pgoutput ");
+		if (use_new_options_syntax)
+			appendStringInfoChar(&cmd, '(');
 		if (two_phase)
-			appendStringInfoString(&cmd, " TWO_PHASE");
+		{
+			appendStringInfoString(&cmd, "TWO_PHASE");
+			if (use_new_options_syntax)
+				appendStringInfoString(&cmd, ", ");
+			else
+				appendStringInfoChar(&cmd, ' ');
+		}
 
-		switch (snapshot_action)
+		if (use_new_options_syntax)
 		{
-			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
-				break;
-			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
-				break;
-			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
-				break;
+			switch (snapshot_action)
+			{
+				case CRS_EXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "SNAPSHOT 'export'");
+					break;
+				case CRS_NOEXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
+					break;
+				case CRS_USE_SNAPSHOT:
+					appendStringInfoString(&cmd, "SNAPSHOT 'use'");
+					break;
+			}
 		}
+		else
+		{
+			switch (snapshot_action)
+			{
+				case CRS_EXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
+					break;
+				case CRS_NOEXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
+					break;
+				case CRS_USE_SNAPSHOT:
+					appendStringInfoString(&cmd, "USE_SNAPSHOT");
+					break;
+			}
+		}
+
+		if (use_new_options_syntax)
+			appendStringInfoChar(&cmd, ')');
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		if (use_new_options_syntax)
+			appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
+		else
+			appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 3b59d62ed8..126380e2df 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -103,8 +103,8 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	plugin_opt_arg
 %type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -243,8 +243,8 @@ base_backup_legacy_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -254,8 +254,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -268,28 +268,33 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("export"), -1);
 				}
 			| K_NOEXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(false), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("nothing"), -1);
 				}
 			| K_USE_SNAPSHOT
 				{
-				  $$ = makeDefElem("use_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("use"), -1);
 				}
 			| K_RESERVE_WAL
 				{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3ca2a11389..b811a5c0ef 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -872,26 +872,30 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 	{
 		DefElem    *defel = (DefElem *) lfirst(lc);
 
-		if (strcmp(defel->defname, "export_snapshot") == 0)
+		if (strcmp(defel->defname, "snapshot") == 0)
 		{
+			char	   *action;
+
 			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 
+			action = defGetString(defel);
 			snapshot_action_given = true;
-			*snapshot_action = defGetBoolean(defel) ? CRS_EXPORT_SNAPSHOT :
-				CRS_NOEXPORT_SNAPSHOT;
-		}
-		else if (strcmp(defel->defname, "use_snapshot") == 0)
-		{
-			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
+
+			if (strcmp(action, "export") == 0)
+				*snapshot_action = CRS_EXPORT_SNAPSHOT;
+			else if (strcmp(action, "nothing") == 0)
+				*snapshot_action = CRS_NOEXPORT_SNAPSHOT;
+			else if (strcmp(action, "use") == 0)
+				*snapshot_action = CRS_USE_SNAPSHOT;
+			else
 				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						 errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
+								defel->defname, action)));
 
-			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -901,7 +905,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
@@ -910,7 +914,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 			two_phase_given = true;
-			*two_phase = true;
+			*two_phase = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -980,7 +984,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));
 
 			need_full_snapshot = true;
 		}
@@ -990,25 +994,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index d782b81adc..37237cd5d9 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -490,6 +490,7 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 {
 	PQExpBuffer query;
 	PGresult   *res;
+	bool		use_new_option_syntax = (PQserverVersion(conn) >= 150000);
 
 	query = createPQExpBuffer();
 
@@ -498,27 +499,54 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	Assert(!(two_phase && is_physical));
 	Assert(slot_name != NULL);
 
-	/* Build query */
+	/* Build base portion of query */
 	appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	if (is_temporary)
 		appendPQExpBufferStr(query, " TEMPORARY");
 	if (is_physical)
-	{
 		appendPQExpBufferStr(query, " PHYSICAL");
+	else
+		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
+
+	/* Add any requested options */
+	if (use_new_option_syntax)
+		appendPQExpBufferStr(query, " (");
+	if (is_physical)
+	{
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "RESERVE_WAL");
 	}
 	else
 	{
-		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (two_phase && PQserverVersion(conn) >= 150000)
-			appendPQExpBufferStr(query, " TWO_PHASE");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "TWO_PHASE");
 
 		if (PQserverVersion(conn) >= 100000)
+		{
 			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+			if (use_new_option_syntax)
+				AppendStringCommandOption(query, use_new_option_syntax,
+										   "SNAPSHOT", "nothing");
+			else
+				AppendPlainCommandOption(query, use_new_option_syntax,
+										 "NOEXPORT_SNAPSHOT");
+		}
+	}
+	if (use_new_option_syntax)
+	{
+		/* Suppress option list if it would be empty, otherwise terminate */
+		if (query->data[query->len - 1] == '(')
+		{
+			query->len -= 2;
+			query->data[query->len] = '\0';
+		}
+		else
+			appendPQExpBufferChar(query, ')');
 	}
 
+	/* Now run the query */
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-- 
2.24.3 (Apple Git-128)

From cd91633f75e49264d36a871b918d992072647092 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 21 Sep 2021 12:22:07 -0400
Subject: [PATCH v9 1/2] Flexible options for BASE_BACKUP.

Previously, BASE_BACKUP used an entirely hard-coded syntax, but that's
hard to extend. Instead, adopt the same kind of syntax we've used for
SQL commands such as VACUUM, ANALYZE, COPY, and EXPLAIN, where it's
not necessary for all of the option names to be parser keywords.

In the new syntax, most of the options now take an optional Boolean
argument. To match our practice in other in places, the options which
the old syntax called NOWAIT and NOVERIFY_CHECKSUMS options are in the
new syntax called WAIT and VERIFY_CHECKUMS, and the default value is
false. The old FAST option in the new syntax has been replaced by a
CHECKSUM option whose value may be 'fast' or 'spread'.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, and makes pg_basebackup prefer
the new syntax when the server is new enough to support it.

Patch by me, reviewed by Fabien Coelho, Sergei Kornilov, and
Fujii Masao.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml            | 67 +++++++++++-------
 src/backend/replication/basebackup.c  | 51 ++++++++------
 src/backend/replication/repl_gram.y   | 97 +++++++++++++++++++++++----
 src/bin/pg_basebackup/pg_basebackup.c | 71 +++++++++++++-------
 src/bin/pg_basebackup/streamutil.c    | 61 +++++++++++++++++
 src/bin/pg_basebackup/streamutil.h    | 12 ++++
 6 files changed, 276 insertions(+), 83 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a232546b1d..8ed8833444 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2517,7 +2517,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-base-backup" xreflabel="BASE_BACKUP">
-    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
+    <term><literal>BASE_BACKUP</literal> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
      <indexterm><primary>BASE_BACKUP</primary></indexterm>
     </term>
     <listitem>
@@ -2540,52 +2540,55 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>PROGRESS</literal></term>
+        <term><literal>PROGRESS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Request information required to generate a progress report. This will
-          send back an approximate size in the header of each tablespace, which
-          can be used to calculate how far along the stream is done. This is
-          calculated by enumerating all the file sizes once before the transfer
-          is even started, and might as such have a negative impact on the
-          performance.  In particular, it might take longer before the first data
+          If set to true, request information required to generate a progress
+          report. This will send back an approximate size in the header of each
+          tablespace, which can be used to calculate how far along the stream
+          is done. This is calculated by enumerating all the file sizes once
+          before the transfer is even started, and might as such have a
+          negative impact on the performance.  In particular, it might take
+          longer before the first data
           is streamed. Since the database files can change during the backup,
           the size is only approximate and might both grow and shrink between
           the time of approximation and the sending of the actual files.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>FAST</literal></term>
+        <term><literal>CHECKPOINT { 'fast' | 'spread' }</literal></term>
         <listitem>
          <para>
-          Request a fast checkpoint.
+          Sets the type of checkpoint to be performed at the beginning of the
+          base backup. The default is <literal>spread</literal>.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>WAL</literal></term>
+        <term><literal>WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include the necessary WAL segments in the backup. This will include
-          all the files between start and stop backup in the
+          If set to true, include the necessary WAL segments in the backup.
+          This will include all the files between start and stop backup in the
           <filename>pg_wal</filename> directory of the base directory tar
-          file.
+          file. The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOWAIT</literal></term>
+        <term><literal>WAIT [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, the backup will wait until the last required WAL
+          If set to true, the backup will wait until the last required WAL
           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.
+          not enabled. If false, the backup will neither wait nor warn,
+          leaving the client responsible for ensuring the required log is
+          available. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2605,25 +2608,25 @@ The commands accepted in replication mode are:
        </varlistentry>
 
        <varlistentry>
-        <term><literal>TABLESPACE_MAP</literal></term>
+        <term><literal>TABLESPACE_MAP [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          Include information about symbolic links present in the directory
-          <filename>pg_tblspc</filename> in a file named
+          If true, include information about symbolic links present in the
+          directory <filename>pg_tblspc</filename> in a file named
           <filename>tablespace_map</filename>. The tablespace map file includes
           each symbolic link name as it exists in the directory
           <filename>pg_tblspc/</filename> and the full path of that symbolic link.
+          The default is false.
          </para>
         </listitem>
        </varlistentry>
 
        <varlistentry>
-        <term><literal>NOVERIFY_CHECKSUMS</literal></term>
+        <term><literal>VERIFY_CHECKSUMS [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
-          By default, checksums are verified during a base backup if they are
-          enabled. Specifying <literal>NOVERIFY_CHECKSUMS</literal> disables
-          this verification.
+          If true, checksums are verified during a base backup if they are
+          enabled. If false, this is skipped. The default is true.
          </para>
         </listitem>
        </varlistentry>
@@ -2708,6 +2711,7 @@ The commands accepted in replication mode are:
        </varlistentry>
       </variablelist>
      </para>
+
      <para>
       After the second regular result set, one or more CopyOutResponse results
       will be sent, one for the main data directory and one for each additional tablespace other
@@ -2788,6 +2792,17 @@ The commands accepted in replication mode are:
      </para>
     </listitem>
   </varlistentry>
+
+  <varlistentry>
+    <term><literal>BASE_BACKUP</literal> [ <literal>LABEL</literal> <replaceable>'label'</replaceable> ] [ <literal>PROGRESS</literal> ] [ <literal>FAST</literal> ] [ <literal>WAL</literal> ] [ <literal>NOWAIT</literal> ] [ <literal>MAX_RATE</literal> <replaceable>rate</replaceable> ] [ <literal>TABLESPACE_MAP</literal> ] [ <literal>NOVERIFY_CHECKSUMS</literal> ] [ <literal>MANIFEST</literal> <replaceable>manifest_option</replaceable> ] [ <literal>MANIFEST_CHECKSUMS</literal> <replaceable>checksum_algorithm</replaceable> ]
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>BASE_BACKUP</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
 </variablelist>
 
 </para>
diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index e09108d0ec..4c97ab7b5a 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -19,6 +19,7 @@
 #include "access/xlog_internal.h"	/* for pg_start/stop_backup */
 #include "catalog/pg_type.h"
 #include "common/file_perm.h"
+#include "commands/defrem.h"
 #include "commands/progress.h"
 #include "lib/stringinfo.h"
 #include "libpq/libpq.h"
@@ -764,7 +765,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 	ListCell   *lopt;
 	bool		o_label = false;
 	bool		o_progress = false;
-	bool		o_fast = false;
+	bool		o_checkpoint = false;
 	bool		o_nowait = false;
 	bool		o_wal = false;
 	bool		o_maxrate = false;
@@ -787,7 +788,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->label = strVal(defel->arg);
+			opt->label = defGetString(defel);
 			o_label = true;
 		}
 		else if (strcmp(defel->defname, "progress") == 0)
@@ -796,25 +797,35 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->progress = true;
+			opt->progress = defGetBoolean(defel);
 			o_progress = true;
 		}
-		else if (strcmp(defel->defname, "fast") == 0)
+		else if (strcmp(defel->defname, "checkpoint") == 0)
 		{
-			if (o_fast)
+			char	   *optval = defGetString(defel);
+
+			if (o_checkpoint)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->fastcheckpoint = true;
-			o_fast = true;
+			if (pg_strcasecmp(optval, "fast") == 0)
+				opt->fastcheckpoint = true;
+			else if (pg_strcasecmp(optval, "spread") == 0)
+				opt->fastcheckpoint = false;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("unrecognized checkpoint type: \"%s\"",
+								optval)));
+			o_checkpoint = true;
 		}
-		else if (strcmp(defel->defname, "nowait") == 0)
+		else if (strcmp(defel->defname, "wait") == 0)
 		{
 			if (o_nowait)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->nowait = true;
+			opt->nowait = !defGetBoolean(defel);
 			o_nowait = true;
 		}
 		else if (strcmp(defel->defname, "wal") == 0)
@@ -823,19 +834,19 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->includewal = true;
+			opt->includewal = defGetBoolean(defel);
 			o_wal = true;
 		}
 		else if (strcmp(defel->defname, "max_rate") == 0)
 		{
-			long		maxrate;
+			int64		maxrate;
 
 			if (o_maxrate)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
 
-			maxrate = intVal(defel->arg);
+			maxrate = defGetInt64(defel);
 			if (maxrate < MAX_RATE_LOWER || maxrate > MAX_RATE_UPPER)
 				ereport(ERROR,
 						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
@@ -851,21 +862,21 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			opt->sendtblspcmapfile = true;
+			opt->sendtblspcmapfile = defGetBoolean(defel);
 			o_tablespace_map = true;
 		}
-		else if (strcmp(defel->defname, "noverify_checksums") == 0)
+		else if (strcmp(defel->defname, "verify_checksums") == 0)
 		{
 			if (o_noverify_checksums)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("duplicate option \"%s\"", defel->defname)));
-			noverify_checksums = true;
+			noverify_checksums = !defGetBoolean(defel);
 			o_noverify_checksums = true;
 		}
 		else if (strcmp(defel->defname, "manifest") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 			bool		manifest_bool;
 
 			if (o_manifest)
@@ -890,7 +901,7 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 		}
 		else if (strcmp(defel->defname, "manifest_checksums") == 0)
 		{
-			char	   *optval = strVal(defel->arg);
+			char	   *optval = defGetString(defel);
 
 			if (o_manifest_checksums)
 				ereport(ERROR,
@@ -905,8 +916,10 @@ parse_basebackup_options(List *options, basebackup_options *opt)
 			o_manifest_checksums = true;
 		}
 		else
-			elog(ERROR, "option \"%s\" not recognized",
-				 defel->defname);
+			ereport(ERROR,
+					errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("option \"%s\" not recognized",
+						   defel->defname));
 	}
 	if (opt->label == NULL)
 		opt->label = "base backup";
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index e1e8ec29cc..3b59d62ed8 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -95,13 +95,13 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
 				timeline_history show sql_cmd
-%type <list>	base_backup_opt_list
-%type <defelt>	base_backup_opt
+%type <list>	base_backup_legacy_opt_list generic_option_list
+%type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
 %type <list>	plugin_options plugin_opt_list
 %type <defelt>	plugin_opt_elem
 %type <node>	plugin_opt_arg
-%type <str>		opt_slot var_name
+%type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
 %type <list>	create_slot_opt_list
 %type <defelt>	create_slot_opt
@@ -157,12 +157,24 @@ var_name:	IDENT	{ $$ = $1; }
 		;
 
 /*
+ * BASE_BACKUP ( option [ 'value' ] [, ...] )
+ *
+ * We also still support the legacy syntax:
+ *
  * BASE_BACKUP [LABEL '<label>'] [PROGRESS] [FAST] [WAL] [NOWAIT]
  * [MAX_RATE %d] [TABLESPACE_MAP] [NOVERIFY_CHECKSUMS]
  * [MANIFEST %s] [MANIFEST_CHECKSUMS %s]
+ *
+ * Future options should be supported only using the new syntax.
  */
 base_backup:
-			K_BASE_BACKUP base_backup_opt_list
+			K_BASE_BACKUP '(' generic_option_list ')'
+				{
+					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
+					cmd->options = $3;
+					$$ = (Node *) cmd;
+				}
+			| K_BASE_BACKUP base_backup_legacy_opt_list
 				{
 					BaseBackupCmd *cmd = makeNode(BaseBackupCmd);
 					cmd->options = $2;
@@ -170,14 +182,14 @@ base_backup:
 				}
 			;
 
-base_backup_opt_list:
-			base_backup_opt_list base_backup_opt
+base_backup_legacy_opt_list:
+			base_backup_legacy_opt_list base_backup_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-base_backup_opt:
+base_backup_legacy_opt:
 			K_LABEL SCONST
 				{
 				  $$ = makeDefElem("label",
@@ -190,8 +202,8 @@ base_backup_opt:
 				}
 			| K_FAST
 				{
-				  $$ = makeDefElem("fast",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("checkpoint",
+								   (Node *)makeString("fast"), -1);
 				}
 			| K_WAL
 				{
@@ -200,8 +212,8 @@ base_backup_opt:
 				}
 			| K_NOWAIT
 				{
-				  $$ = makeDefElem("nowait",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("wait",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MAX_RATE UCONST
 				{
@@ -215,8 +227,8 @@ base_backup_opt:
 				}
 			| K_NOVERIFY_CHECKSUMS
 				{
-				  $$ = makeDefElem("noverify_checksums",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("verify_checksums",
+								   (Node *)makeInteger(false), -1);
 				}
 			| K_MANIFEST SCONST
 				{
@@ -422,6 +434,65 @@ plugin_opt_arg:
 sql_cmd:
 			IDENT							{ $$ = (Node *) make_sqlcmd(); }
 		;
+
+generic_option_list:
+			generic_option_list ',' generic_option
+				{ $$ = lappend($1, $3); }
+			| generic_option
+				{ $$ = list_make1($1); }
+			;
+
+generic_option:
+			ident_or_keyword
+				{
+					$$ = makeDefElem($1, NULL, -1);
+				}
+			| ident_or_keyword IDENT
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword SCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeString($2), -1);
+				}
+			| ident_or_keyword UCONST
+				{
+					$$ = makeDefElem($1, (Node *) makeInteger($2), -1);
+				}
+			;
+
+ident_or_keyword:
+			IDENT							{ $$ = $1; }
+			| K_BASE_BACKUP					{ $$ = "base_backup"; }
+			| K_IDENTIFY_SYSTEM				{ $$ = "identify_system"; }
+			| K_SHOW						{ $$ = "show"; }
+			| K_START_REPLICATION			{ $$ = "start_replication"; }
+			| K_CREATE_REPLICATION_SLOT	{ $$ = "create_replication_slot"; }
+			| K_DROP_REPLICATION_SLOT		{ $$ = "drop_replication_slot"; }
+			| K_TIMELINE_HISTORY			{ $$ = "timeline_history"; }
+			| K_LABEL						{ $$ = "label"; }
+			| K_PROGRESS					{ $$ = "progress"; }
+			| K_FAST						{ $$ = "fast"; }
+			| K_WAIT						{ $$ = "wait"; }
+			| K_NOWAIT						{ $$ = "nowait"; }
+			| K_MAX_RATE					{ $$ = "max_rate"; }
+			| K_WAL							{ $$ = "wal"; }
+			| K_TABLESPACE_MAP				{ $$ = "tablespace_map"; }
+			| K_NOVERIFY_CHECKSUMS			{ $$ = "noverify_checksums"; }
+			| K_TIMELINE					{ $$ = "timeline"; }
+			| K_PHYSICAL					{ $$ = "physical"; }
+			| K_LOGICAL						{ $$ = "logical"; }
+			| K_SLOT						{ $$ = "slot"; }
+			| K_RESERVE_WAL					{ $$ = "reserve_wal"; }
+			| K_TEMPORARY					{ $$ = "temporary"; }
+			| K_TWO_PHASE					{ $$ = "two_phase"; }
+			| K_EXPORT_SNAPSHOT				{ $$ = "export_snapshot"; }
+			| K_NOEXPORT_SNAPSHOT			{ $$ = "noexport_snapshot"; }
+			| K_USE_SNAPSHOT				{ $$ = "use_snapshot"; }
+			| K_MANIFEST					{ $$ = "manifest"; }
+			| K_MANIFEST_CHECKSUMS			{ $$ = "manifest_checksums"; }
+		;
+
 %%
 
 static SQLCmd *
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 669aa207a3..27ee6394cf 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -1804,10 +1804,6 @@ BaseBackup(void)
 	TimeLineID	latesttli;
 	TimeLineID	starttli;
 	char	   *basebkp;
-	char		escaped_label[MAXPGPATH];
-	char	   *maxrate_clause = NULL;
-	char	   *manifest_clause = NULL;
-	char	   *manifest_checksums_clause = "";
 	int			i;
 	char		xlogstart[64];
 	char		xlogend[64];
@@ -1816,8 +1812,11 @@ BaseBackup(void)
 	int			serverVersion,
 				serverMajor;
 	int			writing_to_stdout;
+	bool		use_new_option_syntax = false;
+	PQExpBufferData buf;
 
 	Assert(conn != NULL);
+	initPQExpBuffer(&buf);
 
 	/*
 	 * Check server version. BASE_BACKUP command was introduced in 9.1, so we
@@ -1835,6 +1834,8 @@ BaseBackup(void)
 					 serverver ? serverver : "'unknown'");
 		exit(1);
 	}
+	if (serverMajor >= 1500)
+		use_new_option_syntax = true;
 
 	/*
 	 * If WAL streaming was requested, also check that the server is new
@@ -1865,20 +1866,48 @@ BaseBackup(void)
 	/*
 	 * Start the actual backup
 	 */
-	PQescapeStringConn(conn, escaped_label, label, sizeof(escaped_label), &i);
-
+	AppendStringCommandOption(&buf, use_new_option_syntax, "LABEL", label);
+	if (estimatesize)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "PROGRESS");
+	if (includewal == FETCH_WAL)
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "WAL");
+	if (fastcheckpoint)
+	{
+		if (use_new_option_syntax)
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+									  "CHECKPOINT", "fast");
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "FAST");
+	}
+	if (includewal != NO_WAL)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax, "WAIT", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax, "NOWAIT");
+	}
 	if (maxrate > 0)
-		maxrate_clause = psprintf("MAX_RATE %u", maxrate);
+		AppendIntegerCommandOption(&buf, use_new_option_syntax, "MAX_RATE",
+									  maxrate);
+	if (format == 't')
+		AppendPlainCommandOption(&buf, use_new_option_syntax, "TABLESPACE_MAP");
+	if (!verify_checksums)
+	{
+		if (use_new_option_syntax)
+			AppendIntegerCommandOption(&buf, use_new_option_syntax,
+										  "VERIFY_CHECKSUMS", 0);
+		else
+			AppendPlainCommandOption(&buf, use_new_option_syntax,
+										"NOVERIFY_CHECKSUMS");
+	}
 
 	if (manifest)
 	{
-		if (manifest_force_encode)
-			manifest_clause = "MANIFEST 'force-encode'";
-		else
-			manifest_clause = "MANIFEST 'yes'";
+		AppendStringCommandOption(&buf, use_new_option_syntax, "MANIFEST",
+									 manifest_force_encode ? "force-encode" : "yes");
 		if (manifest_checksums != NULL)
-			manifest_checksums_clause = psprintf("MANIFEST_CHECKSUMS '%s'",
-												 manifest_checksums);
+			AppendStringCommandOption(&buf, use_new_option_syntax,
+										 "MANIFEST_CHECKSUMS", manifest_checksums);
 	}
 
 	if (verbose)
@@ -1893,18 +1922,10 @@ BaseBackup(void)
 			fprintf(stderr, "\n");
 	}
 
-	basebkp =
-		psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",
-				 escaped_label,
-				 estimatesize ? "PROGRESS" : "",
-				 includewal == FETCH_WAL ? "WAL" : "",
-				 fastcheckpoint ? "FAST" : "",
-				 includewal == NO_WAL ? "" : "NOWAIT",
-				 maxrate_clause ? maxrate_clause : "",
-				 format == 't' ? "TABLESPACE_MAP" : "",
-				 verify_checksums ? "" : "NOVERIFY_CHECKSUMS",
-				 manifest_clause ? manifest_clause : "",
-				 manifest_checksums_clause);
+	if (use_new_option_syntax && buf.len > 0)
+		basebkp = psprintf("BASE_BACKUP (%s)", buf.data);
+	else
+		basebkp = psprintf("BASE_BACKUP %s", buf.data);
 
 	if (PQsendQuery(conn, basebkp) == 0)
 	{
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index f5b3b476e5..d782b81adc 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -603,6 +603,67 @@ DropReplicationSlot(PGconn *conn, const char *slot_name)
 	return true;
 }
 
+/*
+ * Append a "plain" option - one with no value - to a server command that
+ * is being constructed.
+ *
+ * In the old syntax, all options were parser keywords, so you could just
+ * write things like SOME_COMMAND OPTION1 OPTION2 'opt2value' OPTION3 42. The
+ * new syntax uses a comma-separated list surrounded by parentheses, so the
+ * equivalent is SOME_COMMAND (OPTION1, OPTION2 'optvalue', OPTION3 42).
+ */
+void
+AppendPlainCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						 char *option_name)
+{
+	if (buf->len > 0 && buf->data[buf->len - 1] != '(')
+	{
+		if (use_new_option_syntax)
+			appendPQExpBufferStr(buf, ", ");
+		else
+			appendPQExpBufferChar(buf, ' ');
+	}
+
+	appendPQExpBuffer(buf, " %s", option_name);
+}
+
+/*
+ * Append an option with an associated string value to a server command that
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendStringCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						  char *option_name, char *option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	if (option_value != NULL)
+	{
+		size_t		length = strlen(option_value);
+		char	   *escaped_value = palloc(1 + 2 * length);
+
+		PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
+		appendPQExpBuffer(buf, " '%s'", escaped_value);
+		pfree(escaped_value);
+	}
+}
+
+/*
+ * Append an option with an associated integer value to a server command
+ * is being constructed.
+ *
+ * See comments for AppendPlainCommandOption, above.
+ */
+void
+AppendIntegerCommandOption(PQExpBuffer buf, bool use_new_option_syntax,
+						   char *option_name, int32 option_value)
+{
+	AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
+
+	appendPQExpBuffer(buf, " %d", option_value);
+}
 
 /*
  * Frontend version of GetCurrentTimestamp(), since we are not linked with
diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h
index 504803b976..65135c79e0 100644
--- a/src/bin/pg_basebackup/streamutil.h
+++ b/src/bin/pg_basebackup/streamutil.h
@@ -15,6 +15,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "libpq-fe.h"
+#include "pqexpbuffer.h"
 
 extern const char *progname;
 extern char *connection_string;
@@ -40,6 +41,17 @@ extern bool RunIdentifySystem(PGconn *conn, char **sysid,
 							  TimeLineID *starttli,
 							  XLogRecPtr *startpos,
 							  char **db_name);
+
+extern void AppendPlainCommandOption(PQExpBuffer buf,
+									 bool use_new_option_syntax,
+									 char *option_value);
+extern void AppendStringCommandOption(PQExpBuffer buf,
+									  bool use_new_option_syntax,
+									  char *option_name, char *option_value);
+extern void AppendIntegerCommandOption(PQExpBuffer buf,
+									   bool use_new_option_syntax,
+									   char *option_name, int32 option_value);
+
 extern bool RetrieveWalSegSize(PGconn *conn);
 extern TimestampTz feGetCurrentTimestamp(void);
 extern void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
-- 
2.24.3 (Apple Git-128)

From d728c23ffee4eb45821ab2a828a18655623a2750 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Thu, 23 Sep 2021 11:03:15 -0400
Subject: [PATCH v9 2/2] Flexible options for CREATE_REPLICATION_SLOT.

Like BASE_BACKUP, CREATE_REPLICATION_SLOT has historically used a
hard-coded syntax.  To improve future extensibility, adopt a flexible
options syntax here, too.

In the new syntax, instead of three mutually exclusive options
EXPORT_SNAPSHOT, USE_SNAPSHOT, and NOEXPORT_SNAPSHOT, there is now a single
SNAPSHOT option with three possible values: 'export', 'use', and 'nothing'.

This commit does not remove support for the old syntax. It just adds
the new one as an additional option, makes pg_receivewal,
pg_recvlogical, and walreceiver processes use it.

Patch by me, reviewed by Fabien Coelho, Sergei Kornilov, and
Fujii Masao.

Discussion: http://postgr.es/m/CA+TgmobAczXDRO_Gr2euo_TxgzaH1JxbNxvFx=HYvBinefNH8Q@mail.gmail.com
Discussion: http://postgr.es/m/CA+TgmoZGwR=ZVWFeecncubEyPdwghnvfkkdBe9BLccLSiqdf9Q@mail.gmail.com
---
 doc/src/sgml/protocol.sgml                    | 37 +++++++----
 .../libpqwalreceiver/libpqwalreceiver.c       | 61 +++++++++++++++----
 src/backend/replication/repl_gram.y           | 35 ++++++-----
 src/backend/replication/walsender.c           | 40 ++++++------
 src/bin/pg_basebackup/streamutil.c            | 40 ++++++++++--
 5 files changed, 150 insertions(+), 63 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 8ed8833444..b95cc88599 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1914,7 +1914,7 @@ The commands accepted in replication mode are:
   </varlistentry>
 
   <varlistentry id="protocol-replication-create-slot" xreflabel="CREATE_REPLICATION_SLOT">
-   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+   <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> | <literal>LOGICAL</literal> } [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
      <indexterm><primary>CREATE_REPLICATION_SLOT</primary></indexterm>
     </term>
     <listitem>
@@ -1954,46 +1954,50 @@ The commands accepted in replication mode are:
         </para>
        </listitem>
       </varlistentry>
+     </variablelist>
+
+     <para>The following options are supported:</para>
 
+     <variablelist>
       <varlistentry>
-       <term><literal>TWO_PHASE</literal></term>
+       <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this logical replication slot supports decoding of two-phase
+         If true, this logical replication slot supports decoding of two-phase
          transactions. With this option, two-phase commands like
          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
          The transaction will be decoded and transmitted at
          <literal>PREPARE TRANSACTION</literal> time.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>RESERVE_WAL</literal></term>
+       <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
        <listitem>
         <para>
-         Specify that this physical replication slot reserves <acronym>WAL</acronym>
+         If true, this physical replication slot reserves <acronym>WAL</acronym>
          immediately.  Otherwise, <acronym>WAL</acronym> is only reserved upon
          connection from a streaming replication client.
+         The default is false.
         </para>
        </listitem>
       </varlistentry>
 
       <varlistentry>
-       <term><literal>EXPORT_SNAPSHOT</literal></term>
-       <term><literal>NOEXPORT_SNAPSHOT</literal></term>
-       <term><literal>USE_SNAPSHOT</literal></term>
+       <term><literal>SNAPSHOT { 'export' | 'use' | 'nothing' }</literal></term>
        <listitem>
         <para>
          Decides what to do with the snapshot created during logical slot
-         initialization. <literal>EXPORT_SNAPSHOT</literal>, which is the default,
+         initialization. <literal>'export'</literal>, which is the default,
          will export the snapshot for use in other sessions. This option can't
-         be used inside a transaction.  <literal>USE_SNAPSHOT</literal> will use the
+         be used inside a transaction.  <literal>'use'</literal> will use the
          snapshot for the current transaction executing the command. This
          option must be used in a transaction, and
          <literal>CREATE_REPLICATION_SLOT</literal> must be the first command
-         run in that transaction.  Finally, <literal>NOEXPORT_SNAPSHOT</literal> will
+         run in that transaction.  Finally, <literal>'nothing'</literal> will
          just use the snapshot for logical decoding as normal but won't do
          anything else with it.
         </para>
@@ -2052,6 +2056,17 @@ The commands accepted in replication mode are:
     </listitem>
   </varlistentry>
 
+  <varlistentry>
+    <term><literal>CREATE_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> [ <literal>TEMPORARY</literal> ] { <literal>PHYSICAL</literal> [ <literal>RESERVE_WAL</literal> ] | <literal>LOGICAL</literal> <replaceable class="parameter">output_plugin</replaceable> [ <literal>EXPORT_SNAPSHOT</literal> | <literal>NOEXPORT_SNAPSHOT</literal> | <literal>USE_SNAPSHOT</literal> | <literal>TWO_PHASE</literal> ] }
+    </term>
+    <listitem>
+     <para>
+      For compatibility with older releases, this alternative syntax for
+      the <literal>CREATE_REPLICATION_SLOT</literal> command is still supported.
+     </para>
+    </listitem>
+  </varlistentry>
+
   <varlistentry>
     <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ]
      <indexterm><primary>START_REPLICATION</primary></indexterm>
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 19ea159af4..5c6e56a5b2 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -862,6 +862,9 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 	PGresult   *res;
 	StringInfoData cmd;
 	char	   *snapshot;
+	int			use_new_options_syntax;
+
+	use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
 
 	initStringInfo(&cmd);
 
@@ -872,26 +875,58 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 
 	if (conn->logical)
 	{
-		appendStringInfoString(&cmd, " LOGICAL pgoutput");
+		appendStringInfoString(&cmd, " LOGICAL pgoutput ");
+		if (use_new_options_syntax)
+			appendStringInfoChar(&cmd, '(');
 		if (two_phase)
-			appendStringInfoString(&cmd, " TWO_PHASE");
+		{
+			appendStringInfoString(&cmd, "TWO_PHASE");
+			if (use_new_options_syntax)
+				appendStringInfoString(&cmd, ", ");
+			else
+				appendStringInfoChar(&cmd, ' ');
+		}
 
-		switch (snapshot_action)
+		if (use_new_options_syntax)
 		{
-			case CRS_EXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
-				break;
-			case CRS_NOEXPORT_SNAPSHOT:
-				appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
-				break;
-			case CRS_USE_SNAPSHOT:
-				appendStringInfoString(&cmd, " USE_SNAPSHOT");
-				break;
+			switch (snapshot_action)
+			{
+				case CRS_EXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "SNAPSHOT 'export'");
+					break;
+				case CRS_NOEXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
+					break;
+				case CRS_USE_SNAPSHOT:
+					appendStringInfoString(&cmd, "SNAPSHOT 'use'");
+					break;
+			}
 		}
+		else
+		{
+			switch (snapshot_action)
+			{
+				case CRS_EXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
+					break;
+				case CRS_NOEXPORT_SNAPSHOT:
+					appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
+					break;
+				case CRS_USE_SNAPSHOT:
+					appendStringInfoString(&cmd, "USE_SNAPSHOT");
+					break;
+			}
+		}
+
+		if (use_new_options_syntax)
+			appendStringInfoChar(&cmd, ')');
 	}
 	else
 	{
-		appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+		if (use_new_options_syntax)
+			appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
+		else
+			appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
 	}
 
 	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 3b59d62ed8..126380e2df 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -103,8 +103,8 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>	plugin_opt_arg
 %type <str>		opt_slot var_name ident_or_keyword
 %type <boolval>	opt_temporary
-%type <list>	create_slot_opt_list
-%type <defelt>	create_slot_opt
+%type <list>	create_slot_options create_slot_legacy_opt_list
+%type <defelt>	create_slot_legacy_opt
 
 %%
 
@@ -243,8 +243,8 @@ base_backup_legacy_opt:
 			;
 
 create_replication_slot:
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL RESERVE_WAL */
-			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL [options] */
+			K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_PHYSICAL create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -254,8 +254,8 @@ create_replication_slot:
 					cmd->options = $5;
 					$$ = (Node *) cmd;
 				}
-			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */
-			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list
+			/* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin [options] */
+			| K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_options
 				{
 					CreateReplicationSlotCmd *cmd;
 					cmd = makeNode(CreateReplicationSlotCmd);
@@ -268,28 +268,33 @@ create_replication_slot:
 				}
 			;
 
-create_slot_opt_list:
-			create_slot_opt_list create_slot_opt
+create_slot_options:
+			'(' generic_option_list ')'			{ $$ = $2; }
+			| create_slot_legacy_opt_list		{ $$ = $1; }
+			;
+
+create_slot_legacy_opt_list:
+			create_slot_legacy_opt_list create_slot_legacy_opt
 				{ $$ = lappend($1, $2); }
 			| /* EMPTY */
 				{ $$ = NIL; }
 			;
 
-create_slot_opt:
+create_slot_legacy_opt:
 			K_EXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("export"), -1);
 				}
 			| K_NOEXPORT_SNAPSHOT
 				{
-				  $$ = makeDefElem("export_snapshot",
-								   (Node *)makeInteger(false), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("nothing"), -1);
 				}
 			| K_USE_SNAPSHOT
 				{
-				  $$ = makeDefElem("use_snapshot",
-								   (Node *)makeInteger(true), -1);
+				  $$ = makeDefElem("snapshot",
+								   (Node *)makeString("use"), -1);
 				}
 			| K_RESERVE_WAL
 				{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3ca2a11389..b811a5c0ef 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -872,26 +872,30 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 	{
 		DefElem    *defel = (DefElem *) lfirst(lc);
 
-		if (strcmp(defel->defname, "export_snapshot") == 0)
+		if (strcmp(defel->defname, "snapshot") == 0)
 		{
+			char	   *action;
+
 			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 
+			action = defGetString(defel);
 			snapshot_action_given = true;
-			*snapshot_action = defGetBoolean(defel) ? CRS_EXPORT_SNAPSHOT :
-				CRS_NOEXPORT_SNAPSHOT;
-		}
-		else if (strcmp(defel->defname, "use_snapshot") == 0)
-		{
-			if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
+
+			if (strcmp(action, "export") == 0)
+				*snapshot_action = CRS_EXPORT_SNAPSHOT;
+			else if (strcmp(action, "nothing") == 0)
+				*snapshot_action = CRS_NOEXPORT_SNAPSHOT;
+			else if (strcmp(action, "use") == 0)
+				*snapshot_action = CRS_USE_SNAPSHOT;
+			else
 				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						 errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
+								defel->defname, action)));
 
-			snapshot_action_given = true;
-			*snapshot_action = CRS_USE_SNAPSHOT;
 		}
 		else if (strcmp(defel->defname, "reserve_wal") == 0)
 		{
@@ -901,7 +905,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						 errmsg("conflicting or redundant options")));
 
 			reserve_wal_given = true;
-			*reserve_wal = true;
+			*reserve_wal = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
@@ -910,7 +914,7 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
 			two_phase_given = true;
-			*two_phase = true;
+			*two_phase = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
@@ -980,7 +984,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... EXPORT_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));
 
 			need_full_snapshot = true;
 		}
@@ -990,25 +994,25 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called inside a transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (XactIsoLevel != XACT_REPEATABLE_READ)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (FirstSnapshotSet)
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must be called before any query",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			if (IsSubTransaction())
 				ereport(ERROR,
 				/*- translator: %s is a CREATE_REPLICATION_SLOT statement */
 						(errmsg("%s must not be called in a subtransaction",
-								"CREATE_REPLICATION_SLOT ... USE_SNAPSHOT")));
+								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
 
 			need_full_snapshot = true;
 		}
diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c
index d782b81adc..37237cd5d9 100644
--- a/src/bin/pg_basebackup/streamutil.c
+++ b/src/bin/pg_basebackup/streamutil.c
@@ -490,6 +490,7 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 {
 	PQExpBuffer query;
 	PGresult   *res;
+	bool		use_new_option_syntax = (PQserverVersion(conn) >= 150000);
 
 	query = createPQExpBuffer();
 
@@ -498,27 +499,54 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin,
 	Assert(!(two_phase && is_physical));
 	Assert(slot_name != NULL);
 
-	/* Build query */
+	/* Build base portion of query */
 	appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
 	if (is_temporary)
 		appendPQExpBufferStr(query, " TEMPORARY");
 	if (is_physical)
-	{
 		appendPQExpBufferStr(query, " PHYSICAL");
+	else
+		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
+
+	/* Add any requested options */
+	if (use_new_option_syntax)
+		appendPQExpBufferStr(query, " (");
+	if (is_physical)
+	{
 		if (reserve_wal)
-			appendPQExpBufferStr(query, " RESERVE_WAL");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "RESERVE_WAL");
 	}
 	else
 	{
-		appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 		if (two_phase && PQserverVersion(conn) >= 150000)
-			appendPQExpBufferStr(query, " TWO_PHASE");
+			AppendPlainCommandOption(query, use_new_option_syntax,
+									 "TWO_PHASE");
 
 		if (PQserverVersion(conn) >= 100000)
+		{
 			/* pg_recvlogical doesn't use an exported snapshot, so suppress */
-			appendPQExpBufferStr(query, " NOEXPORT_SNAPSHOT");
+			if (use_new_option_syntax)
+				AppendStringCommandOption(query, use_new_option_syntax,
+										   "SNAPSHOT", "nothing");
+			else
+				AppendPlainCommandOption(query, use_new_option_syntax,
+										 "NOEXPORT_SNAPSHOT");
+		}
+	}
+	if (use_new_option_syntax)
+	{
+		/* Suppress option list if it would be empty, otherwise terminate */
+		if (query->data[query->len - 1] == '(')
+		{
+			query->len -= 2;
+			query->data[query->len] = '\0';
+		}
+		else
+			appendPQExpBufferChar(query, ')');
 	}
 
+	/* Now run the query */
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-- 
2.24.3 (Apple Git-128)