From d40707f217d672db828bdb898176bccb77818a1d Mon Sep 17 00:00:00 2001 From: Alexander Korotkov 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 Reported-by: Alexander Korotkov 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)