From e3869dff4ad96d0e3ce417767dd02f147c717846 Mon Sep 17 00:00:00 2001 From: Rafia Sabih Date: Thu, 2 Apr 2026 19:28:35 +0200 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 a scan starts instead of creating a cursor to fetch the tuples it just start processing the tuples. Next, when the flow switches to a new scan, all the tuples of the first scan are fetched and saved in a tuplestore. Later when the flow comes to the the last scan and finds tuples in tuplestore, it fetches from there. 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.disable_cursor. By default it is unset, everything works as it was before this patch i.e. with the cursors. At the moment, this does not have support for async mode. 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 | 18 + contrib/postgres_fdw/postgres_fdw.c | 433 +++++++++++++++--- contrib/postgres_fdw/postgres_fdw.h | 7 + contrib/postgres_fdw/sql/postgres_fdw.sql | 29 ++ 6 files changed, 628 insertions(+), 51 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 192f8011160..c43f63fed63 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -988,6 +988,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 ac34a1acacb..179812416d7 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -448,6 +448,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.disable_cursor TO on; +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.disable_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 @@ -914,6 +923,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.disable_cursor TO on; +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.disable_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 @@ -2183,6 +2308,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.disable_cursor TO on; +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.disable_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; @@ -2234,6 +2377,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.disable_cursor TO on; +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.disable_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) @@ -11568,6 +11719,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.disable_cursor TO on; +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.disable_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 b0bd72d1e58..12db4ef0b1c 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 disable_cursor = false; /* * Helper functions @@ -586,5 +587,22 @@ _PG_init(void) NULL, NULL); + /* + * If disable_cursor is set to true, 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 scan switches to other active query.In the next + * call, tuples are fetched from this tuplestore. + */ + DefineCustomBoolVariable("postgres_fdw.disable_cursor", + "If set cursors are not used, otherwise fetches with cursor", + NULL, + &disable_cursor, + false, + 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 41e47cc795b..b0e0b3cdf3b 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -23,6 +23,7 @@ #include "commands/explain_state.h" #include "executor/execAsync.h" #include "executor/instrument.h" +#include "executor/executor.h" #include "foreign/fdwapi.h" #include "funcapi.h" #include "miscadmin.h" @@ -49,6 +50,7 @@ #include "utils/rel.h" #include "utils/sampling.h" #include "utils/selfuncs.h" +#include "utils/tuplestore.h" PG_MODULE_MAGIC_EXT( .name = "postgres_fdw", @@ -166,6 +168,7 @@ typedef struct PgFdwScanState /* batch-level state, for optimizing rewinds and avoiding useless fetch */ int fetch_ct_2; /* Min(# of fetches done, 2) */ bool eof_reached; /* true if last fetch reached EOF */ + bool rescan; /* identify when rescan is done */ /* for asynchronous execution */ bool async_capable; /* engage asynchronous-capable logic? */ @@ -175,6 +178,12 @@ 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 */ + bool scan_init; + 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 */ } PgFdwScanState; /* @@ -452,7 +461,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, @@ -517,7 +526,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, @@ -547,6 +557,11 @@ 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 fetch_from_tuplestore(ForeignScanState *node); +static void save_to_tuplestore(ForeignScanState *node); +static void scan_init(ForeignScanState *node); +static void end_scan(PGconn *conn); /* * Foreign-data wrapper handler function: return a struct with pointers @@ -1544,7 +1559,10 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Assign a unique ID for my cursor */ fsstate->cursor_number = GetCursorNumber(fsstate->conn); - fsstate->cursor_exists = false; + if (!disable_cursor) + fsstate->cursor_exists = false; + else + fsstate->scan_init = false; /* Get private info created by planner functions. */ fsstate->query = strVal(list_nth(fsplan->fdw_private, @@ -1594,6 +1612,13 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Set the async-capable flag */ fsstate->async_capable = node->ss.ps.async_capable; + /* Initially, there is no active_scan */ + fsstate->conn_state->active_scan = NULL; + fsstate->tuplestore = NULL; + fsstate->rescan = false; + /* Async mode not supported in cursors are disabled. */ + if (disable_cursor) + fsstate->async_capable = node->ss.ps.async_capable = false; } /* @@ -1613,8 +1638,10 @@ postgresIterateForeignScan(ForeignScanState *node) * already created the cursor before we get here, even if this is the * first call after Begin or ReScan. */ - if (!fsstate->cursor_exists) + if (!disable_cursor && !fsstate->cursor_exists) create_cursor(node); + else if (disable_cursor && !fsstate->scan_init) + scan_init(node); /* * Get some more tuples, if we've run out. @@ -1626,7 +1653,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); @@ -1653,8 +1681,13 @@ postgresReScanForeignScan(ForeignScanState *node) char sql[64]; PGresult *res; - /* If we haven't created the cursor yet, nothing to do. */ - if (!fsstate->cursor_exists) + /* + * If we haven't created the cursor yet or scan not initialised when + * cursors are disabled, nothing to do. + */ + if (!disable_cursor && !fsstate->cursor_exists) + return; + else if (disable_cursor && !fsstate->scan_init) return; /* @@ -1667,7 +1700,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 @@ -1681,19 +1714,22 @@ postgresReScanForeignScan(ForeignScanState *node) if (node->ss.ps.chgParam != NULL) { fsstate->cursor_exists = false; - snprintf(sql, sizeof(sql), "CLOSE c%u", - fsstate->cursor_number); + if (!disable_cursor) + snprintf(sql, sizeof(sql), "CLOSE c%u", + fsstate->cursor_number); } 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 (!disable_cursor) + snprintf(sql, sizeof(sql), "CLOSE c%u", + fsstate->cursor_number); } } else @@ -1702,10 +1738,20 @@ 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); + if (!disable_cursor) + { + res = pgfdw_exec_query(fsstate->conn, sql, fsstate->conn_state); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pgfdw_report_error(res, fsstate->conn, sql); + } + else + { + res = pgfdw_get_result(fsstate->conn); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pgfdw_report_error(res, fsstate->conn, "in rescan with cursors diabled"); + fsstate->rescan = true; + fsstate->scan_init = false; + } PQclear(res); /* Now force a fresh FETCH. */ @@ -1730,10 +1776,15 @@ postgresEndForeignScan(ForeignScanState *node) return; /* Close the cursor if open, to prevent accumulation of cursors */ - if (fsstate->cursor_exists) + if (!disable_cursor && fsstate->cursor_exists) close_cursor(fsstate->conn, fsstate->cursor_number, fsstate->conn_state); + else if (disable_cursor && fsstate->scan_init) + { + end_scan(fsstate->conn); + fsstate->scan_init = false; + } /* Release remote connection */ ReleaseConnection(fsstate->conn); fsstate->conn = NULL; @@ -3780,37 +3831,104 @@ create_cursor(ForeignScanState *node) pgfdw_report_error(res, conn, fsstate->query); PQclear(res); + /* Clean up */ + pfree(buf.data); + /* Mark the cursor as created, and show no tuples have been retrieved */ - fsstate->cursor_exists = true; fsstate->tuples = NULL; fsstate->num_tuples = 0; fsstate->next_tuple = 0; fsstate->fetch_ct_2 = 0; fsstate->eof_reached = false; + fsstate->cursor_exists = true; +} +static void +scan_init(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; - /* Clean up */ - pfree(buf.data); + /* + * 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 + * memory leak over repeated scans. + */ + if (numParams > 0) + { + MemoryContext oldcontext; + + oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory); + + process_query_params(econtext, + fsstate->param_flinfo, + fsstate->param_exprs, + values); + + MemoryContextSwitchTo(oldcontext); + } + + /* + * 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 save_to_tuplestore in such cases, check if the queries are + * different and tuplestore is not already filled for this query. + */ + if (fsstate->conn_state->active_scan && + fsstate != fsstate->conn_state->active_scan) + save_to_tuplestore(node); + + /* + * To remember the current scan as the last one, when control switches to + * another scan + */ + fsstate->conn_state->active_scan = 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); + + fsstate->tuples = NULL; + fsstate->num_tuples = 0; + fsstate->next_tuple = 0; + fsstate->fetch_ct_2 = 0; + fsstate->eof_reached = false; + fsstate->scan_init = true; } /* * 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 *active_fsstate = fsstate->conn_state->active_scan; */ PGconn *conn = fsstate->conn; PGresult *res; - int numrows; - int i; + int numrows = 0, + i = 0; + bool already_done = false; MemoryContext oldcontext; /* * We'll store the tuples in the batch_cxt. First, flush the previous - * batch. + * batch. When cursors are disabled, the tuplestore is created in + * batch_cxt, so it should not be reset here. */ fsstate->tuples = NULL; - MemoryContextReset(fsstate->batch_cxt); + if (!disable_cursor) + MemoryContextReset(fsstate->batch_cxt); oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt); if (fsstate->async_capable) @@ -3829,7 +3947,7 @@ fetch_more_data(ForeignScanState *node) /* Reset per-connection state */ fsstate->conn_state->pendingAreq = NULL; } - else + else if (!disable_cursor) { char sql[64]; @@ -3842,24 +3960,56 @@ fetch_more_data(ForeignScanState *node) if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, fsstate->query); } + else + { + /* + * When cursor mode is not used, there is a special possibility -- + * reading from the tuplestore. For processing of tuples nothing + * changes in cursor and mode without cursor during fetch. + */ + if (fsstate->tuplestore) + { + /* Reading tuples from tuplestore */ + fetch_from_tuplestore(node); + return; + } + + /* + * Non-cursor mode uses PQSetChunkedRowsMode during scan_init, 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); - /* Convert the data into HeapTuples */ - numrows = PQntuples(res); - fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple)); - fsstate->num_tuples = numrows; - fsstate->next_tuple = 0; - for (i = 0; i < numrows; i++) + } + 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 */ @@ -3874,6 +4024,160 @@ fetch_more_data(ForeignScanState *node) MemoryContextSwitchTo(oldcontext); } +/* + * This is used in non-cursor mode only to fetch the tuples from the tuplestore. + */ +static void +fetch_from_tuplestore(ForeignScanState *node) +{ + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + int numrows = 0, + i = 0; + + /* Retrieve the tuples from the tuplestore instead of actual fetch */ + numrows = tuplestore_tuple_count(fsstate->tuplestore); + fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple)); + fsstate->slot = MakeSingleTupleTableSlot(fsstate->tupdesc, &TTSOpsMinimalTuple); + + 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; + fsstate->eof_reached = true; + + /* Clean up */ + tuplestore_end(fsstate->tuplestore); + ExecDropSingleTupleTableSlot(fsstate->slot); + fsstate->slot = NULL; + fsstate->tuplestore = NULL; + return; +} + +/* + * In non-cursor mode only, save the tuples to tuplestore. + */ +static void +save_to_tuplestore(ForeignScanState *node) +{ + PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; + PgFdwScanState *active_fsstate = fsstate->conn_state->active_scan; + PGconn *conn = fsstate->conn; + PGresult *res; + int numrows = 0, + i = 0; + MemoryContext oldcontext; + + int numParams = active_fsstate->numParams; + const char **values = active_fsstate->param_values; + ExprContext *econtext = node->ss.ps.ps_ExprContext; + + /* + * Construct array of query parameter values in text format, as done in + * scan_init + */ + if (numParams > 0) + { + oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory); + process_query_params(econtext, + active_fsstate->param_flinfo, + active_fsstate->param_exprs, + values); + MemoryContextSwitchTo(oldcontext); + } + if (fsstate->rescan) + { + /* + * When coming from rescan, the connection is not setup. It is then + * required to set up the connection based on the query in + * active_scan. + */ + if (!PQsendQueryParams(conn, active_fsstate->query, active_fsstate->numParams, + NULL, values, NULL, NULL, 0)) + pgfdw_report_error(NULL, conn, active_fsstate->query); + + if (!PQsetChunkedRowsMode(conn, active_fsstate->fetch_size)) + pgfdw_report_error(NULL, conn, active_fsstate->query); + fsstate->rescan = false; + } + oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt); + + res = pgfdw_get_next_result(conn); + + if (PQresultStatus(res) == PGRES_FATAL_ERROR) + pgfdw_report_error(res, conn, fsstate->query); + + else if (PQresultStatus(res) == PGRES_TUPLES_OK) + { + /* + * This signifies there is nothing todo, all is fetched. + */ + active_fsstate->eof_reached = true; + + /* There are no more tuples to fetch */ + res = pgfdw_get_result(conn); + } + else if (PQresultStatus(res) == PGRES_TUPLES_CHUNK) + { + Assert(active_fsstate != NULL); + active_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); + + /* 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, + active_fsstate->rel, + active_fsstate->attinmeta, + active_fsstate->retrieved_attrs, + node, + active_fsstate->temp_cxt, + active_fsstate->tupdesc); + tuplestore_puttuple(active_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, active_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); + } + } + } + + /* + * Remove the active_scan since it is completely fetched, so no need to + * remember it now. + */ + fsstate->conn_state->active_scan = NULL; + + PQclear(res); + MemoryContextSwitchTo(oldcontext); + return; +} + /* * Force assorted GUC parameters to settings that ensure that we'll output * data values in a form that is unambiguous to the remote server. @@ -3949,6 +4253,20 @@ close_cursor(PGconn *conn, unsigned int cursor_number, PQclear(res); } +/* + * When cursors are disabled, end the scan to ensure all the tuples were retieved. + */ +static void +end_scan(PGconn *conn) +{ + PGresult *res; + + res = pgfdw_get_result(conn); + if (res && PQresultStatus(res) != PGRES_TUPLES_OK) + pgfdw_report_error(res, conn, "scan not ended properly in no cursor mode"); + PQclear(res); +} + /* * create_foreign_modify * Construct an execution state of a foreign insert/update/delete @@ -4330,7 +4648,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 @@ -4609,7 +4927,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) @@ -5240,7 +5558,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, for (;;) { int numrows; - int i; + int i = 0; /* Allow users to cancel long query */ CHECK_FOR_INTERRUPTS(); @@ -5270,7 +5588,8 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, } /* Close the cursor, just to be tidy. */ - close_cursor(conn, cursor_number, NULL); + if (!disable_cursor) + close_cursor(conn, cursor_number, NULL); ReleaseConnection(conn); @@ -5361,7 +5680,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); } @@ -7191,6 +7510,10 @@ postgresIsForeignPathAsyncCapable(ForeignPath *path) RelOptInfo *rel = ((Path *) path)->parent; PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private; + /* Async mode not supported when cursors are disabled. */ + if (disable_cursor) + fpinfo->async_capable = false; + return fpinfo->async_capable; } @@ -7315,7 +7638,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); } @@ -7404,9 +7727,12 @@ fetch_more_data_begin(AsyncRequest *areq) Assert(!fsstate->conn_state->pendingAreq); /* Create the cursor synchronously. */ - if (!fsstate->cursor_exists) + if (!disable_cursor && !fsstate->cursor_exists) create_cursor(node); + if (disable_cursor) + ereport(ERROR, errmsg("Async not supported in cursors are disabled")); + /* 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); @@ -7433,7 +7759,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 @@ -7495,7 +7821,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; @@ -7519,9 +7846,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 a2bb1ff352c..d30da230437 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -19,6 +19,7 @@ #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 +132,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; /* last query executed, required for + * non-cursor mode */ } PgFdwConnState; /* @@ -164,6 +169,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 +185,7 @@ extern List *ExtractExtensionList(const char *extensionsString, bool warnOnMissing); extern char *process_pgfdw_appname(const char *appname); extern char *pgfdw_application_name; +extern bool disable_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 0e218b29a29..a2868a5f911 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -307,6 +307,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.disable_cursor TO on; +SELECT * FROM ft1 t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; +RESET postgres_fdw.disable_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 @@ -386,6 +390,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.disable_cursor TO on; +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.disable_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)); @@ -676,6 +689,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.disable_cursor TO on; +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.disable_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; @@ -684,6 +702,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.disable_cursor TO on; +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.disable_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) @@ -3963,6 +3985,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.disable_cursor TO on; +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.disable_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)