diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index e90289e4ab1..eebdf1c4ad8 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -11661,6 +11661,22 @@ SELECT * FROM result_tbl ORDER BY a; (3 rows) DELETE FROM result_tbl; +-- Check that a pending async request on a shared connection isn't corrupted +-- when the Append is rescanned with the subplan holding it pruned out. Here +-- async_p2 and async_p3 share a connection, and per outer row exactly one of +-- them is pruned while async_p1 (a different connection) always matches; the +-- inner LIMIT leaves the other shared-connection request outstanding across +-- the rescan, and reusing that connection for the next round must drain it +-- rather than trip over stale state. +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 dfc58beb0d2..c18b54f1756 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3997,6 +3997,38 @@ INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505; SELECT * FROM result_tbl ORDER BY a; DELETE FROM result_tbl; +-- Check that a pending async request on a shared connection isn't corrupted +-- when the Append is rescanned with the subplan holding it pruned out. Here +-- async_p2 and async_p3 share a connection, and per outer row exactly one of +-- them is pruned while async_p1 (a different connection) always matches; the +-- inner LIMIT leaves the other shared-connection request outstanding across +-- the rescan, and reusing that connection for the next round must drain it +-- rather than trip over stale state. +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; + +-- Expose stale rows from a pending async request that was valid for a +-- previous rescan but is pruned out for the current one. +CREATE VIEW base_tbl2_slow AS + SELECT t.a, t.b, t.c + FROM base_tbl2 t, LATERAL pg_sleep(0.2); + +ALTER FOREIGN TABLE async_p2 OPTIONS (SET table_name 'base_tbl2_slow'); + +SELECT o.x, s.a +FROM (VALUES (2505), (3505)) o(x), + LATERAL ( + SELECT a + FROM async_pt + WHERE a = o.x OR (o.x = 2505 AND a = 1505) + LIMIT 1 + ) s +ORDER BY o.x; + +ALTER FOREIGN TABLE async_p2 OPTIONS (SET table_name 'base_tbl2'); +DROP VIEW base_tbl2_slow; + -- 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..6a5a14cd576 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -469,7 +469,21 @@ ExecReScanAppend(AppendState *node) { AsyncRequest *areq = node->as_asyncrequests[i]; - areq->callback_pending = false; + /* + * Leave a request that is still marked as pending a callback + * alone: it may genuinely still be in flight, or it may have an + * unconsumed result already sitting on a connection shared with + * another subplan (as can happen with postgres_fdw). Blindly + * clearing callback_pending here would desync our bookkeeping + * from the async-capable node's own, which can lead it to + * mishandle that connection later (e.g. postgres_fdw asserts that + * a request it still considers in-process has callback_pending + * set). Such a request is instead drained lazily, right before + * it would be reused, in ExecAppendAsyncBegin(). + */ + if (areq->callback_pending) + continue; + areq->request_complete = false; areq->result = NULL; } @@ -915,7 +929,25 @@ ExecAppendAsyncBegin(AppendState *node) AsyncRequest *areq = node->as_asyncrequests[i]; Assert(areq->request_index == i); - Assert(!areq->callback_pending); + + /* + * This request may still be marked as pending a callback, if + * ExecReScanAppend() left it alone because it might have been + * genuinely in flight (or had an unconsumed result waiting on a + * connection shared with another subplan). Drain it now, before + * reusing it: ExecReScan() lets the async-capable node settle any + * such outstanding state (e.g. postgres_fdw's + * postgresReScanForeignScan() will wait for an in-progress request on + * its own connection and consume its result), after which it's safe + * to reset our own bookkeeping and issue a fresh request. + */ + if (areq->callback_pending) + { + ExecReScan(node->appendplans[i]); + areq->callback_pending = false; + areq->request_complete = false; + areq->result = NULL; + } /* Do the actual work. */ ExecAsyncRequest(areq);