repl_parser.patch

*** a/doc/src/sgml/protocol.sgml
--- b/doc/src/sgml/protocol.sgml
***************
*** 1460,1475 **** The commands accepted in walsender mode are:
    </varlistentry>
  
    <varlistentry>
!     <term>BASE_BACKUP <replaceable>options</><literal>;</><replaceable>label</></term>
      <listitem>
       <para>
        Instructs the server to start streaming a base backup.
!       The system will automatically be put in backup mode with the label
!       specified in <replaceable>label</> before the backup is started, and
!       taken out of it when the backup is complete. The following options
!       are accepted:
        <variablelist>
         <varlistentry>
          <term><literal>PROGRESS</></term>
          <listitem>
           <para>
--- 1460,1486 ----
    </varlistentry>
  
    <varlistentry>
!     <term>BASE_BACKUP [<literal>LABEL</literal> <replaceable>'label'</replaceable>] [<literal>PROGRESS</literal>]</term>
      <listitem>
       <para>
        Instructs the server to start streaming a base backup.
!       The system will automatically be put in backup mode before the backup
!       is started, and taken out of it when the backup is complete. The
!       following options are accepted:
        <variablelist>
         <varlistentry>
+         <term><literal>LABEL</literal> <replaceable>'label'</replaceable></term>
+         <listitem>
+          <para>
+           Sets the label of the backup. If none is specified, a backup label
+           of <literal>base backup</literal> will be used. The quoting rules
+           for the label are the same as a standard SQL string with
+           <xref linkend="guc-standard-conforming-strings"> turned on.
+          </para>
+         </listitem>
+        </varlistentry>
+ 
+        <varlistentry>
          <term><literal>PROGRESS</></term>
          <listitem>
           <para>
*** a/src/backend/replication/Makefile
--- b/src/backend/replication/Makefile
***************
*** 12,17 **** subdir = src/backend/replication
  top_builddir = ../../..
  include $(top_builddir)/src/Makefile.global
  
! OBJS = walsender.o walreceiverfuncs.o walreceiver.o basebackup.o
  
  include $(top_srcdir)/src/backend/common.mk
--- 12,40 ----
  top_builddir = ../../..
  include $(top_builddir)/src/Makefile.global
  
! OBJS = walsender.o walreceiverfuncs.o walreceiver.o basebackup.o \
! 	repl_gram.o
  
  include $(top_srcdir)/src/backend/common.mk
+ 
+ # repl_scanner is compiled as part of repl_gram
+ repl_gram.o: repl_scanner.c
+ 
+ # See notes in src/backend/parser/Makefile about the following two rules
+ 
+ repl_gram.c: repl_gram.y
+ ifdef BISON
+ 	$(BISON) -d $(BISONFLAGS) -o $@ $<
+ else
+ 	@$(missing) bison $< $@
+ endif
+ 
+ repl_scanner.c: repl_scanner.l
+ ifdef FLEX
+ 	$(FLEX) $(FLEXFLAGS) -o'$@' $<
+ else
+ 	@$(missing) flex $< $@
+ endif
+ 
+ # repl_gram.c and repl_scanner.c are in the distribution tarball, so
+ # they are not cleaned here.
*** a/src/backend/replication/basebackup.c
--- b/src/backend/replication/basebackup.c
***************
*** 64,75 **** base_backup_cleanup(int code, Datum arg)
   * pg_stop_backup() for the user.
   */
  void
! SendBaseBackup(const char *options)
  {
  	DIR		   *dir;
  	struct dirent *de;
- 	char	   *backup_label = strchr(options, ';');
- 	bool		progress = false;
  	List	   *tablespaces = NIL;
  	tablespaceinfo *ti;
  	MemoryContext backup_context;
--- 64,73 ----
   * pg_stop_backup() for the user.
   */
  void
! SendBaseBackup(const char *backup_label, bool progress)
  {
  	DIR		   *dir;
  	struct dirent *de;
  	List	   *tablespaces = NIL;
  	tablespaceinfo *ti;
  	MemoryContext backup_context;
***************
*** 83,100 **** SendBaseBackup(const char *options)
  	old_context = MemoryContextSwitchTo(backup_context);
  
  	if (backup_label == NULL)
! 		ereport(FATAL,
! 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
! 				 errmsg("invalid base backup options: %s", options)));
! 	backup_label++;				/* Walk past the semicolon */
! 
! 	/* Currently the only option string supported is PROGRESS */
! 	if (strncmp(options, "PROGRESS", 8) == 0)
! 		progress = true;
! 	else if (options[0] != ';')
! 		ereport(FATAL,
! 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
! 				 errmsg("invalid base backup options: %s", options)));
  
  	/* Make sure we can open the directory with tablespaces in it */
  	dir = AllocateDir("pg_tblspc");
--- 81,87 ----
  	old_context = MemoryContextSwitchTo(backup_context);
  
  	if (backup_label == NULL)
! 		backup_label = "base backup";
  
  	/* Make sure we can open the directory with tablespaces in it */
  	dir = AllocateDir("pg_tblspc");
*** /dev/null
--- b/src/backend/replication/repl_gram.y
***************
*** 0 ****
--- 1,143 ----
+ %{
+ /*-------------------------------------------------------------------------
+  *
+  * repl_gram.y				- Parser for the replication commands
+  *
+  * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+  * Portions Copyright (c) 1994, Regents of the University of California
+  *
+  *
+  * IDENTIFICATION
+  *	  src/backend/replication/repl_gram.y
+  *
+  *-------------------------------------------------------------------------
+  */
+ 
+ #include "postgres.h"
+ 
+ #include "replication/replnodes.h"
+ #include "replication/walsender.h"
+ 
+ /* Result of the parsing is returned here */
+ Node *replication_parse_result;
+ 
+ /* Location tracking support --- simpler than bison's default */
+ #define YYLLOC_DEFAULT(Current, Rhs, N) \
+ 	do { \
+ 		if (N) \
+ 			(Current) = (Rhs)[1]; \
+ 		else \
+ 			(Current) = (Rhs)[0]; \
+ 	} while (0)
+ 
+ /*
+  * Bison doesn't allocate anything that needs to live across parser calls,
+  * so we can easily have it use palloc instead of malloc.  This prevents
+  * memory leaks if we error out during parsing.  Note this only works with
+  * bison >= 2.0.  However, in bison 1.875 the default is to use alloca()
+  * if possible, so there's not really much problem anyhow, at least if
+  * you're building with gcc.
+  */
+ #define YYMALLOC palloc
+ #define YYFREE   pfree
+ 
+ #define parser_yyerror(msg)  replication_yyerror(msg, yyscanner)
+ #define parser_errposition(pos)  replication_scanner_errposition(pos)
+ 
+ %}
+ 
+ %expect 0
+ %name-prefix="replication_yy"
+ 
+ %union {
+ 		char					*str;
+ 		bool					boolval;
+ 
+ 		XLogRecPtr				recptr;
+ 		Node					*node;
+ }
+ 
+ /* Non-keyword tokens */
+ %token <str> SCONST
+ %token <recptr> RECPTR
+ 
+ /* Keyword tokens. */
+ %token K_BASE_BACKUP
+ %token K_IDENTIFY_SYSTEM
+ %token K_LABEL
+ %token K_PROGRESS
+ %token K_START_REPLICATION
+ 
+ %type <node>	command
+ %type <node>	base_backup start_replication identify_system
+ %type <boolval>	opt_progress
+ %type <str>     opt_label
+ 
+ %%
+ 
+ firstcmd: command opt_semicolon
+ 				{
+ 					replication_parse_result = $1;
+ 				}
+ 			;
+ 
+ opt_semicolon:	';'
+ 				| /* EMPTY */
+ 				;
+ 
+ command:
+ 			identify_system
+ 			| base_backup
+ 			| start_replication
+ 			;
+ 
+ /*
+  * IDENTIFY_SYSTEM
+  */
+ identify_system:
+ 			K_IDENTIFY_SYSTEM
+ 				{
+ 					$$ = (Node *) makeNode(IdentifySystemCmd);
+ 				}
+ 			;
+ 
+ /*
+  * BASE_BACKUP [LABEL <label>] [PROGRESS]
+  */
+ base_backup:
+ 			K_BASE_BACKUP opt_label opt_progress
+ 				{
+ 					BaseBackupCmd *cmd = (BaseBackupCmd *) makeNode(BaseBackupCmd);
+ 
+ 					cmd->label = $2;
+ 					cmd->progress = $3;
+ 
+ 					$$ = (Node *) cmd;
+ 				}
+ 			;
+ 
+ opt_label: K_LABEL SCONST { $$ = $2; }
+ 			| /* EMPTY */		{ $$ = NULL; }
+ 			;
+ 
+ opt_progress: K_PROGRESS		{ $$ = true; }
+ 			| /* EMPTY */		{ $$ = false; }
+ 			;
+ 
+ /*
+  * START_REPLICATION %X/%X
+  */
+ start_replication:
+ 			K_START_REPLICATION RECPTR
+ 				{
+ 					StartReplicationCmd *cmd;
+ 
+ 					cmd = makeNode(StartReplicationCmd);
+ 					cmd->startpoint = $2;
+ 
+ 					$$ = (Node *) cmd;
+ 				}
+ 			;
+ %%
+ 
+ #include "repl_scanner.c"
*** /dev/null
--- b/src/backend/replication/repl_scanner.l
***************
*** 0 ****
--- 1,165 ----
+ %{
+ /*-------------------------------------------------------------------------
+  *
+  * repl_scanner.l
+  *	  a lexical scanner for the replication commands
+  *
+  * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+  * Portions Copyright (c) 1994, Regents of the University of California
+  *
+  *
+  * IDENTIFICATION
+  *	  src/backend/replication/repl_scanner.l
+  *
+  *-------------------------------------------------------------------------
+  */
+ #include "postgres.h"
+ 
+ /* Avoid exit() on fatal scanner errors (a bit ugly -- see yy_fatal_error) */
+ #undef fprintf
+ #define fprintf(file, fmt, msg)  ereport(ERROR, (errmsg_internal("%s", msg)))
+ 
+ /* Handle to the buffer that the lexer uses internally */
+ static YY_BUFFER_STATE scanbufhandle;
+ 
+ static StringInfoData litbuf;
+ 
+ static void startlit(void);
+ static char *litbufdup(void);
+ static void addlit(char *ytext, int yleng);
+ static void addlitchar(unsigned char ychar);
+ 
+ %}
+ 
+ %option 8bit
+ %option never-interactive
+ %option nodefault
+ %option noinput
+ %option nounput
+ %option noyywrap
+ %option warn
+ %option prefix="replication_yy"
+ 
+ %x xq
+ 
+ /* Extended quote
+  * xqdouble implements embedded quote, ''''
+  */
+ xqstart			{quote}
+ xqdouble		{quote}{quote}
+ xqinside		[^']+
+ 
+ hexdigit		[0-9A-Za-z]+
+ 
+ quote			'
+ quotestop		{quote}
+ 
+ %%
+ 
+ BASE_BACKUP			{ return K_BASE_BACKUP; }
+ IDENTIFY_SYSTEM		{ return K_IDENTIFY_SYSTEM; }
+ LABEL			{ return K_LABEL; }
+ PROGRESS			{ return K_PROGRESS; }
+ START_REPLICATION	{ return K_START_REPLICATION; }
+ ","				{ return ','; }
+ ";"				{ return ';'; }
+ 
+ [\n]			;
+ [\t]			;
+ " "				;
+ 
+ {hexdigit}+\/{hexdigit}+		{
+ 					yylval.str = pstrdup(yytext);
+ 					if (sscanf(yytext, "%X/%X", &yylval.recptr.xlogid, &yylval.recptr.xrecoff) != 2)
+ 						yyerror("invalid streaming start location");
+ 					return RECPTR;
+ 				}
+ 
+ {xqstart}		{
+ 					BEGIN(xq);
+ 					startlit();
+ 				}
+ <xq>{quotestop}	{
+ 					yyless(1);
+ 					BEGIN(INITIAL);
+ 					yylval.str = litbufdup();
+ 					return SCONST;
+ 				}
+ <xq>{xqdouble} {
+ 					addlitchar('\'');
+ 				}
+ <xq>{xqinside}  {
+ 					addlit(yytext, yyleng);
+ 				}
+ 
+ <xq><<EOF>>		{ yyerror("unterminated quoted string"); }
+ 
+ 
+ <<EOF>>			{
+ 					yyterminate();
+ 				}
+ 
+ .				{
+ 					elog(ERROR, "syntax error: unexpected character \"%s\"", yytext);
+ 				}
+ %%
+ 
+ 
+ static void
+ startlit(void)
+ {
+   initStringInfo(&litbuf);
+ }
+ 
+ static char *
+ litbufdup(void)
+ {
+   return litbuf.data;
+ }
+ 
+ static void
+ addlit(char *ytext, int yleng)
+ {
+ 	appendBinaryStringInfo(&litbuf, ytext, yleng);
+ }
+ 
+ static void
+ addlitchar(unsigned char ychar)
+ {
+     appendStringInfoChar(&litbuf, ychar);
+ }
+ 
+ void
+ yyerror(const char *message)
+ {
+ 	elog(ERROR, "%s", message);
+ }
+ 
+ 
+ void
+ replication_scanner_init(const char *str)
+ {
+ 	Size		slen = strlen(str);
+ 	char	   *scanbuf;
+ 
+ 	/*
+ 	 * Might be left over after ereport()
+ 	 */
+ 	if (YY_CURRENT_BUFFER)
+ 		yy_delete_buffer(YY_CURRENT_BUFFER);
+ 
+ 	/*
+ 	 * Make a scan buffer with special termination needed by flex.
+ 	 */
+ 	scanbuf = (char *) palloc(slen + 2);
+ 	memcpy(scanbuf, str, slen);
+ 	scanbuf[slen] = scanbuf[slen + 1] = YY_END_OF_BUFFER_CHAR;
+ 	scanbufhandle = yy_scan_buffer(scanbuf, slen + 2);
+ }
+ 
+ void
+ replication_scanner_finish()
+ {
+ 	yy_delete_buffer(scanbufhandle);
+ 	scanbufhandle = NULL;
+ }
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 45,50 ****
--- 45,51 ----
  #include "libpq/pqsignal.h"
  #include "miscadmin.h"
  #include "replication/basebackup.h"
+ #include "replication/replnodes.h"
  #include "replication/walprotocol.h"
  #include "replication/walsender.h"
  #include "storage/fd.h"
***************
*** 99,104 **** static void WalSndXLogSendHandler(SIGNAL_ARGS);
--- 100,106 ----
  static void WalSndLastCycleHandler(SIGNAL_ARGS);
  
  /* Prototypes for private functions */
+ static bool HandleReplicationCommand(const char *cmd_string);
  static int	WalSndLoop(void);
  static void InitWalSnd(void);
  static void WalSndHandshake(void);
***************
*** 106,111 **** static void WalSndKill(int code, Datum arg);
--- 108,115 ----
  static void XLogRead(char *buf, XLogRecPtr recptr, Size nbytes);
  static bool XLogSend(char *msgbuf, bool *caughtup);
  static void CheckClosedConnection(void);
+ static void IdentifySystem(void);
+ static void StartReplication(StartReplicationCmd * cmd);
  
  
  /* Main entry point for walsender process */
***************
*** 215,332 **** WalSndHandshake(void)
  			case 'Q':			/* Query message */
  				{
  					const char *query_string;
- 					XLogRecPtr	recptr;
  
  					query_string = pq_getmsgstring(&input_message);
  					pq_getmsgend(&input_message);
  
! 					if (strcmp(query_string, "IDENTIFY_SYSTEM") == 0)
! 					{
! 						StringInfoData buf;
! 						char		sysid[32];
! 						char		tli[11];
! 
! 						/*
! 						 * Reply with a result set with one row, two columns.
! 						 * First col is system ID, and second is timeline ID
! 						 */
! 
! 						snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
! 								 GetSystemIdentifier());
! 						snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
! 
! 						/* Send a RowDescription message */
! 						pq_beginmessage(&buf, 'T');
! 						pq_sendint(&buf, 2, 2); /* 2 fields */
! 
! 						/* first field */
! 						pq_sendstring(&buf, "systemid");		/* col name */
! 						pq_sendint(&buf, 0, 4); /* table oid */
! 						pq_sendint(&buf, 0, 2); /* attnum */
! 						pq_sendint(&buf, TEXTOID, 4);	/* type oid */
! 						pq_sendint(&buf, -1, 2);		/* typlen */
! 						pq_sendint(&buf, 0, 4); /* typmod */
! 						pq_sendint(&buf, 0, 2); /* format code */
! 
! 						/* second field */
! 						pq_sendstring(&buf, "timeline");		/* col name */
! 						pq_sendint(&buf, 0, 4); /* table oid */
! 						pq_sendint(&buf, 0, 2); /* attnum */
! 						pq_sendint(&buf, INT4OID, 4);	/* type oid */
! 						pq_sendint(&buf, 4, 2); /* typlen */
! 						pq_sendint(&buf, 0, 4); /* typmod */
! 						pq_sendint(&buf, 0, 2); /* format code */
! 						pq_endmessage(&buf);
! 
! 						/* Send a DataRow message */
! 						pq_beginmessage(&buf, 'D');
! 						pq_sendint(&buf, 2, 2); /* # of columns */
! 						pq_sendint(&buf, strlen(sysid), 4);		/* col1 len */
! 						pq_sendbytes(&buf, (char *) &sysid, strlen(sysid));
! 						pq_sendint(&buf, strlen(tli), 4);		/* col2 len */
! 						pq_sendbytes(&buf, (char *) tli, strlen(tli));
! 						pq_endmessage(&buf);
! 
! 						/* Send CommandComplete and ReadyForQuery messages */
! 						EndCommand("SELECT", DestRemote);
! 						ReadyForQuery(DestRemote);
! 						/* ReadyForQuery did pq_flush for us */
! 					}
! 					else if (sscanf(query_string, "START_REPLICATION %X/%X",
! 									&recptr.xlogid, &recptr.xrecoff) == 2)
! 					{
! 						StringInfoData buf;
! 
! 						/*
! 						 * Check that we're logging enough information in the
! 						 * WAL for log-shipping.
! 						 *
! 						 * NOTE: This only checks the current value of
! 						 * wal_level. Even if the current setting is not
! 						 * 'minimal', there can be old WAL in the pg_xlog
! 						 * directory that was created with 'minimal'. So this
! 						 * is not bulletproof, the purpose is just to give a
! 						 * user-friendly error message that hints how to
! 						 * configure the system correctly.
! 						 */
! 						if (wal_level == WAL_LEVEL_MINIMAL)
! 							ereport(FATAL,
! 									(errcode(ERRCODE_CANNOT_CONNECT_NOW),
! 									 errmsg("standby connections not allowed because wal_level=minimal")));
! 
! 						/* Send a CopyBothResponse message, and start streaming */
! 						pq_beginmessage(&buf, 'W');
! 						pq_sendbyte(&buf, 0);
! 						pq_sendint(&buf, 0, 2);
! 						pq_endmessage(&buf);
! 						pq_flush();
! 
! 						/*
! 						 * Initialize position to the received one, then the
! 						 * xlog records begin to be shipped from that position
! 						 */
! 						sentPtr = recptr;
! 
! 						/* break out of the loop */
  						replication_started = true;
- 					}
- 					else if (strncmp(query_string, "BASE_BACKUP ", 12) == 0)
- 					{
- 						/* Command is BASE_BACKUP <options>;<label> */
- 						SendBaseBackup(query_string + strlen("BASE_BACKUP "));
- 						/* Send CommandComplete and ReadyForQuery messages */
- 						EndCommand("SELECT", DestRemote);
- 						ReadyForQuery(DestRemote);
- 						/* ReadyForQuery did pq_flush for us */
- 					}
- 					else
- 					{
- 						ereport(FATAL,
- 								(errcode(ERRCODE_PROTOCOL_VIOLATION),
- 								 errmsg("invalid standby query string: %s", query_string)));
- 					}
- 					break;
  				}
  
  			case 'X':
  				/* standby is closing the connection */
--- 219,232 ----
  			case 'Q':			/* Query message */
  				{
  					const char *query_string;
  
  					query_string = pq_getmsgstring(&input_message);
  					pq_getmsgend(&input_message);
  
! 					if (HandleReplicationCommand(query_string))
  						replication_started = true;
  				}
+ 				break;
  
  			case 'X':
  				/* standby is closing the connection */
***************
*** 348,353 **** WalSndHandshake(void)
--- 248,414 ----
  }
  
  /*
+  * IDENTIFY_SYSTEM
+  */
+ static void
+ IdentifySystem(void)
+ {
+ 	StringInfoData buf;
+ 	char		sysid[32];
+ 	char		tli[11];
+ 
+ 	/*
+ 	 * Reply with a result set with one row, two columns. First col is system
+ 	 * ID, and second is timeline ID
+ 	 */
+ 
+ 	snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
+ 			 GetSystemIdentifier());
+ 	snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
+ 
+ 	/* Send a RowDescription message */
+ 	pq_beginmessage(&buf, 'T');
+ 	pq_sendint(&buf, 2, 2);		/* 2 fields */
+ 
+ 	/* first field */
+ 	pq_sendstring(&buf, "systemid");	/* col name */
+ 	pq_sendint(&buf, 0, 4);		/* table oid */
+ 	pq_sendint(&buf, 0, 2);		/* attnum */
+ 	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+ 	pq_sendint(&buf, -1, 2);	/* typlen */
+ 	pq_sendint(&buf, 0, 4);		/* typmod */
+ 	pq_sendint(&buf, 0, 2);		/* format code */
+ 
+ 	/* second field */
+ 	pq_sendstring(&buf, "timeline");	/* col name */
+ 	pq_sendint(&buf, 0, 4);		/* table oid */
+ 	pq_sendint(&buf, 0, 2);		/* attnum */
+ 	pq_sendint(&buf, INT4OID, 4);		/* type oid */
+ 	pq_sendint(&buf, 4, 2);		/* typlen */
+ 	pq_sendint(&buf, 0, 4);		/* typmod */
+ 	pq_sendint(&buf, 0, 2);		/* format code */
+ 	pq_endmessage(&buf);
+ 
+ 	/* Send a DataRow message */
+ 	pq_beginmessage(&buf, 'D');
+ 	pq_sendint(&buf, 2, 2);		/* # of columns */
+ 	pq_sendint(&buf, strlen(sysid), 4); /* col1 len */
+ 	pq_sendbytes(&buf, (char *) &sysid, strlen(sysid));
+ 	pq_sendint(&buf, strlen(tli), 4);	/* col2 len */
+ 	pq_sendbytes(&buf, (char *) tli, strlen(tli));
+ 	pq_endmessage(&buf);
+ 
+ 	/* Send CommandComplete and ReadyForQuery messages */
+ 	EndCommand("SELECT", DestRemote);
+ 	ReadyForQuery(DestRemote);
+ 	/* ReadyForQuery did pq_flush for us */
+ }
+ 
+ /*
+  * START_REPLICATION
+  */
+ static void
+ StartReplication(StartReplicationCmd * cmd)
+ {
+ 	StringInfoData buf;
+ 
+ 	/*
+ 	 * Check that we're logging enough information in the WAL for
+ 	 * log-shipping.
+ 	 *
+ 	 * NOTE: This only checks the current value of wal_level. Even if the
+ 	 * current setting is not 'minimal', there can be old WAL in the pg_xlog
+ 	 * directory that was created with 'minimal'. So this is not bulletproof,
+ 	 * the purpose is just to give a user-friendly error message that hints
+ 	 * how to configure the system correctly.
+ 	 */
+ 	if (wal_level == WAL_LEVEL_MINIMAL)
+ 		ereport(FATAL,
+ 				(errcode(ERRCODE_CANNOT_CONNECT_NOW),
+ 		errmsg("standby connections not allowed because wal_level=minimal")));
+ 
+ 	/* Send a CopyBothResponse message, and start streaming */
+ 	pq_beginmessage(&buf, 'W');
+ 	pq_sendbyte(&buf, 0);
+ 	pq_sendint(&buf, 0, 2);
+ 	pq_endmessage(&buf);
+ 	pq_flush();
+ 
+ 	/*
+ 	 * Initialize position to the received one, then the xlog records begin to
+ 	 * be shipped from that position
+ 	 */
+ 	sentPtr = cmd->startpoint;
+ }
+ 
+ /*
+  * Execute an incoming replication command.
+  */
+ static bool
+ HandleReplicationCommand(const char *cmd_string)
+ {
+ 	bool		replication_started = false;
+ 	int			parse_rc;
+ 	Node	   *cmd_node;
+ 	MemoryContext cmd_context;
+ 	MemoryContext old_context;
+ 
+ 	elog(DEBUG1, "received replication command: %s", cmd_string);
+ 
+ 	cmd_context = AllocSetContextCreate(CurrentMemoryContext,
+ 										"Replication command context",
+ 										ALLOCSET_DEFAULT_MINSIZE,
+ 										ALLOCSET_DEFAULT_INITSIZE,
+ 										ALLOCSET_DEFAULT_MAXSIZE);
+ 	old_context = MemoryContextSwitchTo(cmd_context);
+ 
+ 	replication_scanner_init(cmd_string);
+ 	parse_rc = replication_yyparse();
+ 	if (parse_rc != 0)
+ 		elog(ERROR, "replication command parser returned %d", parse_rc);
+ 
+ 	cmd_node = replication_parse_result;
+ 
+ 	switch (cmd_node->type)
+ 	{
+ 		case T_IdentifySystemCmd:
+ 			IdentifySystem();
+ 			break;
+ 
+ 		case T_StartReplicationCmd:
+ 			StartReplication((StartReplicationCmd *) cmd_node);
+ 
+ 			/* break out of the loop */
+ 			replication_started = true;
+ 			break;
+ 
+ 		case T_BaseBackupCmd:
+ 			{
+ 				BaseBackupCmd *cmd = (BaseBackupCmd *) cmd_node;
+ 
+ 				SendBaseBackup(cmd->label, cmd->progress);
+ 
+ 				/* Send CommandComplete and ReadyForQuery messages */
+ 				EndCommand("SELECT", DestRemote);
+ 				ReadyForQuery(DestRemote);
+ 				/* ReadyForQuery did pq_flush for us */
+ 				break;
+ 			}
+ 
+ 		default:
+ 			ereport(FATAL,
+ 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
+ 					 errmsg("invalid standby query string: %s", cmd_string)));
+ 	}
+ 
+ 	/* done */
+ 	MemoryContextSwitchTo(old_context);
+ 	MemoryContextDelete(cmd_context);
+ 
+ 	return replication_started;
+ }
+ 
+ /*
   * Check if the remote end has closed the connection.
   */
  static void
*** a/src/include/replication/basebackup.h
--- b/src/include/replication/basebackup.h
***************
*** 12,17 ****
  #ifndef _BASEBACKUP_H
  #define _BASEBACKUP_H
  
! extern void SendBaseBackup(const char *options);
  
  #endif   /* _BASEBACKUP_H */
--- 12,17 ----
  #ifndef _BASEBACKUP_H
  #define _BASEBACKUP_H
  
! extern void SendBaseBackup(const char *backup_label, bool progress);
  
  #endif   /* _BASEBACKUP_H */
*** /dev/null
--- b/src/include/replication/replnodes.h
***************
*** 0 ****
--- 1,63 ----
+ /*-------------------------------------------------------------------------
+  *
+  * replnodes.h
+  *	  definitions for replication grammar parse nodes
+  *
+  *
+  * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+  * Portions Copyright (c) 1994, Regents of the University of California
+  *
+  * src/include/replication/replnodes.h
+  *
+  *-------------------------------------------------------------------------
+  */
+ #ifndef REPLNODES_H
+ #define REPLNODES_H
+ 
+ #include "access/xlogdefs.h"
+ #include "nodes/primnodes.h"
+ #include "nodes/value.h"
+ 
+ /*
+  * NodeTags for replication parser
+  */
+ typedef enum ReplNodeTag
+ {
+ 	T_IdentifySystemCmd = 10,
+ 	T_BaseBackupCmd,
+ 	T_StartReplicationCmd
+ }	ReplNodeTag;
+ 
+ /* ----------------------
+  *		IDENTIFY_SYSTEM command
+  * ----------------------
+  */
+ typedef struct IdentifySystemCmd
+ {
+ 	NodeTag		type;
+ }	IdentifySystemCmd;
+ 
+ 
+ /* ----------------------
+  *		BASE_BACKUP command
+  * ----------------------
+  */
+ typedef struct BaseBackupCmd
+ {
+ 	NodeTag		type;
+ 	char	   *label;
+ 	bool		progress;
+ }	BaseBackupCmd;
+ 
+ 
+ /* ----------------------
+  *		START_REPLICATION command
+  * ----------------------
+  */
+ typedef struct StartReplicationCmd
+ {
+ 	NodeTag		type;
+ 	XLogRecPtr	startpoint;
+ }	StartReplicationCmd;
+ 
+ #endif   /* REPLNODES_H */
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 13,18 ****
--- 13,19 ----
  #define _WALSENDER_H
  
  #include "access/xlog.h"
+ #include "nodes/nodes.h"
  #include "storage/latch.h"
  #include "storage/spin.h"
  
***************
*** 56,59 **** extern void WalSndWakeup(void);
--- 57,72 ----
  
  extern Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS);
  
+ /*
+  * Internal functions for parsing the replication grammar, in repl_gram.y and
+  * repl_scanner.l
+  */
+ extern int	replication_yyparse(void);
+ extern int	replication_yylex(void);
+ extern void replication_yyerror(const char *str);
+ extern void replication_scanner_init(const char *query_string);
+ extern void replication_scanner_finish(void);
+ 
+ extern Node *replication_parse_result;
+ 
  #endif   /* _WALSENDER_H */
*** a/src/tools/msvc/Mkvcbuild.pm
--- b/src/tools/msvc/Mkvcbuild.pm
***************
*** 70,75 **** sub mkvcbuild
--- 70,76 ----
      $postgres->AddFiles('src\backend\parser','scan.l','gram.y');
      $postgres->AddFiles('src\backend\bootstrap','bootscanner.l','bootparse.y');
      $postgres->AddFiles('src\backend\utils\misc','guc-file.l');
+     $postgres->AddFiles('src\backend\replication', 'repl_scanner.l', 'repl_gram.y');
      $postgres->AddDefine('BUILDING_DLL');
      $postgres->AddLibrary('wsock32.lib');
      $postgres->AddLibrary('ws2_32.lib');
*** a/src/tools/msvc/pgbison.bat
--- b/src/tools/msvc/pgbison.bat
***************
*** 17,22 **** goto nobison
--- 17,23 ----
  
  if "%1" == "src\backend\parser\gram.y" call :generate %1 src\backend\parser\gram.c src\backend\parser\gram.h
  if "%1" == "src\backend\bootstrap\bootparse.y" call :generate %1 src\backend\bootstrap\bootparse.c
+ if "%1" == "src\backend\replication\repl_gram.y" call :generate %1 src\backend\replication\repl_gram.c
  if "%1" == "src\pl\plpgsql\src\gram.y" call :generate %1 src\pl\plpgsql\src\pl_gram.c src\pl\plpgsql\src\pl_gram.h
  if "%1" == "src\interfaces\ecpg\preproc\preproc.y" call :generate %1 src\interfaces\ecpg\preproc\preproc.c src\interfaces\ecpg\preproc\preproc.h
  if "%1" == "contrib\cube\cubeparse.y" call :generate %1 contrib\cube\cubeparse.c
*** a/src/tools/msvc/pgflex.bat
--- b/src/tools/msvc/pgflex.bat
***************
*** 13,18 **** if errorlevel 1 goto noflex
--- 13,19 ----
  if "%1" == "src\backend\parser\scan.l" call :generate %1 src\backend\parser\scan.c -CF
  if "%1" == "src\backend\bootstrap\bootscanner.l" call :generate %1 src\backend\bootstrap\bootscanner.c
  if "%1" == "src\backend\utils\misc\guc-file.l" call :generate %1 src\backend\utils\misc\guc-file.c
+ if "%1" == "src\backend\replication\repl_scanner.l" call :generate %1 src\backend\replication\repl_scanner.c
  if "%1" == "src\interfaces\ecpg\preproc\pgc.l" call :generate %1 src\interfaces\ecpg\preproc\pgc.c
  if "%1" == "src\bin\psql\psqlscan.l" call :generate %1 src\bin\psql\psqlscan.c
  if "%1" == "contrib\cube\cubescan.l" call :generate %1 contrib\cube\cubescan.c