From 496c0e2c156a6349b6d0f58e5c389998be7b8a31 Mon Sep 17 00:00:00 2001 From: Rafia Sabih Date: Mon, 26 Jan 2026 12:25:08 +0100 Subject: [PATCH] Fetch without cursors This implements a new fetch mechanism for postgres_fdw which does not use cursors. The motivation behind this work is the limitation of cursors to not be able to use parallel query even at the local side. Since, this new fetch mode doesn't use cursors parallel query can now be used and hence improving the performance for postgres_fdw. The way this new mechanism works is, once we start a query and create a cursor to fetch the tuples for a query this mechanism just start processing the tuples for this query without creating any cursor for it. Next, when we need tuples from a different query and issue a call to create a new cursor in old method. At this point, the new mode fetches all the tuples of the first query and saves them in a tuplestore. Moving forward, the next query is processed as is. Next, whenever we need to fetch the tuples of the first query, we do so by reading the associated tuplestore where we saved the tuples earlier. This way we use this tuplestore to keep track of the tuples required instead of cursors. This new mode can be used by a new GUC called postgres_fdw.use_cursor. When it is set, everything works as it was before this patch i.e. with the cursors. At the moment, this does not have support for async mode. When async mode is used and use_cursor is set to false, then it switches to true. Original idea: Bernd Helmle Key suggestions: Robert Haas --- contrib/postgres_fdw/connection.c | 11 + .../postgres_fdw/expected/postgres_fdw.out | 181 +++++++ contrib/postgres_fdw/option.c | 19 + contrib/postgres_fdw/postgres_fdw.c | 443 ++++++++++++++---- contrib/postgres_fdw/postgres_fdw.h | 8 + contrib/postgres_fdw/sql/postgres_fdw.sql | 31 +- 6 files changed, 612 insertions(+), 81 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 953c2e0ab82..0d0aa044ed0 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -966,6 +966,17 @@ pgfdw_get_result(PGconn *conn) return libpqsrv_get_result_last(conn, pgfdw_we_get_result); } +/* + * Wrap libpqsrv_get_result(), adding wait event. + * Used in case of non-cursor mode. + * 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. * diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 48e3185b227..e9d9deead8b 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -418,6 +418,15 @@ SELECT * FROM ft1 t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; 1000 | 0 | 01000 | Thu Jan 01 00:00:00 1970 PST | Thu Jan 01 00:00:00 1970 | 0 | 0 | foo (1 row) +--Test in non-cursor mode to cover process_query_params path +SET postgres_fdw.use_cursor TO off; +SELECT * FROM ft1 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) + +RESET postgres_fdw.use_cursor; -- used in CTE WITH t1 AS (SELECT * FROM ft1 WHERE c1 <= 10) SELECT t2.c1, t2.c2, t2.c3, t2.c4 FROM t1, ft2 t2 WHERE t1.c1 = t2.c1 ORDER BY t1.c1; c1 | c2 | c3 | c4 @@ -884,6 +893,122 @@ 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) +-- Test in non-cursor mode for rescan path +SET postgres_fdw.use_cursor TO off; +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); + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 | c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +-----+----+-------+------------------------------+--------------------------+----+------------+-----+-----+----+-------+------------------------------+--------------------------+----+------------+----- + 6 | 6 | 00006 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 6 | 6 | 00006 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 16 | 6 | 00016 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 16 | 6 | 00016 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 26 | 6 | 00026 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 26 | 6 | 00026 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 36 | 6 | 00036 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 36 | 6 | 00036 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 46 | 6 | 00046 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 46 | 6 | 00046 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 56 | 6 | 00056 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 56 | 6 | 00056 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 66 | 6 | 00066 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 66 | 6 | 00066 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 76 | 6 | 00076 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 76 | 6 | 00076 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 86 | 6 | 00086 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 86 | 6 | 00086 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 96 | 6 | 00096 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 96 | 6 | 00096 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 106 | 6 | 00106 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 106 | 6 | 00106 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 116 | 6 | 00116 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 116 | 6 | 00116 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 126 | 6 | 00126 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 126 | 6 | 00126 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 136 | 6 | 00136 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 136 | 6 | 00136 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 146 | 6 | 00146 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 146 | 6 | 00146 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 156 | 6 | 00156 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 156 | 6 | 00156 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 166 | 6 | 00166 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 166 | 6 | 00166 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 176 | 6 | 00176 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 176 | 6 | 00176 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 186 | 6 | 00186 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 186 | 6 | 00186 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 196 | 6 | 00196 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 196 | 6 | 00196 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 206 | 6 | 00206 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 206 | 6 | 00206 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 216 | 6 | 00216 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 216 | 6 | 00216 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 226 | 6 | 00226 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 226 | 6 | 00226 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 236 | 6 | 00236 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 236 | 6 | 00236 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 246 | 6 | 00246 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 246 | 6 | 00246 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 256 | 6 | 00256 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 256 | 6 | 00256 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 266 | 6 | 00266 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 266 | 6 | 00266 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 276 | 6 | 00276 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 276 | 6 | 00276 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 286 | 6 | 00286 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 286 | 6 | 00286 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 296 | 6 | 00296 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 296 | 6 | 00296 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 306 | 6 | 00306 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 306 | 6 | 00306 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 316 | 6 | 00316 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 316 | 6 | 00316 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 326 | 6 | 00326 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 326 | 6 | 00326 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 336 | 6 | 00336 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 336 | 6 | 00336 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 346 | 6 | 00346 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 346 | 6 | 00346 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 356 | 6 | 00356 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 356 | 6 | 00356 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 366 | 6 | 00366 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 366 | 6 | 00366 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 376 | 6 | 00376 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 376 | 6 | 00376 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 386 | 6 | 00386 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 386 | 6 | 00386 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 396 | 6 | 00396 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 396 | 6 | 00396 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 406 | 6 | 00406 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 406 | 6 | 00406 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 416 | 6 | 00416 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 416 | 6 | 00416 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 426 | 6 | 00426 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 426 | 6 | 00426 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 436 | 6 | 00436 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 436 | 6 | 00436 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 446 | 6 | 00446 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 446 | 6 | 00446 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 456 | 6 | 00456 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 456 | 6 | 00456 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 466 | 6 | 00466 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 466 | 6 | 00466 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 476 | 6 | 00476 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 476 | 6 | 00476 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 486 | 6 | 00486 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 486 | 6 | 00486 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 496 | 6 | 00496 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 496 | 6 | 00496 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 506 | 6 | 00506 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 506 | 6 | 00506 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 516 | 6 | 00516 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 516 | 6 | 00516 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 526 | 6 | 00526 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 526 | 6 | 00526 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 536 | 6 | 00536 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 536 | 6 | 00536 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 546 | 6 | 00546 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 546 | 6 | 00546 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 556 | 6 | 00556 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 556 | 6 | 00556 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 566 | 6 | 00566 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 566 | 6 | 00566 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 576 | 6 | 00576 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 576 | 6 | 00576 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 586 | 6 | 00586 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 586 | 6 | 00586 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 596 | 6 | 00596 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 596 | 6 | 00596 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 606 | 6 | 00606 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 606 | 6 | 00606 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 616 | 6 | 00616 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 616 | 6 | 00616 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 626 | 6 | 00626 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 626 | 6 | 00626 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 636 | 6 | 00636 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 636 | 6 | 00636 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 646 | 6 | 00646 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 646 | 6 | 00646 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 656 | 6 | 00656 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 656 | 6 | 00656 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 666 | 6 | 00666 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 666 | 6 | 00666 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 676 | 6 | 00676 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 676 | 6 | 00676 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 686 | 6 | 00686 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 686 | 6 | 00686 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 696 | 6 | 00696 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 696 | 6 | 00696 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 706 | 6 | 00706 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 706 | 6 | 00706 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 716 | 6 | 00716 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 716 | 6 | 00716 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 726 | 6 | 00726 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 726 | 6 | 00726 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 736 | 6 | 00736 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 736 | 6 | 00736 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 746 | 6 | 00746 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 746 | 6 | 00746 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 756 | 6 | 00756 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 756 | 6 | 00756 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 766 | 6 | 00766 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 766 | 6 | 00766 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 776 | 6 | 00776 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 776 | 6 | 00776 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 786 | 6 | 00786 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 786 | 6 | 00786 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 796 | 6 | 00796 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 796 | 6 | 00796 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 806 | 6 | 00806 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 806 | 6 | 00806 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 816 | 6 | 00816 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 816 | 6 | 00816 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 826 | 6 | 00826 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 826 | 6 | 00826 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 836 | 6 | 00836 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 836 | 6 | 00836 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 846 | 6 | 00846 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 846 | 6 | 00846 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 856 | 6 | 00856 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 856 | 6 | 00856 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 866 | 6 | 00866 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 866 | 6 | 00866 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 876 | 6 | 00876 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 876 | 6 | 00876 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 886 | 6 | 00886 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 886 | 6 | 00886 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo + 896 | 6 | 00896 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo | 896 | 6 | 00896 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo + 906 | 6 | 00906 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo | 906 | 6 | 00906 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo + 916 | 6 | 00916 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo | 916 | 6 | 00916 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo + 926 | 6 | 00926 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo | 926 | 6 | 00926 | Tue Jan 27 00:00:00 1970 PST | Tue Jan 27 00:00:00 1970 | 6 | 6 | foo + 936 | 6 | 00936 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo | 936 | 6 | 00936 | Fri Feb 06 00:00:00 1970 PST | Fri Feb 06 00:00:00 1970 | 6 | 6 | foo + 946 | 6 | 00946 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo | 946 | 6 | 00946 | Mon Feb 16 00:00:00 1970 PST | Mon Feb 16 00:00:00 1970 | 6 | 6 | foo + 956 | 6 | 00956 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo | 956 | 6 | 00956 | Thu Feb 26 00:00:00 1970 PST | Thu Feb 26 00:00:00 1970 | 6 | 6 | foo + 966 | 6 | 00966 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo | 966 | 6 | 00966 | Sun Mar 08 00:00:00 1970 PST | Sun Mar 08 00:00:00 1970 | 6 | 6 | foo + 976 | 6 | 00976 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo | 976 | 6 | 00976 | Wed Mar 18 00:00:00 1970 PST | Wed Mar 18 00:00:00 1970 | 6 | 6 | foo + 986 | 6 | 00986 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 00:00:00 1970 | 6 | 6 | foo | 986 | 6 | 00986 | Sat Mar 28 00:00:00 1970 PST | Sat Mar 28 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 | 996 | 6 | 00996 | Tue Apr 07 00:00:00 1970 PST | Tue Apr 07 00:00:00 1970 | 6 | 6 | foo +(100 rows) + +-- Test for non cursor mode covering rescans and three active cursors +SELECT count(*) FROM ft2 a, ft2 b, ft2 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 +------- + 100 +(1 row) + +RESET postgres_fdw.use_cursor; -- 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 @@ -2153,6 +2278,24 @@ SELECT t1.c1 FROM ft1 t1 WHERE NOT EXISTS (SELECT 1 FROM ft2 t2 WHERE t1.c1 = t2 119 (10 rows) +--Test in non-cursor mode to cover the patch for two simultaneous active cursors +SET postgres_fdw.use_cursor TO off; +SELECT t1.c1 FROM ft1 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) + +RESET postgres_fdw.use_cursor; -- CROSS JOIN can be pushed down EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft1 t1 CROSS JOIN ft2 t2 ORDER BY t1.c1, t2.c1 OFFSET 100 LIMIT 10; @@ -2204,6 +2347,14 @@ SELECT t1.c1, t2.c1 FROM ft5 t1 JOIN ft6 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1, t ----+---- (0 rows) +--Test in non-cursor mode to cover the case with multiple cursors but only one active cursor at a time +SET postgres_fdw.use_cursor TO off; +SELECT t1.c1, t2.c1 FROM ft5 t1 JOIN ft6 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1, t2.c1 OFFSET 100 LIMIT 10; + c1 | c1 +----+---- +(0 rows) + +RESET postgres_fdw.use_cursor; -- unsafe join conditions (c8 has a UDT), not pushed down. Practically a CROSS -- JOIN since c8 in both tables has same value. EXPLAIN (VERBOSE, COSTS OFF) @@ -11506,6 +11657,36 @@ SELECT * FROM result_tbl ORDER BY a; (20 rows) DELETE FROM result_tbl; +-- Test in non-cursor mode, it doesn't support async execution +SET postgres_fdw.use_cursor TO off; +INSERT INTO result_tbl SELECT * FROM async_pt WHERE b % 100 = 0; +SELECT * FROM result_tbl ORDER BY a; + a | b | c +------+-----+------ + 1000 | 0 | 0000 + 1100 | 100 | 0100 + 1200 | 200 | 0200 + 1300 | 300 | 0300 + 1400 | 400 | 0400 + 1500 | 500 | 0500 + 1600 | 600 | 0600 + 1700 | 700 | 0700 + 1800 | 800 | 0800 + 1900 | 900 | 0900 + 2000 | 0 | 0000 + 2100 | 100 | 0100 + 2200 | 200 | 0200 + 2300 | 300 | 0300 + 2400 | 400 | 0400 + 2500 | 500 | 0500 + 2600 | 600 | 0600 + 2700 | 700 | 0700 + 2800 | 800 | 0800 + 2900 | 900 | 0900 +(20 rows) + +DELETE FROM result_tbl; +RESET postgres_fdw.use_cursor; EXPLAIN (VERBOSE, COSTS OFF) INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505; QUERY PLAN diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index 04788b7e8b3..bb06bba8fd5 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -44,6 +44,7 @@ static PgFdwOption *postgres_fdw_options; * GUC parameters */ char *pgfdw_application_name = NULL; +bool pgfdw_use_cursor = true; /* * Helper functions @@ -586,5 +587,23 @@ _PG_init(void) NULL, NULL); + /* + * If use_cursor is set to false, then the new way of fetching is used. In + * this mode, cursors are not used, rather the tuples are stored in a + * tuplestore in case the switch of queries in between execution. So, for + * the next call, tuples are fetched from this tuplestore instead of the + * fetch from cursor. + */ + DefineCustomBoolVariable("postgres_fdw.use_cursor", + "If set uses the cursor, otherwise fetches without cursor", + NULL, + &pgfdw_use_cursor, + true, + PGC_USERSET, + 0, + NULL, + NULL, + NULL); + MarkGUCPrefixReserved("postgres_fdw"); } diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 5e178c21b39..8e0eb048aec 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -22,6 +22,7 @@ #include "commands/explain_format.h" #include "commands/explain_state.h" #include "executor/execAsync.h" +#include "executor/executor.h" #include "foreign/fdwapi.h" #include "funcapi.h" #include "miscadmin.h" @@ -174,6 +175,14 @@ typedef struct PgFdwScanState MemoryContext temp_cxt; /* context for per-tuple temporary data */ int fetch_size; /* number of tuples per fetch */ + /* To be used only in non-cursor mode */ + Tuplestorestate *tuplestore; /* Tuplestore to save the tuples of the + * query for later fetch. */ + TupleTableSlot *slot; /* Slot to be used when reading the tuple from + * the tuplestore */ + bool tuples_ready; /* To indicate when tuplestore is ready to be + * read. */ + int total_tuples; /* total tuples in the tuplestore. */ } PgFdwScanState; /* @@ -451,7 +460,7 @@ static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, EquivalenceClass *ec, EquivalenceMember *em, void *arg); static void create_cursor(ForeignScanState *node); -static void fetch_more_data(ForeignScanState *node); +static void fetch_more_data(ForeignScanState *node, bool use_tuplestore); static void close_cursor(PGconn *conn, unsigned int cursor_number, PgFdwConnState *conn_state); static PgFdwModifyState *create_foreign_modify(EState *estate, @@ -516,7 +525,8 @@ static HeapTuple make_tuple_from_result_row(PGresult *res, AttInMetadata *attinmeta, List *retrieved_attrs, ForeignScanState *fsstate, - MemoryContext temp_context); + MemoryContext temp_context, + TupleDesc last_tupdesc); static void conversion_error_callback(void *arg); static bool foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel, JoinType jointype, RelOptInfo *outerrel, RelOptInfo *innerrel, @@ -546,6 +556,8 @@ static void merge_fdw_options(PgFdwRelationInfo *fpinfo, const PgFdwRelationInfo *fpinfo_i); static int get_batch_size_option(Relation rel); +/* Only required for non-cursor mode */ +static void fill_tuplestore(ForeignScanState *node); /* * Foreign-data wrapper handler function: return a struct with pointers @@ -1593,6 +1605,57 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Set the async-capable flag */ fsstate->async_capable = node->ss.ps.async_capable; + /* Initially, there is no last_query */ + fsstate->conn_state->last_query = NULL; +} + +/* + * This routine fetches all the tuples of a query and saves them in a tuplestore. + * This is required when the result of a query is not completely fetched but the control + * switches to a different query. + * A call to fetch_data is made from here, hence we need complete ForeignScanState here. + */ +static void +fill_tuplestore(ForeignScanState *node) +{ + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + PgFdwScanState *last_fsstate = (PgFdwScanState *) fsstate->conn_state->last_query; + MemoryContext oldcontext; + ExprContext *econtext = node->ss.ps.ps_ExprContext; + PGconn *conn = fsstate->conn; + const char **values = last_fsstate->param_values; + int numParams = last_fsstate->numParams; + + /* + * Construct array of query parameter values in text format, as done in + * create_cursor + */ + if (numParams > 0) + { + oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory); + process_query_params(econtext, + last_fsstate->param_flinfo, + last_fsstate->param_exprs, + values); + MemoryContextSwitchTo(oldcontext); + } + if (conn->asyncStatus == PGASYNC_IDLE) + { + /* If the connection is not active then set up */ + if (!PQsendQueryParams(conn, last_fsstate->query, last_fsstate->numParams, + NULL, values, NULL, NULL, 0)) + pgfdw_report_error(NULL, conn, last_fsstate->query); + + if (!PQsetChunkedRowsMode(conn, last_fsstate->fetch_size)) + pgfdw_report_error(NULL, conn, last_fsstate->query); + } + fetch_more_data(node, true); + + /* + * Remove the last_query since it is completely fetched, so no need to + * remember it now. + */ + fsstate->conn_state->last_query = NULL; } /* @@ -1625,7 +1688,8 @@ postgresIterateForeignScan(ForeignScanState *node) return ExecClearTuple(slot); /* No point in another fetch if we already detected EOF, though. */ if (!fsstate->eof_reached) - fetch_more_data(node); + fetch_more_data(node, false); + /* If we didn't get any tuples, must be end of data. */ if (fsstate->next_tuple >= fsstate->num_tuples) return ExecClearTuple(slot); @@ -1651,6 +1715,7 @@ postgresReScanForeignScan(ForeignScanState *node) PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; char sql[64]; PGresult *res; + bool close_cursor = false; /* If we haven't created the cursor yet, nothing to do. */ if (!fsstate->cursor_exists) @@ -1666,7 +1731,7 @@ postgresReScanForeignScan(ForeignScanState *node) if (fsstate->async_capable && fsstate->conn_state->pendingAreq && fsstate->conn_state->pendingAreq->requestee == (PlanState *) node) - fetch_more_data(node); + fetch_more_data(node, false); /* * If any internal parameters affecting this node have changed, we'd @@ -1680,19 +1745,26 @@ postgresReScanForeignScan(ForeignScanState *node) if (node->ss.ps.chgParam != NULL) { fsstate->cursor_exists = false; - snprintf(sql, sizeof(sql), "CLOSE c%u", - fsstate->cursor_number); + if (pgfdw_use_cursor) + snprintf(sql, sizeof(sql), "CLOSE c%u", + fsstate->cursor_number); + else + close_cursor = true; } else if (fsstate->fetch_ct_2 > 1) { if (PQserverVersion(fsstate->conn) < 150000) + /* TODO: Handle it in non-cursor mode as well */ snprintf(sql, sizeof(sql), "MOVE BACKWARD ALL IN c%u", fsstate->cursor_number); else { fsstate->cursor_exists = false; - snprintf(sql, sizeof(sql), "CLOSE c%u", - fsstate->cursor_number); + if (pgfdw_use_cursor) + snprintf(sql, sizeof(sql), "CLOSE c%u", + fsstate->cursor_number); + else + close_cursor = true; } } else @@ -1701,18 +1773,32 @@ postgresReScanForeignScan(ForeignScanState *node) fsstate->next_tuple = 0; return; } - - 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; + if (pgfdw_use_cursor) + { + 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; + } + else if (!pgfdw_use_cursor && close_cursor) + { + res = pgfdw_get_result(fsstate->conn); + while (res != NULL) + res = pgfdw_get_result(fsstate->conn); + 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; + } } /* @@ -3755,29 +3841,67 @@ create_cursor(ForeignScanState *node) MemoryContextSwitchTo(oldcontext); } - /* Construct the DECLARE CURSOR command */ - initStringInfo(&buf); - appendStringInfo(&buf, "DECLARE c%u CURSOR FOR\n%s", - fsstate->cursor_number, fsstate->query); + if (pgfdw_use_cursor) + { + /* Construct the DECLARE CURSOR command */ + initStringInfo(&buf); + appendStringInfo(&buf, "DECLARE c%u CURSOR FOR\n%s", + fsstate->cursor_number, fsstate->query); - /* - * Notice that we pass NULL for paramTypes, thus forcing the remote server - * to infer types for all parameters. Since we explicitly cast every - * parameter (see deparse.c), the "inference" is trivial and will produce - * the desired result. This allows us to avoid assuming that the remote - * server has the same OIDs we do for the parameters' types. - */ - if (!PQsendQueryParams(conn, buf.data, numParams, - NULL, values, NULL, NULL, 0)) - pgfdw_report_error(NULL, conn, buf.data); + /* + * Notice that we pass NULL for paramTypes, thus forcing the remote + * server to infer types for all parameters. Since we explicitly cast + * every parameter (see deparse.c), the "inference" is trivial and + * will produce the desired result. This allows us to avoid assuming + * that the remote server has the same OIDs we do for the parameters' + * types. + */ + if (!PQsendQueryParams(conn, buf.data, numParams, + NULL, values, NULL, NULL, 0)) + pgfdw_report_error(NULL, conn, buf.data); - /* - * Get the result, and check for success. - */ - res = pgfdw_get_result(conn); - if (PQresultStatus(res) != PGRES_COMMAND_OK) - pgfdw_report_error(res, conn, fsstate->query); - PQclear(res); + /* + * Get the result, and check for success. + */ + res = pgfdw_get_result(conn); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pgfdw_report_error(res, conn, fsstate->query); + PQclear(res); + + /* Clean up */ + pfree(buf.data); + } + else + { + /* + * Finish fetching tuples of the last query. Do this only when there + * is a different query than the current one. In the case of rescan, + * create_cursor is called simultaneously for the same query so to + * avoid calling fill_tuplestore in such cases, check if the queries + * are different and tuplestore is not already filled for this query. + */ + if (fsstate->conn_state->last_query && + fsstate != fsstate->conn_state->last_query && + !fsstate->conn_state->last_query->tuples_ready) + fill_tuplestore(node); + + /* + * To remember the current query as the last one, when control + * switches to another query + */ + fsstate->conn_state->last_query = fsstate; + + if (!PQsendQueryParams(conn, fsstate->query, numParams, + NULL, values, NULL, NULL, 0)) + pgfdw_report_error(NULL, conn, fsstate->query); + + /* + * Call for Chunked rows mode with same size of chunk as the fetch + * size + */ + if (!PQsetChunkedRowsMode(conn, fsstate->fetch_size)) + pgfdw_report_error(NULL, conn, fsstate->query); + } /* Mark the cursor as created, and show no tuples have been retrieved */ fsstate->cursor_exists = true; @@ -3786,30 +3910,29 @@ create_cursor(ForeignScanState *node) fsstate->next_tuple = 0; fsstate->fetch_ct_2 = 0; fsstate->eof_reached = false; - - /* Clean up */ - pfree(buf.data); } /* * Fetch some more rows from the node's cursor. */ static void -fetch_more_data(ForeignScanState *node) +fetch_more_data(ForeignScanState *node, bool use_tuplestore) { PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + PgFdwScanState *last_fsstate = fsstate->conn_state->last_query; PGconn *conn = fsstate->conn; PGresult *res; - int numrows; - int i; MemoryContext oldcontext; + int numrows, + i = 0, + total_tuples = 0; + bool already_done = false; /* * 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); if (fsstate->async_capable) @@ -3828,7 +3951,7 @@ fetch_more_data(ForeignScanState *node) /* Reset per-connection state */ fsstate->conn_state->pendingAreq = NULL; } - else + else if (pgfdw_use_cursor) { char sql[64]; @@ -3841,32 +3964,177 @@ fetch_more_data(ForeignScanState *node) if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, fsstate->query); } + else + { + /* + * In non-cursor mode, there are three options for the further + * processing: 1. If the tuplestore is already filled, retrieve the + * tuples from there. 2. Fetch the tuples till the end of the query + * and store them in tuplestore. 3. Perform a normal fetch and process + * the tuples. + */ + if (fsstate->tuplestore && fsstate->tuples_ready) + { + /* Retrieve the tuples from the tuplestore instead of actual fetch */ + numrows = fsstate->total_tuples; + fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple)); + fsstate->slot = MakeSingleTupleTableSlot(fsstate->tupdesc, &TTSOpsMinimalTuple); - /* Convert the data into HeapTuples */ - numrows = PQntuples(res); - fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple)); - fsstate->num_tuples = numrows; - fsstate->next_tuple = 0; + while (tuplestore_gettupleslot(fsstate->tuplestore, true, true, fsstate->slot)) + { + fsstate->tuples[i++] = ExecFetchSlotHeapTuple(fsstate->slot, true, NULL); + ExecClearTuple(fsstate->slot); + } + fsstate->num_tuples = numrows; + fsstate->next_tuple = 0; + already_done = true; + fsstate->eof_reached = true; + fsstate->tuples_ready = false; + + /* Clean up */ + tuplestore_end(fsstate->tuplestore); + ExecDropSingleTupleTableSlot(fsstate->slot); + fsstate->slot = NULL; + fsstate->tuplestore = NULL; + return; + } + else + { + /* + * Non-cursor mode uses PQSetChunkedRowsMode during create_cursor, + * so just get the result here. + */ + res = pgfdw_get_next_result(conn); + if (PQresultStatus(res) == PGRES_FATAL_ERROR) + pgfdw_report_error(res, conn, fsstate->query); - for (i = 0; i < numrows; i++) + else if (PQresultStatus(res) == PGRES_TUPLES_OK) + { + /* + * This signifies query is completed and there are no more + * tuples left. + */ + if (use_tuplestore) + { + /* + * If we are here to store the tuples in tuplestore then + * this signals we have already fetched all the tuples for + * this query, so nothing to do. Just set the right flags + */ + already_done = true; + last_fsstate->tuples_ready = true; + last_fsstate->eof_reached = true; + } + + /* There are no more tuples to fetch */ + while (res != NULL) + res = pgfdw_get_result(conn); + } + else if (PQresultStatus(res) == PGRES_TUPLES_CHUNK) + { + if (use_tuplestore) + { + /* + * This is to fetch all the tuples of the query in + * last_fsstate and save them in Tuple Slot. + */ + + /* + * We should never be here without a valid last_fsstate in + * the scan state. + */ + Assert(last_fsstate != NULL); + last_fsstate->tuplestore = tuplestore_begin_heap(true, false, work_mem); + + for (;;) + { + /* + * Since it is using PQSetChunkedRowsMode, we only get + * the fsstate->fetch_size tuples in one run, so keep + * on executing till we get NULL in PGresult i.e. all + * the tuples are retrieved. + */ + CHECK_FOR_INTERRUPTS(); + numrows = PQntuples(res); + total_tuples += numrows; + + /* Convert the data into HeapTuples */ + Assert(IsA(node->ss.ps.plan, ForeignScan)); + for (i = 0; i < numrows; i++) + { + HeapTuple temp_tuple; + + temp_tuple = make_tuple_from_result_row(res, i, + last_fsstate->rel, + last_fsstate->attinmeta, + last_fsstate->retrieved_attrs, + node, + last_fsstate->temp_cxt, + last_fsstate->tupdesc); + tuplestore_puttuple(last_fsstate->tuplestore, temp_tuple); + heap_freetuple(temp_tuple); + } + + res = pgfdw_get_next_result(conn); + if (res == NULL) + break; + else if (PQresultStatus(res) == PGRES_FATAL_ERROR) + pgfdw_report_error(res, conn, last_fsstate->query); + else if (PQresultStatus(res) == PGRES_TUPLES_OK) + { + /* This means all the tuples are retreived. */ + numrows = PQntuples(res); + /* If there is nothing to fetch */ + if (numrows == 0) + res = pgfdw_get_result(conn); + } + } + + /* + * EOF is reached because we are storing all tuples to the + * tuplestore. + */ + already_done = true; + last_fsstate->tuples_ready = true; + last_fsstate->total_tuples = total_tuples; + } + } + } + } + if (!already_done) { - Assert(IsA(node->ss.ps.plan, ForeignScan)); + /* + * To fetch tuples for the query in fsstate used in both cursor and + * non-cursor mode. + */ + /* Convert the data into HeapTuples */ + numrows = PQntuples(res); + fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple)); + fsstate->num_tuples = numrows; + fsstate->next_tuple = 0; - fsstate->tuples[i] = - make_tuple_from_result_row(res, i, - fsstate->rel, - fsstate->attinmeta, - fsstate->retrieved_attrs, - node, - fsstate->temp_cxt); + for (i = 0; i < numrows; i++) + { + Assert(IsA(node->ss.ps.plan, ForeignScan)); + + fsstate->tuples[i] = + make_tuple_from_result_row(res, i, + fsstate->rel, + fsstate->attinmeta, + fsstate->retrieved_attrs, + node, + fsstate->temp_cxt, + NULL); + } } /* Update fetch_ct_2 */ - if (fsstate->fetch_ct_2 < 2) + if (!use_tuplestore && 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); + if (!use_tuplestore) + fsstate->eof_reached = (numrows < fsstate->fetch_size); PQclear(res); @@ -3941,11 +4209,16 @@ close_cursor(PGconn *conn, unsigned int cursor_number, char sql[64]; PGresult *res; - snprintf(sql, sizeof(sql), "CLOSE c%u", cursor_number); - res = pgfdw_exec_query(conn, sql, conn_state); - if (PQresultStatus(res) != PGRES_COMMAND_OK) - pgfdw_report_error(res, conn, sql); - PQclear(res); + if (pgfdw_use_cursor) + { + snprintf(sql, sizeof(sql), "CLOSE c%u", cursor_number); + res = pgfdw_exec_query(conn, sql, conn_state); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pgfdw_report_error(res, conn, sql); + PQclear(res); + } + else + while (pgfdw_get_result(conn) != NULL); } /* @@ -4329,7 +4602,7 @@ store_returning_result(PgFdwModifyState *fmstate, fmstate->attinmeta, fmstate->retrieved_attrs, NULL, - fmstate->temp_cxt); + fmstate->temp_cxt, NULL); /* * The returning slot will not necessarily be suitable to store heaptuples @@ -4608,7 +4881,7 @@ get_returning_data(ForeignScanState *node) dmstate->attinmeta, dmstate->retrieved_attrs, node, - dmstate->temp_cxt); + dmstate->temp_cxt, NULL); ExecStoreHeapTuple(newtup, slot, false); /* Get the updated/deleted tuple. */ if (dmstate->rel) @@ -5239,7 +5512,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, for (;;) { int numrows; - int i; + int i = 0; /* Allow users to cancel long query */ CHECK_FOR_INTERRUPTS(); @@ -5360,7 +5633,7 @@ analyze_row_processor(PGresult *res, int row, PgFdwAnalyzeState *astate) astate->attinmeta, astate->retrieved_attrs, NULL, - astate->temp_cxt); + astate->temp_cxt, NULL); MemoryContextSwitchTo(oldcontext); } @@ -7314,7 +7587,7 @@ postgresForeignAsyncNotify(AsyncRequest *areq) if (!PQconsumeInput(fsstate->conn)) pgfdw_report_error(NULL, fsstate->conn, fsstate->query); - fetch_more_data(node); + fetch_more_data(node, false); produce_tuple_asynchronously(areq, true); } @@ -7400,6 +7673,13 @@ fetch_more_data_begin(AsyncRequest *areq) PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; char sql[64]; + /* + * Reset cursor mode when in asynchronous mode as it is not supported in + * non-cursor mode + */ + if (!pgfdw_use_cursor) + pgfdw_use_cursor = true; + Assert(!fsstate->conn_state->pendingAreq); /* Create the cursor synchronously. */ @@ -7432,7 +7712,7 @@ process_pending_request(AsyncRequest *areq) /* The request should be currently in-process */ Assert(fsstate->conn_state->pendingAreq == areq); - fetch_more_data(node); + fetch_more_data(node, false); /* * If we didn't get any tuples, must be end of data; complete the request @@ -7494,7 +7774,8 @@ make_tuple_from_result_row(PGresult *res, AttInMetadata *attinmeta, List *retrieved_attrs, ForeignScanState *fsstate, - MemoryContext temp_context) + MemoryContext temp_context, + TupleDesc last_tupdesc) { HeapTuple tuple; TupleDesc tupdesc; @@ -7518,9 +7799,13 @@ make_tuple_from_result_row(PGresult *res, /* * Get the tuple descriptor for the row. Use the rel's tupdesc if rel is - * provided, otherwise look to the scan node's ScanTupleSlot. + * provided, otherwise look to the scan node's ScanTupleSlot. In case of + * non-cursor mode, use the tupledesc that is already provided, because + * getting from the current fsstate would be wrong in this case. */ - if (rel) + if (last_tupdesc) + tupdesc = last_tupdesc; + else if (rel) tupdesc = RelationGetDescr(rel); else { diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index e69735298d7..6a257252647 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -16,9 +16,11 @@ #include "foreign/foreign.h" #include "lib/stringinfo.h" #include "libpq/libpq-be-fe.h" +#include "libpq-int.h" #include "nodes/execnodes.h" #include "nodes/pathnodes.h" #include "utils/relcache.h" +#include "funcapi.h" /* * FDW-specific planner information kept in RelOptInfo.fdw_private for a @@ -131,12 +133,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 *last_query; /* last query executed, required for + * non-cursor mode */ } PgFdwConnState; /* @@ -164,6 +170,7 @@ 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); pg_noreturn extern void pgfdw_report_error(PGresult *res, PGconn *conn, @@ -179,6 +186,7 @@ extern List *ExtractExtensionList(const char *extensionsString, bool warnOnMissing); extern char *process_pgfdw_appname(const char *appname); extern char *pgfdw_application_name; +extern bool pgfdw_use_cursor; /* in deparse.c */ extern void classifyConditions(PlannerInfo *root, diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 9a8f9e28135..f15c5ae2145 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3,11 +3,9 @@ -- =================================================================== CREATE EXTENSION postgres_fdw; - SELECT current_database() AS current_database, current_setting('port') AS current_port \gset - CREATE SERVER testserver1 FOREIGN DATA WRAPPER postgres_fdw; CREATE SERVER loopback FOREIGN DATA WRAPPER postgres_fdw OPTIONS (dbname :'current_database', port :'current_port'); @@ -277,6 +275,10 @@ SELECT COUNT(*) FROM ft1 t1; SELECT * FROM ft1 t1 WHERE t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 <= 10) ORDER BY c1; -- subquery+MAX SELECT * FROM ft1 t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; +--Test in non-cursor mode to cover process_query_params path +SET postgres_fdw.use_cursor TO off; +SELECT * FROM ft1 t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; +RESET postgres_fdw.use_cursor; -- used in CTE WITH t1 AS (SELECT * FROM ft1 WHERE c1 <= 10) SELECT t2.c1, t2.c2, t2.c3, t2.c4 FROM t1, ft2 t2 WHERE t1.c1 = t2.c1 ORDER BY t1.c1; -- fixed values @@ -356,6 +358,15 @@ 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); + +-- Test in non-cursor mode for rescan path +SET postgres_fdw.use_cursor TO off; +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); +-- Test for non cursor mode covering rescans and three active cursors +SELECT count(*) FROM ft2 a, ft2 b, ft2 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); +RESET postgres_fdw.use_cursor; + -- 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)); @@ -646,6 +657,11 @@ SELECT t1.c1 FROM ft1 t1 WHERE EXISTS (SELECT 1 FROM ft2 t2 WHERE t1.c1 = t2.c1) EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1 FROM ft1 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 t1 WHERE NOT EXISTS (SELECT 1 FROM ft2 t2 WHERE t1.c1 = t2.c2) ORDER BY t1.c1 OFFSET 100 LIMIT 10; + +--Test in non-cursor mode to cover the patch for two simultaneous active cursors +SET postgres_fdw.use_cursor TO off; +SELECT t1.c1 FROM ft1 t1 WHERE NOT EXISTS (SELECT 1 FROM ft2 t2 WHERE t1.c1 = t2.c2) ORDER BY t1.c1 OFFSET 100 LIMIT 10; +RESET postgres_fdw.use_cursor; -- CROSS JOIN can be pushed down EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft1 t1 CROSS JOIN ft2 t2 ORDER BY t1.c1, t2.c1 OFFSET 100 LIMIT 10; @@ -654,6 +670,10 @@ SELECT t1.c1, t2.c1 FROM ft1 t1 CROSS JOIN ft2 t2 ORDER BY t1.c1, t2.c1 OFFSET 1 EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft5 t1 JOIN ft6 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1, t2.c1 OFFSET 100 LIMIT 10; SELECT t1.c1, t2.c1 FROM ft5 t1 JOIN ft6 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1, t2.c1 OFFSET 100 LIMIT 10; +--Test in non-cursor mode to cover the case with multiple cursors but only one active cursor at a time +SET postgres_fdw.use_cursor TO off; +SELECT t1.c1, t2.c1 FROM ft5 t1 JOIN ft6 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c1, t2.c1 OFFSET 100 LIMIT 10; +RESET postgres_fdw.use_cursor; -- unsafe join conditions (c8 has a UDT), not pushed down. Practically a CROSS -- JOIN since c8 in both tables has same value. EXPLAIN (VERBOSE, COSTS OFF) @@ -3907,6 +3927,13 @@ INSERT INTO result_tbl SELECT * FROM async_pt WHERE b % 100 = 0; SELECT * FROM result_tbl ORDER BY a; DELETE FROM result_tbl; +-- Test in non-cursor mode, it doesn't support async execution +SET postgres_fdw.use_cursor TO off; +INSERT INTO result_tbl SELECT * FROM async_pt WHERE b % 100 = 0; +SELECT * FROM result_tbl ORDER BY a; +DELETE FROM result_tbl; +RESET postgres_fdw.use_cursor; + EXPLAIN (VERBOSE, COSTS OFF) INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505; INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505; -- 2.39.5 (Apple Git-154)