agora inbox for pgsql-hackers@postgresql.org  

help / color / mirror / Atom feed

[PATCH 3/7] Modify async execution infrastructure.
6+ messages / 2 participants
[nested] [flat]

* [PATCH 3/7] Modify async execution infrastructure.
@ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 6+ messages in thread

From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw)

---
 contrib/postgres_fdw/expected/postgres_fdw.out |  68 ++++++++--------
 contrib/postgres_fdw/postgres_fdw.c            |   5 +-
 src/backend/executor/execAsync.c               | 105 ++++++++++++++-----------
 src/backend/executor/nodeAppend.c              |  50 ++++++------
 src/backend/executor/nodeForeignscan.c         |   4 +-
 src/backend/nodes/copyfuncs.c                  |   1 +
 src/backend/nodes/outfuncs.c                   |   1 +
 src/backend/nodes/readfuncs.c                  |   1 +
 src/backend/optimizer/plan/createplan.c        |  24 +++++-
 src/backend/utils/adt/ruleutils.c              |   6 +-
 src/include/executor/nodeForeignscan.h         |   2 +-
 src/include/foreign/fdwapi.h                   |   2 +-
 src/include/nodes/execnodes.h                  |  10 ++-
 src/include/nodes/plannodes.h                  |   1 +
 14 files changed, 167 insertions(+), 113 deletions(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 457cfdb..083d947 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -6321,13 +6321,13 @@ insert into bar2 values(4,44,44);
 insert into bar2 values(7,77,77);
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for update;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6335,10 +6335,10 @@ select * from bar where f1 in (select f1 from foo) for update;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6358,13 +6358,13 @@ select * from bar where f1 in (select f1 from foo) for update;
 
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for share;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6372,10 +6372,10 @@ select * from bar where f1 in (select f1 from foo) for share;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6396,22 +6396,22 @@ select * from bar where f1 in (select f1 from foo) for share;
 -- Check UPDATE with inherited target and an inherited source table
 explain (verbose, costs off)
 update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
-                                               QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
  Update on public.bar
    Update on public.bar
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar.f1 = foo2.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Seq Scan on public.bar
                Output: bar.f1, bar.f2, bar.ctid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6419,16 +6419,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
                            ->  Seq Scan on public.foo
                                  Output: foo.ctid, foo.*, foo.tableoid, foo.f1
    ->  Hash Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar2.f1 = foo.f1)
          ->  Foreign Scan on public.bar2
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6462,8 +6462,8 @@ where bar.f1 = ss.f1;
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1))
-         Hash Cond: (foo2.f1 = bar.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1))
+         Hash Cond: (foo.f1 = bar.f1)
          ->  Append
                ->  Foreign Scan on public.foo2
                      Output: ROW(foo2.f1), foo2.f1
@@ -6480,8 +6480,8 @@ where bar.f1 = ss.f1;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid
    ->  Merge Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1))
-         Merge Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1))
+         Merge Cond: (bar2.f1 = foo.f1)
          ->  Sort
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Sort Key: bar2.f1
@@ -6489,8 +6489,8 @@ where bar.f1 = ss.f1;
                      Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                      Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Sort
-               Output: (ROW(foo2.f1)), foo2.f1
-               Sort Key: foo2.f1
+               Output: (ROW(foo.f1)), foo.f1
+               Sort Key: foo.f1
                ->  Append
                      ->  Foreign Scan on public.foo2
                            Output: ROW(foo2.f1), foo2.f1
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index c64ae41..b92b279 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root,
 static bool postgresIsForeignPathAsyncCapable(ForeignPath *path);
 static void postgresForeignAsyncRequest(EState *estate,
 							PendingAsyncRequest *areq);
-static void postgresForeignAsyncConfigureWait(EState *estate,
+static bool postgresForeignAsyncConfigureWait(EState *estate,
 								  PendingAsyncRequest *areq,
 								  bool reinit);
 static void postgresForeignAsyncNotify(EState *estate,
@@ -4479,11 +4479,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq)
 	ExecAsyncRequestDone(estate, areq, (Node *) slot);
 }
 
-static void
+static bool
 postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 								  bool reinit)
 {
 	elog(ERROR, "postgresForeignAsyncConfigureWait");
+	return false;
 }
 
 static void
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index e070c26..33496a9 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -22,7 +22,7 @@
 #include "storage/latch.h"
 
 static bool ExecAsyncEventWait(EState *estate, long timeout);
-static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
+static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 	bool reinit);
 static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq);
 static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq);
@@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 				 PlanState *requestee)
 {
 	PendingAsyncRequest *areq = NULL;
-	int		i = estate->es_num_pending_async;
+	int		nasync = estate->es_num_pending_async;
 
 	/*
 	 * If the number of pending asynchronous nodes exceeds the number of
@@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * We start with 16 slots, and thereafter double the array size each
 	 * time we run out of slots.
 	 */
-	if (i >= estate->es_max_pending_async)
+	if (nasync >= estate->es_max_pending_async)
 	{
 		int	newmax;
 
@@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * PendingAsyncRequest if there is one.  If not, we must allocate a new
 	 * one.
 	 */
-	if (estate->es_pending_async[i] == NULL)
+	if (estate->es_pending_async[nasync] == NULL)
 	{
 		areq = MemoryContextAllocZero(estate->es_query_cxt,
 									  sizeof(PendingAsyncRequest));
-		estate->es_pending_async[i] = areq;
+		estate->es_pending_async[nasync] = areq;
 	}
 	else
 	{
-		areq = estate->es_pending_async[i];
+		areq = estate->es_pending_async[nasync];
 		MemSet(areq, 0, sizeof(PendingAsyncRequest));
 	}
-	areq->myindex = estate->es_num_pending_async++;
+	areq->myindex = estate->es_num_pending_async;
 
 	/* Initialize the new request. */
 	areq->requestor = requestor;
 	areq->request_index = request_index;
 	areq->requestee = requestee;
 
-	/* Give the requestee a chance to do whatever it wants. */
+	/*
+	 * Give the requestee a chance to do whatever it wants.
+	 * Requst functions return true if a result is immediately available.
+	 */
 	switch (nodeTag(requestee))
 	{
 		case T_ForeignScanState:
@@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 			elog(ERROR, "unrecognized node type: %d",
 				(int) nodeTag(requestee));
 	}
+
+	/*
+	 * If a result is available, complete it immediately.
+	 */
+	if (areq->state == ASYNC_COMPLETE)
+	{
+		Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot));
+		ExecAsyncResponse(estate, areq);
+
+		return;
+	}
+
+	/* No result available now, make this node pending */
+	estate->es_num_pending_async++;
 }
 
 /*
@@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 		{
 			PendingAsyncRequest *areq = estate->es_pending_async[i];
 
-			/* Skip it if no callback is pending. */
-			if (!areq->callback_pending)
-				continue;
-
-			/*
-			 * Mark it as no longer needing a callback.  We must do this
-			 * before dispatching the callback in case the callback resets
-			 * the flag.
-			 */
-			areq->callback_pending = false;
-			estate->es_async_callback_pending--;
-
-			/* Perform the actual callback; set request_done if appropraite. */
-			if (!areq->request_complete)
+			/* Skip it if not pending. */
+			if (areq->state == ASYNC_CALLBACK_PENDING)
+			{
+				/*
+				 * Mark it as no longer needing a callback.  We must do this
+				 * before dispatching the callback in case the callback resets
+				 * the flag.
+				 */
+				estate->es_async_callback_pending--;
 				ExecAsyncNotify(estate, areq);
-			else
+			}
+
+			if (areq->state == ASYNC_COMPLETE)
 			{
 				any_node_done = true;
 				if (requestor == areq->requestor)
@@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 				PendingAsyncRequest *head;
 				PendingAsyncRequest *tail = estate->es_pending_async[tidx];
 
-				if (!tail->callback_pending && tail->request_complete)
+				if (tail->state == ASYNC_COMPLETE)
 					continue;
 				head = estate->es_pending_async[hidx];
 				estate->es_pending_async[tidx] = head;
@@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
  * means wait forever, 0 means don't wait at all, and >0 means wait for the
  * indicated number of milliseconds.
  *
- * Returns true if we found some events and false if we timed out.
+ * Returns true if we found some events and false if we timed out or there's
+ * no event to wait. The latter is occur when the areq is processed during
  */
 static bool
 ExecAsyncEventWait(EState *estate, long timeout)
@@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout)
 	int		n;
 	bool	reinit = false;
 	bool	process_latch_set = false;
+	bool	added = false;
 
 	if (estate->es_wait_event_set == NULL)
 	{
@@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		PendingAsyncRequest *areq = estate->es_pending_async[i];
 
 		if (areq->num_fd_events > 0)
-			ExecAsyncConfigureWait(estate, areq, reinit);
+			added |= ExecAsyncConfigureWait(estate, areq, reinit);
 	}
 
+	Assert(added);
+
 	/* Wait for at least one event to occur. */
 	noccurred = WaitEventSetWait(estate->es_wait_event_set, timeout,
 								 occurred_event, EVENT_BUFFER_SIZE,
 								 WAIT_EVENT_ASYNC_WAIT);
+
 	if (noccurred == 0)
 		return false;
 
@@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		{
 			PendingAsyncRequest *areq = w->user_data;
 
-			if (!areq->callback_pending)
-			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
-				estate->es_async_callback_pending++;
-			}
+			Assert(areq->state == ASYNC_WAITING);
+
+			areq->state = ASYNC_CALLBACK_PENDING;
+			estate->es_async_callback_pending++;
 		}
 	}
 
@@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout)
 
 			if (areq->wants_process_latch)
 			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
+				Assert(areq->state == ASYNC_WAITING);
+				areq->state = ASYNC_CALLBACK_PENDING;
 			}
 		}
 	}
@@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout)
  * The events should include only WL_SOCKET_READABLE or WL_SOCKET_WRITEABLE,
  * and the number of calls should not exceed areq->num_fd_events (as
  * prevously set via ExecAsyncSetRequiredEvents).
+ *
+ * Individual requests can omit registering an event but it is a
+ * responsibility of the node driver to set at least one event per one
+ * requestor.
  */
-static void
+static bool
 ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 					   bool reinit)
 {
 	switch (nodeTag(areq->requestee))
 	{
 		case T_ForeignScanState:
-			ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
+			return ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq,
 	estate->es_total_fd_events += num_fd_events - areq->num_fd_events;
 	areq->num_fd_events = num_fd_events;
 	areq->wants_process_latch = wants_process_latch;
+	areq->state = ASYNC_WAITING;
 
 	if (force_reset && estate->es_wait_event_set != NULL)
 	{
@@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result)
 	 * need a callback to remove registered wait events.  It's not clear
 	 * that we would come out ahead, so use brute force for now.
 	 */
+	Assert(areq->state == ASYNC_IDLE || areq->state == ASYNC_CALLBACK_PENDING);
+
 	if (areq->num_fd_events > 0 || areq->wants_process_latch)
 		ExecAsyncSetRequiredEvents(estate, areq, 0, false, true);
 
 	/* Save result and mark request as complete. */
 	areq->result = result;
-	areq->request_complete = true;
-
-	/* Make sure this request is flagged for a callback. */
-	if (!areq->callback_pending)
-	{
-		areq->callback_pending = true;
-		estate->es_async_callback_pending++;
-	}
+	areq->state = ASYNC_COMPLETE;
 }
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index bb06569..c234f1f 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -229,9 +229,15 @@ ExecAppend(AppendState *node)
 		 */
 		while ((i = bms_first_member(node->as_needrequest)) >= 0)
 		{
-			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
 			node->as_nasyncpending++;
+
+			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
+			/* If this request immediately gives a result, take it. */
+			if (node->as_nasyncresult > 0)
+				return node->as_asyncresult[--node->as_nasyncresult];
 		}
+		if (node->as_nasyncpending == 0 && node->as_syncdone)
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 	}
 
 	for (;;)
@@ -246,32 +252,32 @@ ExecAppend(AppendState *node)
 		{
 			long	timeout = node->as_syncdone ? -1 : 0;
 
-			for (;;)
+			while (node->as_nasyncpending > 0)
 			{
-				if (node->as_nasyncpending == 0)
-				{
-					/*
-					 * If there is no asynchronous activity still pending
-					 * and the synchronous activity is also complete, we're
-					 * totally done scanning this node.  Otherwise, we're
-					 * done with the asynchronous stuff but must continue
-					 * scanning the synchronous children.
-					 */
-					if (node->as_syncdone)
-						return ExecClearTuple(node->ps.ps_ResultTupleSlot);
-					break;
-				}
-				if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout))
-				{
-					/* Timeout reached. */
-					break;
-				}
-				if (node->as_nasyncresult > 0)
+				if (ExecAsyncEventLoop(node->ps.state, &node->ps, timeout) &&
+					node->as_nasyncresult > 0)
 				{
 					/* Asynchronous subplan returned a tuple! */
 					--node->as_nasyncresult;
 					return node->as_asyncresult[node->as_nasyncresult];
 				}
+
+				/* Timeout reached. Go through to sync nodes if exists */
+				if (!node->as_syncdone)
+					break;
+			}
+
+			/*
+			 * If there is no asynchronous activity still pending and the
+			 * synchronous activity is also complete, we're totally done
+			 * scanning this node.  Otherwise, we're done with the
+			 * asynchronous stuff but must continue scanning the synchronous
+			 * children.
+			 */
+			if (node->as_syncdone)
+			{
+				Assert(node->as_nasyncpending == 0);
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 			}
 		}
 
@@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq)
 	TupleTableSlot *slot;
 
 	/* We shouldn't be called until the request is complete. */
-	Assert(areq->request_complete);
+	Assert(areq->state == ASYNC_COMPLETE);
 
 	/* Our result slot shouldn't already be occupied. */
 	Assert(TupIsNull(node->ps.ps_ResultTupleSlot));
diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c
index 85d436f..d3567bb 100644
--- a/src/backend/executor/nodeForeignscan.c
+++ b/src/backend/executor/nodeForeignscan.c
@@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq)
  *		In async mode, configure for a wait
  * ----------------------------------------------------------------
  */
-void
+bool
 ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit)
 {
@@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate,
 	FdwRoutine *fdwroutine = node->fdwroutine;
 
 	Assert(fdwroutine->ForeignAsyncConfigureWait != NULL);
-	fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
+	return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index e4a103f..27ccf9d 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -219,6 +219,7 @@ _copyAppend(const Append *from)
 	 */
 	COPY_NODE_FIELD(appendplans);
 	COPY_SCALAR_FIELD(nasyncplans);
+	COPY_SCALAR_FIELD(referent);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 1566e0d..c8b9f31 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node)
 
 	WRITE_NODE_FIELD(appendplans);
 	WRITE_INT_FIELD(nasyncplans);
+	WRITE_INT_FIELD(referent);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 69453b5..8443a62 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1520,6 +1520,7 @@ _readAppend(void)
 
 	READ_NODE_FIELD(appendplans);
 	READ_INT_FIELD(nasyncplans);
+	READ_INT_FIELD(referent);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 7caa8d3..ff1d663 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual,
 			 Index scanrelid, int ctePlanId, int cteParam);
 static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual,
 				   Index scanrelid, int wtParam);
-static Append *make_append(List *asyncplans, int nasyncplans, List *tlist);
+static Append *make_append(List *asyncplans, int nasyncplans,
+						   int referent, List *tlist);
 static RecursiveUnion *make_recursive_union(List *tlist,
 					 Plan *lefttree,
 					 Plan *righttree,
@@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	List	   *syncplans = NIL;
 	ListCell   *subpaths;
 	int			nasyncplans = 0;
+	bool		first = true;
+	bool		referent_is_sync = true;
 
 	/*
 	 * The subpaths list could be empty, if every child was proven empty by
@@ -985,7 +988,14 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		return plan;
 	}
 
-	/* Build the plan for each child */
+	/*
+	 * Build the plan for each child
+
+	 * The first child in an inheritance set is the representative in
+	 * explaining tlist entries (see set_deparse_planstate). We should keep
+	 * the first child in best_path->subpaths at the head of the subplan list
+	 * for the reason.
+	 */
 	foreach(subpaths, best_path->subpaths)
 	{
 		Path	   *subpath = (Path *) lfirst(subpaths);
@@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		{
 			asyncplans = lappend(asyncplans, subplan);
 			++nasyncplans;
+			if (first)
+				referent_is_sync = false;
 		}
 		else
 			syncplans = lappend(syncplans, subplan);
+
+		first = false;
 	}
 
 	/*
@@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	 * parent-rel Vars it'll be asked to emit.
 	 */
 
-	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist);
+	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans,
+					   referent_is_sync ? nasyncplans : 0, tlist);
 
 	copy_generic_path_info(&plan->plan, (Path *) best_path);
 
@@ -4951,7 +4966,7 @@ make_foreignscan(List *qptlist,
 }
 
 static Append *
-make_append(List *appendplans, int nasyncplans, List *tlist)
+make_append(List *appendplans, int nasyncplans,	int referent, List *tlist)
 {
 	Append	   *node = makeNode(Append);
 	Plan	   *plan = &node->plan;
@@ -4962,6 +4977,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist)
 	plan->righttree = NULL;
 	node->appendplans = appendplans;
 	node->nasyncplans = nasyncplans;
+	node->referent = referent;
 
 	return node;
 }
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index a3a4174..9a2ee83 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -4079,7 +4079,11 @@ set_deparse_planstate(deparse_namespace *dpns, PlanState *ps)
 	 * lists containing references to non-target relations.
 	 */
 	if (IsA(ps, AppendState))
-		dpns->outer_planstate = ((AppendState *) ps)->appendplans[0];
+	{
+		int idx = ((Append*)(((AppendState *) ps)->ps.plan))->referent;
+		dpns->outer_planstate =
+			((AppendState *) ps)->appendplans[idx];
+	}
 	else if (IsA(ps, MergeAppendState))
 		dpns->outer_planstate = ((MergeAppendState *) ps)->mergeplans[0];
 	else if (IsA(ps, ModifyTableState))
diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h
index 3e69ab0..47a3920 100644
--- a/src/include/executor/nodeForeignscan.h
+++ b/src/include/executor/nodeForeignscan.h
@@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node,
 
 extern void ExecAsyncForeignScanRequest(EState *estate,
 	PendingAsyncRequest *areq);
-extern void ExecAsyncForeignScanConfigureWait(EState *estate,
+extern bool ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit);
 extern void ExecAsyncForeignScanNotify(EState *estate,
 	PendingAsyncRequest *areq);
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 88feb9a..65517fd 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root,
 typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path);
 typedef void (*ForeignAsyncRequest_function) (EState *estate,
 											PendingAsyncRequest *areq);
-typedef void (*ForeignAsyncConfigureWait_function) (EState *estate,
+typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate,
 											PendingAsyncRequest *areq,
 											bool reinit);
 typedef void (*ForeignAsyncNotify_function) (EState *estate,
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b50b41c..0c6af86 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -352,6 +352,13 @@ typedef struct ResultRelInfo
  * State for an asynchronous tuple request.
  * ----------------
  */
+typedef enum AsyncRequestState
+{
+	ASYNC_IDLE,
+	ASYNC_WAITING,
+	ASYNC_CALLBACK_PENDING,
+	ASYNC_COMPLETE
+} AsyncRequestState;
 typedef struct PendingAsyncRequest
 {
 	int			myindex;			/* Index in es_pending_async. */
@@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest
 	int			request_index;	/* Scratch space for requestor. */
 	int			num_fd_events;	/* Max number of FD events requestee needs. */
 	bool		wants_process_latch;	/* Requestee cares about MyLatch. */
-	bool		callback_pending;			/* Callback is needed. */
-	bool		request_complete;		/* Request complete, result valid. */
+	AsyncRequestState state;
 	Node	   *result;			/* Result (NULL if no more tuples). */
 } PendingAsyncRequest;
 
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 327119b..1df6693 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -209,6 +209,7 @@ typedef struct Append
 	Plan		plan;
 	List	   *appendplans;
 	int			nasyncplans;	/* # of async plans, always at start of list */
+	int			referent; 		/* index of inheritance tree referent */
 } Append;
 
 /* ----------------
-- 
2.9.2


----Next_Part(Tue_Nov_15_20_25_13_2016_668)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0004-Make-postgres_fdw-async-capable.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH 3/4] Modify async execution infrastructure.
@ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 6+ messages in thread

From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw)

---
 contrib/postgres_fdw/expected/postgres_fdw.out |  68 ++++++++--------
 contrib/postgres_fdw/postgres_fdw.c            |   5 +-
 src/backend/executor/execAsync.c               | 105 ++++++++++++++-----------
 src/backend/executor/nodeAppend.c              |  50 ++++++------
 src/backend/executor/nodeForeignscan.c         |   4 +-
 src/backend/nodes/copyfuncs.c                  |   1 +
 src/backend/nodes/outfuncs.c                   |   1 +
 src/backend/nodes/readfuncs.c                  |   1 +
 src/backend/optimizer/plan/createplan.c        |  24 +++++-
 src/backend/utils/adt/ruleutils.c              |   6 +-
 src/include/executor/nodeForeignscan.h         |   2 +-
 src/include/foreign/fdwapi.h                   |   2 +-
 src/include/nodes/execnodes.h                  |  10 ++-
 src/include/nodes/plannodes.h                  |   1 +
 14 files changed, 167 insertions(+), 113 deletions(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 6677bc4..d429790 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -5230,13 +5230,13 @@ insert into bar2 values(4,44,44);
 insert into bar2 values(7,77,77);
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for update;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -5244,10 +5244,10 @@ select * from bar where f1 in (select f1 from foo) for update;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -5267,13 +5267,13 @@ select * from bar where f1 in (select f1 from foo) for update;
 
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for share;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -5281,10 +5281,10 @@ select * from bar where f1 in (select f1 from foo) for share;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -5305,22 +5305,22 @@ select * from bar where f1 in (select f1 from foo) for share;
 -- Check UPDATE with inherited target and an inherited source table
 explain (verbose, costs off)
 update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
-                                               QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
  Update on public.bar
    Update on public.bar
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar.f1 = foo2.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Seq Scan on public.bar
                Output: bar.f1, bar.f2, bar.ctid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -5328,16 +5328,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
                            ->  Seq Scan on public.foo
                                  Output: foo.ctid, foo.*, foo.tableoid, foo.f1
    ->  Hash Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar2.f1 = foo.f1)
          ->  Foreign Scan on public.bar2
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -5371,8 +5371,8 @@ where bar.f1 = ss.f1;
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1))
-         Hash Cond: (foo2.f1 = bar.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1))
+         Hash Cond: (foo.f1 = bar.f1)
          ->  Append
                ->  Foreign Scan on public.foo2
                      Output: ROW(foo2.f1), foo2.f1
@@ -5389,8 +5389,8 @@ where bar.f1 = ss.f1;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid
    ->  Merge Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1))
-         Merge Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1))
+         Merge Cond: (bar2.f1 = foo.f1)
          ->  Sort
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Sort Key: bar2.f1
@@ -5398,8 +5398,8 @@ where bar.f1 = ss.f1;
                      Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                      Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Sort
-               Output: (ROW(foo2.f1)), foo2.f1
-               Sort Key: foo2.f1
+               Output: (ROW(foo.f1)), foo.f1
+               Sort Key: foo.f1
                ->  Append
                      ->  Foreign Scan on public.foo2
                            Output: ROW(foo2.f1), foo2.f1
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 6da5843..997bd6c 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -348,7 +348,7 @@ static bool postgresRecheckForeignScan(ForeignScanState *node,
 static bool postgresIsForeignPathAsyncCapable(ForeignPath *path);
 static void postgresForeignAsyncRequest(EState *estate,
 							PendingAsyncRequest *areq);
-static void postgresForeignAsyncConfigureWait(EState *estate,
+static bool postgresForeignAsyncConfigureWait(EState *estate,
 								  PendingAsyncRequest *areq,
 								  bool reinit);
 static void postgresForeignAsyncNotify(EState *estate,
@@ -4379,11 +4379,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq)
 	ExecAsyncRequestDone(estate, areq, (Node *) slot);
 }
 
-static void
+static bool
 postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 								  bool reinit)
 {
 	elog(ERROR, "postgresForeignAsyncConfigureWait");
+	return false;
 }
 
 static void
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index e070c26..33496a9 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -22,7 +22,7 @@
 #include "storage/latch.h"
 
 static bool ExecAsyncEventWait(EState *estate, long timeout);
-static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
+static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 	bool reinit);
 static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq);
 static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq);
@@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 				 PlanState *requestee)
 {
 	PendingAsyncRequest *areq = NULL;
-	int		i = estate->es_num_pending_async;
+	int		nasync = estate->es_num_pending_async;
 
 	/*
 	 * If the number of pending asynchronous nodes exceeds the number of
@@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * We start with 16 slots, and thereafter double the array size each
 	 * time we run out of slots.
 	 */
-	if (i >= estate->es_max_pending_async)
+	if (nasync >= estate->es_max_pending_async)
 	{
 		int	newmax;
 
@@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * PendingAsyncRequest if there is one.  If not, we must allocate a new
 	 * one.
 	 */
-	if (estate->es_pending_async[i] == NULL)
+	if (estate->es_pending_async[nasync] == NULL)
 	{
 		areq = MemoryContextAllocZero(estate->es_query_cxt,
 									  sizeof(PendingAsyncRequest));
-		estate->es_pending_async[i] = areq;
+		estate->es_pending_async[nasync] = areq;
 	}
 	else
 	{
-		areq = estate->es_pending_async[i];
+		areq = estate->es_pending_async[nasync];
 		MemSet(areq, 0, sizeof(PendingAsyncRequest));
 	}
-	areq->myindex = estate->es_num_pending_async++;
+	areq->myindex = estate->es_num_pending_async;
 
 	/* Initialize the new request. */
 	areq->requestor = requestor;
 	areq->request_index = request_index;
 	areq->requestee = requestee;
 
-	/* Give the requestee a chance to do whatever it wants. */
+	/*
+	 * Give the requestee a chance to do whatever it wants.
+	 * Requst functions return true if a result is immediately available.
+	 */
 	switch (nodeTag(requestee))
 	{
 		case T_ForeignScanState:
@@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 			elog(ERROR, "unrecognized node type: %d",
 				(int) nodeTag(requestee));
 	}
+
+	/*
+	 * If a result is available, complete it immediately.
+	 */
+	if (areq->state == ASYNC_COMPLETE)
+	{
+		Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot));
+		ExecAsyncResponse(estate, areq);
+
+		return;
+	}
+
+	/* No result available now, make this node pending */
+	estate->es_num_pending_async++;
 }
 
 /*
@@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 		{
 			PendingAsyncRequest *areq = estate->es_pending_async[i];
 
-			/* Skip it if no callback is pending. */
-			if (!areq->callback_pending)
-				continue;
-
-			/*
-			 * Mark it as no longer needing a callback.  We must do this
-			 * before dispatching the callback in case the callback resets
-			 * the flag.
-			 */
-			areq->callback_pending = false;
-			estate->es_async_callback_pending--;
-
-			/* Perform the actual callback; set request_done if appropraite. */
-			if (!areq->request_complete)
+			/* Skip it if not pending. */
+			if (areq->state == ASYNC_CALLBACK_PENDING)
+			{
+				/*
+				 * Mark it as no longer needing a callback.  We must do this
+				 * before dispatching the callback in case the callback resets
+				 * the flag.
+				 */
+				estate->es_async_callback_pending--;
 				ExecAsyncNotify(estate, areq);
-			else
+			}
+
+			if (areq->state == ASYNC_COMPLETE)
 			{
 				any_node_done = true;
 				if (requestor == areq->requestor)
@@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 				PendingAsyncRequest *head;
 				PendingAsyncRequest *tail = estate->es_pending_async[tidx];
 
-				if (!tail->callback_pending && tail->request_complete)
+				if (tail->state == ASYNC_COMPLETE)
 					continue;
 				head = estate->es_pending_async[hidx];
 				estate->es_pending_async[tidx] = head;
@@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
  * means wait forever, 0 means don't wait at all, and >0 means wait for the
  * indicated number of milliseconds.
  *
- * Returns true if we found some events and false if we timed out.
+ * Returns true if we found some events and false if we timed out or there's
+ * no event to wait. The latter is occur when the areq is processed during
  */
 static bool
 ExecAsyncEventWait(EState *estate, long timeout)
@@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout)
 	int		n;
 	bool	reinit = false;
 	bool	process_latch_set = false;
+	bool	added = false;
 
 	if (estate->es_wait_event_set == NULL)
 	{
@@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		PendingAsyncRequest *areq = estate->es_pending_async[i];
 
 		if (areq->num_fd_events > 0)
-			ExecAsyncConfigureWait(estate, areq, reinit);
+			added |= ExecAsyncConfigureWait(estate, areq, reinit);
 	}
 
+	Assert(added);
+
 	/* Wait for at least one event to occur. */
 	noccurred = WaitEventSetWait(estate->es_wait_event_set, timeout,
 								 occurred_event, EVENT_BUFFER_SIZE,
 								 WAIT_EVENT_ASYNC_WAIT);
+
 	if (noccurred == 0)
 		return false;
 
@@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		{
 			PendingAsyncRequest *areq = w->user_data;
 
-			if (!areq->callback_pending)
-			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
-				estate->es_async_callback_pending++;
-			}
+			Assert(areq->state == ASYNC_WAITING);
+
+			areq->state = ASYNC_CALLBACK_PENDING;
+			estate->es_async_callback_pending++;
 		}
 	}
 
@@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout)
 
 			if (areq->wants_process_latch)
 			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
+				Assert(areq->state == ASYNC_WAITING);
+				areq->state = ASYNC_CALLBACK_PENDING;
 			}
 		}
 	}
@@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout)
  * The events should include only WL_SOCKET_READABLE or WL_SOCKET_WRITEABLE,
  * and the number of calls should not exceed areq->num_fd_events (as
  * prevously set via ExecAsyncSetRequiredEvents).
+ *
+ * Individual requests can omit registering an event but it is a
+ * responsibility of the node driver to set at least one event per one
+ * requestor.
  */
-static void
+static bool
 ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 					   bool reinit)
 {
 	switch (nodeTag(areq->requestee))
 	{
 		case T_ForeignScanState:
-			ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
+			return ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq,
 	estate->es_total_fd_events += num_fd_events - areq->num_fd_events;
 	areq->num_fd_events = num_fd_events;
 	areq->wants_process_latch = wants_process_latch;
+	areq->state = ASYNC_WAITING;
 
 	if (force_reset && estate->es_wait_event_set != NULL)
 	{
@@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result)
 	 * need a callback to remove registered wait events.  It's not clear
 	 * that we would come out ahead, so use brute force for now.
 	 */
+	Assert(areq->state == ASYNC_IDLE || areq->state == ASYNC_CALLBACK_PENDING);
+
 	if (areq->num_fd_events > 0 || areq->wants_process_latch)
 		ExecAsyncSetRequiredEvents(estate, areq, 0, false, true);
 
 	/* Save result and mark request as complete. */
 	areq->result = result;
-	areq->request_complete = true;
-
-	/* Make sure this request is flagged for a callback. */
-	if (!areq->callback_pending)
-	{
-		areq->callback_pending = true;
-		estate->es_async_callback_pending++;
-	}
+	areq->state = ASYNC_COMPLETE;
 }
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index bb06569..c234f1f 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -229,9 +229,15 @@ ExecAppend(AppendState *node)
 		 */
 		while ((i = bms_first_member(node->as_needrequest)) >= 0)
 		{
-			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
 			node->as_nasyncpending++;
+
+			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
+			/* If this request immediately gives a result, take it. */
+			if (node->as_nasyncresult > 0)
+				return node->as_asyncresult[--node->as_nasyncresult];
 		}
+		if (node->as_nasyncpending == 0 && node->as_syncdone)
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 	}
 
 	for (;;)
@@ -246,32 +252,32 @@ ExecAppend(AppendState *node)
 		{
 			long	timeout = node->as_syncdone ? -1 : 0;
 
-			for (;;)
+			while (node->as_nasyncpending > 0)
 			{
-				if (node->as_nasyncpending == 0)
-				{
-					/*
-					 * If there is no asynchronous activity still pending
-					 * and the synchronous activity is also complete, we're
-					 * totally done scanning this node.  Otherwise, we're
-					 * done with the asynchronous stuff but must continue
-					 * scanning the synchronous children.
-					 */
-					if (node->as_syncdone)
-						return ExecClearTuple(node->ps.ps_ResultTupleSlot);
-					break;
-				}
-				if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout))
-				{
-					/* Timeout reached. */
-					break;
-				}
-				if (node->as_nasyncresult > 0)
+				if (ExecAsyncEventLoop(node->ps.state, &node->ps, timeout) &&
+					node->as_nasyncresult > 0)
 				{
 					/* Asynchronous subplan returned a tuple! */
 					--node->as_nasyncresult;
 					return node->as_asyncresult[node->as_nasyncresult];
 				}
+
+				/* Timeout reached. Go through to sync nodes if exists */
+				if (!node->as_syncdone)
+					break;
+			}
+
+			/*
+			 * If there is no asynchronous activity still pending and the
+			 * synchronous activity is also complete, we're totally done
+			 * scanning this node.  Otherwise, we're done with the
+			 * asynchronous stuff but must continue scanning the synchronous
+			 * children.
+			 */
+			if (node->as_syncdone)
+			{
+				Assert(node->as_nasyncpending == 0);
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 			}
 		}
 
@@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq)
 	TupleTableSlot *slot;
 
 	/* We shouldn't be called until the request is complete. */
-	Assert(areq->request_complete);
+	Assert(areq->state == ASYNC_COMPLETE);
 
 	/* Our result slot shouldn't already be occupied. */
 	Assert(TupIsNull(node->ps.ps_ResultTupleSlot));
diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c
index 85d436f..d3567bb 100644
--- a/src/backend/executor/nodeForeignscan.c
+++ b/src/backend/executor/nodeForeignscan.c
@@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq)
  *		In async mode, configure for a wait
  * ----------------------------------------------------------------
  */
-void
+bool
 ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit)
 {
@@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate,
 	FdwRoutine *fdwroutine = node->fdwroutine;
 
 	Assert(fdwroutine->ForeignAsyncConfigureWait != NULL);
-	fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
+	return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 23b4e18..72d8cd6 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -219,6 +219,7 @@ _copyAppend(const Append *from)
 	 */
 	COPY_NODE_FIELD(appendplans);
 	COPY_SCALAR_FIELD(nasyncplans);
+	COPY_SCALAR_FIELD(referent);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index dc5b938..1ebdc48 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node)
 
 	WRITE_NODE_FIELD(appendplans);
 	WRITE_INT_FIELD(nasyncplans);
+	WRITE_INT_FIELD(referent);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 69453b5..8443a62 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1520,6 +1520,7 @@ _readAppend(void)
 
 	READ_NODE_FIELD(appendplans);
 	READ_INT_FIELD(nasyncplans);
+	READ_INT_FIELD(referent);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index e7e55c0..c73bbb3 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual,
 			 Index scanrelid, int ctePlanId, int cteParam);
 static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual,
 				   Index scanrelid, int wtParam);
-static Append *make_append(List *asyncplans, int nasyncplans, List *tlist);
+static Append *make_append(List *asyncplans, int nasyncplans,
+						   int referent, List *tlist);
 static RecursiveUnion *make_recursive_union(List *tlist,
 					 Plan *lefttree,
 					 Plan *righttree,
@@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	List	   *syncplans = NIL;
 	ListCell   *subpaths;
 	int			nasyncplans = 0;
+	bool		first = true;
+	bool		referent_is_sync = true;
 
 	/*
 	 * The subpaths list could be empty, if every child was proven empty by
@@ -985,7 +988,14 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		return plan;
 	}
 
-	/* Build the plan for each child */
+	/*
+	 * Build the plan for each child
+
+	 * The first child in an inheritance set is the representative in
+	 * explaining tlist entries (see set_deparse_planstate). We should keep
+	 * the first child in best_path->subpaths at the head of the subplan list
+	 * for the reason.
+	 */
 	foreach(subpaths, best_path->subpaths)
 	{
 		Path	   *subpath = (Path *) lfirst(subpaths);
@@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		{
 			asyncplans = lappend(asyncplans, subplan);
 			++nasyncplans;
+			if (first)
+				referent_is_sync = false;
 		}
 		else
 			syncplans = lappend(syncplans, subplan);
+
+		first = false;
 	}
 
 	/*
@@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	 * parent-rel Vars it'll be asked to emit.
 	 */
 
-	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist);
+	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans,
+					   referent_is_sync ? nasyncplans : 0, tlist);
 
 	copy_generic_path_info(&plan->plan, (Path *) best_path);
 
@@ -4944,7 +4959,7 @@ make_foreignscan(List *qptlist,
 }
 
 static Append *
-make_append(List *appendplans, int nasyncplans, List *tlist)
+make_append(List *appendplans, int nasyncplans,	int referent, List *tlist)
 {
 	Append	   *node = makeNode(Append);
 	Plan	   *plan = &node->plan;
@@ -4955,6 +4970,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist)
 	plan->righttree = NULL;
 	node->appendplans = appendplans;
 	node->nasyncplans = nasyncplans;
+	node->referent = referent;
 
 	return node;
 }
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 8a81d7a..de0e96c 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -4056,7 +4056,11 @@ set_deparse_planstate(deparse_namespace *dpns, PlanState *ps)
 	 * lists containing references to non-target relations.
 	 */
 	if (IsA(ps, AppendState))
-		dpns->outer_planstate = ((AppendState *) ps)->appendplans[0];
+	{
+		int idx = ((Append*)(((AppendState *) ps)->ps.plan))->referent;
+		dpns->outer_planstate =
+			((AppendState *) ps)->appendplans[idx];
+	}
 	else if (IsA(ps, MergeAppendState))
 		dpns->outer_planstate = ((MergeAppendState *) ps)->mergeplans[0];
 	else if (IsA(ps, ModifyTableState))
diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h
index 3e69ab0..47a3920 100644
--- a/src/include/executor/nodeForeignscan.h
+++ b/src/include/executor/nodeForeignscan.h
@@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node,
 
 extern void ExecAsyncForeignScanRequest(EState *estate,
 	PendingAsyncRequest *areq);
-extern void ExecAsyncForeignScanConfigureWait(EState *estate,
+extern bool ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit);
 extern void ExecAsyncForeignScanNotify(EState *estate,
 	PendingAsyncRequest *areq);
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 88feb9a..65517fd 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root,
 typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path);
 typedef void (*ForeignAsyncRequest_function) (EState *estate,
 											PendingAsyncRequest *areq);
-typedef void (*ForeignAsyncConfigureWait_function) (EState *estate,
+typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate,
 											PendingAsyncRequest *areq,
 											bool reinit);
 typedef void (*ForeignAsyncNotify_function) (EState *estate,
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b50b41c..0c6af86 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -352,6 +352,13 @@ typedef struct ResultRelInfo
  * State for an asynchronous tuple request.
  * ----------------
  */
+typedef enum AsyncRequestState
+{
+	ASYNC_IDLE,
+	ASYNC_WAITING,
+	ASYNC_CALLBACK_PENDING,
+	ASYNC_COMPLETE
+} AsyncRequestState;
 typedef struct PendingAsyncRequest
 {
 	int			myindex;			/* Index in es_pending_async. */
@@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest
 	int			request_index;	/* Scratch space for requestor. */
 	int			num_fd_events;	/* Max number of FD events requestee needs. */
 	bool		wants_process_latch;	/* Requestee cares about MyLatch. */
-	bool		callback_pending;			/* Callback is needed. */
-	bool		request_complete;		/* Request complete, result valid. */
+	AsyncRequestState state;
 	Node	   *result;			/* Result (NULL if no more tuples). */
 } PendingAsyncRequest;
 
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 327119b..1df6693 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -209,6 +209,7 @@ typedef struct Append
 	Plan		plan;
 	List	   *appendplans;
 	int			nasyncplans;	/* # of async plans, always at start of list */
+	int			referent; 		/* index of inheritance tree referent */
 } Append;
 
 /* ----------------
-- 
2.9.2


----Next_Part(Tue_Oct_18_10_30_51_2016_252)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0004-Make-postgres_fdw-async-capable.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH 3/6] Modify async execution infrastructure.
@ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 6+ messages in thread

From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw)

---
 contrib/postgres_fdw/expected/postgres_fdw.out |  68 ++++++++--------
 contrib/postgres_fdw/postgres_fdw.c            |   5 +-
 src/backend/executor/execAsync.c               | 105 ++++++++++++++-----------
 src/backend/executor/nodeAppend.c              |  50 ++++++------
 src/backend/executor/nodeForeignscan.c         |   4 +-
 src/backend/nodes/copyfuncs.c                  |   1 +
 src/backend/nodes/outfuncs.c                   |   1 +
 src/backend/nodes/readfuncs.c                  |   1 +
 src/backend/optimizer/plan/createplan.c        |  24 +++++-
 src/backend/utils/adt/ruleutils.c              |   6 +-
 src/include/executor/nodeForeignscan.h         |   2 +-
 src/include/foreign/fdwapi.h                   |   2 +-
 src/include/nodes/execnodes.h                  |  10 ++-
 src/include/nodes/plannodes.h                  |   1 +
 14 files changed, 167 insertions(+), 113 deletions(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index f9fd172..4b76e41 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -6329,13 +6329,13 @@ insert into bar2 values(4,44,44);
 insert into bar2 values(7,77,77);
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for update;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6343,10 +6343,10 @@ select * from bar where f1 in (select f1 from foo) for update;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6366,13 +6366,13 @@ select * from bar where f1 in (select f1 from foo) for update;
 
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for share;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6380,10 +6380,10 @@ select * from bar where f1 in (select f1 from foo) for share;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6404,22 +6404,22 @@ select * from bar where f1 in (select f1 from foo) for share;
 -- Check UPDATE with inherited target and an inherited source table
 explain (verbose, costs off)
 update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
-                                               QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
  Update on public.bar
    Update on public.bar
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar.f1 = foo2.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Seq Scan on public.bar
                Output: bar.f1, bar.f2, bar.ctid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6427,16 +6427,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
                            ->  Seq Scan on public.foo
                                  Output: foo.ctid, foo.*, foo.tableoid, foo.f1
    ->  Hash Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar2.f1 = foo.f1)
          ->  Foreign Scan on public.bar2
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6470,8 +6470,8 @@ where bar.f1 = ss.f1;
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1))
-         Hash Cond: (foo2.f1 = bar.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1))
+         Hash Cond: (foo.f1 = bar.f1)
          ->  Append
                ->  Foreign Scan on public.foo2
                      Output: ROW(foo2.f1), foo2.f1
@@ -6488,8 +6488,8 @@ where bar.f1 = ss.f1;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid
    ->  Merge Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1))
-         Merge Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1))
+         Merge Cond: (bar2.f1 = foo.f1)
          ->  Sort
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Sort Key: bar2.f1
@@ -6497,8 +6497,8 @@ where bar.f1 = ss.f1;
                      Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                      Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Sort
-               Output: (ROW(foo2.f1)), foo2.f1
-               Sort Key: foo2.f1
+               Output: (ROW(foo.f1)), foo.f1
+               Sort Key: foo.f1
                ->  Append
                      ->  Foreign Scan on public.foo2
                            Output: ROW(foo2.f1), foo2.f1
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index e75f8a1..830212f 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root,
 static bool postgresIsForeignPathAsyncCapable(ForeignPath *path);
 static void postgresForeignAsyncRequest(EState *estate,
 							PendingAsyncRequest *areq);
-static void postgresForeignAsyncConfigureWait(EState *estate,
+static bool postgresForeignAsyncConfigureWait(EState *estate,
 								  PendingAsyncRequest *areq,
 								  bool reinit);
 static void postgresForeignAsyncNotify(EState *estate,
@@ -4479,11 +4479,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq)
 	ExecAsyncRequestDone(estate, areq, (Node *) slot);
 }
 
-static void
+static bool
 postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 								  bool reinit)
 {
 	elog(ERROR, "postgresForeignAsyncConfigureWait");
+	return false;
 }
 
 static void
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index e070c26..33496a9 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -22,7 +22,7 @@
 #include "storage/latch.h"
 
 static bool ExecAsyncEventWait(EState *estate, long timeout);
-static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
+static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 	bool reinit);
 static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq);
 static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq);
@@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 				 PlanState *requestee)
 {
 	PendingAsyncRequest *areq = NULL;
-	int		i = estate->es_num_pending_async;
+	int		nasync = estate->es_num_pending_async;
 
 	/*
 	 * If the number of pending asynchronous nodes exceeds the number of
@@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * We start with 16 slots, and thereafter double the array size each
 	 * time we run out of slots.
 	 */
-	if (i >= estate->es_max_pending_async)
+	if (nasync >= estate->es_max_pending_async)
 	{
 		int	newmax;
 
@@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * PendingAsyncRequest if there is one.  If not, we must allocate a new
 	 * one.
 	 */
-	if (estate->es_pending_async[i] == NULL)
+	if (estate->es_pending_async[nasync] == NULL)
 	{
 		areq = MemoryContextAllocZero(estate->es_query_cxt,
 									  sizeof(PendingAsyncRequest));
-		estate->es_pending_async[i] = areq;
+		estate->es_pending_async[nasync] = areq;
 	}
 	else
 	{
-		areq = estate->es_pending_async[i];
+		areq = estate->es_pending_async[nasync];
 		MemSet(areq, 0, sizeof(PendingAsyncRequest));
 	}
-	areq->myindex = estate->es_num_pending_async++;
+	areq->myindex = estate->es_num_pending_async;
 
 	/* Initialize the new request. */
 	areq->requestor = requestor;
 	areq->request_index = request_index;
 	areq->requestee = requestee;
 
-	/* Give the requestee a chance to do whatever it wants. */
+	/*
+	 * Give the requestee a chance to do whatever it wants.
+	 * Requst functions return true if a result is immediately available.
+	 */
 	switch (nodeTag(requestee))
 	{
 		case T_ForeignScanState:
@@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 			elog(ERROR, "unrecognized node type: %d",
 				(int) nodeTag(requestee));
 	}
+
+	/*
+	 * If a result is available, complete it immediately.
+	 */
+	if (areq->state == ASYNC_COMPLETE)
+	{
+		Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot));
+		ExecAsyncResponse(estate, areq);
+
+		return;
+	}
+
+	/* No result available now, make this node pending */
+	estate->es_num_pending_async++;
 }
 
 /*
@@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 		{
 			PendingAsyncRequest *areq = estate->es_pending_async[i];
 
-			/* Skip it if no callback is pending. */
-			if (!areq->callback_pending)
-				continue;
-
-			/*
-			 * Mark it as no longer needing a callback.  We must do this
-			 * before dispatching the callback in case the callback resets
-			 * the flag.
-			 */
-			areq->callback_pending = false;
-			estate->es_async_callback_pending--;
-
-			/* Perform the actual callback; set request_done if appropraite. */
-			if (!areq->request_complete)
+			/* Skip it if not pending. */
+			if (areq->state == ASYNC_CALLBACK_PENDING)
+			{
+				/*
+				 * Mark it as no longer needing a callback.  We must do this
+				 * before dispatching the callback in case the callback resets
+				 * the flag.
+				 */
+				estate->es_async_callback_pending--;
 				ExecAsyncNotify(estate, areq);
-			else
+			}
+
+			if (areq->state == ASYNC_COMPLETE)
 			{
 				any_node_done = true;
 				if (requestor == areq->requestor)
@@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 				PendingAsyncRequest *head;
 				PendingAsyncRequest *tail = estate->es_pending_async[tidx];
 
-				if (!tail->callback_pending && tail->request_complete)
+				if (tail->state == ASYNC_COMPLETE)
 					continue;
 				head = estate->es_pending_async[hidx];
 				estate->es_pending_async[tidx] = head;
@@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
  * means wait forever, 0 means don't wait at all, and >0 means wait for the
  * indicated number of milliseconds.
  *
- * Returns true if we found some events and false if we timed out.
+ * Returns true if we found some events and false if we timed out or there's
+ * no event to wait. The latter is occur when the areq is processed during
  */
 static bool
 ExecAsyncEventWait(EState *estate, long timeout)
@@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout)
 	int		n;
 	bool	reinit = false;
 	bool	process_latch_set = false;
+	bool	added = false;
 
 	if (estate->es_wait_event_set == NULL)
 	{
@@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		PendingAsyncRequest *areq = estate->es_pending_async[i];
 
 		if (areq->num_fd_events > 0)
-			ExecAsyncConfigureWait(estate, areq, reinit);
+			added |= ExecAsyncConfigureWait(estate, areq, reinit);
 	}
 
+	Assert(added);
+
 	/* Wait for at least one event to occur. */
 	noccurred = WaitEventSetWait(estate->es_wait_event_set, timeout,
 								 occurred_event, EVENT_BUFFER_SIZE,
 								 WAIT_EVENT_ASYNC_WAIT);
+
 	if (noccurred == 0)
 		return false;
 
@@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		{
 			PendingAsyncRequest *areq = w->user_data;
 
-			if (!areq->callback_pending)
-			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
-				estate->es_async_callback_pending++;
-			}
+			Assert(areq->state == ASYNC_WAITING);
+
+			areq->state = ASYNC_CALLBACK_PENDING;
+			estate->es_async_callback_pending++;
 		}
 	}
 
@@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout)
 
 			if (areq->wants_process_latch)
 			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
+				Assert(areq->state == ASYNC_WAITING);
+				areq->state = ASYNC_CALLBACK_PENDING;
 			}
 		}
 	}
@@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout)
  * The events should include only WL_SOCKET_READABLE or WL_SOCKET_WRITEABLE,
  * and the number of calls should not exceed areq->num_fd_events (as
  * prevously set via ExecAsyncSetRequiredEvents).
+ *
+ * Individual requests can omit registering an event but it is a
+ * responsibility of the node driver to set at least one event per one
+ * requestor.
  */
-static void
+static bool
 ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 					   bool reinit)
 {
 	switch (nodeTag(areq->requestee))
 	{
 		case T_ForeignScanState:
-			ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
+			return ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq,
 	estate->es_total_fd_events += num_fd_events - areq->num_fd_events;
 	areq->num_fd_events = num_fd_events;
 	areq->wants_process_latch = wants_process_latch;
+	areq->state = ASYNC_WAITING;
 
 	if (force_reset && estate->es_wait_event_set != NULL)
 	{
@@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result)
 	 * need a callback to remove registered wait events.  It's not clear
 	 * that we would come out ahead, so use brute force for now.
 	 */
+	Assert(areq->state == ASYNC_IDLE || areq->state == ASYNC_CALLBACK_PENDING);
+
 	if (areq->num_fd_events > 0 || areq->wants_process_latch)
 		ExecAsyncSetRequiredEvents(estate, areq, 0, false, true);
 
 	/* Save result and mark request as complete. */
 	areq->result = result;
-	areq->request_complete = true;
-
-	/* Make sure this request is flagged for a callback. */
-	if (!areq->callback_pending)
-	{
-		areq->callback_pending = true;
-		estate->es_async_callback_pending++;
-	}
+	areq->state = ASYNC_COMPLETE;
 }
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index bb06569..c234f1f 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -229,9 +229,15 @@ ExecAppend(AppendState *node)
 		 */
 		while ((i = bms_first_member(node->as_needrequest)) >= 0)
 		{
-			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
 			node->as_nasyncpending++;
+
+			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
+			/* If this request immediately gives a result, take it. */
+			if (node->as_nasyncresult > 0)
+				return node->as_asyncresult[--node->as_nasyncresult];
 		}
+		if (node->as_nasyncpending == 0 && node->as_syncdone)
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 	}
 
 	for (;;)
@@ -246,32 +252,32 @@ ExecAppend(AppendState *node)
 		{
 			long	timeout = node->as_syncdone ? -1 : 0;
 
-			for (;;)
+			while (node->as_nasyncpending > 0)
 			{
-				if (node->as_nasyncpending == 0)
-				{
-					/*
-					 * If there is no asynchronous activity still pending
-					 * and the synchronous activity is also complete, we're
-					 * totally done scanning this node.  Otherwise, we're
-					 * done with the asynchronous stuff but must continue
-					 * scanning the synchronous children.
-					 */
-					if (node->as_syncdone)
-						return ExecClearTuple(node->ps.ps_ResultTupleSlot);
-					break;
-				}
-				if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout))
-				{
-					/* Timeout reached. */
-					break;
-				}
-				if (node->as_nasyncresult > 0)
+				if (ExecAsyncEventLoop(node->ps.state, &node->ps, timeout) &&
+					node->as_nasyncresult > 0)
 				{
 					/* Asynchronous subplan returned a tuple! */
 					--node->as_nasyncresult;
 					return node->as_asyncresult[node->as_nasyncresult];
 				}
+
+				/* Timeout reached. Go through to sync nodes if exists */
+				if (!node->as_syncdone)
+					break;
+			}
+
+			/*
+			 * If there is no asynchronous activity still pending and the
+			 * synchronous activity is also complete, we're totally done
+			 * scanning this node.  Otherwise, we're done with the
+			 * asynchronous stuff but must continue scanning the synchronous
+			 * children.
+			 */
+			if (node->as_syncdone)
+			{
+				Assert(node->as_nasyncpending == 0);
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 			}
 		}
 
@@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq)
 	TupleTableSlot *slot;
 
 	/* We shouldn't be called until the request is complete. */
-	Assert(areq->request_complete);
+	Assert(areq->state == ASYNC_COMPLETE);
 
 	/* Our result slot shouldn't already be occupied. */
 	Assert(TupIsNull(node->ps.ps_ResultTupleSlot));
diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c
index 85d436f..d3567bb 100644
--- a/src/backend/executor/nodeForeignscan.c
+++ b/src/backend/executor/nodeForeignscan.c
@@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq)
  *		In async mode, configure for a wait
  * ----------------------------------------------------------------
  */
-void
+bool
 ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit)
 {
@@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate,
 	FdwRoutine *fdwroutine = node->fdwroutine;
 
 	Assert(fdwroutine->ForeignAsyncConfigureWait != NULL);
-	fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
+	return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 23b4e18..72d8cd6 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -219,6 +219,7 @@ _copyAppend(const Append *from)
 	 */
 	COPY_NODE_FIELD(appendplans);
 	COPY_SCALAR_FIELD(nasyncplans);
+	COPY_SCALAR_FIELD(referent);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index dc5b938..1ebdc48 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node)
 
 	WRITE_NODE_FIELD(appendplans);
 	WRITE_INT_FIELD(nasyncplans);
+	WRITE_INT_FIELD(referent);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 69453b5..8443a62 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1520,6 +1520,7 @@ _readAppend(void)
 
 	READ_NODE_FIELD(appendplans);
 	READ_INT_FIELD(nasyncplans);
+	READ_INT_FIELD(referent);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 7caa8d3..ff1d663 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual,
 			 Index scanrelid, int ctePlanId, int cteParam);
 static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual,
 				   Index scanrelid, int wtParam);
-static Append *make_append(List *asyncplans, int nasyncplans, List *tlist);
+static Append *make_append(List *asyncplans, int nasyncplans,
+						   int referent, List *tlist);
 static RecursiveUnion *make_recursive_union(List *tlist,
 					 Plan *lefttree,
 					 Plan *righttree,
@@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	List	   *syncplans = NIL;
 	ListCell   *subpaths;
 	int			nasyncplans = 0;
+	bool		first = true;
+	bool		referent_is_sync = true;
 
 	/*
 	 * The subpaths list could be empty, if every child was proven empty by
@@ -985,7 +988,14 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		return plan;
 	}
 
-	/* Build the plan for each child */
+	/*
+	 * Build the plan for each child
+
+	 * The first child in an inheritance set is the representative in
+	 * explaining tlist entries (see set_deparse_planstate). We should keep
+	 * the first child in best_path->subpaths at the head of the subplan list
+	 * for the reason.
+	 */
 	foreach(subpaths, best_path->subpaths)
 	{
 		Path	   *subpath = (Path *) lfirst(subpaths);
@@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		{
 			asyncplans = lappend(asyncplans, subplan);
 			++nasyncplans;
+			if (first)
+				referent_is_sync = false;
 		}
 		else
 			syncplans = lappend(syncplans, subplan);
+
+		first = false;
 	}
 
 	/*
@@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	 * parent-rel Vars it'll be asked to emit.
 	 */
 
-	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist);
+	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans,
+					   referent_is_sync ? nasyncplans : 0, tlist);
 
 	copy_generic_path_info(&plan->plan, (Path *) best_path);
 
@@ -4951,7 +4966,7 @@ make_foreignscan(List *qptlist,
 }
 
 static Append *
-make_append(List *appendplans, int nasyncplans, List *tlist)
+make_append(List *appendplans, int nasyncplans,	int referent, List *tlist)
 {
 	Append	   *node = makeNode(Append);
 	Plan	   *plan = &node->plan;
@@ -4962,6 +4977,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist)
 	plan->righttree = NULL;
 	node->appendplans = appendplans;
 	node->nasyncplans = nasyncplans;
+	node->referent = referent;
 
 	return node;
 }
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 8a81d7a..de0e96c 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -4056,7 +4056,11 @@ set_deparse_planstate(deparse_namespace *dpns, PlanState *ps)
 	 * lists containing references to non-target relations.
 	 */
 	if (IsA(ps, AppendState))
-		dpns->outer_planstate = ((AppendState *) ps)->appendplans[0];
+	{
+		int idx = ((Append*)(((AppendState *) ps)->ps.plan))->referent;
+		dpns->outer_planstate =
+			((AppendState *) ps)->appendplans[idx];
+	}
 	else if (IsA(ps, MergeAppendState))
 		dpns->outer_planstate = ((MergeAppendState *) ps)->mergeplans[0];
 	else if (IsA(ps, ModifyTableState))
diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h
index 3e69ab0..47a3920 100644
--- a/src/include/executor/nodeForeignscan.h
+++ b/src/include/executor/nodeForeignscan.h
@@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node,
 
 extern void ExecAsyncForeignScanRequest(EState *estate,
 	PendingAsyncRequest *areq);
-extern void ExecAsyncForeignScanConfigureWait(EState *estate,
+extern bool ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit);
 extern void ExecAsyncForeignScanNotify(EState *estate,
 	PendingAsyncRequest *areq);
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 88feb9a..65517fd 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root,
 typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path);
 typedef void (*ForeignAsyncRequest_function) (EState *estate,
 											PendingAsyncRequest *areq);
-typedef void (*ForeignAsyncConfigureWait_function) (EState *estate,
+typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate,
 											PendingAsyncRequest *areq,
 											bool reinit);
 typedef void (*ForeignAsyncNotify_function) (EState *estate,
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b50b41c..0c6af86 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -352,6 +352,13 @@ typedef struct ResultRelInfo
  * State for an asynchronous tuple request.
  * ----------------
  */
+typedef enum AsyncRequestState
+{
+	ASYNC_IDLE,
+	ASYNC_WAITING,
+	ASYNC_CALLBACK_PENDING,
+	ASYNC_COMPLETE
+} AsyncRequestState;
 typedef struct PendingAsyncRequest
 {
 	int			myindex;			/* Index in es_pending_async. */
@@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest
 	int			request_index;	/* Scratch space for requestor. */
 	int			num_fd_events;	/* Max number of FD events requestee needs. */
 	bool		wants_process_latch;	/* Requestee cares about MyLatch. */
-	bool		callback_pending;			/* Callback is needed. */
-	bool		request_complete;		/* Request complete, result valid. */
+	AsyncRequestState state;
 	Node	   *result;			/* Result (NULL if no more tuples). */
 } PendingAsyncRequest;
 
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 327119b..1df6693 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -209,6 +209,7 @@ typedef struct Append
 	Plan		plan;
 	List	   *appendplans;
 	int			nasyncplans;	/* # of async plans, always at start of list */
+	int			referent; 		/* index of inheritance tree referent */
 } Append;
 
 /* ----------------
-- 
2.9.2


----Next_Part(Tue_Oct_25_18_21_50_2016_530)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0004-Make-postgres_fdw-async-capable.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH 3/7] Modify async execution infrastructure.
@ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 6+ messages in thread

From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw)

---
 contrib/postgres_fdw/expected/postgres_fdw.out |  68 ++++++++--------
 contrib/postgres_fdw/postgres_fdw.c            |   5 +-
 src/backend/executor/execAsync.c               | 105 ++++++++++++++-----------
 src/backend/executor/nodeAppend.c              |  50 ++++++------
 src/backend/executor/nodeForeignscan.c         |   4 +-
 src/backend/nodes/copyfuncs.c                  |   1 +
 src/backend/nodes/outfuncs.c                   |   1 +
 src/backend/nodes/readfuncs.c                  |   1 +
 src/backend/optimizer/plan/createplan.c        |  24 +++++-
 src/backend/utils/adt/ruleutils.c              |   6 +-
 src/include/executor/nodeForeignscan.h         |   2 +-
 src/include/foreign/fdwapi.h                   |   2 +-
 src/include/nodes/execnodes.h                  |  10 ++-
 src/include/nodes/plannodes.h                  |   1 +
 14 files changed, 167 insertions(+), 113 deletions(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 1b36579..a98e138 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -6321,13 +6321,13 @@ insert into bar2 values(4,44,44);
 insert into bar2 values(7,77,77);
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for update;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6335,10 +6335,10 @@ select * from bar where f1 in (select f1 from foo) for update;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6358,13 +6358,13 @@ select * from bar where f1 in (select f1 from foo) for update;
 
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for share;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6372,10 +6372,10 @@ select * from bar where f1 in (select f1 from foo) for share;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6396,22 +6396,22 @@ select * from bar where f1 in (select f1 from foo) for share;
 -- Check UPDATE with inherited target and an inherited source table
 explain (verbose, costs off)
 update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
-                                               QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
  Update on public.bar
    Update on public.bar
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar.f1 = foo2.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Seq Scan on public.bar
                Output: bar.f1, bar.f2, bar.ctid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6419,16 +6419,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
                            ->  Seq Scan on public.foo
                                  Output: foo.ctid, foo.*, foo.tableoid, foo.f1
    ->  Hash Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar2.f1 = foo.f1)
          ->  Foreign Scan on public.bar2
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6462,8 +6462,8 @@ where bar.f1 = ss.f1;
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1))
-         Hash Cond: (foo2.f1 = bar.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1))
+         Hash Cond: (foo.f1 = bar.f1)
          ->  Append
                ->  Foreign Scan on public.foo2
                      Output: ROW(foo2.f1), foo2.f1
@@ -6480,8 +6480,8 @@ where bar.f1 = ss.f1;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid
    ->  Merge Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1))
-         Merge Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1))
+         Merge Cond: (bar2.f1 = foo.f1)
          ->  Sort
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Sort Key: bar2.f1
@@ -6489,8 +6489,8 @@ where bar.f1 = ss.f1;
                      Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                      Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Sort
-               Output: (ROW(foo2.f1)), foo2.f1
-               Sort Key: foo2.f1
+               Output: (ROW(foo.f1)), foo.f1
+               Sort Key: foo.f1
                ->  Append
                      ->  Foreign Scan on public.foo2
                            Output: ROW(foo2.f1), foo2.f1
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index e75f8a1..830212f 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root,
 static bool postgresIsForeignPathAsyncCapable(ForeignPath *path);
 static void postgresForeignAsyncRequest(EState *estate,
 							PendingAsyncRequest *areq);
-static void postgresForeignAsyncConfigureWait(EState *estate,
+static bool postgresForeignAsyncConfigureWait(EState *estate,
 								  PendingAsyncRequest *areq,
 								  bool reinit);
 static void postgresForeignAsyncNotify(EState *estate,
@@ -4479,11 +4479,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq)
 	ExecAsyncRequestDone(estate, areq, (Node *) slot);
 }
 
-static void
+static bool
 postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 								  bool reinit)
 {
 	elog(ERROR, "postgresForeignAsyncConfigureWait");
+	return false;
 }
 
 static void
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index e070c26..33496a9 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -22,7 +22,7 @@
 #include "storage/latch.h"
 
 static bool ExecAsyncEventWait(EState *estate, long timeout);
-static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
+static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 	bool reinit);
 static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq);
 static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq);
@@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 				 PlanState *requestee)
 {
 	PendingAsyncRequest *areq = NULL;
-	int		i = estate->es_num_pending_async;
+	int		nasync = estate->es_num_pending_async;
 
 	/*
 	 * If the number of pending asynchronous nodes exceeds the number of
@@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * We start with 16 slots, and thereafter double the array size each
 	 * time we run out of slots.
 	 */
-	if (i >= estate->es_max_pending_async)
+	if (nasync >= estate->es_max_pending_async)
 	{
 		int	newmax;
 
@@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * PendingAsyncRequest if there is one.  If not, we must allocate a new
 	 * one.
 	 */
-	if (estate->es_pending_async[i] == NULL)
+	if (estate->es_pending_async[nasync] == NULL)
 	{
 		areq = MemoryContextAllocZero(estate->es_query_cxt,
 									  sizeof(PendingAsyncRequest));
-		estate->es_pending_async[i] = areq;
+		estate->es_pending_async[nasync] = areq;
 	}
 	else
 	{
-		areq = estate->es_pending_async[i];
+		areq = estate->es_pending_async[nasync];
 		MemSet(areq, 0, sizeof(PendingAsyncRequest));
 	}
-	areq->myindex = estate->es_num_pending_async++;
+	areq->myindex = estate->es_num_pending_async;
 
 	/* Initialize the new request. */
 	areq->requestor = requestor;
 	areq->request_index = request_index;
 	areq->requestee = requestee;
 
-	/* Give the requestee a chance to do whatever it wants. */
+	/*
+	 * Give the requestee a chance to do whatever it wants.
+	 * Requst functions return true if a result is immediately available.
+	 */
 	switch (nodeTag(requestee))
 	{
 		case T_ForeignScanState:
@@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 			elog(ERROR, "unrecognized node type: %d",
 				(int) nodeTag(requestee));
 	}
+
+	/*
+	 * If a result is available, complete it immediately.
+	 */
+	if (areq->state == ASYNC_COMPLETE)
+	{
+		Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot));
+		ExecAsyncResponse(estate, areq);
+
+		return;
+	}
+
+	/* No result available now, make this node pending */
+	estate->es_num_pending_async++;
 }
 
 /*
@@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 		{
 			PendingAsyncRequest *areq = estate->es_pending_async[i];
 
-			/* Skip it if no callback is pending. */
-			if (!areq->callback_pending)
-				continue;
-
-			/*
-			 * Mark it as no longer needing a callback.  We must do this
-			 * before dispatching the callback in case the callback resets
-			 * the flag.
-			 */
-			areq->callback_pending = false;
-			estate->es_async_callback_pending--;
-
-			/* Perform the actual callback; set request_done if appropraite. */
-			if (!areq->request_complete)
+			/* Skip it if not pending. */
+			if (areq->state == ASYNC_CALLBACK_PENDING)
+			{
+				/*
+				 * Mark it as no longer needing a callback.  We must do this
+				 * before dispatching the callback in case the callback resets
+				 * the flag.
+				 */
+				estate->es_async_callback_pending--;
 				ExecAsyncNotify(estate, areq);
-			else
+			}
+
+			if (areq->state == ASYNC_COMPLETE)
 			{
 				any_node_done = true;
 				if (requestor == areq->requestor)
@@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 				PendingAsyncRequest *head;
 				PendingAsyncRequest *tail = estate->es_pending_async[tidx];
 
-				if (!tail->callback_pending && tail->request_complete)
+				if (tail->state == ASYNC_COMPLETE)
 					continue;
 				head = estate->es_pending_async[hidx];
 				estate->es_pending_async[tidx] = head;
@@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
  * means wait forever, 0 means don't wait at all, and >0 means wait for the
  * indicated number of milliseconds.
  *
- * Returns true if we found some events and false if we timed out.
+ * Returns true if we found some events and false if we timed out or there's
+ * no event to wait. The latter is occur when the areq is processed during
  */
 static bool
 ExecAsyncEventWait(EState *estate, long timeout)
@@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout)
 	int		n;
 	bool	reinit = false;
 	bool	process_latch_set = false;
+	bool	added = false;
 
 	if (estate->es_wait_event_set == NULL)
 	{
@@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		PendingAsyncRequest *areq = estate->es_pending_async[i];
 
 		if (areq->num_fd_events > 0)
-			ExecAsyncConfigureWait(estate, areq, reinit);
+			added |= ExecAsyncConfigureWait(estate, areq, reinit);
 	}
 
+	Assert(added);
+
 	/* Wait for at least one event to occur. */
 	noccurred = WaitEventSetWait(estate->es_wait_event_set, timeout,
 								 occurred_event, EVENT_BUFFER_SIZE,
 								 WAIT_EVENT_ASYNC_WAIT);
+
 	if (noccurred == 0)
 		return false;
 
@@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		{
 			PendingAsyncRequest *areq = w->user_data;
 
-			if (!areq->callback_pending)
-			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
-				estate->es_async_callback_pending++;
-			}
+			Assert(areq->state == ASYNC_WAITING);
+
+			areq->state = ASYNC_CALLBACK_PENDING;
+			estate->es_async_callback_pending++;
 		}
 	}
 
@@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout)
 
 			if (areq->wants_process_latch)
 			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
+				Assert(areq->state == ASYNC_WAITING);
+				areq->state = ASYNC_CALLBACK_PENDING;
 			}
 		}
 	}
@@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout)
  * The events should include only WL_SOCKET_READABLE or WL_SOCKET_WRITEABLE,
  * and the number of calls should not exceed areq->num_fd_events (as
  * prevously set via ExecAsyncSetRequiredEvents).
+ *
+ * Individual requests can omit registering an event but it is a
+ * responsibility of the node driver to set at least one event per one
+ * requestor.
  */
-static void
+static bool
 ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 					   bool reinit)
 {
 	switch (nodeTag(areq->requestee))
 	{
 		case T_ForeignScanState:
-			ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
+			return ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq,
 	estate->es_total_fd_events += num_fd_events - areq->num_fd_events;
 	areq->num_fd_events = num_fd_events;
 	areq->wants_process_latch = wants_process_latch;
+	areq->state = ASYNC_WAITING;
 
 	if (force_reset && estate->es_wait_event_set != NULL)
 	{
@@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result)
 	 * need a callback to remove registered wait events.  It's not clear
 	 * that we would come out ahead, so use brute force for now.
 	 */
+	Assert(areq->state == ASYNC_IDLE || areq->state == ASYNC_CALLBACK_PENDING);
+
 	if (areq->num_fd_events > 0 || areq->wants_process_latch)
 		ExecAsyncSetRequiredEvents(estate, areq, 0, false, true);
 
 	/* Save result and mark request as complete. */
 	areq->result = result;
-	areq->request_complete = true;
-
-	/* Make sure this request is flagged for a callback. */
-	if (!areq->callback_pending)
-	{
-		areq->callback_pending = true;
-		estate->es_async_callback_pending++;
-	}
+	areq->state = ASYNC_COMPLETE;
 }
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index bb06569..c234f1f 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -229,9 +229,15 @@ ExecAppend(AppendState *node)
 		 */
 		while ((i = bms_first_member(node->as_needrequest)) >= 0)
 		{
-			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
 			node->as_nasyncpending++;
+
+			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
+			/* If this request immediately gives a result, take it. */
+			if (node->as_nasyncresult > 0)
+				return node->as_asyncresult[--node->as_nasyncresult];
 		}
+		if (node->as_nasyncpending == 0 && node->as_syncdone)
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 	}
 
 	for (;;)
@@ -246,32 +252,32 @@ ExecAppend(AppendState *node)
 		{
 			long	timeout = node->as_syncdone ? -1 : 0;
 
-			for (;;)
+			while (node->as_nasyncpending > 0)
 			{
-				if (node->as_nasyncpending == 0)
-				{
-					/*
-					 * If there is no asynchronous activity still pending
-					 * and the synchronous activity is also complete, we're
-					 * totally done scanning this node.  Otherwise, we're
-					 * done with the asynchronous stuff but must continue
-					 * scanning the synchronous children.
-					 */
-					if (node->as_syncdone)
-						return ExecClearTuple(node->ps.ps_ResultTupleSlot);
-					break;
-				}
-				if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout))
-				{
-					/* Timeout reached. */
-					break;
-				}
-				if (node->as_nasyncresult > 0)
+				if (ExecAsyncEventLoop(node->ps.state, &node->ps, timeout) &&
+					node->as_nasyncresult > 0)
 				{
 					/* Asynchronous subplan returned a tuple! */
 					--node->as_nasyncresult;
 					return node->as_asyncresult[node->as_nasyncresult];
 				}
+
+				/* Timeout reached. Go through to sync nodes if exists */
+				if (!node->as_syncdone)
+					break;
+			}
+
+			/*
+			 * If there is no asynchronous activity still pending and the
+			 * synchronous activity is also complete, we're totally done
+			 * scanning this node.  Otherwise, we're done with the
+			 * asynchronous stuff but must continue scanning the synchronous
+			 * children.
+			 */
+			if (node->as_syncdone)
+			{
+				Assert(node->as_nasyncpending == 0);
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 			}
 		}
 
@@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq)
 	TupleTableSlot *slot;
 
 	/* We shouldn't be called until the request is complete. */
-	Assert(areq->request_complete);
+	Assert(areq->state == ASYNC_COMPLETE);
 
 	/* Our result slot shouldn't already be occupied. */
 	Assert(TupIsNull(node->ps.ps_ResultTupleSlot));
diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c
index 85d436f..d3567bb 100644
--- a/src/backend/executor/nodeForeignscan.c
+++ b/src/backend/executor/nodeForeignscan.c
@@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq)
  *		In async mode, configure for a wait
  * ----------------------------------------------------------------
  */
-void
+bool
 ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit)
 {
@@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate,
 	FdwRoutine *fdwroutine = node->fdwroutine;
 
 	Assert(fdwroutine->ForeignAsyncConfigureWait != NULL);
-	fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
+	return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 23b4e18..72d8cd6 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -219,6 +219,7 @@ _copyAppend(const Append *from)
 	 */
 	COPY_NODE_FIELD(appendplans);
 	COPY_SCALAR_FIELD(nasyncplans);
+	COPY_SCALAR_FIELD(referent);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index dc5b938..1ebdc48 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node)
 
 	WRITE_NODE_FIELD(appendplans);
 	WRITE_INT_FIELD(nasyncplans);
+	WRITE_INT_FIELD(referent);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 69453b5..8443a62 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1520,6 +1520,7 @@ _readAppend(void)
 
 	READ_NODE_FIELD(appendplans);
 	READ_INT_FIELD(nasyncplans);
+	READ_INT_FIELD(referent);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 7caa8d3..ff1d663 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual,
 			 Index scanrelid, int ctePlanId, int cteParam);
 static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual,
 				   Index scanrelid, int wtParam);
-static Append *make_append(List *asyncplans, int nasyncplans, List *tlist);
+static Append *make_append(List *asyncplans, int nasyncplans,
+						   int referent, List *tlist);
 static RecursiveUnion *make_recursive_union(List *tlist,
 					 Plan *lefttree,
 					 Plan *righttree,
@@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	List	   *syncplans = NIL;
 	ListCell   *subpaths;
 	int			nasyncplans = 0;
+	bool		first = true;
+	bool		referent_is_sync = true;
 
 	/*
 	 * The subpaths list could be empty, if every child was proven empty by
@@ -985,7 +988,14 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		return plan;
 	}
 
-	/* Build the plan for each child */
+	/*
+	 * Build the plan for each child
+
+	 * The first child in an inheritance set is the representative in
+	 * explaining tlist entries (see set_deparse_planstate). We should keep
+	 * the first child in best_path->subpaths at the head of the subplan list
+	 * for the reason.
+	 */
 	foreach(subpaths, best_path->subpaths)
 	{
 		Path	   *subpath = (Path *) lfirst(subpaths);
@@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		{
 			asyncplans = lappend(asyncplans, subplan);
 			++nasyncplans;
+			if (first)
+				referent_is_sync = false;
 		}
 		else
 			syncplans = lappend(syncplans, subplan);
+
+		first = false;
 	}
 
 	/*
@@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	 * parent-rel Vars it'll be asked to emit.
 	 */
 
-	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist);
+	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans,
+					   referent_is_sync ? nasyncplans : 0, tlist);
 
 	copy_generic_path_info(&plan->plan, (Path *) best_path);
 
@@ -4951,7 +4966,7 @@ make_foreignscan(List *qptlist,
 }
 
 static Append *
-make_append(List *appendplans, int nasyncplans, List *tlist)
+make_append(List *appendplans, int nasyncplans,	int referent, List *tlist)
 {
 	Append	   *node = makeNode(Append);
 	Plan	   *plan = &node->plan;
@@ -4962,6 +4977,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist)
 	plan->righttree = NULL;
 	node->appendplans = appendplans;
 	node->nasyncplans = nasyncplans;
+	node->referent = referent;
 
 	return node;
 }
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 8a81d7a..de0e96c 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -4056,7 +4056,11 @@ set_deparse_planstate(deparse_namespace *dpns, PlanState *ps)
 	 * lists containing references to non-target relations.
 	 */
 	if (IsA(ps, AppendState))
-		dpns->outer_planstate = ((AppendState *) ps)->appendplans[0];
+	{
+		int idx = ((Append*)(((AppendState *) ps)->ps.plan))->referent;
+		dpns->outer_planstate =
+			((AppendState *) ps)->appendplans[idx];
+	}
 	else if (IsA(ps, MergeAppendState))
 		dpns->outer_planstate = ((MergeAppendState *) ps)->mergeplans[0];
 	else if (IsA(ps, ModifyTableState))
diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h
index 3e69ab0..47a3920 100644
--- a/src/include/executor/nodeForeignscan.h
+++ b/src/include/executor/nodeForeignscan.h
@@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node,
 
 extern void ExecAsyncForeignScanRequest(EState *estate,
 	PendingAsyncRequest *areq);
-extern void ExecAsyncForeignScanConfigureWait(EState *estate,
+extern bool ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit);
 extern void ExecAsyncForeignScanNotify(EState *estate,
 	PendingAsyncRequest *areq);
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 88feb9a..65517fd 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root,
 typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path);
 typedef void (*ForeignAsyncRequest_function) (EState *estate,
 											PendingAsyncRequest *areq);
-typedef void (*ForeignAsyncConfigureWait_function) (EState *estate,
+typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate,
 											PendingAsyncRequest *areq,
 											bool reinit);
 typedef void (*ForeignAsyncNotify_function) (EState *estate,
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b50b41c..0c6af86 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -352,6 +352,13 @@ typedef struct ResultRelInfo
  * State for an asynchronous tuple request.
  * ----------------
  */
+typedef enum AsyncRequestState
+{
+	ASYNC_IDLE,
+	ASYNC_WAITING,
+	ASYNC_CALLBACK_PENDING,
+	ASYNC_COMPLETE
+} AsyncRequestState;
 typedef struct PendingAsyncRequest
 {
 	int			myindex;			/* Index in es_pending_async. */
@@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest
 	int			request_index;	/* Scratch space for requestor. */
 	int			num_fd_events;	/* Max number of FD events requestee needs. */
 	bool		wants_process_latch;	/* Requestee cares about MyLatch. */
-	bool		callback_pending;			/* Callback is needed. */
-	bool		request_complete;		/* Request complete, result valid. */
+	AsyncRequestState state;
 	Node	   *result;			/* Result (NULL if no more tuples). */
 } PendingAsyncRequest;
 
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 327119b..1df6693 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -209,6 +209,7 @@ typedef struct Append
 	Plan		plan;
 	List	   *appendplans;
 	int			nasyncplans;	/* # of async plans, always at start of list */
+	int			referent; 		/* index of inheritance tree referent */
 } Append;
 
 /* ----------------
-- 
2.9.2


----Next_Part(Mon_Oct_31_10_39_12_2016_712)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="0004-Make-postgres_fdw-async-capable.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH 03/13] Modify async execution infrastructure.
@ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
  0 siblings, 0 replies; 6+ messages in thread

From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw)

---
 contrib/postgres_fdw/expected/postgres_fdw.out |  68 ++++++++--------
 contrib/postgres_fdw/postgres_fdw.c            |   5 +-
 src/backend/executor/execAsync.c               | 105 ++++++++++++++-----------
 src/backend/executor/nodeAppend.c              |  50 ++++++------
 src/backend/executor/nodeForeignscan.c         |   4 +-
 src/backend/nodes/copyfuncs.c                  |   1 +
 src/backend/nodes/outfuncs.c                   |   1 +
 src/backend/nodes/readfuncs.c                  |   1 +
 src/backend/optimizer/plan/createplan.c        |  24 +++++-
 src/backend/utils/adt/ruleutils.c              |   6 +-
 src/include/executor/nodeForeignscan.h         |   2 +-
 src/include/foreign/fdwapi.h                   |   2 +-
 src/include/nodes/execnodes.h                  |  10 ++-
 src/include/nodes/plannodes.h                  |   1 +
 14 files changed, 167 insertions(+), 113 deletions(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index df22beb..9180afe 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -6402,13 +6402,13 @@ insert into bar2 values(4,44,44);
 insert into bar2 values(7,77,77);
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for update;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6416,10 +6416,10 @@ select * from bar where f1 in (select f1 from foo) for update;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6439,13 +6439,13 @@ select * from bar where f1 in (select f1 from foo) for update;
 
 explain (verbose, costs off)
 select * from bar where f1 in (select f1 from foo) for share;
-                                                       QUERY PLAN                                                       
-------------------------------------------------------------------------------------------------------------------------
+                                          QUERY PLAN                                          
+----------------------------------------------------------------------------------------------
  LockRows
-   Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
+   Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
    ->  Hash Join
-         Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Append
                ->  Foreign Scan on public.bar2
                      Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid
@@ -6453,10 +6453,10 @@ select * from bar where f1 in (select f1 from foo) for share;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6477,22 +6477,22 @@ select * from bar where f1 in (select f1 from foo) for share;
 -- Check UPDATE with inherited target and an inherited source table
 explain (verbose, costs off)
 update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
-                                               QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
  Update on public.bar
    Update on public.bar
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar.f1 = foo2.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar.f1 = foo.f1)
          ->  Seq Scan on public.bar
                Output: bar.f1, bar.f2, bar.ctid
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6500,16 +6500,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo);
                            ->  Seq Scan on public.foo
                                  Output: foo.ctid, foo.*, foo.tableoid, foo.f1
    ->  Hash Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid
-         Hash Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid
+         Hash Cond: (bar2.f1 = foo.f1)
          ->  Foreign Scan on public.bar2
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Hash
-               Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
+               Output: foo.ctid, foo.*, foo.tableoid, foo.f1
                ->  HashAggregate
-                     Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1
-                     Group Key: foo2.f1
+                     Output: foo.ctid, foo.*, foo.tableoid, foo.f1
+                     Group Key: foo.f1
                      ->  Append
                            ->  Foreign Scan on public.foo2
                                  Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1
@@ -6543,8 +6543,8 @@ where bar.f1 = ss.f1;
    Foreign Update on public.bar2
      Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1
    ->  Hash Join
-         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1))
-         Hash Cond: (foo2.f1 = bar.f1)
+         Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1))
+         Hash Cond: (foo.f1 = bar.f1)
          ->  Append
                ->  Foreign Scan on public.foo2
                      Output: ROW(foo2.f1), foo2.f1
@@ -6561,8 +6561,8 @@ where bar.f1 = ss.f1;
                ->  Seq Scan on public.bar
                      Output: bar.f1, bar.f2, bar.ctid
    ->  Merge Join
-         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1))
-         Merge Cond: (bar2.f1 = foo2.f1)
+         Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1))
+         Merge Cond: (bar2.f1 = foo.f1)
          ->  Sort
                Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                Sort Key: bar2.f1
@@ -6570,8 +6570,8 @@ where bar.f1 = ss.f1;
                      Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid
                      Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE
          ->  Sort
-               Output: (ROW(foo2.f1)), foo2.f1
-               Sort Key: foo2.f1
+               Output: (ROW(foo.f1)), foo.f1
+               Sort Key: foo.f1
                ->  Append
                      ->  Foreign Scan on public.foo2
                            Output: ROW(foo2.f1), foo2.f1
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index f180838..abb256b 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root,
 static bool postgresIsForeignPathAsyncCapable(ForeignPath *path);
 static void postgresForeignAsyncRequest(EState *estate,
 							PendingAsyncRequest *areq);
-static void postgresForeignAsyncConfigureWait(EState *estate,
+static bool postgresForeignAsyncConfigureWait(EState *estate,
 								  PendingAsyncRequest *areq,
 								  bool reinit);
 static void postgresForeignAsyncNotify(EState *estate,
@@ -4477,11 +4477,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq)
 	ExecAsyncRequestDone(estate, areq, (Node *) slot);
 }
 
-static void
+static bool
 postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 								  bool reinit)
 {
 	elog(ERROR, "postgresForeignAsyncConfigureWait");
+	return false;
 }
 
 static void
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index e070c26..33496a9 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -22,7 +22,7 @@
 #include "storage/latch.h"
 
 static bool ExecAsyncEventWait(EState *estate, long timeout);
-static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
+static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 	bool reinit);
 static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq);
 static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq);
@@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 				 PlanState *requestee)
 {
 	PendingAsyncRequest *areq = NULL;
-	int		i = estate->es_num_pending_async;
+	int		nasync = estate->es_num_pending_async;
 
 	/*
 	 * If the number of pending asynchronous nodes exceeds the number of
@@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * We start with 16 slots, and thereafter double the array size each
 	 * time we run out of slots.
 	 */
-	if (i >= estate->es_max_pending_async)
+	if (nasync >= estate->es_max_pending_async)
 	{
 		int	newmax;
 
@@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 	 * PendingAsyncRequest if there is one.  If not, we must allocate a new
 	 * one.
 	 */
-	if (estate->es_pending_async[i] == NULL)
+	if (estate->es_pending_async[nasync] == NULL)
 	{
 		areq = MemoryContextAllocZero(estate->es_query_cxt,
 									  sizeof(PendingAsyncRequest));
-		estate->es_pending_async[i] = areq;
+		estate->es_pending_async[nasync] = areq;
 	}
 	else
 	{
-		areq = estate->es_pending_async[i];
+		areq = estate->es_pending_async[nasync];
 		MemSet(areq, 0, sizeof(PendingAsyncRequest));
 	}
-	areq->myindex = estate->es_num_pending_async++;
+	areq->myindex = estate->es_num_pending_async;
 
 	/* Initialize the new request. */
 	areq->requestor = requestor;
 	areq->request_index = request_index;
 	areq->requestee = requestee;
 
-	/* Give the requestee a chance to do whatever it wants. */
+	/*
+	 * Give the requestee a chance to do whatever it wants.
+	 * Requst functions return true if a result is immediately available.
+	 */
 	switch (nodeTag(requestee))
 	{
 		case T_ForeignScanState:
@@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index,
 			elog(ERROR, "unrecognized node type: %d",
 				(int) nodeTag(requestee));
 	}
+
+	/*
+	 * If a result is available, complete it immediately.
+	 */
+	if (areq->state == ASYNC_COMPLETE)
+	{
+		Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot));
+		ExecAsyncResponse(estate, areq);
+
+		return;
+	}
+
+	/* No result available now, make this node pending */
+	estate->es_num_pending_async++;
 }
 
 /*
@@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 		{
 			PendingAsyncRequest *areq = estate->es_pending_async[i];
 
-			/* Skip it if no callback is pending. */
-			if (!areq->callback_pending)
-				continue;
-
-			/*
-			 * Mark it as no longer needing a callback.  We must do this
-			 * before dispatching the callback in case the callback resets
-			 * the flag.
-			 */
-			areq->callback_pending = false;
-			estate->es_async_callback_pending--;
-
-			/* Perform the actual callback; set request_done if appropraite. */
-			if (!areq->request_complete)
+			/* Skip it if not pending. */
+			if (areq->state == ASYNC_CALLBACK_PENDING)
+			{
+				/*
+				 * Mark it as no longer needing a callback.  We must do this
+				 * before dispatching the callback in case the callback resets
+				 * the flag.
+				 */
+				estate->es_async_callback_pending--;
 				ExecAsyncNotify(estate, areq);
-			else
+			}
+
+			if (areq->state == ASYNC_COMPLETE)
 			{
 				any_node_done = true;
 				if (requestor == areq->requestor)
@@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
 				PendingAsyncRequest *head;
 				PendingAsyncRequest *tail = estate->es_pending_async[tidx];
 
-				if (!tail->callback_pending && tail->request_complete)
+				if (tail->state == ASYNC_COMPLETE)
 					continue;
 				head = estate->es_pending_async[hidx];
 				estate->es_pending_async[tidx] = head;
@@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout)
  * means wait forever, 0 means don't wait at all, and >0 means wait for the
  * indicated number of milliseconds.
  *
- * Returns true if we found some events and false if we timed out.
+ * Returns true if we found some events and false if we timed out or there's
+ * no event to wait. The latter is occur when the areq is processed during
  */
 static bool
 ExecAsyncEventWait(EState *estate, long timeout)
@@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout)
 	int		n;
 	bool	reinit = false;
 	bool	process_latch_set = false;
+	bool	added = false;
 
 	if (estate->es_wait_event_set == NULL)
 	{
@@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		PendingAsyncRequest *areq = estate->es_pending_async[i];
 
 		if (areq->num_fd_events > 0)
-			ExecAsyncConfigureWait(estate, areq, reinit);
+			added |= ExecAsyncConfigureWait(estate, areq, reinit);
 	}
 
+	Assert(added);
+
 	/* Wait for at least one event to occur. */
 	noccurred = WaitEventSetWait(estate->es_wait_event_set, timeout,
 								 occurred_event, EVENT_BUFFER_SIZE,
 								 WAIT_EVENT_ASYNC_WAIT);
+
 	if (noccurred == 0)
 		return false;
 
@@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout)
 		{
 			PendingAsyncRequest *areq = w->user_data;
 
-			if (!areq->callback_pending)
-			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
-				estate->es_async_callback_pending++;
-			}
+			Assert(areq->state == ASYNC_WAITING);
+
+			areq->state = ASYNC_CALLBACK_PENDING;
+			estate->es_async_callback_pending++;
 		}
 	}
 
@@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout)
 
 			if (areq->wants_process_latch)
 			{
-				Assert(!areq->request_complete);
-				areq->callback_pending = true;
+				Assert(areq->state == ASYNC_WAITING);
+				areq->state = ASYNC_CALLBACK_PENDING;
 			}
 		}
 	}
@@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout)
  * The events should include only WL_SOCKET_READABLE or WL_SOCKET_WRITEABLE,
  * and the number of calls should not exceed areq->num_fd_events (as
  * prevously set via ExecAsyncSetRequiredEvents).
+ *
+ * Individual requests can omit registering an event but it is a
+ * responsibility of the node driver to set at least one event per one
+ * requestor.
  */
-static void
+static bool
 ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq,
 					   bool reinit)
 {
 	switch (nodeTag(areq->requestee))
 	{
 		case T_ForeignScanState:
-			ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
+			return ExecAsyncForeignScanConfigureWait(estate, areq, reinit);
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq,
 	estate->es_total_fd_events += num_fd_events - areq->num_fd_events;
 	areq->num_fd_events = num_fd_events;
 	areq->wants_process_latch = wants_process_latch;
+	areq->state = ASYNC_WAITING;
 
 	if (force_reset && estate->es_wait_event_set != NULL)
 	{
@@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result)
 	 * need a callback to remove registered wait events.  It's not clear
 	 * that we would come out ahead, so use brute force for now.
 	 */
+	Assert(areq->state == ASYNC_IDLE || areq->state == ASYNC_CALLBACK_PENDING);
+
 	if (areq->num_fd_events > 0 || areq->wants_process_latch)
 		ExecAsyncSetRequiredEvents(estate, areq, 0, false, true);
 
 	/* Save result and mark request as complete. */
 	areq->result = result;
-	areq->request_complete = true;
-
-	/* Make sure this request is flagged for a callback. */
-	if (!areq->callback_pending)
-	{
-		areq->callback_pending = true;
-		estate->es_async_callback_pending++;
-	}
+	areq->state = ASYNC_COMPLETE;
 }
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index e61218a..568fa25 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -229,9 +229,15 @@ ExecAppend(AppendState *node)
 		 */
 		while ((i = bms_first_member(node->as_needrequest)) >= 0)
 		{
-			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
 			node->as_nasyncpending++;
+
+			ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]);
+			/* If this request immediately gives a result, take it. */
+			if (node->as_nasyncresult > 0)
+				return node->as_asyncresult[--node->as_nasyncresult];
 		}
+		if (node->as_nasyncpending == 0 && node->as_syncdone)
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 	}
 
 	for (;;)
@@ -246,32 +252,32 @@ ExecAppend(AppendState *node)
 		{
 			long	timeout = node->as_syncdone ? -1 : 0;
 
-			for (;;)
+			while (node->as_nasyncpending > 0)
 			{
-				if (node->as_nasyncpending == 0)
-				{
-					/*
-					 * If there is no asynchronous activity still pending
-					 * and the synchronous activity is also complete, we're
-					 * totally done scanning this node.  Otherwise, we're
-					 * done with the asynchronous stuff but must continue
-					 * scanning the synchronous children.
-					 */
-					if (node->as_syncdone)
-						return ExecClearTuple(node->ps.ps_ResultTupleSlot);
-					break;
-				}
-				if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout))
-				{
-					/* Timeout reached. */
-					break;
-				}
-				if (node->as_nasyncresult > 0)
+				if (ExecAsyncEventLoop(node->ps.state, &node->ps, timeout) &&
+					node->as_nasyncresult > 0)
 				{
 					/* Asynchronous subplan returned a tuple! */
 					--node->as_nasyncresult;
 					return node->as_asyncresult[node->as_nasyncresult];
 				}
+
+				/* Timeout reached. Go through to sync nodes if exists */
+				if (!node->as_syncdone)
+					break;
+			}
+
+			/*
+			 * If there is no asynchronous activity still pending and the
+			 * synchronous activity is also complete, we're totally done
+			 * scanning this node.  Otherwise, we're done with the
+			 * asynchronous stuff but must continue scanning the synchronous
+			 * children.
+			 */
+			if (node->as_syncdone)
+			{
+				Assert(node->as_nasyncpending == 0);
+				return ExecClearTuple(node->ps.ps_ResultTupleSlot);
 			}
 		}
 
@@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq)
 	TupleTableSlot *slot;
 
 	/* We shouldn't be called until the request is complete. */
-	Assert(areq->request_complete);
+	Assert(areq->state == ASYNC_COMPLETE);
 
 	/* Our result slot shouldn't already be occupied. */
 	Assert(TupIsNull(node->ps.ps_ResultTupleSlot));
diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c
index 61899d1..85dad79 100644
--- a/src/backend/executor/nodeForeignscan.c
+++ b/src/backend/executor/nodeForeignscan.c
@@ -376,7 +376,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq)
  *		In async mode, configure for a wait
  * ----------------------------------------------------------------
  */
-void
+bool
 ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit)
 {
@@ -384,7 +384,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate,
 	FdwRoutine *fdwroutine = node->fdwroutine;
 
 	Assert(fdwroutine->ForeignAsyncConfigureWait != NULL);
-	fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
+	return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit);
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index a8cabdf..c62aaf2 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -237,6 +237,7 @@ _copyAppend(const Append *from)
 	 */
 	COPY_NODE_FIELD(appendplans);
 	COPY_SCALAR_FIELD(nasyncplans);
+	COPY_SCALAR_FIELD(referent);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index a894a9d..c2e34a8 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -370,6 +370,7 @@ _outAppend(StringInfo str, const Append *node)
 
 	WRITE_NODE_FIELD(appendplans);
 	WRITE_INT_FIELD(nasyncplans);
+	WRITE_INT_FIELD(referent);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 67439ec..9837eff 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1540,6 +1540,7 @@ _readAppend(void)
 
 	READ_NODE_FIELD(appendplans);
 	READ_INT_FIELD(nasyncplans);
+	READ_INT_FIELD(referent);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 2140094..0575541 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -194,7 +194,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual,
 			 Index scanrelid, int ctePlanId, int cteParam);
 static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual,
 				   Index scanrelid, int wtParam);
-static Append *make_append(List *asyncplans, int nasyncplans, List *tlist);
+static Append *make_append(List *asyncplans, int nasyncplans,
+						   int referent, List *tlist);
 static RecursiveUnion *make_recursive_union(List *tlist,
 					 Plan *lefttree,
 					 Plan *righttree,
@@ -966,6 +967,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	List	   *syncplans = NIL;
 	ListCell   *subpaths;
 	int			nasyncplans = 0;
+	bool		first = true;
+	bool		referent_is_sync = true;
 
 	/*
 	 * The subpaths list could be empty, if every child was proven empty by
@@ -991,7 +994,14 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		return plan;
 	}
 
-	/* Build the plan for each child */
+	/*
+	 * Build the plan for each child
+
+	 * The first child in an inheritance set is the representative in
+	 * explaining tlist entries (see set_deparse_planstate). We should keep
+	 * the first child in best_path->subpaths at the head of the subplan list
+	 * for the reason.
+	 */
 	foreach(subpaths, best_path->subpaths)
 	{
 		Path	   *subpath = (Path *) lfirst(subpaths);
@@ -1005,9 +1015,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 		{
 			asyncplans = lappend(asyncplans, subplan);
 			++nasyncplans;
+			if (first)
+				referent_is_sync = false;
 		}
 		else
 			syncplans = lappend(syncplans, subplan);
+
+		first = false;
 	}
 
 	/*
@@ -1017,7 +1031,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path)
 	 * parent-rel Vars it'll be asked to emit.
 	 */
 
-	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist);
+	plan = make_append(list_concat(asyncplans, syncplans), nasyncplans,
+					   referent_is_sync ? nasyncplans : 0, tlist);
 
 	copy_generic_path_info(&plan->plan, (Path *) best_path);
 
@@ -5019,7 +5034,7 @@ make_foreignscan(List *qptlist,
 }
 
 static Append *
-make_append(List *appendplans, int nasyncplans, List *tlist)
+make_append(List *appendplans, int nasyncplans,	int referent, List *tlist)
 {
 	Append	   *node = makeNode(Append);
 	Plan	   *plan = &node->plan;
@@ -5030,6 +5045,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist)
 	plan->righttree = NULL;
 	node->appendplans = appendplans;
 	node->nasyncplans = nasyncplans;
+	node->referent = referent;
 
 	return node;
 }
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index f355954..76dd07a 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -4242,7 +4242,11 @@ set_deparse_planstate(deparse_namespace *dpns, PlanState *ps)
 	 * lists containing references to non-target relations.
 	 */
 	if (IsA(ps, AppendState))
-		dpns->outer_planstate = ((AppendState *) ps)->appendplans[0];
+	{
+		int idx = ((Append*)(((AppendState *) ps)->ps.plan))->referent;
+		dpns->outer_planstate =
+			((AppendState *) ps)->appendplans[idx];
+	}
 	else if (IsA(ps, MergeAppendState))
 		dpns->outer_planstate = ((MergeAppendState *) ps)->mergeplans[0];
 	else if (IsA(ps, ModifyTableState))
diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h
index 5a61306..2d9a62b 100644
--- a/src/include/executor/nodeForeignscan.h
+++ b/src/include/executor/nodeForeignscan.h
@@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node,
 
 extern void ExecAsyncForeignScanRequest(EState *estate,
 	PendingAsyncRequest *areq);
-extern void ExecAsyncForeignScanConfigureWait(EState *estate,
+extern bool ExecAsyncForeignScanConfigureWait(EState *estate,
 	PendingAsyncRequest *areq, bool reinit);
 extern void ExecAsyncForeignScanNotify(EState *estate,
 	PendingAsyncRequest *areq);
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 4c50f1e..41fc76f 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root,
 typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path);
 typedef void (*ForeignAsyncRequest_function) (EState *estate,
 											PendingAsyncRequest *areq);
-typedef void (*ForeignAsyncConfigureWait_function) (EState *estate,
+typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate,
 											PendingAsyncRequest *areq,
 											bool reinit);
 typedef void (*ForeignAsyncNotify_function) (EState *estate,
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 81e997e..5afcd34 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -361,6 +361,13 @@ typedef struct ResultRelInfo
  * State for an asynchronous tuple request.
  * ----------------
  */
+typedef enum AsyncRequestState
+{
+	ASYNC_IDLE,
+	ASYNC_WAITING,
+	ASYNC_CALLBACK_PENDING,
+	ASYNC_COMPLETE
+} AsyncRequestState;
 typedef struct PendingAsyncRequest
 {
 	int			myindex;			/* Index in es_pending_async. */
@@ -369,8 +376,7 @@ typedef struct PendingAsyncRequest
 	int			request_index;	/* Scratch space for requestor. */
 	int			num_fd_events;	/* Max number of FD events requestee needs. */
 	bool		wants_process_latch;	/* Requestee cares about MyLatch. */
-	bool		callback_pending;			/* Callback is needed. */
-	bool		request_complete;		/* Request complete, result valid. */
+	AsyncRequestState state;
 	Node	   *result;			/* Result (NULL if no more tuples). */
 } PendingAsyncRequest;
 
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index f0daada..ebbc78d 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -229,6 +229,7 @@ typedef struct Append
 	Plan		plan;
 	List	   *appendplans;
 	int			nasyncplans;	/* # of async plans, always at start of list */
+	int			referent; 		/* index of inheritance tree referent */
 } Append;
 
 /* ----------------
-- 
2.9.2


----Next_Part(Wed_Feb_22_17_39_45_2017_090)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline; filename="0002-Fix-some-bugs.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH v51 02/10] Rename cluster.c/h -> repack.c/h
@ 2026-03-31 16:55 Álvaro Herrera <alvherre@kurilemu.de>
  0 siblings, 0 replies; 6+ messages in thread

From: Álvaro Herrera @ 2026-03-31 16:55 UTC (permalink / raw)

---
 src/backend/commands/Makefile                |  2 +-
 src/backend/commands/matview.c               |  2 +-
 src/backend/commands/meson.build             |  2 +-
 src/backend/commands/{cluster.c => repack.c} |  6 +++---
 src/backend/commands/tablecmds.c             |  2 +-
 src/backend/commands/vacuum.c                |  6 +++---
 src/backend/storage/ipc/procsignal.c         |  1 +
 src/backend/tcop/postgres.c                  |  1 +
 src/backend/tcop/utility.c                   |  2 +-
 src/include/commands/{cluster.h => repack.h} | 12 ++++++------
 10 files changed, 19 insertions(+), 17 deletions(-)
 rename src/backend/commands/{cluster.c => repack.c} (99%)
 rename src/include/commands/{cluster.h => repack.h} (90%)

diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index c10fdba2bbb..fe1bba3a9b9 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -18,7 +18,6 @@ OBJS = \
 	amcmds.o \
 	analyze.o \
 	async.o \
-	cluster.o \
 	collationcmds.o \
 	comment.o \
 	constraint.o \
@@ -51,6 +50,7 @@ OBJS = \
 	proclang.o \
 	propgraphcmds.o \
 	publicationcmds.o \
+	repack.o \
 	schemacmds.o \
 	seclabel.o \
 	sequence.o \
diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c
index d3be8939011..5db4fe75dce 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -24,8 +24,8 @@
 #include "catalog/namespace.h"
 #include "catalog/pg_am.h"
 #include "catalog/pg_opclass.h"
-#include "commands/cluster.h"
 #include "commands/matview.h"
+#include "commands/repack.h"
 #include "commands/tablecmds.h"
 #include "commands/tablespace.h"
 #include "executor/executor.h"
diff --git a/src/backend/commands/meson.build b/src/backend/commands/meson.build
index 90c7e37a429..f624aae74af 100644
--- a/src/backend/commands/meson.build
+++ b/src/backend/commands/meson.build
@@ -6,7 +6,6 @@ backend_sources += files(
   'amcmds.c',
   'analyze.c',
   'async.c',
-  'cluster.c',
   'collationcmds.c',
   'comment.c',
   'constraint.c',
@@ -39,6 +38,7 @@ backend_sources += files(
   'proclang.c',
   'propgraphcmds.c',
   'publicationcmds.c',
+  'repack.c',
   'schemacmds.c',
   'seclabel.c',
   'sequence.c',
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/repack.c
similarity index 99%
rename from src/backend/commands/cluster.c
rename to src/backend/commands/repack.c
index f241e18b153..20f0a572236 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/repack.c
@@ -1,6 +1,6 @@
 /*-------------------------------------------------------------------------
  *
- * cluster.c
+ * repack.c
  *    REPACK a table; formerly known as CLUSTER.  VACUUM FULL also uses
  *    parts of this code.
  *
@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  src/backend/commands/cluster.c
+ *	  src/backend/commands/repack.c
  *
  *-------------------------------------------------------------------------
  */
@@ -33,9 +33,9 @@
 #include "catalog/pg_am.h"
 #include "catalog/pg_inherits.h"
 #include "catalog/toasting.h"
-#include "commands/cluster.h"
 #include "commands/defrem.h"
 #include "commands/progress.h"
+#include "commands/repack.h"
 #include "commands/tablecmds.h"
 #include "commands/vacuum.h"
 #include "miscadmin.h"
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 0ce2e81f9c2..e2882a50b3b 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -57,10 +57,10 @@
 #include "catalog/storage.h"
 #include "catalog/storage_xlog.h"
 #include "catalog/toasting.h"
-#include "commands/cluster.h"
 #include "commands/comment.h"
 #include "commands/defrem.h"
 #include "commands/event_trigger.h"
+#include "commands/repack.h"
 #include "commands/sequence.h"
 #include "commands/tablecmds.h"
 #include "commands/tablespace.h"
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 0ed363d1c85..b179b62b5c8 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -9,7 +9,7 @@
  *
  * VACUUM for heap AM is implemented in vacuumlazy.c, parallel vacuum in
  * vacuumparallel.c, ANALYZE in analyze.c, and VACUUM FULL is a variant of
- * CLUSTER, handled in cluster.c.
+ * REPACK, handled in repack.c.
  *
  *
  * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
@@ -38,9 +38,9 @@
 #include "catalog/pg_database.h"
 #include "catalog/pg_inherits.h"
 #include "commands/async.h"
-#include "commands/cluster.h"
 #include "commands/defrem.h"
 #include "commands/progress.h"
+#include "commands/repack.h"
 #include "commands/vacuum.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
@@ -2293,7 +2293,7 @@ vacuum_rel(Oid relid, RangeVar *relation, VacuumParams params,
 			if ((params.options & VACOPT_VERBOSE) != 0)
 				cluster_params.options |= CLUOPT_VERBOSE;
 
-			/* VACUUM FULL is a variant of REPACK; see cluster.c */
+			/* VACUUM FULL is a variant of REPACK; see repack.c */
 			cluster_rel(REPACK_COMMAND_VACUUMFULL, rel, InvalidOid,
 						&cluster_params);
 			/* cluster_rel closes the relation, but keeps lock */
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 7e017c8d53b..7cef6e43661 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -19,6 +19,7 @@
 
 #include "access/parallel.h"
 #include "commands/async.h"
+#include "commands/repack.h"
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "port/pg_bitutils.h"
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 10be60011ad..9fbaa5c00f0 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -39,6 +39,7 @@
 #include "commands/event_trigger.h"
 #include "commands/explain_state.h"
 #include "commands/prepare.h"
+#include "commands/repack.h"
 #include "common/pg_prng.h"
 #include "jit/jit.h"
 #include "libpq/libpq.h"
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 2b609bfc824..5f8c766c4be 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -26,7 +26,6 @@
 #include "catalog/toasting.h"
 #include "commands/alter.h"
 #include "commands/async.h"
-#include "commands/cluster.h"
 #include "commands/collationcmds.h"
 #include "commands/comment.h"
 #include "commands/conversioncmds.h"
@@ -46,6 +45,7 @@
 #include "commands/proclang.h"
 #include "commands/propgraphcmds.h"
 #include "commands/publicationcmds.h"
+#include "commands/repack.h"
 #include "commands/schemacmds.h"
 #include "commands/seclabel.h"
 #include "commands/sequence.h"
diff --git a/src/include/commands/cluster.h b/src/include/commands/repack.h
similarity index 90%
rename from src/include/commands/cluster.h
rename to src/include/commands/repack.h
index d6b62c747e8..85061158b0c 100644
--- a/src/include/commands/cluster.h
+++ b/src/include/commands/repack.h
@@ -1,17 +1,17 @@
 /*-------------------------------------------------------------------------
  *
- * cluster.h
- *	  header file for postgres cluster command stuff
+ * repack.h
+ *	  header file for the REPACK command
  *
  * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994-5, Regents of the University of California
  *
- * src/include/commands/cluster.h
+ * src/include/commands/repack.h
  *
  *-------------------------------------------------------------------------
  */
-#ifndef CLUSTER_H
-#define CLUSTER_H
+#ifndef REPACK_H
+#define REPACK_H
 
 #include "nodes/parsenodes.h"
 #include "parser/parse_node.h"
@@ -52,4 +52,4 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap,
 							 MultiXactId cutoffMulti,
 							 char newrelpersistence);
 
-#endif							/* CLUSTER_H */
+#endif							/* REPACK_H */
-- 
2.47.3


--r2slln3zpmilwu22
Content-Type: text/x-diff; charset=utf-8
Content-Disposition: attachment;
	filename="v51-0003-Add-CONCURRENTLY-option-to-REPACK-command.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

end of thread, other threads:[~2026-03-31 16:55 UTC | newest]

Thread overview: 6+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2016-10-17 06:54 [PATCH 3/7] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2016-10-17 06:54 [PATCH 3/4] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2016-10-17 06:54 [PATCH 3/6] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2016-10-17 06:54 [PATCH 3/7] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2016-10-17 06:54 [PATCH 03/13] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2026-03-31 16:55 [PATCH v51 02/10] Rename cluster.c/h -> repack.c/h Álvaro Herrera <alvherre@kurilemu.de>

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox