From 8642dd6119492bf0e6c748f7d957a37afb15b359 Mon Sep 17 00:00:00 2001 From: Rafia Sabih Date: Mon, 2 Nov 2026 15:04:37 +0100 Subject: [PATCH v19 2/2] postgres_fdw: Add streaming_fetch option for cursor-free fetching postgres_fdw normally fetches remote tuples via a cursor, which prevents the remote side from using parallel query. The new boolean option streaming_fetch (server- and table-level, default false) instead runs the scan with libpq's chunked-rows mode, using fetch_size as the chunk size, leaving the remote query free to run in parallel. Asynchronous execution is unsupported in this mode and is disabled automatically when it's enabled. Scans that fetch the remote ctid (e.g. for a non-direct UPDATE or DELETE) or that lock rows with FOR UPDATE/SHARE still use a cursor. Since a connection can hold only one chunked-rows result at a time, a second scan starting on the same connection while another is still active drains the first scan's remaining tuples into a work_mem-bounded tuplestore and replays them when execution returns to it, so nested or interleaved scans stay correct without a second connection. If a drain is interrupted by an error, the scan is marked so that a later read errors out instead of silently returning short results, since the drain may have aborted only a subtransaction. A scan that stops early (an un-pushed-down LIMIT, a rescan) must drain the rest of the remote result before its connection can be reused, and a parameterized rescan re-runs the query from scratch, so streaming_fetch can lose to the cursor path when few rows are read. Original idea: Bernd Helmle Key suggestions and review: Robert Haas --- contrib/postgres_fdw/connection.c | 74 +- .../postgres_fdw/expected/postgres_fdw.out | 803 ++++++++++++++++++ .../postgres_fdw/expected/query_cancel.out | 21 + contrib/postgres_fdw/option.c | 4 + contrib/postgres_fdw/postgres_fdw.c | 634 ++++++++++++-- contrib/postgres_fdw/postgres_fdw.h | 8 + contrib/postgres_fdw/sql/postgres_fdw.sql | 341 ++++++++ contrib/postgres_fdw/sql/query_cancel.sql | 11 + doc/src/sgml/postgres-fdw.sgml | 52 ++ 9 files changed, 1860 insertions(+), 88 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 17f8c1e8f31..0fcb156efda 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -177,11 +177,10 @@ static void pgfdw_inval_callback(Datum arg, SysCacheIdentifier cacheid, uint32 hashvalue); static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry); static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel); -static bool pgfdw_cancel_query(PGconn *conn); static bool pgfdw_cancel_query_begin(PGconn *conn, TimestampTz endtime); static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, TimestampTz retrycanceltime, - bool consume_input); + bool consume_input, PgFdwConnState *state); static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); static bool pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query); @@ -313,6 +312,9 @@ GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state) /* Process a pending asynchronous request if any. */ if (entry->state.pendingAreq) process_pending_request(entry->state.pendingAreq); + + /* Likewise, finish any streaming_fetch scan using the connection. */ + drain_active_scan(&entry->state); /* Start a new transaction or subtransaction if needed. */ begin_remote_xact(entry); } @@ -1083,14 +1085,20 @@ GetPrepStmtNumber(PGconn *conn) * ignore that for now. * * Caller is responsible for the error handling on the result. + * + * state must not be NULL, since any streaming_fetch scan still using the + * connection has to be drained first. */ PGresult * pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state) { /* First, process a pending asynchronous request, if any. */ - if (state && state->pendingAreq) + if (state->pendingAreq) process_pending_request(state->pendingAreq); + /* Likewise, finish any streaming_fetch scan using the connection. */ + drain_active_scan(state); + if (!PQsendQuery(conn, query)) return NULL; return pgfdw_get_result(conn); @@ -1107,6 +1115,18 @@ pgfdw_get_result(PGconn *conn) return libpqsrv_get_result_last(conn, pgfdw_we_get_result); } +/* + * Wrap libpqsrv_get_result(), adding wait event. Used to read a + * streaming_fetch result one chunk at a time. + * + * Caller is responsible for the error handling on the result. + */ +PGresult * +pgfdw_get_next_result(PGconn *conn) +{ + return libpqsrv_get_result(conn, pgfdw_we_get_result); +} + /* * Report an error we got from the remote server. * @@ -1280,7 +1300,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) if (entry->have_prep_stmt && entry->have_error) { res = pgfdw_exec_query(entry->conn, "DEALLOCATE ALL", - NULL); + &entry->state); PQclear(res); } entry->have_prep_stmt = false; @@ -1410,6 +1430,14 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, */ pgfdw_reject_incomplete_xact_state_change(entry); + /* + * A cursor that outlives this subtransaction may still be + * streaming; drain it before RELEASE SAVEPOINT. Do this before + * setting changing_xact_state, so an error only aborts the + * subtransaction. + */ + drain_active_scan(&entry->state); + /* Commit all remote subtransactions during pre-commit */ snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); entry->changing_xact_state = true; @@ -1606,9 +1634,11 @@ pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel) * XXX: if the query was one sent by fetch_more_data_begin(), we could get the * query text from the pendingAreq saved in the per-connection state, then * report the query using it. + * + * On success, pgfdw_cancel_query_end() also clears state->active_scan. */ -static bool -pgfdw_cancel_query(PGconn *conn) +bool +pgfdw_cancel_query(PGconn *conn, PgFdwConnState *state) { TimestampTz now = GetCurrentTimestamp(); TimestampTz endtime; @@ -1628,7 +1658,7 @@ pgfdw_cancel_query(PGconn *conn) if (!pgfdw_cancel_query_begin(conn, endtime)) return false; - return pgfdw_cancel_query_end(conn, endtime, retrycanceltime, false); + return pgfdw_cancel_query_end(conn, endtime, retrycanceltime, false, state); } /* @@ -1655,7 +1685,8 @@ pgfdw_cancel_query_begin(PGconn *conn, TimestampTz endtime) static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, - TimestampTz retrycanceltime, bool consume_input) + TimestampTz retrycanceltime, bool consume_input, + PgFdwConnState *state) { PGresult *result; bool timed_out; @@ -1692,6 +1723,9 @@ pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, } PQclear(result); + /* The canceled query's result is consumed; nothing is streaming now. */ + state->active_scan = NULL; + return true; } @@ -1943,7 +1977,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) * remote server, and if so, request cancellation of the command. */ if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE && - !pgfdw_cancel_query(entry->conn)) + !pgfdw_cancel_query(entry->conn, &entry->state)) return; /* Unable to cancel running query */ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); @@ -1966,10 +2000,12 @@ 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(). Likewise, active_scan may still point at a + * streaming_fetch scan that is going away with this (sub)transaction. + * Reset both, so the connection's next user sees no stale state. */ - if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; + entry->state.active_scan = NULL; /* Disarm changing_xact_state if it all worked */ entry->changing_xact_state = false; @@ -2193,7 +2229,7 @@ pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, RETRY_CANCEL_TIMEOUT); if (!pgfdw_cancel_query_end(entry->conn, endtime, - retrycanceltime, true)) + retrycanceltime, true, &entry->state)) { /* Unable to cancel running query */ pgfdw_reset_xact_state(entry, toplevel); @@ -2263,9 +2299,9 @@ pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, entry->have_error = false; } - /* Reset the per-connection state if needed */ - if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + /* Reset the async request and any undrained streaming_fetch scan */ + entry->state.pendingAreq = NULL; + entry->state.active_scan = NULL; /* We're done with this entry; unset the changing_xact_state flag */ entry->changing_xact_state = false; @@ -2308,9 +2344,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 */ - if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + /* Reset the async request and any undrained streaming_fetch scan */ + entry->state.pendingAreq = NULL; + entry->state.active_scan = 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..3a90e5332a2 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -255,6 +255,49 @@ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1; -- should work again ANALYZE ft1; ALTER FOREIGN TABLE ft2 OPTIONS (use_remote_estimate 'true'); -- =================================================================== +-- test streaming_fetch option +-- =================================================================== +CREATE SERVER fetch_stream_srv FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (streaming_fetch 'true'); +CREATE USER MAPPING FOR CURRENT_USER SERVER fetch_stream_srv; +CREATE FOREIGN TABLE ft_stream (c1 int) SERVER fetch_stream_srv; +-- Set on the server; EXPLAIN VERBOSE shows "Streaming Fetch: true". +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft_stream; + QUERY PLAN +----------------------------------------------- + Foreign Scan on public.ft_stream + Output: c1 + Remote SQL: SELECT c1 FROM public.ft_stream + Streaming Fetch: true +(4 rows) + +-- A table-level setting overrides the server's, in either direction. +ALTER FOREIGN TABLE ft_stream OPTIONS (ADD streaming_fetch 'false'); +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft_stream; + QUERY PLAN +----------------------------------------------- + Foreign Scan on public.ft_stream + Output: c1 + Remote SQL: SELECT c1 FROM public.ft_stream +(3 rows) + +ALTER SERVER fetch_stream_srv OPTIONS (SET streaming_fetch 'false'); +ALTER FOREIGN TABLE ft_stream OPTIONS (SET streaming_fetch 'true'); +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft_stream; + QUERY PLAN +----------------------------------------------- + Foreign Scan on public.ft_stream + Output: c1 + Remote SQL: SELECT c1 FROM public.ft_stream + Streaming Fetch: true +(4 rows) + +ALTER FOREIGN TABLE ft_stream OPTIONS (SET streaming_fetch 'yes'); -- ERROR +ERROR: streaming_fetch requires a Boolean value +DROP FOREIGN TABLE ft_stream; +DROP USER MAPPING FOR CURRENT_USER SERVER fetch_stream_srv; +DROP SERVER fetch_stream_srv; +-- =================================================================== -- test subscription -- =================================================================== CREATE SUBSCRIPTION regress_pgfdw_subscription SERVER testserver1 @@ -892,6 +935,690 @@ WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); 996 | 6 | 00996 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 996 | 6 | 00996 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo (100 rows) +-- =================================================================== +-- streaming_fetch scans sharing a connection +-- =================================================================== +-- ft1_s and ft2_s stream the same remote table as ft1 and ft2, and all four +-- share the loopback connection. ft2_s gets its rows in chunks of 5. +CREATE FOREIGN TABLE ft1_s ( + c1 int NOT NULL, + c2 int NOT NULL, + c3 text, + c4 timestamptz, + c5 timestamp, + c6 varchar(10), + c7 char(10) default 'ft1', + c8 user_enum +) SERVER loopback OPTIONS (schema_name 'S 1', table_name 'T 1', + streaming_fetch 'true'); +ALTER FOREIGN TABLE ft1_s ALTER COLUMN c1 OPTIONS (column_name 'C 1'); +CREATE FOREIGN TABLE ft2_s ( + c1 int NOT NULL, + c2 int NOT NULL, + c3 text, + c4 timestamptz, + c5 timestamp, + c6 varchar(10), + c7 char(10) default 'ft2', + c8 user_enum +) SERVER loopback OPTIONS (schema_name 'S 1', table_name 'T 1', + streaming_fetch 'true', fetch_size '5', + use_remote_estimate 'true'); +ALTER FOREIGN TABLE ft2_s ALTER COLUMN c1 OPTIONS (column_name 'C 1'); +-- One row per chunk, and chunks of 5 + 5 + 2. The c8 filter can't be pushed +-- down, so the aggregate runs locally over the streamed rows. +ALTER FOREIGN TABLE ft1_s OPTIONS (ADD fetch_size '1'); +SELECT array_agg(c1 ORDER BY c1) FROM ft1_s WHERE c1 <= 12 AND c8 = 'foo'; + array_agg +------------------------------ + {1,2,3,4,5,6,7,8,9,10,11,12} +(1 row) + +ALTER FOREIGN TABLE ft1_s OPTIONS (DROP fetch_size); +SELECT array_agg(c1 ORDER BY c1) FROM ft2_s WHERE c1 <= 12 AND c8 = 'foo'; + array_agg +------------------------------ + {1,2,3,4,5,6,7,8,9,10,11,12} +(1 row) + +-- Streaming and cursor scans in one query, and two streaming scans at once. +SELECT * FROM ft1_s t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +------+----+-------+------------------------------+--------------------------+----+------------+----- + 1000 | 0 | 01000 | Thu Jan 01 00:00:00 1970 PST | Thu Jan 01 00:00:00 1970 | 0 | 0 | foo +(1 row) + +SELECT * FROM ft1_s t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2_s t2) ORDER BY c1; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +------+----+-------+------------------------------+--------------------------+----+------------+----- + 1000 | 0 | 01000 | Thu Jan 01 00:00:00 1970 PST | Thu Jan 01 00:00:00 1970 | 0 | 0 | foo +(1 row) + +SELECT t1.c1, t2."C 1" FROM ft2_s t1 JOIN "S 1"."T 1" t2 ON (t1.c1 = t2."C 1") OFFSET 100 LIMIT 10; + c1 | C 1 +-----+----- + 101 | 101 + 102 | 102 + 103 | 103 + 104 | 104 + 105 | 105 + 106 | 106 + 107 | 107 + 108 | 108 + 109 | 109 + 110 | 110 +(10 rows) + +SELECT t1.c1 FROM ft1_s t1 WHERE NOT EXISTS (SELECT 1 FROM ft2 t2 WHERE t1.c1 = t2.c2) ORDER BY t1.c1 OFFSET 100 LIMIT 10; + c1 +----- + 110 + 111 + 112 + 113 + 114 + 115 + 116 + 117 + 118 + 119 +(10 rows) + +SELECT t1.c1 FROM ft1_s t1 WHERE NOT EXISTS (SELECT 1 FROM ft2_s t2 WHERE t1.c1 = t2.c2) ORDER BY t1.c1 OFFSET 100 LIMIT 10; + c1 +----- + 110 + 111 + 112 + 113 + 114 + 115 + 116 + 117 + 118 + 119 +(10 rows) + +-- Nested loops: each time the inner scan starts, the rest of the outer +-- scan's result is saved locally, and later read back in fetch_size batches. +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM ft2_s a, ft2_s b +WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); + QUERY PLAN +------------------------------------------------------------------------------------------------------------- + Nested Loop + Output: a.c1, a.c2, a.c3, a.c4, a.c5, a.c6, a.c7, a.c8, b.c1, b.c2, b.c3, b.c4, b.c5, b.c6, b.c7, b.c8 + -> Foreign Scan on public.ft2_s a + Output: a.c1, a.c2, a.c3, a.c4, a.c5, a.c6, a.c7, a.c8 + Filter: (a.c8 = 'foo'::user_enum) + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((c2 = 6)) + Streaming Fetch: true + -> Foreign Scan on public.ft2_s b + Output: b.c1, b.c2, b.c3, b.c4, b.c5, b.c6, b.c7, b.c8 + Filter: ((b.c7)::text = upper((a.c7)::text)) + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (($1::integer = "C 1")) + Streaming Fetch: true +(12 rows) + +SELECT count(*), sum(b.c1) FROM ft2_s a, ft2_s b +WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); + count | sum +-------+------- + 100 | 50100 +(1 row) + +SELECT count(*), sum(c.c1) FROM ft2_s a, ft2_s b, ft2_s c +WHERE a.c2 = 6 AND b.c1 = a.c1 AND c.c1 = b.c1 +AND a.c8 = 'foo' AND b.c7 = upper(a.c7); + count | sum +-------+------- + 100 | 50100 +(1 row) + +-- An EXISTS subquery stops its streaming scan at the first match, leaving +-- the rest of its result unread. The connection must still work for the +-- outer cursor's later FETCHes (ft1 needs several with fetch_size 5), for a +-- per-row UPDATE, and for a pushed-down UPDATE on ft4. +CREATE FUNCTION direct_modify_ft4(id int) RETURNS void AS $$ + UPDATE ft4 SET c3 = c3 WHERE c1 = id; +$$ LANGUAGE sql; +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '5'); +SET enable_hashagg = off; +SET enable_hashjoin = off; +SET enable_mergejoin = off; +SET enable_material = off; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT a.c1, a.c2 FROM ft1 a +WHERE a.c1 < 20 AND a.c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = a.c2) +ORDER BY a.c1; + QUERY PLAN +-------------------------------------------------------------------------------------------------------------- + Nested Loop Semi Join + Output: a.c1, a.c2 + -> Foreign Scan on public.ft1 a + Output: a.c1, a.c2 + Filter: (a.c8 = 'foo'::user_enum) + Remote SQL: SELECT "C 1", c2, c8 FROM "S 1"."T 1" WHERE (("C 1" < 20)) ORDER BY "C 1" ASC NULLS LAST + -> Foreign Scan on public.ft2_s b + Output: b.c1, b.c2, b.c3, b.c4, b.c5, b.c6, b.c7, b.c8 + Remote SQL: SELECT c2 FROM "S 1"."T 1" WHERE ((c2 = $1::integer)) + Streaming Fetch: true +(10 rows) + +SELECT a.c1, a.c2 FROM ft1 a +WHERE a.c1 < 20 AND a.c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = a.c2) +ORDER BY a.c1; + c1 | c2 +----+---- + 1 | 1 + 2 | 2 + 3 | 3 + 4 | 4 + 5 | 5 + 6 | 6 + 7 | 7 + 8 | 8 + 9 | 9 + 10 | 0 + 11 | 1 + 12 | 2 + 13 | 3 + 14 | 4 + 15 | 5 + 16 | 6 + 17 | 7 + 18 | 8 + 19 | 9 +(19 rows) + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE ft1 SET c3 = c3 +WHERE c1 < 20 AND c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = ft1.c2); + QUERY PLAN +--------------------------------------------------------------------------------------------------------------------------- + Update on public.ft1 + Remote SQL: UPDATE "S 1"."T 1" SET c3 = $2 WHERE ctid = $1 + -> Nested Loop Semi Join + Output: ft1.c3, ft1.ctid, ft1.*, b.* + -> Foreign Scan on public.ft1 + Output: ft1.c3, ft1.ctid, ft1.*, ft1.c2 + Filter: (ft1.c8 = 'foo'::user_enum) + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8, ctid FROM "S 1"."T 1" WHERE (("C 1" < 20)) FOR UPDATE + -> Foreign Scan on public.ft2_s b + Output: b.*, b.c2 + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((c2 = $1::integer)) + Streaming Fetch: true +(12 rows) + +EXPLAIN (VERBOSE, COSTS OFF) UPDATE ft4 SET c3 = c3 WHERE c1 = 1; + QUERY PLAN +--------------------------------------------------------------------- + Update on public.ft4 + -> Foreign Update on public.ft4 + Remote SQL: UPDATE "S 1"."T 3" SET c3 = c3 WHERE ((c1 = 1)) +(3 rows) + +BEGIN; +WITH u AS (UPDATE ft1 SET c3 = c3 + WHERE c1 < 20 AND c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = ft1.c2) + RETURNING c1) +SELECT count(*) FROM u; + count +------- + 19 +(1 row) + +SELECT count(*) FROM (SELECT direct_modify_ft4(a.c1) FROM ft1 a + WHERE a.c1 < 20 AND a.c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = a.c2) + OFFSET 0) s; + count +------- + 19 +(1 row) + +ROLLBACK; +RESET enable_hashagg; +RESET enable_hashjoin; +RESET enable_mergejoin; +RESET enable_material; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); +DROP FUNCTION direct_modify_ft4(int); +-- A LIMIT that can't be pushed down stops the scan early; the rest of the +-- remote result is read and discarded. The second query stops partway +-- through a chunk. +SELECT c1 FROM ft1_s WHERE c8 = 'foo' ORDER BY c1 LIMIT 5; + c1 +---- + 1 + 2 + 3 + 4 + 5 +(5 rows) + +SELECT c1 FROM ft2_s WHERE c8 = 'foo' ORDER BY c1 LIMIT 7; + c1 +---- + 1 + 2 + 3 + 4 + 5 + 6 + 7 +(7 rows) + +SELECT count(*) FROM ft1_s; + count +------- + 1000 +(1 row) + +-- Scans that fetch ctid or lock rows use a cursor (no "Streaming Fetch"). +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft2_s WHERE c1 = 1 FOR UPDATE; + QUERY PLAN +-------------------------------------------------------------------------------------------------------- + Foreign Scan on public.ft2_s + Output: c1, c2, c3, c4, c5, c6, c7, c8, ft2_s.* + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" = 1)) FOR UPDATE +(3 rows) + +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft2_s WHERE c1 = 1 FOR SHARE; + QUERY PLAN +------------------------------------------------------------------------------------------------------- + Foreign Scan on public.ft2_s + Output: c1, c2, c3, c4, c5, c6, c7, c8, ft2_s.* + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" = 1)) FOR SHARE +(3 rows) + +EXPLAIN (VERBOSE, COSTS OFF) DELETE FROM ft2_s WHERE c1 = 1 AND random() >= 0; + QUERY PLAN +--------------------------------------------------------------------------------- + Delete on public.ft2_s + Remote SQL: DELETE FROM "S 1"."T 1" WHERE ctid = $1 + -> Foreign Scan on public.ft2_s + Output: ctid + Filter: (random() >= '0'::double precision) + Remote SQL: SELECT ctid FROM "S 1"."T 1" WHERE (("C 1" = 1)) FOR UPDATE +(6 rows) + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE ft2_s f1 SET c3 = f2.c3 FROM ft1_s f2 +WHERE f1.c1 = f2.c1 AND f1.c1 = 3 AND random() >= 0; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Update on public.ft2_s f1 + Remote SQL: UPDATE "S 1"."T 1" SET c3 = $2 WHERE ctid = $1 + -> Foreign Scan + Output: f2.c3, f1.ctid, f1.*, f2.* + Filter: (random() >= '0'::double precision) + Relations: (public.ft2_s f1) INNER JOIN (public.ft1_s f2) + Remote SQL: SELECT r2.c3, r1.ctid, CASE WHEN (r1.*)::text IS NOT NULL THEN ROW(r1."C 1", r1.c2, r1.c3, r1.c4, r1.c5, r1.c6, r1.c7, r1.c8) END, CASE WHEN (r2.*)::text IS NOT NULL THEN ROW(r2."C 1", r2.c2, r2.c3, r2.c4, r2.c5, r2.c6, r2.c7, r2.c8) END FROM ("S 1"."T 1" r1 INNER JOIN "S 1"."T 1" r2 ON (((r2."C 1" = 3)) AND ((r1."C 1" = 3)))) FOR UPDATE OF r1 + -> Nested Loop + Output: f2.c3, f1.ctid, f1.*, f2.* + -> Foreign Scan on public.ft2_s f1 + Output: f1.ctid, f1.*, f1.c1 + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8, ctid FROM "S 1"."T 1" WHERE (("C 1" = 3)) FOR UPDATE + -> Foreign Scan on public.ft1_s f2 + Output: f2.c3, f2.*, f2.c1 + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" = 3)) + Streaming Fetch: true +(16 rows) + +-- So a non-direct UPDATE whose rows arrive in several chunks updates them all. +BEGIN; +WITH u AS (UPDATE ft2_s SET c3 = c3 WHERE c1 <= 20 AND random() >= 0 RETURNING 1) +SELECT count(*) FROM u; + count +------- + 20 +(1 row) + +ROLLBACK; +DROP FOREIGN TABLE ft1_s, ft2_s; +-- If saving a scan's rows locally fails partway (row 6 can't be converted) +-- inside a subtransaction, the scan must fail at the outer level rather than +-- return short results, even if a later attempt to save the rest succeeds. +CREATE TABLE loct_lost (id int, x text); +INSERT INTO loct_lost SELECT g, g::text FROM generate_series(1, 5) g; +INSERT INTO loct_lost VALUES (6, 'not_an_int'), (7, '7'), (8, '8'); +CREATE FOREIGN TABLE ft_lost (id int, x int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_lost', + streaming_fetch 'true', fetch_size '2'); +CREATE FOREIGN TABLE ft_lost_other (id int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_lost'); +\set VERBOSITY terse +BEGIN; +DECLARE c_lost CURSOR FOR SELECT id, x FROM ft_lost ORDER BY id; +FETCH 2 FROM c_lost; + id | x +----+--- + 1 | 1 + 2 | 2 +(2 rows) + +SAVEPOINT sp; +SELECT id FROM ft_lost_other ORDER BY id LIMIT 1; -- hits row 6 +ERROR: invalid input syntax for type integer: "not_an_int" +ROLLBACK TO SAVEPOINT sp; +FETCH ALL FROM c_lost; -- ERROR +ERROR: could not obtain all rows from remote server +COMMIT; +BEGIN; +DECLARE c_lost CURSOR FOR SELECT id, x FROM ft_lost ORDER BY id; +FETCH 2 FROM c_lost; + id | x +----+--- + 1 | 1 + 2 | 2 +(2 rows) + +SAVEPOINT sp; +SELECT count(*) FROM ft_lost_other; -- hits row 6 +ERROR: invalid input syntax for type integer: "not_an_int" +ROLLBACK TO SAVEPOINT sp; +SELECT count(*) FROM ft_lost_other; -- saves rows 7 and 8 + count +------- + 8 +(1 row) + +FETCH ALL FROM c_lost; -- ERROR +ERROR: could not obtain all rows from remote server +COMMIT; +SELECT count(*) FROM ft_lost_other; -- connection still works + count +------- + 8 +(1 row) + +\set VERBOSITY default +DROP FOREIGN TABLE ft_lost, ft_lost_other; +DROP TABLE loct_lost; +-- Transaction control with a cursor whose remote query is still streaming +-- when postgres_fdw sends RELEASE SAVEPOINT, ROLLBACK TO SAVEPOINT or +-- COMMIT. Each case must return every row or fail cleanly, and leave the +-- connection usable. +CREATE TABLE loct_xact (id int, x text); +INSERT INTO loct_xact SELECT g, g::text FROM generate_series(1, 10) g; +CREATE FOREIGN TABLE ft_xact (id int, x int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_xact', + streaming_fetch 'true', fetch_size '2'); +CREATE FOREIGN TABLE ft_xact_other (id int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_xact'); +-- Cursor opened in a subtransaction that is then released. +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +RELEASE SAVEPOINT s; +FETCH ALL FROM c; + id +---- + 4 + 5 + 6 + 7 + 8 + 9 + 10 +(7 rows) + +COMMIT; +-- Cursor declared outside a subtransaction, but first read inside it. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +SAVEPOINT s; +SELECT count(*) FROM ft_xact_other; + count +------- + 10 +(1 row) + +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +RELEASE SAVEPOINT s; +FETCH ALL FROM c; + id +---- + 4 + 5 + 6 + 7 + 8 + 9 + 10 +(7 rows) + +COMMIT; +-- Nested savepoints, released one at a time; parallel_commit sends each +-- RELEASE SAVEPOINT asynchronously. +ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true'); +BEGIN; +SAVEPOINT s1; +SAVEPOINT s2; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +RELEASE SAVEPOINT s2; +FETCH 2 FROM c; + id +---- + 4 + 5 +(2 rows) + +RELEASE SAVEPOINT s1; +FETCH ALL FROM c; + id +---- + 6 + 7 + 8 + 9 + 10 +(5 rows) + +COMMIT; +ALTER SERVER loopback OPTIONS (DROP parallel_commit); +-- Rolling back the subtransaction in which the cursor was first read +-- cancels the remote query, so the cursor must report an error. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +SAVEPOINT s; +SELECT count(*) FROM ft_xact_other; + count +------- + 10 +(1 row) + +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +ROLLBACK TO SAVEPOINT s; +FETCH ALL FROM c; -- ERROR +ERROR: could not obtain remaining rows from remote server +DETAIL: The remote query was canceled when a subtransaction was rolled back. +COMMIT; +SELECT count(*) FROM ft_xact_other; -- connection still works + count +------- + 10 +(1 row) + +-- Cursor opened in a subtransaction that is rolled back goes away with it. +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +ROLLBACK TO SAVEPOINT s; +SELECT count(*) FROM ft_xact_other; + count +------- + 10 +(1 row) + +COMMIT; +-- Saving the rest of the cursor's rows at RELEASE SAVEPOINT fails on a bad +-- row; the connection must work again after rollback. +INSERT INTO loct_xact VALUES (11, 'not_an_int'); +\set VERBOSITY terse +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT id, x FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id | x +----+--- + 1 | 1 + 2 | 2 + 3 | 3 +(3 rows) + +RELEASE SAVEPOINT s; -- ERROR +ERROR: invalid input syntax for type integer: "not_an_int" +ROLLBACK; +\set VERBOSITY default +SELECT count(*) FROM ft_xact_other; -- connection still works + count +------- + 11 +(1 row) + +DELETE FROM loct_xact WHERE id = 11; +-- Top-level COMMIT closes the cursor before the remote COMMIT. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +COMMIT; +SELECT count(*) FROM ft_xact_other; + count +------- + 10 +(1 row) + +-- A holdable cursor is read to the end at COMMIT and keeps all its rows. +BEGIN; +DECLARE c CURSOR WITH HOLD FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +COMMIT; +FETCH ALL FROM c; + id +---- + 4 + 5 + 6 + 7 + 8 + 9 + 10 +(7 rows) + +CLOSE c; +-- Top-level ROLLBACK. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; + id +---- + 1 + 2 + 3 +(3 rows) + +ROLLBACK; +SELECT count(*) FROM ft_xact_other; + count +------- + 10 +(1 row) + +DROP FOREIGN TABLE ft_xact, ft_xact_other; +DROP TABLE loct_xact; +-- Rows saved locally on behalf of a cursor may spill to a temp file. That +-- file belongs to the cursor, so it must survive the rollback of the +-- subtransaction in which it was written. +CREATE TABLE loct_spill (id int, pad text); +INSERT INTO loct_spill SELECT g, repeat('x', 100) FROM generate_series(1, 5000) g; +CREATE FOREIGN TABLE ft_spill (id int, pad text) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_spill', + streaming_fetch 'true', fetch_size '10'); +CREATE FOREIGN TABLE ft_spill_other (id int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_spill'); +BEGIN; +SET LOCAL work_mem = '64kB'; +DECLARE c CURSOR FOR SELECT id, length(pad) FROM ft_spill ORDER BY id; +FETCH 1 FROM c; + id | length +----+-------- + 1 | 100 +(1 row) + +SAVEPOINT s; +SELECT count(*) FROM ft_spill_other; -- saves c's rows, spilling to disk + count +------- + 5000 +(1 row) + +ROLLBACK TO SAVEPOINT s; +MOVE 4995 FROM c; +FETCH ALL FROM c; + id | length +------+-------- + 4997 | 100 + 4998 | 100 + 4999 | 100 + 5000 | 100 +(4 rows) + +COMMIT; +DROP FOREIGN TABLE ft_spill, ft_spill_other; +DROP TABLE loct_spill; -- bug before 9.3.5 due to sloppy handling of remote-estimate parameters SELECT * FROM ft1 WHERE c1 = ANY (ARRAY(SELECT c1 FROM ft2 WHERE c1 < 5)); c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 @@ -11587,6 +12314,43 @@ SELECT 1 FROM ft1 LIMIT 1; -- should fail ERROR: 08006 \set VERBOSITY default COMMIT; +-- =================================================================== +-- streaming_fetch: reconnecting after the remote backend terminates +-- =================================================================== +ALTER FOREIGN TABLE ft1 OPTIONS (ADD streaming_fetch 'true'); +SELECT 1 FROM ft1 LIMIT 1; -- open the connection + ?column? +---------- + 1 +(1 row) + +DO $$ BEGIN +PERFORM pg_terminate_backend(pid, 180000) FROM pg_stat_activity + WHERE application_name = 'fdw_retry_check'; +END $$; +-- Outside a subtransaction, the broken connection is reestablished. +BEGIN; +SELECT c1 FROM ft1 ORDER BY c1 LIMIT 3; + c1 +---- + 1 + 3 + 4 +(3 rows) + +DO $$ BEGIN +PERFORM pg_terminate_backend(pid, 180000) FROM pg_stat_activity + WHERE application_name = 'fdw_retry_check'; +END $$; +-- Inside one, it isn't, and the query fails. +SAVEPOINT s2; +-- The text of the error might vary across platforms, so only show SQLSTATE. +\set VERBOSITY sqlstate +SELECT 1 FROM ft1 LIMIT 1; -- should fail +ERROR: 08006 +\set VERBOSITY default +COMMIT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP streaming_fetch); -- ============================================================================= -- test connection invalidation cases and postgres_fdw_get_connections function -- ============================================================================= @@ -12816,6 +13580,45 @@ SELECT * FROM local_tbl, async_pt WHERE local_tbl.a = async_pt.a AND local_tbl.c ALTER FOREIGN TABLE async_p1 OPTIONS (DROP use_remote_estimate); ALTER FOREIGN TABLE async_p2 OPTIONS (DROP use_remote_estimate); +-- streaming_fetch disables asynchronous execution, whether it is set on the +-- table (async_p1) or on the server (loopback2, used by async_p2). +ALTER FOREIGN TABLE async_p1 OPTIONS (ADD streaming_fetch 'true'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM async_pt; + QUERY PLAN +---------------------------------------------------------- + Append + -> Foreign Scan on public.async_p1 async_pt_1 + Output: async_pt_1.a, async_pt_1.b, async_pt_1.c + Remote SQL: SELECT a, b, c FROM public.base_tbl1 + Streaming Fetch: true + -> Async Foreign Scan on public.async_p2 async_pt_2 + Output: async_pt_2.a, async_pt_2.b, async_pt_2.c + Remote SQL: SELECT a, b, c FROM public.base_tbl2 + -> Seq Scan on public.async_p3 async_pt_3 + Output: async_pt_3.a, async_pt_3.b, async_pt_3.c +(10 rows) + +ALTER SERVER loopback2 OPTIONS (ADD streaming_fetch 'true'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM async_pt; + QUERY PLAN +---------------------------------------------------------- + Append + -> Foreign Scan on public.async_p1 async_pt_1 + Output: async_pt_1.a, async_pt_1.b, async_pt_1.c + Remote SQL: SELECT a, b, c FROM public.base_tbl1 + Streaming Fetch: true + -> Foreign Scan on public.async_p2 async_pt_2 + Output: async_pt_2.a, async_pt_2.b, async_pt_2.c + Remote SQL: SELECT a, b, c FROM public.base_tbl2 + Streaming Fetch: true + -> Seq Scan on public.async_p3 async_pt_3 + Output: async_pt_3.a, async_pt_3.b, async_pt_3.c +(11 rows) + +ALTER FOREIGN TABLE async_p1 OPTIONS (DROP streaming_fetch); +ALTER SERVER loopback2 OPTIONS (DROP streaming_fetch); DROP TABLE local_tbl; DROP INDEX base_tbl1_idx; DROP INDEX base_tbl2_idx; diff --git a/contrib/postgres_fdw/expected/query_cancel.out b/contrib/postgres_fdw/expected/query_cancel.out index c3fc585d49f..ba976931eac 100644 --- a/contrib/postgres_fdw/expected/query_cancel.out +++ b/contrib/postgres_fdw/expected/query_cancel.out @@ -32,3 +32,24 @@ SET LOCAL statement_timeout = '10ms'; SELECT count(*) FROM ft1 a CROSS JOIN ft1 b CROSS JOIN ft1 c CROSS JOIN ft1 d; ERROR: canceling statement due to statement timeout COMMIT; +-- Same with streaming_fetch: canceling mid-stream must leave the connection +-- usable. +ALTER FOREIGN TABLE ft1 OPTIONS (ADD streaming_fetch 'true'); +BEGIN; +SELECT count(*) FROM ft1 a; + count +------- + 822 +(1 row) + +SET LOCAL statement_timeout = '10ms'; +SELECT count(*) FROM ft1 a CROSS JOIN ft1 b CROSS JOIN ft1 c CROSS JOIN ft1 d; +ERROR: canceling statement due to statement timeout +COMMIT; +SELECT count(*) FROM ft1; + count +------- + 822 +(1 row) + +ALTER FOREIGN TABLE ft1 OPTIONS (DROP streaming_fetch); diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index 5b539c4eeef..14e5c5cd32f 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -122,6 +122,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS) strcmp(def->defname, "parallel_abort") == 0 || strcmp(def->defname, "keep_connections") == 0 || strcmp(def->defname, "import_stats") == 0 || + strcmp(def->defname, "streaming_fetch") == 0 || strcmp(def->defname, "use_scram_passthrough") == 0) { /* these accept only boolean values */ @@ -262,6 +263,9 @@ InitPgFdwOptions(void) /* fetch_size is available on both server and table */ {"fetch_size", ForeignServerRelationId, false}, {"fetch_size", ForeignTableRelationId, false}, + /* streaming_fetch is available on both server and table */ + {"streaming_fetch", ForeignServerRelationId, false}, + {"streaming_fetch", ForeignTableRelationId, false}, /* batch_size is available on both server and table */ {"batch_size", ForeignServerRelationId, false}, {"batch_size", ForeignTableRelationId, false}, diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index afee95f9b76..2652395ef48 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -52,10 +52,12 @@ #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/rel.h" +#include "utils/resowner.h" #include "utils/sampling.h" #include "utils/selfuncs.h" #include "utils/timestamp.h" #include "utils/typcache.h" +#include "utils/tuplestore.h" PG_MODULE_MAGIC_EXT( .name = "postgres_fdw", @@ -86,6 +88,8 @@ enum FdwScanPrivateIndex FdwScanPrivateRetrievedAttrs, /* Integer representing the desired fetch_size */ FdwScanPrivateFetchSize, + /* Boolean indicating whether streaming_fetch mode is enabled */ + FdwScanPrivateStreamingFetch, /* * String describing join i.e. names of relations being joined and types @@ -169,7 +173,7 @@ enum FdwDirectModifyPrivateIndex /* * Execution state of a foreign scan using postgres_fdw. */ -typedef struct PgFdwScanState +struct PgFdwScanState { Relation rel; /* relcache entry for the foreign table. NULL * for a foreign join scan. */ @@ -207,7 +211,14 @@ typedef struct PgFdwScanState MemoryContext temp_cxt; /* context for per-tuple temporary data */ int fetch_size; /* number of tuples per fetch */ -} PgFdwScanState; + /* for streaming_fetch mode */ + bool streaming_fetch; /* use streaming_fetch mode? */ + Tuplestorestate *tuplestore; /* rows saved by drain_active_scan */ + bool tuplestore_lost_rows; /* rows lost by a failed drain */ + ForeignScanState *fsnode; /* back-pointer, for drain_active_scan */ + ResourceOwner resowner; /* owner of the tuplestore's temp files */ + TupleTableSlot *tuplestore_slot; /* slot for reading the tuplestore */ +}; /* * Execution state of a foreign insert/update/delete operation. @@ -520,6 +531,7 @@ static void estimate_path_cost_size(PlannerInfo *root, Cost *p_startup_cost, Cost *p_total_cost); static void get_remote_estimate(const char *sql, PGconn *conn, + PgFdwConnState *conn_state, double *rows, int *width, Cost *startup_cost, @@ -535,7 +547,9 @@ static void adjust_foreign_grouping_path_cost(PlannerInfo *root, static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, EquivalenceClass *ec, EquivalenceMember *em, void *arg); +static void prepare_query(ForeignScanState *node); static void create_cursor(ForeignScanState *node); +static void reset_batch_state(PgFdwScanState *fsstate); static void fetch_more_data(ForeignScanState *node); static void close_cursor(PGconn *conn, unsigned int cursor_number, PgFdwConnState *conn_state); @@ -601,8 +615,10 @@ static bool fetch_remote_statistics(Relation relation, RemoteStatsResults *remstats, RemoteAttributeMapping **p_remattrmap, int *p_attrcnt); -static PGresult *fetch_relstats(PGconn *conn, Relation relation); -static PGresult *fetch_attstats(PGconn *conn, int server_version_num, +static PGresult *fetch_relstats(PGconn *conn, PgFdwConnState *conn_state, + Relation relation); +static PGresult *fetch_attstats(PGconn *conn, PgFdwConnState *conn_state, + int server_version_num, const char *remote_schemaname, const char *remote_relname, const char *column_list); static RemoteAttributeMapping *build_remattrmap(Relation relation, List *va_cols, @@ -670,6 +686,16 @@ static void merge_fdw_options(PgFdwRelationInfo *fpinfo, const PgFdwRelationInfo *fpinfo_i); static int get_batch_size_option(Relation rel); +/* streaming_fetch mode support */ +static void init_scan(ForeignScanState *node); +static PGresult *fetch_stream_result(PgFdwScanState *fsstate); +static void discard_stream_result(PgFdwScanState *fsstate); +static void check_stream_rows_lost(PgFdwScanState *fsstate); +static void fetch_from_tuplestore(ForeignScanState *node); +static bool is_active_scan(PgFdwScanState *fsstate); +static bool scan_fetches_ctid(RelOptInfo *foreignrel, List *fdw_scan_tlist, + List *retrieved_attrs); +static bool scan_locks_rows(PlannerInfo *root, RelOptInfo *foreignrel); /* * Foreign-data wrapper handler function: return a struct with pointers @@ -779,9 +805,12 @@ postgresGetForeignRelSize(PlannerInfo *root, fpinfo->shippable_extensions = NIL; fpinfo->fetch_size = 100; fpinfo->async_capable = false; + fpinfo->streaming_fetch = false; apply_server_options(fpinfo); apply_table_options(fpinfo); + if (fpinfo->streaming_fetch) + fpinfo->async_capable = false; /* * If the table or the server is configured to use remote estimates, @@ -1372,6 +1401,7 @@ postgresGetForeignPlan(PlannerInfo *root, StringInfoData sql; bool has_final_sort = false; bool has_limit = false; + bool streaming_fetch; ListCell *lc; /* @@ -1537,13 +1567,23 @@ postgresGetForeignPlan(PlannerInfo *root, /* Remember remote_exprs for possible use by postgresPlanDirectModify */ fpinfo->final_remote_exprs = remote_exprs; + /* + * Use a cursor anyway for scans that fetch ctid (rows saved in a + * tuplestore lose it), or that lock rows (a streamed query locks every + * row up front, not just the ones fetched). + */ + streaming_fetch = fpinfo->streaming_fetch && + !scan_fetches_ctid(foreignrel, fdw_scan_tlist, retrieved_attrs) && + !scan_locks_rows(root, foreignrel); + /* * Build the fdw_private list that will be available to the executor. * Items in the list must match order in enum FdwScanPrivateIndex. */ - fdw_private = list_make3(makeString(sql.data), + fdw_private = list_make4(makeString(sql.data), retrieved_attrs, - makeInteger(fpinfo->fetch_size)); + makeInteger(fpinfo->fetch_size), + makeBoolean(streaming_fetch)); /* * Position FdwScanPrivateRelations: either the EXPLAIN relation string @@ -1770,6 +1810,8 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) FdwScanPrivateRetrievedAttrs); fsstate->fetch_size = intVal(list_nth(fsplan->fdw_private, FdwScanPrivateFetchSize)); + fsstate->streaming_fetch = boolVal(list_nth(fsplan->fdw_private, + FdwScanPrivateStreamingFetch)); /* Create contexts for batches of tuples and per-tuple temp workspace. */ fsstate->batch_cxt = AllocSetContextCreate(estate->es_query_cxt, @@ -1822,6 +1864,23 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Set the async-capable flag */ fsstate->async_capable = node->ss.ps.async_capable; + fsstate->tuplestore = NULL; + fsstate->tuplestore_lost_rows = false; + fsstate->fsnode = node; + + /* + * A drain may be triggered by another query; its temp files must belong + * to this scan's resource owner (e.g. the cursor's portal), not that one. + */ + fsstate->resowner = CurrentResourceOwner; + + /* Slot for reading back drained rows; freed with the executor state. */ + if (fsstate->streaming_fetch) + fsstate->tuplestore_slot = ExecInitExtraTupleSlot(estate, + fsstate->tupdesc, + &TTSOpsMinimalTuple); + else + fsstate->tuplestore_slot = NULL; } /* @@ -1842,7 +1901,12 @@ postgresIterateForeignScan(ForeignScanState *node) * first call after Begin or ReScan. */ if (!fsstate->scan_in_progress) - create_cursor(node); + { + if (fsstate->streaming_fetch) + init_scan(node); + else + create_cursor(node); + } /* * Get some more tuples, if we've run out. @@ -1870,6 +1934,75 @@ postgresIterateForeignScan(ForeignScanState *node) return slot; } +/* + * Reset a scan's batch bookkeeping so the next fetch starts fresh. The + * caller handles scan_in_progress. + */ +static void +reset_batch_state(PgFdwScanState *fsstate) +{ + fsstate->tuples = NULL; + fsstate->num_tuples = 0; + fsstate->next_tuple = 0; + fsstate->fetch_ct_2 = 0; + fsstate->eof_reached = false; +} + +/* + * Does the given foreign scan fetch the remote ctid? For a join or upper + * relation, retrieved_attrs indexes fdw_scan_tlist, so check that instead. + */ +static bool +scan_fetches_ctid(RelOptInfo *foreignrel, List *fdw_scan_tlist, + List *retrieved_attrs) +{ + ListCell *lc; + + if (IS_SIMPLE_REL(foreignrel)) + return list_member_int(retrieved_attrs, SelfItemPointerAttributeNumber); + + foreach(lc, fdw_scan_tlist) + { + TargetEntry *tle = lfirst_node(TargetEntry, lc); + + if (IsA(tle->expr, Var) && + ((Var *) tle->expr)->varattno == SelfItemPointerAttributeNumber) + return true; + } + + return false; +} + +/* + * Does the given foreign scan lock rows on the remote side? This must cover + * every case where deparseLockingClause() adds FOR UPDATE/SHARE; covering a + * few more (e.g. lower subquery rels) only costs an unneeded cursor. + */ +static bool +scan_locks_rows(PlannerInfo *root, RelOptInfo *foreignrel) +{ + int relid = -1; + + if (IS_UPPER_REL(foreignrel)) + return false; + + while ((relid = bms_next_member(foreignrel->relids, relid)) >= 0) + { + PlanRowMark *rc; + + if (bms_is_member(relid, root->all_result_relids) && + (root->parse->commandType == CMD_UPDATE || + root->parse->commandType == CMD_DELETE)) + return true; + + rc = get_plan_rowmark(root->rowMarks, relid); + if (rc && rc->strength != LCS_NONE) + return true; + } + + return false; +} + /* * postgresReScanForeignScan * Restart the scan. @@ -1880,6 +2013,7 @@ postgresReScanForeignScan(ForeignScanState *node) PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; char sql[64]; PGresult *res; + bool reinitialize_scan = false; /* If no scan is in progress, nothing to do. */ if (!fsstate->scan_in_progress) @@ -1903,40 +2037,67 @@ postgresReScanForeignScan(ForeignScanState *node) */ if (node->ss.ps.chgParam != NULL) { - fsstate->scan_in_progress = false; - snprintf(sql, sizeof(sql), "CLOSE c%u", - fsstate->cursor_number); + reinitialize_scan = true; } else if (fsstate->fetch_ct_2 > 1) { - if (PQserverVersion(fsstate->conn) < 150000) + if (!fsstate->streaming_fetch && + PQserverVersion(fsstate->conn) < 150000) + { snprintf(sql, sizeof(sql), "MOVE BACKWARD ALL IN c%u", fsstate->cursor_number); + res = pgfdw_exec_query(fsstate->conn, sql, fsstate->conn_state); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pgfdw_report_error(res, fsstate->conn, sql); + + PQclear(res); + + /* Now force a fresh FETCH. */ + reset_batch_state(fsstate); + return; + } + else + reinitialize_scan = true; + } + else + { + /* + * Easy: just rescan what we already have in memory, if anything. But + * rows saved in a tuplestore can't be rewound, so start over then. + */ + if (fsstate->streaming_fetch && fsstate->tuplestore) + reinitialize_scan = true; else { - fsstate->scan_in_progress = false; - snprintf(sql, sizeof(sql), "CLOSE c%u", - fsstate->cursor_number); + fsstate->next_tuple = 0; + return; } } - else + if (reinitialize_scan) { - /* Easy: just rescan what we already have in memory, if anything */ - fsstate->next_tuple = 0; - return; + if (fsstate->streaming_fetch) + { + discard_stream_result(fsstate); + if (fsstate->tuplestore) + { + tuplestore_end(fsstate->tuplestore); + fsstate->tuplestore = NULL; + fsstate->tuplestore_lost_rows = false; + } + } + else + { + snprintf(sql, sizeof(sql), "CLOSE c%u", + fsstate->cursor_number); + res = pgfdw_exec_query(fsstate->conn, sql, fsstate->conn_state); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pgfdw_report_error(res, fsstate->conn, sql); + PQclear(res); + } + /* Now force a fresh FETCH, and re-open the scan on the next iterate. */ + reset_batch_state(fsstate); + fsstate->scan_in_progress = false; } - - res = pgfdw_exec_query(fsstate->conn, sql, fsstate->conn_state); - if (PQresultStatus(res) != PGRES_COMMAND_OK) - pgfdw_report_error(res, fsstate->conn, sql); - PQclear(res); - - /* Now force a fresh FETCH. */ - fsstate->tuples = NULL; - fsstate->num_tuples = 0; - fsstate->next_tuple = 0; - fsstate->fetch_ct_2 = 0; - fsstate->eof_reached = false; } /* @@ -1954,9 +2115,19 @@ postgresEndForeignScan(ForeignScanState *node) /* Close the cursor if open, to prevent accumulation of cursors */ if (fsstate->scan_in_progress) - close_cursor(fsstate->conn, fsstate->cursor_number, - fsstate->conn_state); - + { + if (fsstate->streaming_fetch) + { + discard_stream_result(fsstate); + if (fsstate->tuplestore) + tuplestore_end(fsstate->tuplestore); + } + else + { + close_cursor(fsstate->conn, fsstate->cursor_number, + fsstate->conn_state); + } + } /* Release remote connection */ ReleaseConnection(fsstate->conn); fsstate->conn = NULL; @@ -3236,9 +3407,14 @@ postgresExplainForeignScan(ForeignScanState *node, ExplainState *es) if (es->verbose) { char *sql; + bool streaming_fetch; sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql)); + streaming_fetch = boolVal(list_nth(fdw_private, + FdwScanPrivateStreamingFetch)); ExplainPropertyText("Remote SQL", sql, es); + if (streaming_fetch) + ExplainPropertyBool("Streaming Fetch", true, es); } } @@ -3432,6 +3608,7 @@ estimate_path_cost_size(PlannerInfo *root, List *local_param_join_conds; StringInfoData sql; PGconn *conn; + PgFdwConnState *conn_state; Selectivity local_sel; QualCost local_cost; List *fdw_scan_tlist = NIL; @@ -3475,8 +3652,8 @@ estimate_path_cost_size(PlannerInfo *root, false, &retrieved_attrs, NULL); /* Get the remote estimate */ - conn = GetConnection(fpinfo->user, false, NULL); - get_remote_estimate(sql.data, conn, &rows, &width, + conn = GetConnection(fpinfo->user, false, &conn_state); + get_remote_estimate(sql.data, conn, conn_state, &rows, &width, &startup_cost, &total_cost); ReleaseConnection(conn); @@ -3922,7 +4099,7 @@ estimate_path_cost_size(PlannerInfo *root, * The given "sql" must be an EXPLAIN command. */ static void -get_remote_estimate(const char *sql, PGconn *conn, +get_remote_estimate(const char *sql, PGconn *conn, PgFdwConnState *conn_state, double *rows, int *width, Cost *startup_cost, Cost *total_cost) { @@ -3934,7 +4111,7 @@ get_remote_estimate(const char *sql, PGconn *conn, /* * Execute EXPLAIN remotely. */ - res = pgfdw_exec_query(conn, sql, NULL); + res = pgfdw_exec_query(conn, sql, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, sql); @@ -4041,23 +4218,24 @@ ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, } /* - * Create cursor for node's query with current parameter values. + * Work common to create_cursor() and init_scan(): make the connection idle, + * and convert this scan's parameter values to text. */ static void -create_cursor(ForeignScanState *node) +prepare_query(ForeignScanState *node) { PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; ExprContext *econtext = node->ss.ps.ps_ExprContext; int numParams = fsstate->numParams; const char **values = fsstate->param_values; - PGconn *conn = fsstate->conn; - StringInfoData buf; - PGresult *res; /* First, process a pending asynchronous request, if any. */ if (fsstate->conn_state->pendingAreq) process_pending_request(fsstate->conn_state->pendingAreq); + /* Likewise, finish any streaming_fetch scan using the connection. */ + drain_active_scan(fsstate->conn_state); + /* * 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 @@ -4076,6 +4254,22 @@ create_cursor(ForeignScanState *node) MemoryContextSwitchTo(oldcontext); } +} + +/* + * Create cursor for node's query with current parameter values. + */ +static void +create_cursor(ForeignScanState *node) +{ + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + PGconn *conn = fsstate->conn; + int numParams = fsstate->numParams; + const char **values = fsstate->param_values; + StringInfoData buf; + PGresult *res; + + prepare_query(node); /* Construct the DECLARE CURSOR command */ initStringInfo(&buf); @@ -4113,6 +4307,42 @@ create_cursor(ForeignScanState *node) pfree(buf.data); } +/* + * Start the remote query for a streaming_fetch scan in chunked-rows mode; + * the counterpart of create_cursor(). + */ +static void +init_scan(ForeignScanState *node) +{ + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + PGconn *conn = fsstate->conn; + + prepare_query(node); + + if (!PQsendQueryParams(conn, fsstate->query, fsstate->numParams, + NULL, fsstate->param_values, NULL, NULL, 0)) + pgfdw_report_error(NULL, conn, fsstate->query); + + /* Receive the rows in chunks of fetch_size */ + if (!PQsetChunkedRowsMode(conn, fsstate->fetch_size)) + { + pgfdw_cancel_query(fsstate->conn, fsstate->conn_state); + pgfdw_report_error(NULL, conn, fsstate->query); + } + + /* Mark the scan as started, and show no tuples have been retrieved */ + fsstate->scan_in_progress = true; + fsstate->tuples = NULL; + fsstate->num_tuples = 0; + fsstate->next_tuple = 0; + fsstate->fetch_ct_2 = 0; + fsstate->eof_reached = false; + fsstate->tuplestore_lost_rows = false; + + /* Until its result is consumed, this scan owns the connection */ + fsstate->conn_state->active_scan = fsstate; +} + /* * Fetch some more rows from the node's cursor. */ @@ -4126,14 +4356,18 @@ fetch_more_data(ForeignScanState *node) int i; MemoryContext oldcontext; - /* - * We'll store the tuples in the batch_cxt. First, flush the previous - * batch. - */ fsstate->tuples = NULL; MemoryContextReset(fsstate->batch_cxt); oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt); + /* + * Count every fill of tuples[], whichever branch below supplies it, so + * that rescan knows how to rewind. Counting a fetch that then fails is + * harmless, as the error prevents any rescan. + */ + if (fsstate->fetch_ct_2 < 2) + fsstate->fetch_ct_2++; + if (fsstate->async_capable) { Assert(fsstate->conn_state->pendingAreq); @@ -4150,7 +4384,7 @@ fetch_more_data(ForeignScanState *node) /* Reset per-connection state */ fsstate->conn_state->pendingAreq = NULL; } - else + else if (!fsstate->streaming_fetch) { char sql[64]; @@ -4163,7 +4397,37 @@ fetch_more_data(ForeignScanState *node) if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, fsstate->query); } + else if (fsstate->tuplestore) + { + /* Our remaining rows were saved locally by a drain */ + fetch_from_tuplestore(node); + MemoryContextSwitchTo(oldcontext); + return; + } + else + { + /* Don't read past rows that a failed drain consumed but never saved */ + check_stream_rows_lost(fsstate); + + /* + * Not at EOF, but no longer the active scan: abort cleanup (e.g. + * ROLLBACK TO SAVEPOINT) has discarded our result. + */ + if (!is_active_scan(fsstate)) + ereport(ERROR, + errcode(ERRCODE_INVALID_CURSOR_STATE), + errmsg("could not obtain remaining rows from remote server"), + errdetail("The remote query was canceled when a subtransaction was rolled back.")); + res = fetch_stream_result(fsstate); + if (res == NULL) + { + /* Nothing is saved locally either, so this is really EOF */ + fsstate->eof_reached = true; + MemoryContextSwitchTo(oldcontext); + return; + } + } /* Convert the data into HeapTuples */ numrows = PQntuples(res); fsstate->tuples = palloc0_array(HeapTuple, numrows); @@ -4183,18 +4447,224 @@ fetch_more_data(ForeignScanState *node) fsstate->temp_cxt); } - /* Update fetch_ct_2 */ - if (fsstate->fetch_ct_2 < 2) - fsstate->fetch_ct_2++; - /* Must be EOF if we didn't get as many tuples as we asked for. */ fsstate->eof_reached = (numrows < fsstate->fetch_size); PQclear(res); + /* + * After a partial chunk, only the end of the stream is left; consume it + * now so the connection is free. (Any unexpected rows would be saved.) + */ + if (fsstate->streaming_fetch && fsstate->eof_reached) + drain_active_scan(fsstate->conn_state); + MemoryContextSwitchTo(oldcontext); } +/* + * Get the next chunked-rows-mode result for a streaming_fetch scan. + * Returns the result to convert into tuples, or NULL once the end-of-stream + * marker has been consumed and active_scan cleared. + */ +static PGresult * +fetch_stream_result(PgFdwScanState *fsstate) +{ + PGconn *conn = fsstate->conn; + PGresult *res; + + res = pgfdw_get_next_result(conn); + + if (!res || PQresultStatus(res) == PGRES_FATAL_ERROR) + pgfdw_report_error(res, conn, fsstate->query); + + /* + * PGRES_TUPLES_OK ends the stream: consume the final NULL and clear + * active_scan. The caller sets eof_reached, since rows it has already + * saved in a tuplestore may still be unread. + */ + if (PQresultStatus(res) == PGRES_TUPLES_OK) + { + Assert(PQntuples(res) == 0); + PQclear(res); + res = pgfdw_get_next_result(conn); + if (res != NULL) + pgfdw_report_error(res, conn, fsstate->query); + fsstate->conn_state->active_scan = NULL; + return NULL; + } + + if (PQresultStatus(res) != PGRES_TUPLES_CHUNK) + pgfdw_report_error(res, conn, fsstate->query); + + return res; +} + +/* + * Consume and discard every remaining chunked-rows result for this scan. + * Used on rescan or early end of scan. We don't cancel the query instead, + * since that would abort the remote transaction. + */ +static void +discard_stream_result(PgFdwScanState *fsstate) +{ + PGresult *res; + + /* Nothing to do if the stream was already consumed */ + if (!is_active_scan(fsstate)) + return; + + for (;;) + { + CHECK_FOR_INTERRUPTS(); + + res = fetch_stream_result(fsstate); + if (res == NULL) + break; + PQclear(res); + } +} + +/* + * Error out if a failed drain may have lost some of this scan's rows, rather + * than silently return short results. + */ +static void +check_stream_rows_lost(PgFdwScanState *fsstate) +{ + if (fsstate->tuplestore_lost_rows) + ereport(ERROR, + errcode(ERRCODE_INVALID_CURSOR_STATE), + errmsg("could not obtain all rows from remote server"), + errdetail("An earlier error interrupted reading the remote query's result.")); +} + +/* + * Fetch the next batch of up to fetch_size rows from the scan's tuplestore, + * releasing the tuplestore once it is empty. + */ +static void +fetch_from_tuplestore(ForeignScanState *node) +{ + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + TupleTableSlot *slot = fsstate->tuplestore_slot; + int numrows = 0; + + check_stream_rows_lost(fsstate); + + /* ExecFetchSlotHeapTuple() copies each tuple into batch_cxt */ + fsstate->tuples = palloc0_array(HeapTuple, fsstate->fetch_size); + while (numrows < fsstate->fetch_size && + tuplestore_gettupleslot(fsstate->tuplestore, true, false, slot)) + { + fsstate->tuples[numrows++] = ExecFetchSlotHeapTuple(slot, false, NULL); + ExecClearTuple(slot); + } + fsstate->num_tuples = numrows; + fsstate->next_tuple = 0; + + /* Must be EOF if we didn't get as many tuples as we asked for. */ + fsstate->eof_reached = (numrows < fsstate->fetch_size); + + if (fsstate->eof_reached) + { + tuplestore_end(fsstate->tuplestore); + fsstate->tuplestore = NULL; + } +} + +/* + * A connection can carry only one query result at a time. If a + * streaming_fetch scan's result is still being read, save the rest of it in + * the scan's tuplestore, so that the connection can be used for something + * else. Every path that sends a query must call this first. + */ +void +drain_active_scan(PgFdwConnState *conn_state) +{ + PgFdwScanState *active_fsstate = conn_state->active_scan; + MemoryContext oldcontext; + ResourceOwner oldowner; + EState *estate; + bool rows_already_lost; + + if (!active_fsstate) + return; + + /* Rows lost by an earlier failed drain stay lost */ + rows_already_lost = active_fsstate->tuplestore_lost_rows; + + /* The tuplestore and its temp files must last as long as the scan */ + estate = active_fsstate->fsnode->ss.ps.state; + oldcontext = MemoryContextSwitchTo(estate->es_query_cxt); + oldowner = CurrentResourceOwner; + CurrentResourceOwner = active_fsstate->resowner; + + /* Save each chunk until fetch_stream_result() reports the end */ + for (;;) + { + PGresult *res; + int numrows; + int i; + + CHECK_FOR_INTERRUPTS(); + + /* + * An error before this chunk is fully saved loses rows, and may abort + * only a subtransaction, leaving the scan readable at an outer level. + * Flag that until the chunk is stored. + */ + active_fsstate->tuplestore_lost_rows = true; + + res = fetch_stream_result(active_fsstate); + if (res == NULL) + { + active_fsstate->tuplestore_lost_rows = rows_already_lost; + break; + } + + if (active_fsstate->tuplestore == NULL) + active_fsstate->tuplestore = tuplestore_begin_heap(false, false, + work_mem); + + numrows = PQntuples(res); + + /* Convert the data into HeapTuples */ + for (i = 0; i < numrows; i++) + { + HeapTuple temp_tuple; + + temp_tuple = make_tuple_from_result_row(res, i, + active_fsstate->rel, + active_fsstate->attinmeta, + active_fsstate->retrieved_attrs, + active_fsstate->fsnode, + active_fsstate->temp_cxt); + tuplestore_puttuple(active_fsstate->tuplestore, temp_tuple); + heap_freetuple(temp_tuple); + } + PQclear(res); + active_fsstate->tuplestore_lost_rows = rows_already_lost; + } + + /* The scan is at EOF only if nothing was left to save */ + active_fsstate->eof_reached = (active_fsstate->tuplestore == NULL); + + CurrentResourceOwner = oldowner; + MemoryContextSwitchTo(oldcontext); +} + +/* + * Return true if fsstate is the connection's currently active + * streaming_fetch scan. + */ +static bool +is_active_scan(PgFdwScanState *fsstate) +{ + return fsstate->streaming_fetch && + fsstate->conn_state->active_scan == fsstate; +} + /* * Force assorted GUC parameters to settings that ensure that we'll output * data values in a form that is unambiguous to the remote server. @@ -4419,6 +4889,9 @@ execute_foreign_modify(EState *estate, if (fmstate->conn_state->pendingAreq) process_pending_request(fmstate->conn_state->pendingAreq); + /* Likewise, finish any streaming_fetch scan using the connection. */ + drain_active_scan(fmstate->conn_state); + /* * If the existing query was deparsed and prepared for a different number * of rows, rebuild it for the proper number. @@ -4845,6 +5318,9 @@ execute_dml_stmt(ForeignScanState *node) if (dmstate->conn_state->pendingAreq) process_pending_request(dmstate->conn_state->pendingAreq); + /* Likewise, finish any streaming_fetch scan using the connection. */ + drain_active_scan(dmstate->conn_state); + /* * Construct array of query parameter values in text format. */ @@ -5215,6 +5691,7 @@ postgresAnalyzeForeignTable(Relation relation, ForeignTable *table; UserMapping *user; PGconn *conn; + PgFdwConnState *conn_state; StringInfoData sql; PGresult *res; @@ -5234,7 +5711,7 @@ postgresAnalyzeForeignTable(Relation relation, */ table = GetForeignTable(RelationGetRelid(relation)); user = GetUserMapping(relation->rd_rel->relowner, table->serverid); - conn = GetConnection(user, false, NULL); + conn = GetConnection(user, false, &conn_state); /* * Construct command to get page count for relation. @@ -5242,7 +5719,7 @@ postgresAnalyzeForeignTable(Relation relation, initStringInfo(&sql); deparseAnalyzeSizeSql(&sql, relation); - res = pgfdw_exec_query(conn, sql.data, NULL); + res = pgfdw_exec_query(conn, sql.data, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, sql.data); @@ -5269,6 +5746,7 @@ postgresGetAnalyzeInfoForForeignTable(Relation relation, bool *can_tablesample) ForeignTable *table; UserMapping *user; PGconn *conn; + PgFdwConnState *conn_state; StringInfoData sql; PGresult *res; double reltuples; @@ -5283,7 +5761,7 @@ postgresGetAnalyzeInfoForForeignTable(Relation relation, bool *can_tablesample) */ table = GetForeignTable(RelationGetRelid(relation)); user = GetUserMapping(relation->rd_rel->relowner, table->serverid); - conn = GetConnection(user, false, NULL); + conn = GetConnection(user, false, &conn_state); /* * Construct command to get page count for relation. @@ -5291,7 +5769,7 @@ postgresGetAnalyzeInfoForForeignTable(Relation relation, bool *can_tablesample) initStringInfo(&sql); deparseAnalyzeInfoSql(&sql, relation); - res = pgfdw_exec_query(conn, sql.data, NULL); + res = pgfdw_exec_query(conn, sql.data, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, sql.data); @@ -5337,6 +5815,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, ForeignServer *server; UserMapping *user; PGconn *conn; + PgFdwConnState *conn_state; int server_version_num; PgFdwSamplingMethod method = ANALYZE_SAMPLE_AUTO; /* auto is default */ double sample_frac = -1.0; @@ -5372,7 +5851,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, table = GetForeignTable(RelationGetRelid(relation)); server = GetForeignServer(table->serverid); user = GetUserMapping(relation->rd_rel->relowner, table->serverid); - conn = GetConnection(user, false, NULL); + conn = GetConnection(user, false, &conn_state); /* We'll need server version, so fetch it now. */ server_version_num = PQserverVersion(conn); @@ -5522,7 +6001,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, deparseAnalyzeSql(&sql, relation, method, sample_frac, &astate.retrieved_attrs); - res = pgfdw_exec_query(conn, sql.data, NULL); + res = pgfdw_exec_query(conn, sql.data, conn_state); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(res, conn, sql.data); PQclear(res); @@ -5573,7 +6052,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, */ /* Fetch some rows */ - res = pgfdw_exec_query(conn, fetch_sql, NULL); + res = pgfdw_exec_query(conn, fetch_sql, conn_state); /* On error, report the original query, not the FETCH. */ if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, sql.data); @@ -5591,7 +6070,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, } /* Close the cursor, just to be tidy. */ - close_cursor(conn, cursor_number, NULL); + close_cursor(conn, cursor_number, conn_state); ReleaseConnection(conn); @@ -5807,6 +6286,7 @@ fetch_remote_statistics(Relation relation, const char *remote_relname = NULL; UserMapping *user; PGconn *conn; + PgFdwConnState *conn_state; PGresult *relstats = NULL; PGresult *attstats = NULL; int server_version_num; @@ -5838,11 +6318,11 @@ fetch_remote_statistics(Relation relation, * from ImportForeignStatistics(). */ user = GetUserMapping(GetUserId(), table->serverid); - conn = GetConnection(user, false, NULL); + conn = GetConnection(user, false, &conn_state); remstats->version = server_version_num = PQserverVersion(conn); /* Fetch relation stats. */ - remstats->rel = relstats = fetch_relstats(conn, relation); + remstats->rel = relstats = fetch_relstats(conn, conn_state, relation); /* * Verify that the remote table is the sort that can have meaningful stats @@ -5928,7 +6408,7 @@ fetch_remote_statistics(Relation relation, } /* Fetch attribute stats. */ - remstats->att = attstats = fetch_attstats(conn, + remstats->att = attstats = fetch_attstats(conn, conn_state, server_version_num, remote_schemaname, remote_relname, @@ -5976,7 +6456,8 @@ fetch_cleanup: * Attempt to fetch remote relation stats. */ static PGresult * -fetch_relstats(PGconn *conn, Relation relation) +fetch_relstats(PGconn *conn, PgFdwConnState *conn_state, + Relation relation) { StringInfoData sql; PGresult *res; @@ -5984,7 +6465,7 @@ fetch_relstats(PGconn *conn, Relation relation) initStringInfo(&sql); deparseAnalyzeInfoSql(&sql, relation); - res = pgfdw_exec_query(conn, sql.data, NULL); + res = pgfdw_exec_query(conn, sql.data, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, sql.data); @@ -5998,7 +6479,8 @@ fetch_relstats(PGconn *conn, Relation relation) * Attempt to fetch remote attribute stats. */ static PGresult * -fetch_attstats(PGconn *conn, int server_version_num, +fetch_attstats(PGconn *conn, PgFdwConnState *conn_state, + int server_version_num, const char *remote_schemaname, const char *remote_relname, const char *column_list) { @@ -6072,7 +6554,7 @@ fetch_attstats(PGconn *conn, int server_version_num, appendStringInfoString(&sql, " ORDER BY attname::text COLLATE \"C\", inherited DESC"); - res = pgfdw_exec_query(conn, sql.data, NULL); + res = pgfdw_exec_query(conn, sql.data, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, sql.data); @@ -6526,6 +7008,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) ForeignServer *server; UserMapping *mapping; PGconn *conn; + PgFdwConnState *conn_state; StringInfoData buf; PGresult *res; int numrows, @@ -6557,7 +7040,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) */ server = GetForeignServer(serverOid); mapping = GetUserMapping(GetUserId(), server->serverid); - conn = GetConnection(mapping, false, NULL); + conn = GetConnection(mapping, false, &conn_state); /* Don't attempt to import collation if remote server hasn't got it */ if (PQserverVersion(conn) < 90100) @@ -6570,7 +7053,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) appendStringInfoString(&buf, "SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = "); deparseStringLiteral(&buf, stmt->remote_schema); - res = pgfdw_exec_query(conn, buf.data, NULL); + res = pgfdw_exec_query(conn, buf.data, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, buf.data); @@ -6684,7 +7167,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) appendStringInfoString(&buf, " ORDER BY c.relname, a.attnum"); /* Fetch the data */ - res = pgfdw_exec_query(conn, buf.data, NULL); + res = pgfdw_exec_query(conn, buf.data, conn_state); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, buf.data); @@ -7601,6 +8084,8 @@ apply_server_options(PgFdwRelationInfo *fpinfo) (void) parse_int(defGetString(def), &fpinfo->fetch_size, 0, NULL); else if (strcmp(def->defname, "async_capable") == 0) fpinfo->async_capable = defGetBoolean(def); + else if (strcmp(def->defname, "streaming_fetch") == 0) + fpinfo->streaming_fetch = defGetBoolean(def); } } @@ -7624,6 +8109,8 @@ apply_table_options(PgFdwRelationInfo *fpinfo) (void) parse_int(defGetString(def), &fpinfo->fetch_size, 0, NULL); else if (strcmp(def->defname, "async_capable") == 0) fpinfo->async_capable = defGetBoolean(def); + else if (strcmp(def->defname, "streaming_fetch") == 0) + fpinfo->streaming_fetch = defGetBoolean(def); } } @@ -7659,6 +8146,7 @@ merge_fdw_options(PgFdwRelationInfo *fpinfo, fpinfo->use_remote_estimate = fpinfo_o->use_remote_estimate; fpinfo->fetch_size = fpinfo_o->fetch_size; fpinfo->async_capable = fpinfo_o->async_capable; + fpinfo->streaming_fetch = fpinfo_o->streaming_fetch; /* Merge the table level options from either side of the join. */ if (fpinfo_i) @@ -7690,6 +8178,11 @@ merge_fdw_options(PgFdwRelationInfo *fpinfo, */ fpinfo->async_capable = fpinfo_o->async_capable || fpinfo_i->async_capable; + fpinfo->streaming_fetch = fpinfo_o->streaming_fetch || + fpinfo_i->streaming_fetch; + /* streaming_fetch and async execution are mutually exclusive */ + if (fpinfo->streaming_fetch) + fpinfo->async_capable = false; } } @@ -8844,6 +9337,9 @@ fetch_more_data_begin(AsyncRequest *areq) if (!fsstate->scan_in_progress) create_cursor(node); + /* PQsendQuery() below doesn't drain a streaming_fetch scan; do it here */ + drain_active_scan(fsstate->conn_state); + /* We will send this query, but not wait for the response. */ snprintf(sql, sizeof(sql), "FETCH %d FROM c%u", fsstate->fetch_size, fsstate->cursor_number); diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index da7da1c2ea9..60c23611e74 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -88,6 +88,7 @@ typedef struct PgFdwRelationInfo UserMapping *user; /* only set in use_remote_estimate mode */ int fetch_size; /* fetch size for this remote table */ + bool streaming_fetch; /* true if cursor-free fetch is enabled */ /* * Name of the relation, for use while EXPLAINing ForeignScan. It is used @@ -141,12 +142,16 @@ typedef struct PgFdwRelationInfo int relation_index; } PgFdwRelationInfo; +typedef struct PgFdwScanState PgFdwScanState; + /* * Extra control information relating to a connection. */ typedef struct PgFdwConnState { AsyncRequest *pendingAreq; /* pending async request */ + PgFdwScanState *active_scan; /* the streaming_fetch scan, if any, that + * currently has an unconsumed result */ } PgFdwConnState; /* @@ -165,6 +170,7 @@ 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 drain_active_scan(PgFdwConnState *conn_state); /* in connection.c */ extern PGconn *GetConnection(UserMapping *user, bool will_prep_stmt, @@ -174,8 +180,10 @@ extern unsigned int GetCursorNumber(PGconn *conn); extern unsigned int GetPrepStmtNumber(PGconn *conn); extern void do_sql_command(PGconn *conn, const char *sql); extern PGresult *pgfdw_get_result(PGconn *conn); +extern PGresult *pgfdw_get_next_result(PGconn *conn); extern PGresult *pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state); +extern bool pgfdw_cancel_query(PGconn *conn, PgFdwConnState *state); pg_noreturn extern void pgfdw_report_error(PGresult *res, PGconn *conn, const char *sql); extern void pgfdw_report(int elevel, PGresult *res, PGconn *conn, diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 77c9a25a890..b0dca68d060 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -244,6 +244,26 @@ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1; -- should work again ANALYZE ft1; ALTER FOREIGN TABLE ft2 OPTIONS (use_remote_estimate 'true'); +-- =================================================================== +-- test streaming_fetch option +-- =================================================================== +CREATE SERVER fetch_stream_srv FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (streaming_fetch 'true'); +CREATE USER MAPPING FOR CURRENT_USER SERVER fetch_stream_srv; +CREATE FOREIGN TABLE ft_stream (c1 int) SERVER fetch_stream_srv; +-- Set on the server; EXPLAIN VERBOSE shows "Streaming Fetch: true". +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft_stream; +-- A table-level setting overrides the server's, in either direction. +ALTER FOREIGN TABLE ft_stream OPTIONS (ADD streaming_fetch 'false'); +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft_stream; +ALTER SERVER fetch_stream_srv OPTIONS (SET streaming_fetch 'false'); +ALTER FOREIGN TABLE ft_stream OPTIONS (SET streaming_fetch 'true'); +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft_stream; +ALTER FOREIGN TABLE ft_stream OPTIONS (SET streaming_fetch 'yes'); -- ERROR +DROP FOREIGN TABLE ft_stream; +DROP USER MAPPING FOR CURRENT_USER SERVER fetch_stream_srv; +DROP SERVER fetch_stream_srv; + -- =================================================================== -- test subscription -- =================================================================== @@ -363,6 +383,292 @@ EXPLAIN (VERBOSE, COSTS OFF) WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); SELECT * FROM ft2 a, ft2 b WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); + +-- =================================================================== +-- streaming_fetch scans sharing a connection +-- =================================================================== +-- ft1_s and ft2_s stream the same remote table as ft1 and ft2, and all four +-- share the loopback connection. ft2_s gets its rows in chunks of 5. +CREATE FOREIGN TABLE ft1_s ( + c1 int NOT NULL, + c2 int NOT NULL, + c3 text, + c4 timestamptz, + c5 timestamp, + c6 varchar(10), + c7 char(10) default 'ft1', + c8 user_enum +) SERVER loopback OPTIONS (schema_name 'S 1', table_name 'T 1', + streaming_fetch 'true'); +ALTER FOREIGN TABLE ft1_s ALTER COLUMN c1 OPTIONS (column_name 'C 1'); +CREATE FOREIGN TABLE ft2_s ( + c1 int NOT NULL, + c2 int NOT NULL, + c3 text, + c4 timestamptz, + c5 timestamp, + c6 varchar(10), + c7 char(10) default 'ft2', + c8 user_enum +) SERVER loopback OPTIONS (schema_name 'S 1', table_name 'T 1', + streaming_fetch 'true', fetch_size '5', + use_remote_estimate 'true'); +ALTER FOREIGN TABLE ft2_s ALTER COLUMN c1 OPTIONS (column_name 'C 1'); + +-- One row per chunk, and chunks of 5 + 5 + 2. The c8 filter can't be pushed +-- down, so the aggregate runs locally over the streamed rows. +ALTER FOREIGN TABLE ft1_s OPTIONS (ADD fetch_size '1'); +SELECT array_agg(c1 ORDER BY c1) FROM ft1_s WHERE c1 <= 12 AND c8 = 'foo'; +ALTER FOREIGN TABLE ft1_s OPTIONS (DROP fetch_size); +SELECT array_agg(c1 ORDER BY c1) FROM ft2_s WHERE c1 <= 12 AND c8 = 'foo'; + +-- Streaming and cursor scans in one query, and two streaming scans at once. +SELECT * FROM ft1_s t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; +SELECT * FROM ft1_s t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2_s t2) ORDER BY c1; +SELECT t1.c1, t2."C 1" FROM ft2_s t1 JOIN "S 1"."T 1" t2 ON (t1.c1 = t2."C 1") OFFSET 100 LIMIT 10; +SELECT t1.c1 FROM ft1_s t1 WHERE NOT EXISTS (SELECT 1 FROM ft2 t2 WHERE t1.c1 = t2.c2) ORDER BY t1.c1 OFFSET 100 LIMIT 10; +SELECT t1.c1 FROM ft1_s t1 WHERE NOT EXISTS (SELECT 1 FROM ft2_s t2 WHERE t1.c1 = t2.c2) ORDER BY t1.c1 OFFSET 100 LIMIT 10; + +-- Nested loops: each time the inner scan starts, the rest of the outer +-- scan's result is saved locally, and later read back in fetch_size batches. +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM ft2_s a, ft2_s b +WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); +SELECT count(*), sum(b.c1) FROM ft2_s a, ft2_s b +WHERE a.c2 = 6 AND b.c1 = a.c1 AND a.c8 = 'foo' AND b.c7 = upper(a.c7); +SELECT count(*), sum(c.c1) FROM ft2_s a, ft2_s b, ft2_s c +WHERE a.c2 = 6 AND b.c1 = a.c1 AND c.c1 = b.c1 +AND a.c8 = 'foo' AND b.c7 = upper(a.c7); + +-- An EXISTS subquery stops its streaming scan at the first match, leaving +-- the rest of its result unread. The connection must still work for the +-- outer cursor's later FETCHes (ft1 needs several with fetch_size 5), for a +-- per-row UPDATE, and for a pushed-down UPDATE on ft4. +CREATE FUNCTION direct_modify_ft4(id int) RETURNS void AS $$ + UPDATE ft4 SET c3 = c3 WHERE c1 = id; +$$ LANGUAGE sql; +ALTER FOREIGN TABLE ft1 OPTIONS (ADD fetch_size '5'); +SET enable_hashagg = off; +SET enable_hashjoin = off; +SET enable_mergejoin = off; +SET enable_material = off; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT a.c1, a.c2 FROM ft1 a +WHERE a.c1 < 20 AND a.c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = a.c2) +ORDER BY a.c1; +SELECT a.c1, a.c2 FROM ft1 a +WHERE a.c1 < 20 AND a.c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = a.c2) +ORDER BY a.c1; +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE ft1 SET c3 = c3 +WHERE c1 < 20 AND c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = ft1.c2); +EXPLAIN (VERBOSE, COSTS OFF) UPDATE ft4 SET c3 = c3 WHERE c1 = 1; +BEGIN; +WITH u AS (UPDATE ft1 SET c3 = c3 + WHERE c1 < 20 AND c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = ft1.c2) + RETURNING c1) +SELECT count(*) FROM u; +SELECT count(*) FROM (SELECT direct_modify_ft4(a.c1) FROM ft1 a + WHERE a.c1 < 20 AND a.c8 = 'foo' AND EXISTS (SELECT 1 FROM ft2_s b WHERE b.c2 = a.c2) + OFFSET 0) s; +ROLLBACK; +RESET enable_hashagg; +RESET enable_hashjoin; +RESET enable_mergejoin; +RESET enable_material; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP fetch_size); +DROP FUNCTION direct_modify_ft4(int); + +-- A LIMIT that can't be pushed down stops the scan early; the rest of the +-- remote result is read and discarded. The second query stops partway +-- through a chunk. +SELECT c1 FROM ft1_s WHERE c8 = 'foo' ORDER BY c1 LIMIT 5; +SELECT c1 FROM ft2_s WHERE c8 = 'foo' ORDER BY c1 LIMIT 7; +SELECT count(*) FROM ft1_s; + +-- Scans that fetch ctid or lock rows use a cursor (no "Streaming Fetch"). +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft2_s WHERE c1 = 1 FOR UPDATE; +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft2_s WHERE c1 = 1 FOR SHARE; +EXPLAIN (VERBOSE, COSTS OFF) DELETE FROM ft2_s WHERE c1 = 1 AND random() >= 0; +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE ft2_s f1 SET c3 = f2.c3 FROM ft1_s f2 +WHERE f1.c1 = f2.c1 AND f1.c1 = 3 AND random() >= 0; +-- So a non-direct UPDATE whose rows arrive in several chunks updates them all. +BEGIN; +WITH u AS (UPDATE ft2_s SET c3 = c3 WHERE c1 <= 20 AND random() >= 0 RETURNING 1) +SELECT count(*) FROM u; +ROLLBACK; + +DROP FOREIGN TABLE ft1_s, ft2_s; + +-- If saving a scan's rows locally fails partway (row 6 can't be converted) +-- inside a subtransaction, the scan must fail at the outer level rather than +-- return short results, even if a later attempt to save the rest succeeds. +CREATE TABLE loct_lost (id int, x text); +INSERT INTO loct_lost SELECT g, g::text FROM generate_series(1, 5) g; +INSERT INTO loct_lost VALUES (6, 'not_an_int'), (7, '7'), (8, '8'); + +CREATE FOREIGN TABLE ft_lost (id int, x int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_lost', + streaming_fetch 'true', fetch_size '2'); +CREATE FOREIGN TABLE ft_lost_other (id int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_lost'); + +\set VERBOSITY terse +BEGIN; +DECLARE c_lost CURSOR FOR SELECT id, x FROM ft_lost ORDER BY id; +FETCH 2 FROM c_lost; +SAVEPOINT sp; +SELECT id FROM ft_lost_other ORDER BY id LIMIT 1; -- hits row 6 +ROLLBACK TO SAVEPOINT sp; +FETCH ALL FROM c_lost; -- ERROR +COMMIT; +BEGIN; +DECLARE c_lost CURSOR FOR SELECT id, x FROM ft_lost ORDER BY id; +FETCH 2 FROM c_lost; +SAVEPOINT sp; +SELECT count(*) FROM ft_lost_other; -- hits row 6 +ROLLBACK TO SAVEPOINT sp; +SELECT count(*) FROM ft_lost_other; -- saves rows 7 and 8 +FETCH ALL FROM c_lost; -- ERROR +COMMIT; +SELECT count(*) FROM ft_lost_other; -- connection still works +\set VERBOSITY default + +DROP FOREIGN TABLE ft_lost, ft_lost_other; +DROP TABLE loct_lost; + +-- Transaction control with a cursor whose remote query is still streaming +-- when postgres_fdw sends RELEASE SAVEPOINT, ROLLBACK TO SAVEPOINT or +-- COMMIT. Each case must return every row or fail cleanly, and leave the +-- connection usable. +CREATE TABLE loct_xact (id int, x text); +INSERT INTO loct_xact SELECT g, g::text FROM generate_series(1, 10) g; +CREATE FOREIGN TABLE ft_xact (id int, x int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_xact', + streaming_fetch 'true', fetch_size '2'); +CREATE FOREIGN TABLE ft_xact_other (id int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_xact'); + +-- Cursor opened in a subtransaction that is then released. +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +RELEASE SAVEPOINT s; +FETCH ALL FROM c; +COMMIT; + +-- Cursor declared outside a subtransaction, but first read inside it. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +SAVEPOINT s; +SELECT count(*) FROM ft_xact_other; +FETCH 3 FROM c; +RELEASE SAVEPOINT s; +FETCH ALL FROM c; +COMMIT; + +-- Nested savepoints, released one at a time; parallel_commit sends each +-- RELEASE SAVEPOINT asynchronously. +ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true'); +BEGIN; +SAVEPOINT s1; +SAVEPOINT s2; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +RELEASE SAVEPOINT s2; +FETCH 2 FROM c; +RELEASE SAVEPOINT s1; +FETCH ALL FROM c; +COMMIT; +ALTER SERVER loopback OPTIONS (DROP parallel_commit); + +-- Rolling back the subtransaction in which the cursor was first read +-- cancels the remote query, so the cursor must report an error. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +SAVEPOINT s; +SELECT count(*) FROM ft_xact_other; +FETCH 3 FROM c; +ROLLBACK TO SAVEPOINT s; +FETCH ALL FROM c; -- ERROR +COMMIT; +SELECT count(*) FROM ft_xact_other; -- connection still works + +-- Cursor opened in a subtransaction that is rolled back goes away with it. +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +ROLLBACK TO SAVEPOINT s; +SELECT count(*) FROM ft_xact_other; +COMMIT; + +-- Saving the rest of the cursor's rows at RELEASE SAVEPOINT fails on a bad +-- row; the connection must work again after rollback. +INSERT INTO loct_xact VALUES (11, 'not_an_int'); +\set VERBOSITY terse +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT id, x FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +RELEASE SAVEPOINT s; -- ERROR +ROLLBACK; +\set VERBOSITY default +SELECT count(*) FROM ft_xact_other; -- connection still works +DELETE FROM loct_xact WHERE id = 11; + +-- Top-level COMMIT closes the cursor before the remote COMMIT. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +COMMIT; +SELECT count(*) FROM ft_xact_other; + +-- A holdable cursor is read to the end at COMMIT and keeps all its rows. +BEGIN; +DECLARE c CURSOR WITH HOLD FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +COMMIT; +FETCH ALL FROM c; +CLOSE c; + +-- Top-level ROLLBACK. +BEGIN; +DECLARE c CURSOR FOR SELECT id FROM ft_xact ORDER BY id; +FETCH 3 FROM c; +ROLLBACK; +SELECT count(*) FROM ft_xact_other; + +DROP FOREIGN TABLE ft_xact, ft_xact_other; +DROP TABLE loct_xact; + +-- Rows saved locally on behalf of a cursor may spill to a temp file. That +-- file belongs to the cursor, so it must survive the rollback of the +-- subtransaction in which it was written. +CREATE TABLE loct_spill (id int, pad text); +INSERT INTO loct_spill SELECT g, repeat('x', 100) FROM generate_series(1, 5000) g; +CREATE FOREIGN TABLE ft_spill (id int, pad text) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_spill', + streaming_fetch 'true', fetch_size '10'); +CREATE FOREIGN TABLE ft_spill_other (id int) + SERVER loopback OPTIONS (schema_name 'public', table_name 'loct_spill'); + +BEGIN; +SET LOCAL work_mem = '64kB'; +DECLARE c CURSOR FOR SELECT id, length(pad) FROM ft_spill ORDER BY id; +FETCH 1 FROM c; +SAVEPOINT s; +SELECT count(*) FROM ft_spill_other; -- saves c's rows, spilling to disk +ROLLBACK TO SAVEPOINT s; +MOVE 4995 FROM c; +FETCH ALL FROM c; +COMMIT; + +DROP FOREIGN TABLE ft_spill, ft_spill_other; +DROP TABLE loct_spill; + -- bug before 9.3.5 due to sloppy handling of remote-estimate parameters SELECT * FROM ft1 WHERE c1 = ANY (ARRAY(SELECT c1 FROM ft2 WHERE c1 < 5)); SELECT * FROM ft2 WHERE c1 = ANY (ARRAY(SELECT c1 FROM ft1 WHERE c1 < 5)); @@ -3856,6 +4162,31 @@ SELECT 1 FROM ft1 LIMIT 1; -- should fail \set VERBOSITY default COMMIT; +-- =================================================================== +-- streaming_fetch: reconnecting after the remote backend terminates +-- =================================================================== +ALTER FOREIGN TABLE ft1 OPTIONS (ADD streaming_fetch 'true'); +SELECT 1 FROM ft1 LIMIT 1; -- open the connection +DO $$ BEGIN +PERFORM pg_terminate_backend(pid, 180000) FROM pg_stat_activity + WHERE application_name = 'fdw_retry_check'; +END $$; +-- Outside a subtransaction, the broken connection is reestablished. +BEGIN; +SELECT c1 FROM ft1 ORDER BY c1 LIMIT 3; +DO $$ BEGIN +PERFORM pg_terminate_backend(pid, 180000) FROM pg_stat_activity + WHERE application_name = 'fdw_retry_check'; +END $$; +-- Inside one, it isn't, and the query fails. +SAVEPOINT s2; +-- The text of the error might vary across platforms, so only show SQLSTATE. +\set VERBOSITY sqlstate +SELECT 1 FROM ft1 LIMIT 1; -- should fail +\set VERBOSITY default +COMMIT; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP streaming_fetch); + -- ============================================================================= -- test connection invalidation cases and postgres_fdw_get_connections function -- ============================================================================= @@ -4424,6 +4755,16 @@ SELECT * FROM local_tbl, async_pt WHERE local_tbl.a = async_pt.a AND local_tbl.c ALTER FOREIGN TABLE async_p1 OPTIONS (DROP use_remote_estimate); ALTER FOREIGN TABLE async_p2 OPTIONS (DROP use_remote_estimate); +-- streaming_fetch disables asynchronous execution, whether it is set on the +-- table (async_p1) or on the server (loopback2, used by async_p2). +ALTER FOREIGN TABLE async_p1 OPTIONS (ADD streaming_fetch 'true'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM async_pt; +ALTER SERVER loopback2 OPTIONS (ADD streaming_fetch 'true'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM async_pt; +ALTER FOREIGN TABLE async_p1 OPTIONS (DROP streaming_fetch); +ALTER SERVER loopback2 OPTIONS (DROP streaming_fetch); DROP TABLE local_tbl; DROP INDEX base_tbl1_idx; DROP INDEX base_tbl2_idx; diff --git a/contrib/postgres_fdw/sql/query_cancel.sql b/contrib/postgres_fdw/sql/query_cancel.sql index a6830705843..a7238608e64 100644 --- a/contrib/postgres_fdw/sql/query_cancel.sql +++ b/contrib/postgres_fdw/sql/query_cancel.sql @@ -20,3 +20,14 @@ SET LOCAL statement_timeout = '10ms'; -- This would take very long if not canceled: SELECT count(*) FROM ft1 a CROSS JOIN ft1 b CROSS JOIN ft1 c CROSS JOIN ft1 d; COMMIT; + +-- Same with streaming_fetch: canceling mid-stream must leave the connection +-- usable. +ALTER FOREIGN TABLE ft1 OPTIONS (ADD streaming_fetch 'true'); +BEGIN; +SELECT count(*) FROM ft1 a; +SET LOCAL statement_timeout = '10ms'; +SELECT count(*) FROM ft1 a CROSS JOIN ft1 b CROSS JOIN ft1 c CROSS JOIN ft1 d; +COMMIT; +SELECT count(*) FROM ft1; +ALTER FOREIGN TABLE ft1 OPTIONS (DROP streaming_fetch); diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index 01577d8d69b..6404813ba90 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -484,6 +484,58 @@ OPTIONS (ADD password_required 'false'); + + streaming_fetch (boolean) + + + This option controls whether postgres_fdw reads the + results of a foreign scan without declaring a cursor on the remote + server. Because the execution of a cursor's query can be suspended + between fetches, the remote server never uses a parallel plan for it + (see ). With + streaming_fetch enabled, the remote query instead runs + to completion in a single execution and may be parallelized, which can + significantly speed up large scans. It can be specified for a foreign + table or a foreign server. The option specified on a table overrides an + option specified for the server. The default is false. + Asynchronous execution is not supported in this mode and is disabled + automatically when it is enabled. + + + + Even when this option is enabled, a scan that needs the remote rows' + ctid, such as one feeding an UPDATE + or DELETE that cannot be pushed down to the remote + server, or a scan that locks rows with FOR UPDATE or + FOR SHARE, still uses a cursor. + EXPLAIN VERBOSE shows Streaming Fetch: + true for scans that use this mode. + + + + In this mode the rows of a scan are streamed from the remote server in + chunks of fetch_size rows, and a connection can carry + only one such stream at a time. If another scan or statement needs the + same connection while a streaming scan has not yet returned all of its + rows, postgres_fdw reads the remainder of that + scan's rows into a local tuplestore (kept in memory up to + work_mem and spilling to disk beyond that) and returns + them from there. + + + + If a streaming scan is stopped before all of its rows have been read + (for example by a LIMIT that cannot be pushed to the + remote server, or by a rescan), the remaining rows must still be read + from the connection and discarded before it can be reused; likewise a + rescan with changed parameters re-executes the whole remote query rather + than rewinding a cursor. For queries that read only a small fraction of + their rows, streaming_fetch can therefore be slower + than the default cursor-based mode. + + + + -- 2.39.5 (Apple Git-154)