On Fri Oct 16, 2026 at 9:10 AM -03, Etsuro Fujita wrote: >> 2: stale ForeignScanHash entries after a failed portal >> >> An entry is removed only by postgresEndForeignScan or when the >> subtransaction that created the scan aborts (or at transaction end). But >> if an error happens while the scan of an outer-level cursor is fetched >> inside a savepoint, the portal is marked failed and the executor doesn't >> call EndForeignScan, so the entry is left pointing to memory that is >> freed later. See it: > > I couldn't reproduce this in my environment. Could you provide the > reproducer in more detail including the foreign table definition, the > postgres_fdw options, ...? > create extension postgres_fdw; create server loopback foreign data wrapper postgres_fdw options (dbname 'postgres', port '5432'); create user mapping for current_user server loopback; create table t (a int); insert into t values (1), (2), (3), (4); create foreign table ft (a int) server loopback options (table_name 't');
begin; declare c cursor for select 1 / (a - 3) from ft order by a; fetch 1 from c; savepoint s2; fetch 5 from c; rollback to s2; commit; After the COMMIT command it produces: WARNING: 01000: ForeignScan node 0x106019368 still existing LOCATION: AtEOXact_ForeignScan, postgres_fdw.c:4381 The details that matter are that the first "fetch 1" happens before the savepoint, so IIUC the scan and its cursor belong to the outer level, and that the error happens while fetching inside the savepoint. The expression 1 / (a - 3) is not pushed down, so the division by zero is raised locally while the scan returns rows, and the portal is left in the failed state. If I add this after "rollback to s2" and before "commit": close c; savepoint s3; select 1; release s3; the backend crashes in the subtransaction callback. > Thanks for working on the fix! It would be great if you share it. > I'm attaching the v4 patch. It is your v3-0001 plus the fixes from my previous email that you agree. Since you said you'd remove the restrictions of 3 and 4, and add the comments of 6, I went ahead and did them in this version. A few notes: - for 1, the error no longer uses the relation name when fsstate->rel is NULL. - for 2, every scan registers a memory context reset callback on its executor memory context, which removes its hash entry when that memory is freed, even if EndForeignScan is never called. I also removed the WARNING about leftover scans at commit, since a failed portal can legitimately leave one behind at that point. Please let me know if you disagree with that. - for 3 and 4, scan_allowed is replaced by a cursor_lost flag: when a savepoint rollback removes the remote cursor, the scan can still return the rows that are already in its buffer, and the error is only raised when a new remote fetch is needed. A rescan of such a scan either reuses what it has in memory or forgets the cursor so that the next fetch creates a new one. - for 5, I kept ERRCODE_FEATURE_NOT_SUPPORTED as you noted, and only improved the message. - for 6, I added the comments, and while doing it I dropped fs_level, since with the reset callback it wasn't needed anymore: the hash only holds the scan pointer and all the level tracking is in cursor_level. - I added tests for the pushed-down join, the failed portal case, a small fetch_size where more rows can't be fetched, the rescan of a scan with a lost cursor, sibling and nested savepoints, Nikolay's initplan case, a PL/pgSQL refcursor and an async scan. -- Matheus Alcantara EDB: https://www.enterprisedb.com
From f869b19b9f451f8e56dc0e19e1773b1b2db59512 Mon Sep 17 00:00:00 2001 From: Matheus Alcantara <[email protected]> Date: Fri, 16 Oct 2026 11:04:15 -0300 Subject: [PATCH v4] postgres_fdw: Fix handling of open cursors across subtransaction rollback. postgres_fdw declares the remote cursor for a foreign scan at the first FETCH, inside the remote savepoint matching the local subtransaction that is current at that time. If that subtransaction was rolled back while the local cursor was still alive (e.g., a cursor opened before a savepoint and first fetched after it), the remote cursor was destroyed together with its remote savepoint, but the local scan kept using it. This resulted in an error such as 'cursor "c1" does not exist', which was reported at some later point, e.g., when the cursor was closed at commit. Fix by keeping track of the foreign scans that are open in the backend, and of the nesting level at which each remote cursor was declared, so that the subtransaction callbacks can find and update them: on subtransaction commit, hand the cursors declared in it over to the parent level, matching the remote RELEASE; on abort, mark them as lost. The rows a scan has already fetched remain usable after its cursor is lost. Fetching more rows from the remote server fails with a clear error instead. A rescan of such a scan either reuses the rows in memory or creates a new cursor, and EndForeignScan doesn't try to close a cursor that is already gone. The remote (sub)transaction is now also synchronized with the local one before a cursor is created and before each FETCH, MOVE and CLOSE, so that a cursor opened before a transaction mode change (e.g., SET TRANSACTION READ ONLY) picks up the new mode. Since the registry holds pointers into executor memory, make sure an entry never outlives its scan: besides removing it in EndForeignScan, a memory context callback on the scan's executor memory removes it too. This matters because the executor doesn't call EndForeignScan for a scan whose portal was marked failed, which can happen when an error occurs while an outer-level cursor is being fetched inside a savepoint. Also, don't assume that a scan has a single relation when reporting that no more rows can be fetched, since a scan for a pushed-down join or aggregate has none. --- contrib/postgres_fdw/connection.c | 70 +++- .../postgres_fdw/expected/postgres_fdw.out | 353 ++++++++++++++++++ contrib/postgres_fdw/postgres_fdw.c | 288 +++++++++++++- contrib/postgres_fdw/postgres_fdw.h | 4 + contrib/postgres_fdw/sql/postgres_fdw.sql | 223 +++++++++++ src/tools/pgindent/typedefs.list | 1 + 6 files changed, 926 insertions(+), 13 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 17f8c1e8f31..ef81c636baf 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -411,7 +411,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) entry->mapping_hashvalue = GetSysCacheHashValue1(USERMAPPINGOID, ObjectIdGetDatum(user->umid)); - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; + entry->state.entry = entry; /* * Determine whether to keep the connection that we're about to make here @@ -1075,6 +1076,20 @@ GetPrepStmtNumber(PGconn *conn) return ++prep_stmt_number; } +/* + * Exported version of begin_remote_xact(). + * + * This can be called for connections on which begin_remote_xact() has started + * a remote transaction. + */ +void +pgfdw_begin_remote_xact(ConnCacheEntry *entry) +{ + Assert(entry); + Assert(entry->xact_depth > 0); + begin_remote_xact(entry); +} + /* * Submit a query and wait for the result. * @@ -1091,6 +1106,13 @@ pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state) if (state && state->pendingAreq) process_pending_request(state->pendingAreq); + /* + * Second, synchronize the local/remote transactions. Note that we need + * to do this because this function can be called from open cursors. + */ + if (state) + pgfdw_begin_remote_xact(state->entry); + if (!PQsendQuery(conn, query)) return NULL; return pgfdw_get_result(conn); @@ -1194,6 +1216,16 @@ pgfdw_xact_callback(XactEvent event, void *arg) List *pending_entries = NIL; List *cancel_requested = NIL; + /* + * If we are called for post-commit cleanup, all we need to do is clean up + * foreign scans. + */ + if (event == XACT_EVENT_PARALLEL_COMMIT || event == XACT_EVENT_COMMIT) + { + AtEOXact_ForeignScan(); + return; + } + /* Quick exit if no connections were touched in this transaction. */ if (!xact_got_connection) return; @@ -1357,6 +1389,10 @@ pgfdw_xact_callback(XactEvent event, void *arg) /* Likewise for read_only_level */ read_only_level = 0; + + /* If we are called for abort cleanup, clean up foreign scans as well */ + if (event == XACT_EVENT_PARALLEL_ABORT || event == XACT_EVENT_ABORT) + AtEOXact_ForeignScan(); } /* @@ -1372,8 +1408,9 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, List *pending_entries = NIL; List *cancel_requested = NIL; - /* Nothing to do at subxact start, nor after commit. */ + /* Nothing to do at subxact start. */ if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || + event == SUBXACT_EVENT_COMMIT_SUB || event == SUBXACT_EVENT_ABORT_SUB)) return; @@ -1381,11 +1418,22 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, if (!xact_got_connection) return; + curlevel = GetCurrentTransactionNestLevel(); + + /* + * If we are called for post-subcommit cleanup, all we need to do is clean + * up foreign scans. + */ + if (event == SUBXACT_EVENT_COMMIT_SUB) + { + AtEOSubXact_ForeignScan(curlevel, true); + return; + } + /* * Scan all connection cache entries to find open remote subtransactions * of the current level, and close them. */ - curlevel = GetCurrentTransactionNestLevel(); hash_seq_init(&scan, ConnectionHash); while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) { @@ -1459,6 +1507,10 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, /* If in read_only_level, reset it */ if (curlevel == read_only_level) read_only_level = 0; + + /* If we are called for subabort cleanup, clean up foreign scans as well */ + if (event == SUBXACT_EVENT_ABORT_SUB) + AtEOSubXact_ForeignScan(curlevel, false); } /* @@ -1966,10 +2018,10 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) * If pendingAreq of the per-connection state is not NULL, it means that * an asynchronous fetch begun by fetch_more_data_begin() was not done * successfully and thus the per-connection state was not reset in - * fetch_more_data(); in that case reset the per-connection state here. + * fetch_more_data(); in that case reset pendingAreq here. */ if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; /* Disarm changing_xact_state if it all worked */ entry->changing_xact_state = false; @@ -2263,9 +2315,9 @@ pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, entry->have_error = false; } - /* Reset the per-connection state if needed */ + /* Reset pendingAreq here if any */ if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; /* We're done with this entry; unset the changing_xact_state flag */ entry->changing_xact_state = false; @@ -2308,9 +2360,9 @@ pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, entry->have_prep_stmt = false; entry->have_error = false; - /* Reset the per-connection state if needed */ + /* Reset pendingAreq here if any */ if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; /* We're done with this entry; unset the changing_xact_state flag */ entry->changing_xact_state = false; diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 115ebc7925f..fe2bb8695d8 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -5358,6 +5358,9 @@ ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE user_enum; -- =================================================================== -- subtransaction -- + local/remote error doesn't break cursor +-- + we allow cursors opened before a savepoint to be first fetched +-- after the savepoint; if that subtransaction aborts, the rows already +-- fetched can still be returned, but no more can be fetched. -- =================================================================== BEGIN; DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; @@ -5396,6 +5399,356 @@ SELECT * FROM ft1 ORDER BY c1 LIMIT 1; 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo (1 row) +COMMIT; +-- the rows fetched in the first batch are still available +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + +CLOSE c; +COMMIT; +-- but fetching another batch is not possible +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1'); +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +ERROR: cannot fetch more rows from foreign table "ft1" +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +ABORT; +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +CLOSE c; +COMMIT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); +-- a rescan of a scan whose cursor was lost creates a new one +SET enable_hashjoin = off; +SET enable_mergejoin = off; +SET enable_material = off; +EXPLAIN (COSTS OFF) +SELECT * FROM (VALUES (1), (2), (3)) v(x), + LATERAL (SELECT c1 FROM ft1 WHERE c1 = v.x OFFSET 0) s; + QUERY PLAN +--------------------------------- + Nested Loop + -> Values Scan on "*VALUES*" + -> Foreign Scan on ft1 +(3 rows) + +BEGIN; +DECLARE c CURSOR FOR +SELECT * FROM (VALUES (1), (2), (3)) v(x), + LATERAL (SELECT c1 FROM ft1 WHERE c1 = v.x OFFSET 0) s; +SAVEPOINT s; +FETCH c; + x | c1 +---+---- + 1 | 1 +(1 row) + +ROLLBACK TO s; +FETCH ALL c; + x | c1 +---+---- + 2 | 2 + 3 | 3 +(2 rows) + +COMMIT; +RESET enable_hashjoin; +RESET enable_mergejoin; +RESET enable_material; +-- cursor created in a subtransaction is handed to its parent subtransaction +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 3 | 3 | 00003 | Sun Jan 04 00:00:00 1970 PST | Sun Jan 04 00:00:00 1970 | 3 | 3 | foo +(1 row) + +CLOSE c; +COMMIT; +-- a cursor created in a subtransaction, which is released, can be first +-- fetched in a sibling subtransaction that is then rolled back +BEGIN; +SAVEPOINT s1; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s1; +SAVEPOINT s2; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s2; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + +CLOSE c; +COMMIT; +-- the same for nested subtransactions +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s1; +SAVEPOINT s2; +SAVEPOINT s3; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +RELEASE SAVEPOINT s3; +ROLLBACK TO s2; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + +CLOSE c; +COMMIT; +-- with a small fetch size, no more rows can be fetched if the subtransaction +-- the remote cursor was created in has been rolled back, however deep it was +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1'); +BEGIN; +SAVEPOINT s1; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s1; +SAVEPOINT s2; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s2; +FETCH c; +ERROR: cannot fetch more rows from foreign table "ft1" +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +ABORT; +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s1; +SAVEPOINT s2; +SAVEPOINT s3; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +RELEASE SAVEPOINT s3; +ROLLBACK TO s2; +FETCH c; +ERROR: cannot fetch more rows from foreign table "ft1" +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +ABORT; +-- but the cursor is fine if the subtransactions are released +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s1; +SAVEPOINT s2; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +RELEASE SAVEPOINT s2; +RELEASE SAVEPOINT s1; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + +CLOSE c; +COMMIT; +-- a remote cursor is created within the subtransaction in which parameter +-- evaluation happens, which may involve other foreign scans +BEGIN; +DECLARE c CURSOR FOR +SELECT c1 FROM ft1 WHERE c1 = (SELECT c1 FROM ft2 ORDER BY c1 LIMIT 1); +SAVEPOINT s; +FETCH c; + c1 +---- + 1 +(1 row) + +ROLLBACK TO s; +FETCH c; +ERROR: cannot fetch more rows from foreign table "ft1" +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +ABORT; +-- the same with a cursor opened in PL/pgSQL, fetched within a block +DO $$ +DECLARE c refcursor; r record; n int := 0; +BEGIN + OPEN c FOR SELECT c1 FROM ft1 ORDER BY c1; + BEGIN + FETCH c INTO r; n := n + 1; + RAISE EXCEPTION 'x'; + EXCEPTION WHEN others THEN NULL; + END; + FETCH c INTO r; n := n + 1; + CLOSE c; + RAISE NOTICE 'n = %', n; +END $$; +ERROR: cannot fetch more rows from foreign table "ft1" +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +CONTEXT: PL/pgSQL function inline_code_block line 10 at FETCH +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); +-- ... which works if the rows are already fetched +DO $$ +DECLARE c refcursor; r record; n int := 0; +BEGIN + OPEN c FOR SELECT c1 FROM ft1 ORDER BY c1; + BEGIN + FETCH c INTO r; n := n + 1; + RAISE EXCEPTION 'x'; + EXCEPTION WHEN others THEN NULL; + END; + FETCH c INTO r; n := n + 1; + CLOSE c; + RAISE NOTICE 'n = %', n; +END $$; +NOTICE: n = 2 +-- an asynchronous foreign scan whose cursor was lost +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1', ADD async_capable 'true'); +EXPLAIN (COSTS OFF) +SELECT c1 FROM ft1 UNION ALL SELECT c1 FROM ft1 WHERE c1 < 0; + QUERY PLAN +--------------------------------------- + Append + -> Async Foreign Scan on ft1 + -> Async Foreign Scan on ft1 ft1_1 +(3 rows) + +BEGIN; +DECLARE c CURSOR FOR +SELECT c1 FROM ft1 UNION ALL SELECT c1 FROM ft1 WHERE c1 < 0; +SAVEPOINT s; +FETCH c; + c1 +---- + 1 +(1 row) + +ROLLBACK TO s; +FETCH ALL c; +ERROR: cannot fetch more rows from foreign table "ft1" +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +ABORT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size, DROP async_capable); +-- the same applies even if the scan is for a pushed-down join +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1'); +ALTER FOREIGN TABLE ft2 OPTIONS (ADD fetch_size '1'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: t1.c1, t2.c1 + Relations: (public.ft1 t1) INNER JOIN (public.ft2 t2) + Remote SQL: SELECT r1."C 1", r2."C 1" FROM ("S 1"."T 1" r1 INNER JOIN "S 1"."T 1" r2 ON (((r2."C 1" = r1."C 1")))) ORDER BY r1."C 1" ASC NULLS LAST +(4 rows) + +BEGIN; +DECLARE c CURSOR FOR +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1; +SAVEPOINT s; +FETCH c; + c1 | c1 +----+---- + 1 | 1 +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +ERROR: cannot fetch more rows from this foreign scan +DETAIL: The remote cursor was removed by rolling back the subtransaction in which it was created. +ABORT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); +ALTER FOREIGN TABLE ft2 OPTIONS (DROP fetch_size); +-- an error while a cursor is fetched within a savepoint leaves the portal in +-- the failed state, so the scan is not shut down; make sure we cope with that +BEGIN; +DECLARE c CURSOR FOR SELECT 1 / (c1 - 3) FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH 5 c; -- error +ERROR: division by zero +ROLLBACK TO s; +CLOSE c; +SAVEPOINT s; +RELEASE SAVEPOINT s; +SELECT count(*) FROM ft1; + count +------- + 1000 +(1 row) + +COMMIT; +BEGIN; +DECLARE c CURSOR FOR SELECT 1 / (c1 - 3) FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH 5 c; -- error +ERROR: division by zero +ROLLBACK TO s; COMMIT; -- =================================================================== -- test handling of collations diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 4ca061c9e0b..2e5886059dd 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -17,6 +17,7 @@ #include "access/htup_details.h" #include "access/sysattr.h" #include "access/table.h" +#include "access/xact.h" #include "catalog/pg_opfamily.h" #include "commands/defrem.h" #include "commands/explain_format.h" @@ -185,6 +186,10 @@ typedef struct PgFdwScanState PgFdwConnState *conn_state; /* extra per-connection state */ unsigned int cursor_number; /* quasi-unique ID for my cursor */ bool cursor_exists; /* have we created the cursor? */ + int cursor_level; /* nesting level of the (sub)transaction in + * which the remote cursor was declared, which + * is also the depth of the remote savepoint + * holding it; 0 if no cursor exists */ int numParams; /* number of parameters passed to query */ FmgrInfo *param_flinfo; /* output conversion functions for them */ List *param_exprs; /* executable expressions for param values */ @@ -207,8 +212,46 @@ typedef struct PgFdwScanState MemoryContext temp_cxt; /* context for per-tuple temporary data */ int fetch_size; /* number of tuples per fetch */ + + bool cursor_lost; /* true if the remote cursor was removed by a + * rolled-back savepoint; the tuples already + * fetched are still usable, but no more can + * be fetched; see AtEOSubXact_ForeignScan */ } PgFdwScanState; +/* + * Registry of the foreign scans that are currently open in this backend. + * + * A remote cursor is declared inside the remote savepoint that matches the + * local (sub)transaction in which the first FETCH happens. If that local + * subtransaction is rolled back, the remote cursor is destroyed along with + * its remote savepoint, but the local scan, which may well belong to an + * outer (sub)transaction (e.g., a cursor opened before the savepoint), does + * not go away and would keep using a cursor that no longer exists. + * + * The subtransaction callbacks (see pgfdw_subxact_callback) are the only + * place where we learn about this, but they have no way to reach the scan + * nodes of the executor, so each scan registers itself here. On + * subtransaction commit, AtEOSubXact_ForeignScan hands the cursors declared + * in the subtransaction over to its parent, mirroring the remote RELEASE; on + * abort, it marks them as lost. + * + * Entries point into executor memory, so they must never outlive the scan + * node. Normally postgresEndForeignScan removes the entry, but the executor + * does not call EndForeignScan for a scan whose portal was marked failed by + * an aborted subtransaction, e.g., when an error occurred while the scan was + * active. To cover such cases, every scan also registers a callback on its + * memory context that removes the entry (see ForeignScanContextCallback). + * + * Initialized on first use. + */ +typedef struct ForeignScanEntry +{ + ForeignScanState *fs_node; /* hash key */ +} ForeignScanEntry; + +static HTAB *ForeignScanHash = NULL; + /* * Execution state of a foreign insert/update/delete operation. */ @@ -539,6 +582,10 @@ static void create_cursor(ForeignScanState *node); static void fetch_more_data(ForeignScanState *node); static void close_cursor(PGconn *conn, unsigned int cursor_number, PgFdwConnState *conn_state); +static void ForeignScanHashInsert(ForeignScanState *node); +static void ForeignScanHashDelete(ForeignScanState *node); +static void ForeignScanContextCallback(void *arg); +static void check_cursor_not_lost(PgFdwScanState *fsstate); static PgFdwModifyState *create_foreign_modify(EState *estate, RangeTblEntry *rte, ResultRelInfo *resultRelInfo, @@ -1703,6 +1750,7 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan; EState *estate = node->ss.ps.state; PgFdwScanState *fsstate; + MemoryContextCallback *scan_cb; RangeTblEntry *rte; Oid userid; ForeignTable *table; @@ -1762,6 +1810,7 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Assign a unique ID for my cursor */ fsstate->cursor_number = GetCursorNumber(fsstate->conn); fsstate->cursor_exists = false; + fsstate->cursor_level = 0; /* Get private info created by planner functions. */ fsstate->query = strVal(list_nth(fsplan->fdw_private, @@ -1822,6 +1871,20 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Set the async-capable flag */ fsstate->async_capable = node->ss.ps.async_capable; + + fsstate->cursor_lost = false; + + /* + * Register the scan, and make sure that it is unregistered even if + * EndForeignScan is never called for it. The callback is allocated in + * the context it is registered on, so that it can't outlive it. + */ + ForeignScanHashInsert(node); + scan_cb = (MemoryContextCallback *) + MemoryContextAlloc(estate->es_query_cxt, sizeof(MemoryContextCallback)); + scan_cb->func = ForeignScanContextCallback; + scan_cb->arg = node; + MemoryContextRegisterResetCallback(estate->es_query_cxt, scan_cb); } /* @@ -1892,6 +1955,30 @@ postgresReScanForeignScan(ForeignScanState *node) Assert(!fsstate->async_capable || !fsstate->conn_state->pendingAreq || fsstate->conn_state->pendingAreq->requestee != (PlanState *) node); + /* + * If the remote cursor is gone, there is nothing to close or rewind on + * the remote side. If we can't just rescan what we already have in + * memory, forget the cursor, so that a fresh one is created at the next + * fetch. + */ + if (fsstate->cursor_lost) + { + if (node->ss.ps.chgParam == NULL && + fsstate->fetch_ct_2 <= 1 && fsstate->eof_reached) + fsstate->next_tuple = 0; + else + { + fsstate->cursor_exists = false; + fsstate->cursor_lost = false; + fsstate->tuples = NULL; + fsstate->num_tuples = 0; + fsstate->next_tuple = 0; + fsstate->fetch_ct_2 = 0; + fsstate->eof_reached = false; + } + return; + } + /* * If any internal parameters affecting this node have changed, we'd * better destroy and recreate the cursor. Otherwise, if the remote @@ -1904,6 +1991,7 @@ postgresReScanForeignScan(ForeignScanState *node) if (node->ss.ps.chgParam != NULL) { fsstate->cursor_exists = false; + fsstate->cursor_level = 0; snprintf(sql, sizeof(sql), "CLOSE c%u", fsstate->cursor_number); } @@ -1915,6 +2003,7 @@ postgresReScanForeignScan(ForeignScanState *node) else { fsstate->cursor_exists = false; + fsstate->cursor_level = 0; snprintf(sql, sizeof(sql), "CLOSE c%u", fsstate->cursor_number); } @@ -1952,8 +2041,11 @@ postgresEndForeignScan(ForeignScanState *node) if (fsstate == NULL) return; - /* Close the cursor if open, to prevent accumulation of cursors */ - if (fsstate->cursor_exists) + /* + * Close the cursor if open, to prevent accumulation of cursors; but not + * if it's already gone. + */ + if (fsstate->cursor_exists && !fsstate->cursor_lost) close_cursor(fsstate->conn, fsstate->cursor_number, fsstate->conn_state); @@ -1961,6 +2053,8 @@ postgresEndForeignScan(ForeignScanState *node) ReleaseConnection(fsstate->conn); fsstate->conn = NULL; + ForeignScanHashDelete(node); + /* MemoryContexts will be deleted automatically. */ } @@ -4058,6 +4152,12 @@ create_cursor(ForeignScanState *node) if (fsstate->conn_state->pendingAreq) process_pending_request(fsstate->conn_state->pendingAreq); + /* + * Second, synchronize the local/remote transactions. Note that we need + * to do this because this function can be called from open cursors. + */ + pgfdw_begin_remote_xact(fsstate->conn_state->entry); + /* * Construct array of query parameter values in text format. We do the * conversions in the short-lived per-tuple context, so as not to cause a @@ -4103,6 +4203,8 @@ create_cursor(ForeignScanState *node) /* Mark the cursor as created, and show no tuples have been retrieved */ fsstate->cursor_exists = true; + fsstate->cursor_lost = false; + fsstate->cursor_level = GetCurrentTransactionNestLevel(); fsstate->tuples = NULL; fsstate->num_tuples = 0; fsstate->next_tuple = 0; @@ -4126,6 +4228,12 @@ fetch_more_data(ForeignScanState *node) int i; MemoryContext oldcontext; + /* + * Don't throw away the previous batch if we can't fetch another one. + */ + if (!fsstate->async_capable) + check_cursor_not_lost(fsstate); + /* * We'll store the tuples in the batch_cxt. First, flush the previous * batch. @@ -4147,7 +4255,7 @@ fetch_more_data(ForeignScanState *node) if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, fsstate->query); - /* Reset per-connection state */ + /* Reset the pending asynchronous request */ fsstate->conn_state->pendingAreq = NULL; } else @@ -4270,6 +4378,165 @@ close_cursor(PGconn *conn, unsigned int cursor_number, PQclear(res); } +/* + * Complain if the remote cursor of a scan was removed by rolling back a + * savepoint, in which case no more tuples can be fetched for the scan. + * + * Note that fsstate->rel is NULL if the scan is for a join or an upper + * relation pushed down to the remote server. + */ +static void +check_cursor_not_lost(PgFdwScanState *fsstate) +{ + if (!fsstate->cursor_lost) + return; + + if (fsstate->rel != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot fetch more rows from foreign table \"%s\"", + RelationGetRelationName(fsstate->rel)), + errdetail("The remote cursor was removed by rolling back the subtransaction in which it was created."))); + else + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot fetch more rows from this foreign scan"), + errdetail("The remote cursor was removed by rolling back the subtransaction in which it was created."))); +} + +/* + * Add a ForeignScan node to the ForeignScanHash table + */ +static void +ForeignScanHashInsert(ForeignScanState *node) +{ + ForeignScanEntry *entry; + bool found; + + /* First time through, initialize foreign scan hashtable */ + if (ForeignScanHash == NULL) + { + HASHCTL ctl; + + ctl.keysize = sizeof(ForeignScanState *); + ctl.entrysize = sizeof(ForeignScanEntry); + ForeignScanHash = hash_create("postgres_fdw ForeignScan nodes", + 256L, + &ctl, + HASH_ELEM | HASH_BLOBS); + } + + entry = (ForeignScanEntry *) hash_search(ForeignScanHash, + &node, + HASH_ENTER, + &found); + if (found) + elog(ERROR, "duplicate ForeignScan node"); + entry->fs_node = node; +} + +/* + * Remove a ForeignScan node from the ForeignScanHash table, if it is there + */ +static void +ForeignScanHashDelete(ForeignScanState *node) +{ + if (ForeignScanHash == NULL) + return; + + (void) hash_search(ForeignScanHash, &node, HASH_REMOVE, NULL); +} + +/* + * Memory context callback to unregister a foreign scan whose executor memory + * is being freed. This must not throw errors. + */ +static void +ForeignScanContextCallback(void *arg) +{ + ForeignScanHashDelete((ForeignScanState *) arg); +} + +/* + * Clean up foreign scans after a transaction commit/abort + */ +void +AtEOXact_ForeignScan(void) +{ + HASH_SEQ_STATUS scan; + ForeignScanEntry *entry; + + if (ForeignScanHash == NULL) + return; + + hash_seq_init(&scan, ForeignScanHash); + while ((entry = hash_seq_search(&scan)) != NULL) + { + ForeignScanState *node = entry->fs_node; + + /* + * Scans of portals that were marked failed may still be around at + * this point; their memory is freed later, at which point the context + * callback finds nothing to remove. + */ + ForeignScanHashDelete(node); + } +} + +/* + * Clean up foreign scans after a subtransaction subcommit/subabort + * + * "level" is the nesting level of the subtransaction that ended. + * + * We only care about the remote cursors that were declared in that + * subtransaction, i.e., those whose cursor_level is at least "level" (it can't + * be deeper, since deeper subtransactions have already ended and handed their + * cursors over to this one). + * + * On subcommit, the remote savepoint of that level is released, so such + * cursors now belong to the parent level. + * + * On subabort, the remote savepoint is rolled back, so such cursors are gone + * on the remote server. Mark them as lost. The tuples the scan has already + * fetched remain usable, but trying to fetch more fails with a clear error + * (see check_cursor_not_lost), rather than with a remote "cursor does not + * exist" error at some later point, such as when the cursor is closed. Note + * that we needn't tell apart the scans created in the aborted subtransaction, + * which are going away anyway; marking their cursors is harmless. + */ +void +AtEOSubXact_ForeignScan(int level, bool is_subcommit) +{ + HASH_SEQ_STATUS scan; + ForeignScanEntry *entry; + + if (ForeignScanHash == NULL) + return; + + hash_seq_init(&scan, ForeignScanHash); + while ((entry = hash_seq_search(&scan)) != NULL) + { + ForeignScanState *node = entry->fs_node; + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + + /* Skip unless the remote cursor was declared in this subxact */ + if (!fsstate->cursor_exists || fsstate->cursor_level < level) + continue; + + if (is_subcommit) + { + /* The cursor now belongs to the parent subxact */ + fsstate->cursor_level = level - 1; + } + else + { + /* The cursor is gone; remember that */ + fsstate->cursor_lost = true; + fsstate->cursor_level = 0; + } + } +} + /* * create_foreign_modify * Construct an execution state of a foreign insert/update/delete @@ -4394,6 +4661,9 @@ create_foreign_modify(EState *estate, * result if any. (This is the shared guts of postgresExecForeignInsert, * postgresExecForeignBatchInsert, postgresExecForeignUpdate, and * postgresExecForeignDelete.) + * + * Note: this function is never called from open cursors, thus no need to + * synchronize the local/remote transactions. */ static TupleTableSlot ** execute_foreign_modify(EState *estate, @@ -4832,6 +5102,9 @@ rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist) /* * Execute a direct UPDATE/DELETE statement. + * + * Note: this function is never called from open cursors, thus no need to + * synchronize the local/remote transactions. */ static void execute_dml_stmt(ForeignScanState *node) @@ -8840,9 +9113,16 @@ fetch_more_data_begin(AsyncRequest *areq) Assert(!fsstate->conn_state->pendingAreq); - /* Create the cursor synchronously. */ + check_cursor_not_lost(fsstate); + + /* + * Create the cursor synchronously if not already done. Otherwise, + * synchronize the local/remote transactions before the data fetch. + */ if (!fsstate->cursor_exists) create_cursor(node); + else + pgfdw_begin_remote_xact(fsstate->conn_state->entry); /* We will send this query, but not wait for the response. */ snprintf(sql, sizeof(sql), "FETCH %d FROM c%u", diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index da7da1c2ea9..7ed1fb2c8e5 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -147,6 +147,7 @@ typedef struct PgFdwRelationInfo typedef struct PgFdwConnState { AsyncRequest *pendingAreq; /* pending async request */ + struct ConnCacheEntry *entry; /* link to containing ConnCacheEntry */ } PgFdwConnState; /* @@ -165,6 +166,8 @@ typedef enum PgFdwSamplingMethod extern int set_transmission_modes(void); extern void reset_transmission_modes(int nestlevel); extern void process_pending_request(AsyncRequest *areq); +extern void AtEOXact_ForeignScan(void); +extern void AtEOSubXact_ForeignScan(int level, bool is_subcommit); /* in connection.c */ extern PGconn *GetConnection(UserMapping *user, bool will_prep_stmt, @@ -173,6 +176,7 @@ extern void ReleaseConnection(PGconn *conn); extern unsigned int GetCursorNumber(PGconn *conn); extern unsigned int GetPrepStmtNumber(PGconn *conn); extern void do_sql_command(PGconn *conn, const char *sql); +extern void pgfdw_begin_remote_xact(struct ConnCacheEntry *entry); extern PGresult *pgfdw_get_result(PGconn *conn); extern PGresult *pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state); diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 77c9a25a890..3ca6d467d6c 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -1640,6 +1640,9 @@ ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE user_enum; -- =================================================================== -- subtransaction -- + local/remote error doesn't break cursor +-- + we allow cursors opened before a savepoint to be first fetched +-- after the savepoint; if that subtransaction aborts, the rows already +-- fetched can still be returned, but no more can be fetched. -- =================================================================== BEGIN; DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; @@ -1655,6 +1658,226 @@ FETCH c; SELECT * FROM ft1 ORDER BY c1 LIMIT 1; COMMIT; +-- the rows fetched in the first batch are still available +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +CLOSE c; +COMMIT; + +-- but fetching another batch is not possible +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1'); +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +ABORT; + +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +CLOSE c; +COMMIT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); + +-- a rescan of a scan whose cursor was lost creates a new one +SET enable_hashjoin = off; +SET enable_mergejoin = off; +SET enable_material = off; +EXPLAIN (COSTS OFF) +SELECT * FROM (VALUES (1), (2), (3)) v(x), + LATERAL (SELECT c1 FROM ft1 WHERE c1 = v.x OFFSET 0) s; +BEGIN; +DECLARE c CURSOR FOR +SELECT * FROM (VALUES (1), (2), (3)) v(x), + LATERAL (SELECT c1 FROM ft1 WHERE c1 = v.x OFFSET 0) s; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +FETCH ALL c; +COMMIT; +RESET enable_hashjoin; +RESET enable_mergejoin; +RESET enable_material; + +-- cursor created in a subtransaction is handed to its parent subtransaction +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s; +FETCH c; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +CLOSE c; +COMMIT; + +-- a cursor created in a subtransaction, which is released, can be first +-- fetched in a sibling subtransaction that is then rolled back +BEGIN; +SAVEPOINT s1; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s1; +SAVEPOINT s2; +FETCH c; +ROLLBACK TO s2; +FETCH c; +CLOSE c; +COMMIT; + +-- the same for nested subtransactions +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s1; +SAVEPOINT s2; +SAVEPOINT s3; +FETCH c; +RELEASE SAVEPOINT s3; +ROLLBACK TO s2; +FETCH c; +CLOSE c; +COMMIT; + +-- with a small fetch size, no more rows can be fetched if the subtransaction +-- the remote cursor was created in has been rolled back, however deep it was +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1'); +BEGIN; +SAVEPOINT s1; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s1; +SAVEPOINT s2; +FETCH c; +ROLLBACK TO s2; +FETCH c; +ABORT; + +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s1; +SAVEPOINT s2; +SAVEPOINT s3; +FETCH c; +RELEASE SAVEPOINT s3; +ROLLBACK TO s2; +FETCH c; +ABORT; + +-- but the cursor is fine if the subtransactions are released +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s1; +SAVEPOINT s2; +FETCH c; +RELEASE SAVEPOINT s2; +RELEASE SAVEPOINT s1; +FETCH c; +CLOSE c; +COMMIT; + +-- a remote cursor is created within the subtransaction in which parameter +-- evaluation happens, which may involve other foreign scans +BEGIN; +DECLARE c CURSOR FOR +SELECT c1 FROM ft1 WHERE c1 = (SELECT c1 FROM ft2 ORDER BY c1 LIMIT 1); +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +FETCH c; +ABORT; + +-- the same with a cursor opened in PL/pgSQL, fetched within a block +DO $$ +DECLARE c refcursor; r record; n int := 0; +BEGIN + OPEN c FOR SELECT c1 FROM ft1 ORDER BY c1; + BEGIN + FETCH c INTO r; n := n + 1; + RAISE EXCEPTION 'x'; + EXCEPTION WHEN others THEN NULL; + END; + FETCH c INTO r; n := n + 1; + CLOSE c; + RAISE NOTICE 'n = %', n; +END $$; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); + +-- ... which works if the rows are already fetched +DO $$ +DECLARE c refcursor; r record; n int := 0; +BEGIN + OPEN c FOR SELECT c1 FROM ft1 ORDER BY c1; + BEGIN + FETCH c INTO r; n := n + 1; + RAISE EXCEPTION 'x'; + EXCEPTION WHEN others THEN NULL; + END; + FETCH c INTO r; n := n + 1; + CLOSE c; + RAISE NOTICE 'n = %', n; +END $$; + +-- an asynchronous foreign scan whose cursor was lost +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1', ADD async_capable 'true'); +EXPLAIN (COSTS OFF) +SELECT c1 FROM ft1 UNION ALL SELECT c1 FROM ft1 WHERE c1 < 0; +BEGIN; +DECLARE c CURSOR FOR +SELECT c1 FROM ft1 UNION ALL SELECT c1 FROM ft1 WHERE c1 < 0; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +FETCH ALL c; +ABORT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size, DROP async_capable); + +-- the same applies even if the scan is for a pushed-down join +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '1'); +ALTER FOREIGN TABLE ft2 OPTIONS (ADD fetch_size '1'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1; +BEGIN; +DECLARE c CURSOR FOR +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +ABORT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); +ALTER FOREIGN TABLE ft2 OPTIONS (DROP fetch_size); + +-- an error while a cursor is fetched within a savepoint leaves the portal in +-- the failed state, so the scan is not shut down; make sure we cope with that +BEGIN; +DECLARE c CURSOR FOR SELECT 1 / (c1 - 3) FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH 5 c; -- error +ROLLBACK TO s; +CLOSE c; +SAVEPOINT s; +RELEASE SAVEPOINT s; +SELECT count(*) FROM ft1; +COMMIT; +BEGIN; +DECLARE c CURSOR FOR SELECT 1 / (c1 - 3) FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH 5 c; -- error +ROLLBACK TO s; +COMMIT; + -- =================================================================== -- test handling of collations -- =================================================================== diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a7a8dbdd23f..6c4d1096b59 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -887,6 +887,7 @@ ForeignKeyCacheInfo ForeignKeyOptInfo ForeignPath ForeignScan +ForeignScanEntry ForeignScanState ForeignServer ForeignServerInfo -- 2.50.1 (Apple Git-155)
