From cec48686a0d781806a262c548adb7d0fb06868fc Mon Sep 17 00:00:00 2001 From: Etsuro Fujita Date: Fri, 7 Aug 2026 17:30:06 +0900 Subject: [PATCH] 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 Co-authored-by: Alexander Korotkov Co-authored-by: Gleb Kashkin Co-authored-by: Etsuro Fujita Reviewed-by: Alexander Pyhalov Reviewed-by: Gleb Kashkin Discussion: https://postgr.es/m/CAPpHfduMOTnV5Zj2KGJ7zanL_10QvccZHtPUaDfJvBhsh9axnQ%40mail.gmail.com Backpatch-through: 14 --- .../postgres_fdw/expected/postgres_fdw.out | 4 +- contrib/postgres_fdw/postgres_fdw.c | 13 +-- contrib/postgres_fdw/sql/postgres_fdw.sql | 4 +- src/backend/executor/nodeAppend.c | 97 +++++++++++++++---- 4 files changed, 86 insertions(+), 32 deletions(-) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 4b82c214aea..66c74f55a65 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/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) diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 60723eb520e..860c2e9824c 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/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 diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 9c7e2a326fe..af286d9c33d 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/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) diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index e48a547e237..57df488d57e 100644 --- a/src/backend/executor/nodeAppend.c +++ b/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 *