asyncmix.patch

text/x-patch

Download

Patch

Format: unified

File	+	−
contrib/postgres_fdw/postgres_fdw.c	14	9

diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 14824368cc..613d406982 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -455,7 +455,7 @@ static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
 									  void *arg);
 static void create_cursor(ForeignScanState *node);
 static void request_more_data(ForeignScanState *node);
-static void fetch_received_data(ForeignScanState *node);
+static void fetch_received_data(ForeignScanState *node, bool vacateconn);
 static void vacate_connection(PgFdwState *fdwconn, bool clear_queue);
 static void close_cursor(PGconn *conn, unsigned int cursor_number);
 static PgFdwModifyState *create_foreign_modify(EState *estate,
@@ -1706,15 +1706,19 @@ postgresIterateForeignScan(ForeignScanState *node)
 		{
 			/*
 			 * finish the running query before sending the next command for
-			 * this node
+			 * this node.
+			 * When the plan contains both asynchronous subplans and non-async
+			 * subplans backend could request more data in async mode and want to
+			 * get data in sync mode by the same connection. Here it must wait
+			 * for async data before request another.
 			 */
-			if (!fsstate->s.commonstate->busy)
-				vacate_connection((PgFdwState *)fsstate, false);
+			if (fsstate->s.commonstate->busy)
+				vacate_connection(&fsstate->s, false);
 
 			request_more_data(node);
 
 			/* Fetch the result immediately. */
-			fetch_received_data(node);
+			fetch_received_data(node, false);
 		}
 		else if (!fsstate->s.commonstate->busy)
 		{
@@ -1749,7 +1753,7 @@ postgresIterateForeignScan(ForeignScanState *node)
 			/* fetch the leader's data and enqueue it for the next request */
 			if (available)
 			{
-				fetch_received_data(leader);
+				fetch_received_data(leader, false);
 				add_async_waiter(leader);
 			}
 		}
@@ -3729,7 +3733,7 @@ request_more_data(ForeignScanState *node)
  * Fetches received data and automatically send requests of the next waiter.
  */
 static void
-fetch_received_data(ForeignScanState *node)
+fetch_received_data(ForeignScanState *node, bool vacateconn)
 {
 	PgFdwScanState *fsstate = GetPgFdwScanState(node);
 	PGresult   *volatile res = NULL;
@@ -3817,7 +3821,8 @@ fetch_received_data(ForeignScanState *node)
 	waiter = move_to_next_waiter(node);
 
 	/* send the next request if any */
-	if (waiter)
+	if (waiter && (!vacateconn ||
+		GetPgFdwScanState(node)->s.conn != GetPgFdwScanState(waiter)->s.conn))
 		request_more_data(waiter);
 
 	MemoryContextSwitchTo(oldcontext);
@@ -3843,7 +3848,7 @@ vacate_connection(PgFdwState *fdwstate, bool clear_queue)
 	 * query
 	 */
 	leader = commonstate->leader;
-	fetch_received_data(leader);
+	fetch_received_data(leader, true);
 
 	/* let the first waiter be the next leader of this connection */
 	move_to_next_waiter(leader);