Drain pending asynchronous requests during ExecReScanAppend.

The logic for asynchronous Append assumes that pending requests made for
subplans of an Append are drained during ExecReScanAppend.  To ensure
that, commit 9e283fc85 modified postgresReScanForeignScan to drain such
a request if any, but failed to take into account that if such a request
was made for a subplan that is re-scanned with parameter changes or
pruned in the next round by runtime pruning, the postgres_fdw callback
function is called after ExecReScanAppend or never called, respectively.
This would cause such a request to remain even after ExecReScanAppend,
leading to incorrect results, an infinite loop, or an assertion failure.

To fix, modify ExecReScanAppend to, for each of the pending requests,
give the FDW a chance to drain that request using the existing
ForeignAsyncConfigureWait/ForeignAsyncNotify callback functions.  This
makes the change made to postgresReScanForeignScan useless, so remove it
as well.

Back-patch to v14 where asynchronous Append was added.

Reported-by: Alexander Korotkov <aekorotkov@gmail.com>
Co-authored-by: Alexander Korotkov <aekorotkov@gmail.com>
Co-authored-by: Gleb Kashkin <g.kashkin@postgrespro.ru>
Co-authored-by: Etsuro Fujita <etsuro.fujita@gmail.com>
Reviewed-by: Alexander Pyhalov <a.pyhalov@postgrespro.ru>
Reviewed-by: Gleb Kashkin <g.kashkin@postgrespro.ru>
Discussion: https://postgr.es/m/CAPpHfduMOTnV5Zj2KGJ7zanL_10QvccZHtPUaDfJvBhsh9axnQ%40mail.gmail.com
Backpatch-through: 14

contrib/postgres_fdw/expected/postgres_fdw.out

@@ -11231,8 +11231,8 @@ DROP TABLE base_tbl1;
 DROP TABLE base_tbl2;
 DROP TABLE result_tbl;
 DROP TABLE join_tbl;
--- Test that an asynchronous fetch is processed before restarting the scan in
--- ReScanForeignScan
+-- Test ExecAppendAsyncReset code path that drains outstanding async requests
+-- (case where subplans are re-scanned without parameter changes)
 CREATE TABLE base_tbl (a int, b int);
 INSERT INTO base_tbl VALUES (1, 11), (2, 22), (3, 33);
 CREATE FOREIGN TABLE foreign_tbl (b int)

contrib/postgres_fdw/postgres_fdw.c

@@ -1651,16 +1651,11 @@ postgresReScanForeignScan(ForeignScanState *node)
 		return;
 
 	/*
-	 * If the node is async-capable, and an asynchronous fetch for it has been
-	 * begun, the asynchronous fetch might not have yet completed.  Check if
-	 * the node is async-capable, and an asynchronous fetch for it is still in
-	 * progress; if so, complete the asynchronous fetch before restarting the
-	 * scan.
+	 * If the node is async-capable, any asynchronous fetch made for it should
+	 * have been processed before we get here (see ExecAppendAsyncReset()).
 	 */
-	if (fsstate->async_capable &&
-		fsstate->conn_state->pendingAreq &&
-		fsstate->conn_state->pendingAreq->requestee == (PlanState *) node)
-		fetch_more_data(node);
+	Assert(!fsstate->async_capable || !fsstate->conn_state->pendingAreq ||
+		   fsstate->conn_state->pendingAreq->requestee != (PlanState *) node);
 
 	/*
 	 * If any internal parameters affecting this node have changed, we'd

contrib/postgres_fdw/sql/postgres_fdw.sql

@@ -3664,8 +3664,8 @@ DROP TABLE base_tbl2;
 DROP TABLE result_tbl;
 DROP TABLE join_tbl;
 
--- Test that an asynchronous fetch is processed before restarting the scan in
--- ReScanForeignScan
+-- Test ExecAppendAsyncReset code path that drains outstanding async requests
+-- (case where subplans are re-scanned without parameter changes)
 CREATE TABLE base_tbl (a int, b int);
 INSERT INTO base_tbl VALUES (1, 11), (2, 22), (3, 33);
 CREATE FOREIGN TABLE foreign_tbl (b int)

src/backend/executor/nodeAppend.c

@@ -92,6 +92,7 @@ static void ExecAppendAsyncBegin(AppendState *node);
 static bool ExecAppendAsyncGetNext(AppendState *node, TupleTableSlot **result);
 static bool ExecAppendAsyncRequest(AppendState *node, TupleTableSlot **result);
 static void ExecAppendAsyncEventWait(AppendState *node);
+static void ExecAppendAsyncReset(AppendState *node);
 static void classify_matching_subplans(AppendState *node);
 
 /* ----------------------------------------------------------------
@@ -417,6 +418,10 @@ ExecReScanAppend(AppendState *node)
 	int			nasyncplans = node->as_nasyncplans;
 	int			i;
 
+	/* If there are any async subplans, reset async requests made for them. */
+	if (nasyncplans > 0)
+		ExecAppendAsyncReset(node);
+
 	/*
 	 * If any PARAM_EXEC Params used in pruning expressions have changed, then
 	 * we'd better unset the valid subplans so that they are reselected for
@@ -454,25 +459,6 @@ ExecReScanAppend(AppendState *node)
 			ExecReScan(subnode);
 	}
 
-	/* Reset async state */
-	if (nasyncplans > 0)
-	{
-		i = -1;
-		while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
-		{
-			AsyncRequest *areq = node->as_asyncrequests[i];
-
-			areq->callback_pending = false;
-			areq->request_complete = false;
-			areq->result = NULL;
-		}
-
-		node->as_nasyncresults = 0;
-		node->as_nasyncremain = 0;
-		bms_free(node->as_needrequest);
-		node->as_needrequest = NULL;
-	}
-
 	/* Let choose_next_subplan_* function handle setting the first subplan */
 	node->as_whichplan = INVALID_SUBPLAN_INDEX;
 	node->as_syncdone = false;
@@ -1127,6 +1113,79 @@ ExecAppendAsyncEventWait(AppendState *node)
 	}
 }
 
+/* ----------------------------------------------------------------
+ *		ExecAppendAsyncReset
+ *
+ *		Reset asynchronous requests made for async-capable subplans.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecAppendAsyncReset(AppendState *node)
+{
+	int			i;
+
+	/* We should never be called when there are no async subplans. */
+	Assert(node->as_nasyncplans > 0);
+
+	/*
+	 * Drain pending async requests if any.  We force the as_syncdone flag to
+	 * be true so that ExecAppendAsyncEventWait() waits until at least one
+	 * event occurs.
+	 */
+	node->as_syncdone = true;
+	for (;;)
+	{
+		bool		found = false;
+
+		/*
+		 * When called from ExecAppendAsyncEventWait(), postgres_fdw (and
+		 * possibly other FDWs) will skip configuration of events for pending
+		 * requests in some cases if as_needrequest isn't empty.  To avoid
+		 * that, discard results we already have.  Note that we need to do
+		 * this on every iteration, as the call to that function may produce
+		 * new results.
+		 */
+		node->as_nasyncresults = 0;
+		bms_free(node->as_needrequest);
+		node->as_needrequest = NULL;
+
+		i = -1;
+		while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
+		{
+			AsyncRequest *areq = node->as_asyncrequests[i];
+
+			if (areq->callback_pending)
+			{
+				found = true;
+				break;
+			}
+		}
+		if (!found)
+			break;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* Wait or poll for async events. */
+		ExecAppendAsyncEventWait(node);
+	}
+
+	/* Reset async requests. */
+	i = -1;
+	while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
+	{
+		AsyncRequest *areq = node->as_asyncrequests[i];
+
+		Assert(!areq->callback_pending);
+		areq->request_complete = false;
+		areq->result = NULL;
+	}
+
+	/* Reset state variables. */
+	Assert(node->as_nasyncresults == 0);
+	Assert(node->as_needrequest == NULL);
+	node->as_nasyncremain = 0;
+}
+
 /* ----------------------------------------------------------------
  *		ExecAsyncAppendResponse
  *