libpqrcv_PQexec_v2.patch

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index e451c55..5e759b6 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -54,6 +54,7 @@ static void libpqrcv_disconnect(void);
 
 /* Prototypes for private functions */
 static bool libpq_select(int timeout_ms);
+static PGresult *libpqrcv_PQexec(const char *query);
 
 /*
  * Module load callback
@@ -97,10 +98,11 @@ 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)
+	res = libpqrcv_PQexec("IDENTIFY_SYSTEM");
+	if (res == NULL || PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-		PQclear(res);
+		if (res)
+			PQclear(res);
 		ereport(ERROR,
 				(errmsg("could not receive database system identifier and timeline ID from "
 						"the primary server: %s",
@@ -149,11 +151,15 @@ 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)
+	res = libpqrcv_PQexec(cmd);
+	if (res == NULL || PQresultStatus(res) != PGRES_COPY_OUT)
+	{
+		if (res)
+			PQclear(res);
 		ereport(ERROR,
 				(errmsg("could not start WAL streaming: %s",
 						PQerrorMessage(streamConn))));
+	}
 	PQclear(res);
 
 	justconnected = true;
@@ -225,6 +231,66 @@ libpq_select(int timeout_ms)
 }
 
 /*
+ * 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)
+{
+	/*
+	 * 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 the risk is relatively small, so we ignore that
+	 * for now.
+	 */
+	if (!PQsendQuery(streamConn, query))
+		return NULL;
+
+	/*
+	 * 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. Just return the first result
+	 * available on the connection.
+	 */
+	return PQgetResult(streamConn);
+}
+
+/*
  * Disconnect connection to primary, if any.
  */
 static void
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index ff5f293..5ab8f68 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -86,8 +86,8 @@ 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.
+ * 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