libpqrcv_PQexec_v3.patch

*** a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
--- b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
***************
*** 54,59 **** static void libpqrcv_disconnect(void);
--- 54,60 ----
  
  /* Prototypes for private functions */
  static bool libpq_select(int timeout_ms);
+ static PGresult *libpqrcv_PQexec(const char *query);
  
  /*
   * Module load callback
***************
*** 97,103 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
  	 * Get the system identifier and timeline ID as a DataRow message from the
  	 * primary server.
  	 */
! 	res = PQexec(streamConn, "IDENTIFY_SYSTEM");
  	if (PQresultStatus(res) != PGRES_TUPLES_OK)
  	{
  		PQclear(res);
--- 98,104 ----
  	 * Get the system identifier and timeline ID as a DataRow message from the
  	 * primary server.
  	 */
! 	res = libpqrcv_PQexec("IDENTIFY_SYSTEM");
  	if (PQresultStatus(res) != PGRES_TUPLES_OK)
  	{
  		PQclear(res);
***************
*** 149,159 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
  	/* Start streaming from the point requested by startup process */
  	snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X",
  			 startpoint.xlogid, startpoint.xrecoff);
! 	res = PQexec(streamConn, cmd);
  	if (PQresultStatus(res) != PGRES_COPY_OUT)
  		ereport(ERROR,
  				(errmsg("could not start WAL streaming: %s",
  						PQerrorMessage(streamConn))));
  	PQclear(res);
  
  	justconnected = true;
--- 150,163 ----
  	/* Start streaming from the point requested by startup process */
  	snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X",
  			 startpoint.xlogid, startpoint.xrecoff);
! 	res = libpqrcv_PQexec(cmd);
  	if (PQresultStatus(res) != PGRES_COPY_OUT)
+ 	{
+ 		PQclear(res);
  		ereport(ERROR,
  				(errmsg("could not start WAL streaming: %s",
  						PQerrorMessage(streamConn))));
+ 	}
  	PQclear(res);
  
  	justconnected = true;
***************
*** 225,230 **** libpq_select(int timeout_ms)
--- 229,311 ----
  }
  
  /*
+  * Send a query and wait for the results by using the asynchronous libpq
+  * functions and the backend version of select().
+  *
+  * We must not use the regular blocking libpq functions like PQexec()
+  * since they are uninterruptible by signals on some platforms, such as
+  * Windows.
+  *
+  * We must also not use vanilla select() here since it cannot handle the
+  * signal emulation layer on Windows.
+  *
+  * The function is modeled on PQexec() in libpq, but only implements
+  * those parts that are in use in the walreceiver.
+  *
+  * Queries are always executed on the connection in streamConn.
+  */
+ static PGresult *
+ libpqrcv_PQexec(const char *query)
+ {
+ 	PGresult   *res = NULL;
+ 
+ 	/*
+ 	 * PQexec() silently discards any prior query results on the
+ 	 * connection. This is not required for walreceiver since it's
+ 	 * expected that walsender won't generate any such junk results.
+ 	 */
+ 
+ 	/*
+ 	 * Submit a query. Since we don't use non-blocking mode, this also
+ 	 * can block. But its risk is relatively small, so we ignore that
+ 	 * for now.
+ 	 */
+ 	if (!PQsendQuery(streamConn, query))
+ 		return NULL;
+ 
+ 	for (;;)
+ 	{
+ 		PGresult   *next;
+ 
+ 		/*
+ 		 * Receive data until PQgetResult is ready to get the result
+ 		 * without blocking.
+ 		 */
+ 		while (PQisBusy(streamConn))
+ 		{
+ 			/*
+ 			 * We don't need to break down the sleep into smaller increments,
+ 			 * and check for interrupts after each nap, since we can just
+ 			 * elog(FATAL) within SIGTERM signal handler if the signal
+ 			 * arrives in the middle of establishment of replication connection.
+ 			 */
+ 			if (!libpq_select(-1))
+ 				continue;		/* interrupted */
+ 			if (PQconsumeInput(streamConn) == 0)
+ 				return NULL;	/* trouble */
+ 		}
+ 
+ 		/*
+ 		 * Don't emulate the PQexec()'s behavior of returning the last
+ 		 * result when there are many, since walreceiver never sends a
+ 		 * query returning multiple results.
+ 		 */
+ 		if ((next = PQgetResult(streamConn)) == NULL)
+ 			break;				/* query is complete */
+ 		if (PQresultStatus(next) == PGRES_FATAL_ERROR)
+ 			return next;
+ 		PQclear(res);
+ 		res = next;
+ 		if (PQresultStatus(res) == PGRES_COPY_IN ||
+ 			PQresultStatus(res) == PGRES_COPY_OUT ||
+ 			PQstatus(streamConn) == CONNECTION_BAD)
+ 			break;
+ 	}
+ 
+ 	return res;
+ }
+ 
+ /*
   * Disconnect connection to primary, if any.
   */
  static void
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 86,93 **** static void DisableWalRcvImmediateExit(void);
   * We can't just exit(1) within SIGTERM signal handler, because the signal
   * might arrive in the middle of some critical operation, like while we're
   * holding a spinlock. We also can't just set a flag in signal handler and
!  * check it in the main loop, because we perform some blocking libpq
!  * operations like PQexec(), which can take a long time to finish.
   *
   * We use a combined approach: When WalRcvImmediateInterruptOK is true, it's
   * safe for the signal handler to elog(FATAL) immediately. Otherwise it just
--- 86,93 ----
   * We can't just exit(1) within SIGTERM signal handler, because the signal
   * might arrive in the middle of some critical operation, like while we're
   * holding a spinlock. We also can't just set a flag in signal handler and
!  * check it in the main loop, because we perform some blocking operations
!  * like libpqrcv_PQexec(), which can take a long time to finish.
   *
   * We use a combined approach: When WalRcvImmediateInterruptOK is true, it's
   * safe for the signal handler to elog(FATAL) immediately. Otherwise it just