diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index d19121b05da..c6b1871d483 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -11836,6 +11836,72 @@ SELECT * FROM result_tbl ORDER BY a; (3 rows) DELETE FROM result_tbl; +-- Test ExecAppendAsyncReset code path that drains outstanding async requests +-- (case where subplans are re-scanned with parameter changes) +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) + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE (a = 1505 AND o.x = 2505) 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) AND ($1::integer = 2505)) 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) AND ($1::integer = 2505)) 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) AND ($1::integer = 2505)) OR (a = $1::integer))) +(19 rows) + +SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE (a = 1505 AND o.x = 2505) 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" @@ -12621,8 +12687,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 8b660a6c02c..b9739610131 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -1751,16 +1751,11 @@ postgresReScanForeignScan(ForeignScanState *node) return; /* - * If the node is async-capable, and an asynchronous fetch for it has - * 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 (fsstate->async_capable && - fsstate->conn_state->pendingAreq && - fsstate->conn_state->pendingAreq->requestee == (PlanState *) node) - fetch_more_data(node); + * If the node is async-capable, any asynchronous fetch made for it should + * have been processed before we get here (see ExecAppendAsyncReset()). + */ + 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 e7019952173..d429d0a16c4 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -4090,6 +4090,16 @@ INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505; SELECT * FROM result_tbl ORDER BY a; DELETE FROM result_tbl; +-- Test ExecAppendAsyncReset code path that drains outstanding async requests +-- (case where subplans are re-scanned with parameter changes) +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; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE (a = 1505 AND o.x = 2505) 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 AND o.x = 2505) 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 @@ -4329,8 +4339,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 987358e27fa..c9b7870d315 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 ExecAppendAsyncReset(AppendState *node); static void classify_matching_subplans(AppendState *node); /* ---------------------------------------------------------------- @@ -426,6 +427,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 @@ -461,25 +466,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; @@ -1135,6 +1121,78 @@ 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; + + /* + * postgres_fdw (and possibly other FDWs) will skip configuration of + * events 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, since the call to ExecAppendAsyncEventWait() 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 *