From d40707f217d672db828bdb898176bccb77818a1d Mon Sep 17 00:00:00 2001
From: Alexander Korotkov <akorotkov@postgresql.org>
Date: Fri, 31 Jul 2026 01:38:37 +0200
Subject: [PATCH v20 1/6] Fix corruption of async request state on Append
 rescan

ExecReScanAppend() unconditionally cleared callback_pending for every
async subplan, regardless of whether a request was genuinely still in
flight.  For a subplan whose remote fetch had been sent but not yet
consumed, this desynchronized our local bookkeeping from the
async-capable node's own view of the same fact -- e.g. postgres_fdw's
PgFdwConnState.pendingAreq, which tracks the outstanding request on a
possibly shared connection independently of our
AsyncRequest.callback_pending flag.

If that connection is later reused by another subplan (which happens
routinely under runtime partition pruning, once the subplan holding
the stale request is excluded from a round and a sibling sharing its
connection gets its own ReScan), postgres_fdw drains the
still-registered pendingAreq before sending a new command and asserts
that its callback_pending flag is set.  Since we had already cleared
it, this assertion fails; in a non-assert build the connection's state
is left inconsistent instead, which can expose stale rows.

Fix this by processing every still-pending async request during
ExecReScanAppend(), before we touch any of the derived state, in the
new ExecAppendAsyncProcessPending().  It waits for and delivers the
outstanding requests via the normal ExecAppendAsyncEventWait() path
until none remain callback_pending, so the async-capable node's own
per-connection state is settled and no request is left dangling.  This
must be done during the rescan rather than lazily when a request is
next reused, because a subplan pruned out of the coming scan would
never be reused and so would never get a chance to drain.

Author: Etsuro Fujita <etsuro.fujita@gmail.com>
Reported-by: Alexander Korotkov <aekorotkov@gmail.com>
Discussion: https://postgr.es/m/CAPpHfduMOTnV5Zj2KGJ7zanL_10QvccZHtPUaDfJvBhsh9axnQ%40mail.gmail.com
---
 .../postgres_fdw/expected/postgres_fdw.out    | 33 +++++++++++++++++
 contrib/postgres_fdw/sql/postgres_fdw.sql     |  5 +++
 src/backend/executor/nodeAppend.c             | 35 ++++++++++++++++++-
 3 files changed, 72 insertions(+), 1 deletion(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index d19121b05da..5a1740d21cf 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -11836,6 +11836,39 @@ SELECT * FROM result_tbl ORDER BY a;
 (3 rows)
 
 DELETE FROM result_tbl;
+-- Test ExecAppendAsyncProcessPending code path
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE a = 1505 OR a = o.x LIMIT 1) s ORDER BY o.x;
+                                                    QUERY PLAN                                                     
+-------------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: "*VALUES*".column1
+   Sort Key: "*VALUES*".column1
+   ->  Nested Loop
+         Output: "*VALUES*".column1
+         ->  Values Scan on "*VALUES*"
+               Output: "*VALUES*".column1
+         ->  Limit
+               Output: NULL::integer
+               ->  Append
+                     ->  Async Foreign Scan on public.async_p1 async_pt_1
+                           Output: NULL::integer
+                           Remote SQL: SELECT NULL FROM public.base_tbl1 WHERE (((a = 1505) OR (a = $1::integer)))
+                     ->  Async Foreign Scan on public.async_p2 async_pt_2
+                           Output: NULL::integer
+                           Remote SQL: SELECT NULL FROM public.base_tbl2 WHERE (((a = 1505) OR (a = $1::integer)))
+                     ->  Async Foreign Scan on public.async_p3 async_pt_3
+                           Output: NULL::integer
+                           Remote SQL: SELECT NULL FROM public.base_tbl3 WHERE (((a = 1505) OR (a = $1::integer)))
+(19 rows)
+
+SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE a = 1505 OR a = o.x LIMIT 1) s ORDER BY o.x;
+  x   
+------
+ 2505
+ 3505
+(2 rows)
+
 -- Test COPY TO when foreign table is partition
 COPY async_pt TO stdout; --error
 ERROR:  cannot copy from foreign table "async_p1"
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index e7019952173..113a1cd0185 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -4090,6 +4090,11 @@ INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505;
 SELECT * FROM result_tbl ORDER BY a;
 DELETE FROM result_tbl;
 
+-- Test ExecAppendAsyncProcessPending code path
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE a = 1505 OR a = o.x LIMIT 1) s ORDER BY o.x;
+SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE a = 1505 OR a = o.x LIMIT 1) s ORDER BY o.x;
+
 -- Test COPY TO when foreign table is partition
 COPY async_pt TO stdout; --error
 
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index 987358e27fa..d17833dc5f7 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -95,6 +95,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 ExecAppendAsyncProcessPending(AppendState *node);
 static void classify_matching_subplans(AppendState *node);
 
 /* ----------------------------------------------------------------
@@ -426,6 +427,9 @@ ExecReScanAppend(AppendState *node)
 	int			nasyncplans = node->as_nasyncplans;
 	int			i;
 
+	/* Process asynchronous requests still pending */
+	ExecAppendAsyncProcessPending(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
@@ -469,7 +473,7 @@ ExecReScanAppend(AppendState *node)
 		{
 			AsyncRequest *areq = node->as_asyncrequests[i];
 
-			areq->callback_pending = false;
+			Assert(!areq->callback_pending);
 			areq->request_complete = false;
 			areq->result = NULL;
 		}
@@ -1135,6 +1139,35 @@ ExecAppendAsyncEventWait(AppendState *node)
 	}
 }
 
+/* ----------------------------------------------------------------
+ *		ExecAppendAsyncProcessPending
+ *
+ *		Process all of the asynchronous requests still pending.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecAppendAsyncProcessPending(AppendState *node)
+{
+	for (;;)
+	{
+		bool		found = false;
+		int			i;
+
+		i = -1;
+		while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
+		{
+			AsyncRequest *areq = node->as_asyncrequests[i];
+
+			if (areq->callback_pending)
+				found = true;
+		}
+		if (!found)
+			return;
+
+		ExecAppendAsyncEventWait(node);
+	}
+}
+
 /* ----------------------------------------------------------------
  *		ExecAsyncAppendResponse
  *
-- 
2.50.1 (Apple Git-155)

