From b0affb96df0eec357dac96f61bb12a412b112605 Mon Sep 17 00:00:00 2001 From: Sehrope Sarkuni Date: Thu, 10 Sep 2026 21:26:15 +0000 Subject: [PATCH v5 5/5] Add libpq_pipeline tests for _pq_.cursor edge cases Cover SCROLL over SELECT 1, a hash join, an aggregate, and a parallel plan, WITH HOLD on non-SELECT statements, FOR UPDATE with SCROLL and WITH HOLD, a fetch count of INT64_MIN, and nParams out of range. --- .../modules/libpq_pipeline/libpq_pipeline.c | 344 ++++++++++++++++++ 1 file changed, 344 insertions(+) diff --git a/src/test/modules/libpq_pipeline/libpq_pipeline.c b/src/test/modules/libpq_pipeline/libpq_pipeline.c index 43f1cc38ea0..0d54fdbbd7c 100644 --- a/src/test/modules/libpq_pipeline/libpq_pipeline.c +++ b/src/test/modules/libpq_pipeline/libpq_pipeline.c @@ -24,6 +24,8 @@ static void exit_nicely(PGconn *conn); +static void confirm_fetch_result(PGconn *conn, const char *what, int nrows, + int firstvalue); pg_noreturn static void pg_fatal_impl(int line, const char *fmt, ...) pg_attribute_printf(2, 3); static bool process_result(PGconn *conn, PGresult *res, int results, @@ -2514,6 +2516,303 @@ test_cursor_bind_dml(PGconn *conn) /* * Test client-side validation of cursor bind options. */ +/* + * Outside pipeline mode, so each rejection is its own command. + */ +static void +expect_bind_rejected(PGconn *conn, const char *stmt, int cursorOptions, + const char *what) +{ + if (PQsendBindWithCursorOptions(conn, stmt, 0, NULL, NULL, NULL, 0, + "rejected_portal", cursorOptions) != 1) + pg_fatal("%s: PQsendBindWithCursorOptions failed: %s", + what, PQerrorMessage(conn)); + consume_result_status(conn, PGRES_FATAL_ERROR); + consume_null_result(conn); +} + +/* + * SCROLL and WITH HOLD are rejected with FOR UPDATE/SHARE; NO SCROLL is + * fine. + */ +static void +test_cursor_bind_for_update(PGconn *conn) +{ + PGresult *res; + + fprintf(stderr, "test_cursor_bind_for_update... "); + + res = PQexec(conn, "CREATE TEMP TABLE forupd_test AS SELECT generate_series(1, 3) AS id"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("CREATE TABLE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "forupd_stmt", + "SELECT id FROM forupd_test ORDER BY id FOR UPDATE", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + expect_bind_rejected(conn, "forupd_stmt", PQ_BIND_CURSOR_SCROLL, + "SCROLL with FOR UPDATE"); + expect_bind_rejected(conn, "forupd_stmt", PQ_BIND_CURSOR_HOLD, + "HOLD with FOR UPDATE"); + expect_bind_rejected(conn, "forupd_stmt", + PQ_BIND_CURSOR_HOLD | PQ_BIND_CURSOR_SCROLL, + "HOLD | SCROLL with FOR UPDATE"); + + res = PQexec(conn, "BEGIN"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("BEGIN failed: %s", PQerrorMessage(conn)); + PQclear(res); + + if (PQsendBindAndExecutePortal(conn, "forupd_stmt", 0, NULL, NULL, NULL, 0, + "forupd_portal", PQ_BIND_CURSOR_NO_SCROLL, + PQ_FETCH_FORWARD, 2) != 1) + pg_fatal("PQsendBindAndExecutePortal failed: %s", PQerrorMessage(conn)); + confirm_fetch_result(conn, "NO SCROLL with FOR UPDATE", 2, 1); + + res = PQexec(conn, "ROLLBACK"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("ROLLBACK failed: %s", PQerrorMessage(conn)); + PQclear(res); + + fprintf(stderr, "ok\n"); +} + +/* + * WITH HOLD is only allowed for a single SELECT. + */ +static void +test_cursor_bind_hold_non_select(PGconn *conn) +{ + PGresult *res; + + fprintf(stderr, "test_cursor_bind_hold_non_select... "); + + res = PQexec(conn, "CREATE TEMP TABLE holdns_test(id int)"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("CREATE TABLE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "holdns_returning", + "INSERT INTO holdns_test VALUES (1) RETURNING id", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "holdns_update", + "UPDATE holdns_test SET id = id", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "holdns_utility", "SHOW work_mem", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "holdns_modcte", + "WITH t AS (INSERT INTO holdns_test VALUES (2) RETURNING id) SELECT id FROM t", + 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + + expect_bind_rejected(conn, "holdns_returning", PQ_BIND_CURSOR_HOLD, + "HOLD on INSERT RETURNING"); + expect_bind_rejected(conn, "holdns_update", PQ_BIND_CURSOR_HOLD, + "HOLD on UPDATE"); + expect_bind_rejected(conn, "holdns_utility", PQ_BIND_CURSOR_HOLD, + "HOLD on SHOW"); + expect_bind_rejected(conn, "holdns_modcte", PQ_BIND_CURSOR_HOLD, + "HOLD on SELECT with data-modifying CTE"); + + /* nothing was inserted */ + res = PQexec(conn, "SELECT count(*) FROM holdns_test"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("SELECT failed: %s", PQerrorMessage(conn)); + if (strcmp(PQgetvalue(res, 0, 0), "0") != 0) + pg_fatal("expected 0 rows, got %s", PQgetvalue(res, 0, 0)); + PQclear(res); + + fprintf(stderr, "ok\n"); +} + +/* + * Bind stmt as a SCROLL portal and read it in every direction. + */ +static void +check_scroll_plan(PGconn *conn, const char *stmt, const char *what) +{ + PGresult *res; + char first[64]; + char portal[64]; + + snprintf(portal, sizeof(portal), "%s_portal", stmt); + + if (PQsendBindWithCursorOptions(conn, stmt, 0, NULL, NULL, NULL, 0, + portal, PQ_BIND_CURSOR_SCROLL) != 1) + pg_fatal("%s: PQsendBindWithCursorOptions failed: %s", + what, PQerrorMessage(conn)); + res = confirm_result_status(conn, PGRES_COMMAND_OK); + PQclear(res); + consume_null_result(conn); + + /* forward 2, then backward 1 returns the first row either way */ + if (PQsendExecutePortal(conn, portal, PQ_FETCH_FORWARD, 2) != 1) + pg_fatal("%s: forward fetch failed: %s", what, PQerrorMessage(conn)); + res = confirm_result_status(conn, PGRES_TUPLES_OK); + if (PQntuples(res) < 1) + pg_fatal("%s: expected at least 1 row, got %d", what, PQntuples(res)); + strlcpy(first, PQgetvalue(res, 0, 0), sizeof(first)); + PQclear(res); + consume_null_result(conn); + + if (PQsendExecutePortal(conn, portal, PQ_FETCH_BACKWARD, 1) != 1) + pg_fatal("%s: backward fetch failed: %s", what, PQerrorMessage(conn)); + res = confirm_result_status(conn, PGRES_TUPLES_OK); + if (PQntuples(res) != 1) + pg_fatal("%s: expected 1 row from backward fetch, got %d", + what, PQntuples(res)); + if (strcmp(PQgetvalue(res, 0, 0), first) != 0) + pg_fatal("%s: expected \"%s\" from backward fetch, got \"%s\"", + what, first, PQgetvalue(res, 0, 0)); + PQclear(res); + consume_null_result(conn); + + /* full forward pass; this is where a parallel plan broke */ + if (PQsendExecutePortal(conn, portal, PQ_FETCH_FORWARD | PQ_FETCH_MOVE, + PQ_FETCH_ALL) != 1) + pg_fatal("%s: move failed: %s", what, PQerrorMessage(conn)); + res = confirm_result_status(conn, PGRES_TUPLES_OK); + if (PQntuples(res) != 0) + pg_fatal("%s: expected no rows from move, got %d", what, PQntuples(res)); + PQclear(res); + consume_null_result(conn); + + if (PQsendExecutePortal(conn, portal, PQ_FETCH_BACKWARD, 1) != 1) + pg_fatal("%s: backward fetch failed: %s", what, PQerrorMessage(conn)); + res = confirm_result_status(conn, PGRES_TUPLES_OK); + if (PQntuples(res) != 1) + pg_fatal("%s: expected 1 row from backward fetch at end, got %d", + what, PQntuples(res)); + PQclear(res); + consume_null_result(conn); + + if (PQsendExecutePortal(conn, portal, PQ_FETCH_ABSOLUTE, 1) != 1) + pg_fatal("%s: absolute fetch failed: %s", what, PQerrorMessage(conn)); + res = confirm_result_status(conn, PGRES_TUPLES_OK); + if (PQntuples(res) != 1) + pg_fatal("%s: expected 1 row from absolute fetch, got %d", + what, PQntuples(res)); + if (strcmp(PQgetvalue(res, 0, 0), first) != 0) + pg_fatal("%s: expected \"%s\" from absolute fetch, got \"%s\"", + what, first, PQgetvalue(res, 0, 0)); + PQclear(res); + consume_null_result(conn); + + if (PQsendClosePortal(conn, portal) != 1) + pg_fatal("%s: PQsendClosePortal failed: %s", what, PQerrorMessage(conn)); + consume_result_status(conn, PGRES_COMMAND_OK); + consume_null_result(conn); +} + +/* + * SCROLL on plans that need replanning: SELECT 1, hash join, aggregate, + * parallel. + */ +static void +test_cursor_bind_scroll_plans(PGconn *conn) +{ + PGresult *res; + + fprintf(stderr, "test_cursor_bind_scroll_plans... "); + + res = PQexec(conn, "BEGIN"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("BEGIN failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "scrollplan_select1", "SELECT 1", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + check_scroll_plan(conn, "scrollplan_select1", "SELECT 1"); + + res = PQexec(conn, "SET enable_nestloop = off; SET enable_mergejoin = off"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("SET failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "scrollplan_hashjoin", + "SELECT a.g FROM generate_series(1, 5) a(g) " + "JOIN generate_series(1, 5) b(g) USING (g)", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + check_scroll_plan(conn, "scrollplan_hashjoin", "hash join"); + + res = PQprepare(conn, "scrollplan_agg", + "SELECT count(*) FROM generate_series(1, 5)", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + check_scroll_plan(conn, "scrollplan_agg", "aggregate"); + + res = PQexec(conn, "SET debug_parallel_query = on"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("SET failed: %s", PQerrorMessage(conn)); + PQclear(res); + + res = PQprepare(conn, "scrollplan_parallel", + "SELECT g FROM generate_series(1, 5) g", 0, NULL); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("PREPARE failed: %s", PQerrorMessage(conn)); + PQclear(res); + check_scroll_plan(conn, "scrollplan_parallel", "parallel"); + + res = PQexec(conn, "ROLLBACK"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("ROLLBACK failed: %s", PQerrorMessage(conn)); + PQclear(res); + + /* HOLD over a replanned plan survives commit */ + res = PQexec(conn, "BEGIN"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("BEGIN failed: %s", PQerrorMessage(conn)); + PQclear(res); + + if (PQsendBindAndExecutePortal(conn, "scrollplan_hashjoin", 0, NULL, NULL, + NULL, 0, "scrollplan_hold", + PQ_BIND_CURSOR_SCROLL | PQ_BIND_CURSOR_HOLD, + PQ_FETCH_FORWARD, 2) != 1) + pg_fatal("PQsendBindAndExecutePortal failed: %s", PQerrorMessage(conn)); + confirm_fetch_result(conn, "hold+scroll before commit", 2, -1); + + res = PQexec(conn, "COMMIT"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("COMMIT failed: %s", PQerrorMessage(conn)); + PQclear(res); + + if (PQsendExecutePortal(conn, "scrollplan_hold", PQ_FETCH_BACKWARD, + PQ_FETCH_ALL) != 1) + pg_fatal("backward fetch failed: %s", PQerrorMessage(conn)); + confirm_fetch_result(conn, "hold+scroll after commit", 1, -1); + + if (PQsendExecutePortal(conn, "scrollplan_hold", PQ_FETCH_FORWARD, + PQ_FETCH_ALL) != 1) + pg_fatal("forward fetch failed: %s", PQerrorMessage(conn)); + confirm_fetch_result(conn, "hold+scroll forward all", 5, -1); + + if (PQsendClosePortal(conn, "scrollplan_hold") != 1) + pg_fatal("PQsendClosePortal failed: %s", PQerrorMessage(conn)); + consume_result_status(conn, PGRES_COMMAND_OK); + consume_null_result(conn); + + fprintf(stderr, "ok\n"); +} + static void test_cursor_bind_validation(PGconn *conn) { @@ -2555,6 +2854,19 @@ test_cursor_bind_validation(PGconn *conn) PQ_BIND_CURSOR_SCROLL | PQ_BIND_CURSOR_NO_SCROLL) != 0) pg_fatal("expected rejection of SCROLL | NO_SCROLL"); + /* parameter count out of range */ + if (PQsendBindWithCursorOptions(conn, "valstmt", -1, NULL, NULL, NULL, 0, + "p", 0) != 0) + pg_fatal("expected rejection of negative parameter count"); + + if (PQsendBindWithCursorOptions(conn, "valstmt", 65536, NULL, NULL, NULL, 0, + "p", 0) != 0) + pg_fatal("expected rejection of too many parameters"); + + if (PQsendBindAndExecutePortal(conn, "valstmt", 65536, NULL, NULL, NULL, 0, + "p", 0, PQ_FETCH_FORWARD, 1) != 0) + pg_fatal("expected rejection of too many parameters"); + if (PQexitPipelineMode(conn) != 1) pg_fatal("failed to exit pipeline mode: %s", PQerrorMessage(conn)); @@ -3000,6 +3312,29 @@ test_cursor_execute_validation(PGconn *conn) if (PQexitPipelineMode(conn) != 1) pg_fatal("failed to exit pipeline mode: %s", PQerrorMessage(conn)); + /* INT64_MIN is rejected by the server */ + res = PQexec(conn, "BEGIN"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("BEGIN failed: %s", PQerrorMessage(conn)); + PQclear(res); + + if (PQsendBindWithCursorOptions(conn, "exvalstmt", 0, NULL, NULL, NULL, 0, + "exvalportal", PQ_BIND_CURSOR_SCROLL) != 1) + pg_fatal("PQsendBindWithCursorOptions failed: %s", PQerrorMessage(conn)); + consume_result_status(conn, PGRES_COMMAND_OK); + consume_null_result(conn); + + if (PQsendExecutePortal(conn, "exvalportal", PQ_FETCH_FORWARD, + INT64_MIN) != 1) + pg_fatal("PQsendExecutePortal failed: %s", PQerrorMessage(conn)); + consume_result_status(conn, PGRES_FATAL_ERROR); + consume_null_result(conn); + + res = PQexec(conn, "ROLLBACK"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + pg_fatal("ROLLBACK failed: %s", PQerrorMessage(conn)); + PQclear(res); + fprintf(stderr, "ok\n"); } @@ -3058,10 +3393,13 @@ print_test_list(void) { printf("cancel\n"); printf("cursor_bind_dml\n"); + printf("cursor_bind_for_update\n"); + printf("cursor_bind_hold_non_select\n"); printf("cursor_bind_holdable\n"); printf("cursor_bind_holdable_scroll\n"); printf("cursor_bind_no_scroll\n"); printf("cursor_bind_scroll\n"); + printf("cursor_bind_scroll_plans\n"); printf("cursor_bind_validation\n"); printf("cursor_bind_without_extension\n"); printf("cursor_execute_bind_and_fetch\n"); @@ -3178,6 +3516,10 @@ main(int argc, char **argv) test_cancel(conn); else if (strcmp(testname, "cursor_bind_dml") == 0) test_cursor_bind_dml(conn); + else if (strcmp(testname, "cursor_bind_for_update") == 0) + test_cursor_bind_for_update(conn); + else if (strcmp(testname, "cursor_bind_hold_non_select") == 0) + test_cursor_bind_hold_non_select(conn); else if (strcmp(testname, "cursor_bind_holdable") == 0) test_cursor_bind_holdable(conn); else if (strcmp(testname, "cursor_bind_holdable_scroll") == 0) @@ -3186,6 +3528,8 @@ main(int argc, char **argv) test_cursor_bind_no_scroll(conn); else if (strcmp(testname, "cursor_bind_scroll") == 0) test_cursor_bind_scroll(conn); + else if (strcmp(testname, "cursor_bind_scroll_plans") == 0) + test_cursor_bind_scroll_plans(conn); else if (strcmp(testname, "cursor_bind_validation") == 0) test_cursor_bind_validation(conn); else if (strcmp(testname, "cursor_bind_without_extension") == 0) -- 2.17.1