Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * libpq_pipeline.c
4 : : * Verify libpq pipeline execution functionality
5 : : *
6 : : * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : *
10 : : * IDENTIFICATION
11 : : * src/test/modules/libpq_pipeline/libpq_pipeline.c
12 : : *
13 : : *-------------------------------------------------------------------------
14 : : */
15 : :
16 : : #include "postgres_fe.h"
17 : :
18 : : #include <sys/select.h>
19 : : #include <sys/time.h>
20 : :
21 : : #include "catalog/pg_type_d.h"
22 : : #include "libpq-fe.h"
23 : : #include "pg_getopt.h"
24 : :
25 : :
26 : : static void exit_nicely(PGconn *conn);
27 : : pg_noreturn static void pg_fatal_impl(int line, const char *fmt,...)
28 : : pg_attribute_printf(2, 3);
29 : : static bool process_result(PGconn *conn, PGresult *res, int results,
30 : : int numsent);
31 : :
32 : : static const char *const progname = "libpq_pipeline";
33 : :
34 : : /* Options and defaults */
35 : : static char *tracefile = NULL; /* path to PQtrace() file */
36 : :
37 : :
38 : : #ifdef DEBUG_OUTPUT
39 : : #define pg_debug(...) do { fprintf(stderr, __VA_ARGS__); } while (0)
40 : : #else
41 : : #define pg_debug(...)
42 : : #endif
43 : :
44 : : static const char *const drop_table_sql =
45 : : "DROP TABLE IF EXISTS pq_pipeline_demo";
46 : : static const char *const create_table_sql =
47 : : "CREATE UNLOGGED TABLE pq_pipeline_demo(id serial primary key, itemno integer,"
48 : : "int8filler int8);";
49 : : static const char *const insert_sql =
50 : : "INSERT INTO pq_pipeline_demo(itemno) VALUES ($1)";
51 : : static const char *const insert_sql2 =
52 : : "INSERT INTO pq_pipeline_demo(itemno,int8filler) VALUES ($1, $2)";
53 : :
54 : : /* max char length of an int32/64, plus sign and null terminator */
55 : : #define MAXINTLEN 12
56 : : #define MAXINT8LEN 20
57 : :
58 : : static void
1688 alvherre@alvh.no-ip. 59 :UBC 0 : exit_nicely(PGconn *conn)
60 : : {
61 : 0 : PQfinish(conn);
62 : 0 : exit(1);
63 : : }
64 : :
65 : : /*
66 : : * The following few functions are wrapped in macros to make the reported line
67 : : * number in an error match the line number of the invocation.
68 : : */
69 : :
70 : : /*
71 : : * Print an error to stderr and terminate the program.
72 : : */
73 : : #define pg_fatal(...) pg_fatal_impl(__LINE__, __VA_ARGS__)
74 : : pg_noreturn static void
75 : 0 : pg_fatal_impl(int line, const char *fmt,...)
76 : : {
77 : : va_list args;
78 : :
79 : 0 : fflush(stdout);
80 : :
81 : 0 : fprintf(stderr, "\n%s:%d: ", progname, line);
82 : 0 : va_start(args, fmt);
83 : 0 : vfprintf(stderr, fmt, args);
84 : 0 : va_end(args);
85 [ # # ]: 0 : Assert(fmt[strlen(fmt) - 1] != '\n');
86 : 0 : fprintf(stderr, "\n");
87 : 0 : exit(1);
88 : : }
89 : :
90 : : /*
91 : : * Check that libpq next returns a PGresult with the specified status,
92 : : * returning the PGresult so that caller can perform additional checks.
93 : : */
94 : : #define confirm_result_status(conn, status) confirm_result_status_impl(__LINE__, conn, status)
95 : : static PGresult *
55 tgl@sss.pgh.pa.us 96 :GNC 56 : confirm_result_status_impl(int line, PGconn *conn, ExecStatusType status)
97 : : {
98 : : PGresult *res;
99 : :
596 alvherre@alvh.no-ip. 100 :CBC 56 : res = PQgetResult(conn);
101 [ - + ]: 56 : if (res == NULL)
55 tgl@sss.pgh.pa.us 102 :UNC 0 : pg_fatal_impl(line, "PQgetResult returned null unexpectedly: %s",
103 : : PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 104 [ - + ]:GNC 56 : if (PQresultStatus(res) != status)
55 tgl@sss.pgh.pa.us 105 :UNC 0 : pg_fatal_impl(line, "PQgetResult returned status %s, expected %s: %s",
106 : : PQresStatus(PQresultStatus(res)),
107 : : PQresStatus(status),
108 : : PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 109 :GNC 56 : return res;
110 : : }
111 : :
112 : : /*
113 : : * Check that libpq next returns a PGresult with the specified status,
114 : : * then free the PGresult.
115 : : */
116 : : #define consume_result_status(conn, status) consume_result_status_impl(__LINE__, conn, status)
117 : : static void
118 : 39 : consume_result_status_impl(int line, PGconn *conn, ExecStatusType status)
119 : : {
120 : : PGresult *res;
121 : :
122 : 39 : res = confirm_result_status_impl(line, conn, status);
123 : 39 : PQclear(res);
124 : 39 : }
125 : :
126 : : /*
127 : : * Check that libpq next returns a null PGresult.
128 : : */
129 : : #define consume_null_result(conn) consume_null_result_impl(__LINE__, conn)
130 : : static void
131 : 631 : consume_null_result_impl(int line, PGconn *conn)
132 : : {
133 : : PGresult *res;
134 : :
135 : 631 : res = PQgetResult(conn);
136 [ - + ]: 631 : if (res != NULL)
55 tgl@sss.pgh.pa.us 137 :UNC 0 : pg_fatal_impl(line, "expected NULL PGresult, got %s: %s",
138 : : PQresStatus(PQresultStatus(res)),
139 : : PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 140 :GNC 631 : }
141 : :
142 : : /*
143 : : * Check that the query on the given connection got canceled.
144 : : */
145 : : #define consume_query_cancel(conn) consume_query_cancel_impl(__LINE__, conn)
146 : : static void
147 : 12 : consume_query_cancel_impl(int line, PGconn *conn)
148 : : {
149 : : PGresult *res;
150 : :
151 : 12 : res = confirm_result_status_impl(line, conn, PGRES_FATAL_ERROR);
596 alvherre@alvh.no-ip. 152 [ - + ]:CBC 12 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "57014") != 0)
596 alvherre@alvh.no-ip. 153 :UBC 0 : pg_fatal_impl(line, "query failed with a different error than cancellation: %s",
154 : : PQerrorMessage(conn));
596 alvherre@alvh.no-ip. 155 :CBC 12 : PQclear(res);
156 : :
157 [ - + ]: 12 : while (PQisBusy(conn))
596 alvherre@alvh.no-ip. 158 :UBC 0 : PQconsumeInput(conn);
596 alvherre@alvh.no-ip. 159 :CBC 12 : }
160 : :
161 : : /*
162 : : * Using monitorConn, query pg_stat_activity to see that the connection with
163 : : * the given PID is either in the given state, or waiting on the given event
164 : : * (only one of them can be given).
165 : : */
166 : : static void
589 167 : 24 : wait_for_connection_state(int line, PGconn *monitorConn, int procpid,
168 : : char *state, char *event)
169 : : {
594 170 : 24 : const Oid paramTypes[] = {INT4OID, TEXTOID};
171 : : const char *paramValues[2];
172 : 24 : char *pidstr = psprintf("%d", procpid);
173 : :
589 174 [ - + ]: 24 : Assert((state == NULL) ^ (event == NULL));
175 : :
594 176 : 24 : paramValues[0] = pidstr;
589 177 [ + + ]: 24 : paramValues[1] = state ? state : event;
178 : :
179 : : while (true)
596 alvherre@alvh.no-ip. 180 :UBC 0 : {
181 : : PGresult *res;
182 : : char *value;
183 : :
589 alvherre@alvh.no-ip. 184 [ + + ]:CBC 24 : if (state != NULL)
185 : 12 : res = PQexecParams(monitorConn,
186 : : "SELECT count(*) FROM pg_stat_activity WHERE "
187 : : "pid = $1 AND state = $2",
188 : : 2, paramTypes, paramValues, NULL, NULL, 0);
189 : : else
190 : 12 : res = PQexecParams(monitorConn,
191 : : "SELECT count(*) FROM pg_stat_activity WHERE "
192 : : "pid = $1 AND wait_event = $2",
193 : : 2, paramTypes, paramValues, NULL, NULL, 0);
194 : :
596 195 [ - + ]: 24 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
594 alvherre@alvh.no-ip. 196 :UBC 0 : pg_fatal_impl(line, "could not query pg_stat_activity: %s", PQerrorMessage(monitorConn));
596 alvherre@alvh.no-ip. 197 [ - + ]:CBC 24 : if (PQntuples(res) != 1)
594 alvherre@alvh.no-ip. 198 :UBC 0 : pg_fatal_impl(line, "unexpected number of rows received: %d", PQntuples(res));
596 alvherre@alvh.no-ip. 199 [ - + ]:CBC 24 : if (PQnfields(res) != 1)
594 alvherre@alvh.no-ip. 200 :UBC 0 : pg_fatal_impl(line, "unexpected number of columns received: %d", PQnfields(res));
596 alvherre@alvh.no-ip. 201 :CBC 24 : value = PQgetvalue(res, 0, 0);
589 202 [ + - ]: 24 : if (strcmp(value, "0") != 0)
203 : : {
596 204 : 24 : PQclear(res);
205 : 24 : break;
206 : : }
596 alvherre@alvh.no-ip. 207 :UBC 0 : PQclear(res);
208 : :
209 : : /* wait 10ms before polling again */
210 : 0 : pg_usleep(10000);
211 : : }
212 : :
594 alvherre@alvh.no-ip. 213 :CBC 24 : pfree(pidstr);
214 : 24 : }
215 : :
216 : : #define send_cancellable_query(conn, monitorConn) \
217 : : send_cancellable_query_impl(__LINE__, conn, monitorConn)
218 : : static void
219 : 12 : send_cancellable_query_impl(int line, PGconn *conn, PGconn *monitorConn)
220 : : {
221 : : const char *env_wait;
222 : 12 : const Oid paramTypes[1] = {INT4OID};
223 : :
224 : : /*
225 : : * Wait for the connection to be idle, so that our check for an active
226 : : * connection below is reliable, instead of possibly seeing an outdated
227 : : * state.
228 : : */
589 229 : 12 : wait_for_connection_state(line, monitorConn, PQbackendPID(conn), "idle", NULL);
230 : :
594 231 : 12 : env_wait = getenv("PG_TEST_TIMEOUT_DEFAULT");
232 [ + - ]: 12 : if (env_wait == NULL)
233 : 12 : env_wait = "180";
234 : :
235 [ - + ]: 12 : if (PQsendQueryParams(conn, "SELECT pg_sleep($1)", 1, paramTypes,
236 : : &env_wait, NULL, NULL, 0) != 1)
594 alvherre@alvh.no-ip. 237 :UBC 0 : pg_fatal_impl(line, "failed to send query: %s", PQerrorMessage(conn));
238 : :
239 : : /*
240 : : * Wait for the sleep to be active, because if the query is not running
241 : : * yet, the cancel request that we send won't have any effect.
242 : : */
589 alvherre@alvh.no-ip. 243 :CBC 12 : wait_for_connection_state(line, monitorConn, PQbackendPID(conn), NULL, "PgSleep");
596 244 : 12 : }
245 : :
246 : : /*
247 : : * Create a new connection with the same conninfo as the given one.
248 : : */
249 : : static PGconn *
250 : 2 : copy_connection(PGconn *conn)
251 : : {
252 : : PGconn *copyConn;
253 : 2 : PQconninfoOption *opts = PQconninfo(conn);
254 : : const char **keywords;
255 : : const char **vals;
209 heikki.linnakangas@i 256 : 2 : int nopts = 0;
257 : : int i;
258 : :
596 alvherre@alvh.no-ip. 259 [ + + ]: 104 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
260 : 102 : nopts++;
209 heikki.linnakangas@i 261 : 2 : nopts++; /* for the NULL terminator */
262 : :
596 alvherre@alvh.no-ip. 263 : 2 : keywords = pg_malloc(sizeof(char *) * nopts);
264 : 2 : vals = pg_malloc(sizeof(char *) * nopts);
265 : :
209 heikki.linnakangas@i 266 : 2 : i = 0;
596 alvherre@alvh.no-ip. 267 [ + + ]: 104 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
268 : : {
269 [ + + ]: 102 : if (opt->val)
270 : : {
271 : 40 : keywords[i] = opt->keyword;
272 : 40 : vals[i] = opt->val;
273 : 40 : i++;
274 : : }
275 : : }
276 : 2 : keywords[i] = vals[i] = NULL;
277 : :
278 : 2 : copyConn = PQconnectdbParams(keywords, vals, false);
279 : :
280 [ - + ]: 2 : if (PQstatus(copyConn) != CONNECTION_OK)
596 alvherre@alvh.no-ip. 281 :UBC 0 : pg_fatal("Connection to database failed: %s",
282 : : PQerrorMessage(copyConn));
283 : :
55 tgl@sss.pgh.pa.us 284 :GNC 2 : pfree(keywords);
285 : 2 : pfree(vals);
286 : 2 : PQconninfoFree(opts);
287 : :
596 alvherre@alvh.no-ip. 288 :CBC 2 : return copyConn;
289 : : }
290 : :
291 : : /*
292 : : * Test query cancellation routines
293 : : */
294 : : static void
295 : 2 : test_cancel(PGconn *conn)
296 : : {
297 : : PGcancel *cancel;
298 : : PGcancelConn *cancelConn;
299 : : PGconn *monitorConn;
300 : : char errorbuf[256];
301 : :
302 : 2 : fprintf(stderr, "test cancellations... ");
303 : :
304 [ - + ]: 2 : if (PQsetnonblocking(conn, 1) != 0)
596 alvherre@alvh.no-ip. 305 :UBC 0 : pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn));
306 : :
307 : : /*
308 : : * Make a separate connection to the database to monitor the query on the
309 : : * main connection.
310 : : */
596 alvherre@alvh.no-ip. 311 :CBC 2 : monitorConn = copy_connection(conn);
312 [ - + ]: 2 : Assert(PQstatus(monitorConn) == CONNECTION_OK);
313 : :
314 : : /* test PQcancel */
315 : 2 : send_cancellable_query(conn, monitorConn);
316 : 2 : cancel = PQgetCancel(conn);
317 [ - + ]: 2 : if (!PQcancel(cancel, errorbuf, sizeof(errorbuf)))
596 alvherre@alvh.no-ip. 318 :UBC 0 : pg_fatal("failed to run PQcancel: %s", errorbuf);
55 tgl@sss.pgh.pa.us 319 :GNC 2 : consume_query_cancel(conn);
320 : :
321 : : /* PGcancel object can be reused for the next query */
596 alvherre@alvh.no-ip. 322 :CBC 2 : send_cancellable_query(conn, monitorConn);
323 [ - + ]: 2 : if (!PQcancel(cancel, errorbuf, sizeof(errorbuf)))
596 alvherre@alvh.no-ip. 324 :UBC 0 : pg_fatal("failed to run PQcancel: %s", errorbuf);
55 tgl@sss.pgh.pa.us 325 :GNC 2 : consume_query_cancel(conn);
326 : :
596 alvherre@alvh.no-ip. 327 :CBC 2 : PQfreeCancel(cancel);
328 : :
329 : : /* test PQrequestCancel */
330 : 2 : send_cancellable_query(conn, monitorConn);
331 [ - + ]: 2 : if (!PQrequestCancel(conn))
596 alvherre@alvh.no-ip. 332 :UBC 0 : pg_fatal("failed to run PQrequestCancel: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 333 :GNC 2 : consume_query_cancel(conn);
334 : :
335 : : /* test PQcancelBlocking */
595 alvherre@alvh.no-ip. 336 :CBC 2 : send_cancellable_query(conn, monitorConn);
337 : 2 : cancelConn = PQcancelCreate(conn);
338 [ - + ]: 2 : if (!PQcancelBlocking(cancelConn))
595 alvherre@alvh.no-ip. 339 :UBC 0 : pg_fatal("failed to run PQcancelBlocking: %s", PQcancelErrorMessage(cancelConn));
55 tgl@sss.pgh.pa.us 340 :GNC 2 : consume_query_cancel(conn);
595 alvherre@alvh.no-ip. 341 :CBC 2 : PQcancelFinish(cancelConn);
342 : :
343 : : /* test PQcancelCreate and then polling with PQcancelPoll */
344 : 2 : send_cancellable_query(conn, monitorConn);
345 : 2 : cancelConn = PQcancelCreate(conn);
346 [ - + ]: 2 : if (!PQcancelStart(cancelConn))
595 alvherre@alvh.no-ip. 347 :UBC 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
348 : : while (true)
595 alvherre@alvh.no-ip. 349 :CBC 2 : {
350 : : struct timeval tv;
351 : : fd_set input_mask;
352 : : fd_set output_mask;
353 : 4 : PostgresPollingStatusType pollres = PQcancelPoll(cancelConn);
354 : 4 : int sock = PQcancelSocket(cancelConn);
355 : :
356 [ + + ]: 4 : if (pollres == PGRES_POLLING_OK)
357 : 2 : break;
358 : :
359 [ + + ]: 34 : FD_ZERO(&input_mask);
360 [ + + ]: 34 : FD_ZERO(&output_mask);
361 [ + - - ]: 2 : switch (pollres)
362 : : {
363 : 2 : case PGRES_POLLING_READING:
364 : : pg_debug("polling for reads\n");
365 : 2 : FD_SET(sock, &input_mask);
366 : 2 : break;
595 alvherre@alvh.no-ip. 367 :UBC 0 : case PGRES_POLLING_WRITING:
368 : : pg_debug("polling for writes\n");
369 : 0 : FD_SET(sock, &output_mask);
370 : 0 : break;
371 : 0 : default:
372 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
373 : : }
374 : :
595 alvherre@alvh.no-ip. 375 [ - + ]:CBC 2 : if (sock < 0)
595 alvherre@alvh.no-ip. 376 :UBC 0 : pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn));
377 : :
595 alvherre@alvh.no-ip. 378 :CBC 2 : tv.tv_sec = 3;
379 : 2 : tv.tv_usec = 0;
380 : :
381 : : while (true)
382 : : {
383 [ - + ]: 2 : if (select(sock + 1, &input_mask, &output_mask, NULL, &tv) < 0)
384 : : {
595 alvherre@alvh.no-ip. 385 [ # # ]:UBC 0 : if (errno == EINTR)
386 : 0 : continue;
387 : 0 : pg_fatal("select() failed: %m");
388 : : }
595 alvherre@alvh.no-ip. 389 :CBC 2 : break;
390 : : }
391 : : }
392 [ - + ]: 2 : if (PQcancelStatus(cancelConn) != CONNECTION_OK)
595 alvherre@alvh.no-ip. 393 :UBC 0 : pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn));
55 tgl@sss.pgh.pa.us 394 :GNC 2 : consume_query_cancel(conn);
395 : :
396 : : /*
397 : : * test PQcancelReset works on the cancel connection and it can be reused
398 : : * afterwards
399 : : */
595 alvherre@alvh.no-ip. 400 :CBC 2 : PQcancelReset(cancelConn);
401 : :
402 : 2 : send_cancellable_query(conn, monitorConn);
403 [ - + ]: 2 : if (!PQcancelStart(cancelConn))
595 alvherre@alvh.no-ip. 404 :UBC 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
405 : : while (true)
595 alvherre@alvh.no-ip. 406 :CBC 2 : {
407 : : struct timeval tv;
408 : : fd_set input_mask;
409 : : fd_set output_mask;
410 : 4 : PostgresPollingStatusType pollres = PQcancelPoll(cancelConn);
411 : 4 : int sock = PQcancelSocket(cancelConn);
412 : :
413 [ + + ]: 4 : if (pollres == PGRES_POLLING_OK)
414 : 2 : break;
415 : :
416 [ + + ]: 34 : FD_ZERO(&input_mask);
417 [ + + ]: 34 : FD_ZERO(&output_mask);
418 [ + - - ]: 2 : switch (pollres)
419 : : {
420 : 2 : case PGRES_POLLING_READING:
421 : : pg_debug("polling for reads\n");
422 : 2 : FD_SET(sock, &input_mask);
423 : 2 : break;
595 alvherre@alvh.no-ip. 424 :UBC 0 : case PGRES_POLLING_WRITING:
425 : : pg_debug("polling for writes\n");
426 : 0 : FD_SET(sock, &output_mask);
427 : 0 : break;
428 : 0 : default:
429 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
430 : : }
431 : :
595 alvherre@alvh.no-ip. 432 [ - + ]:CBC 2 : if (sock < 0)
595 alvherre@alvh.no-ip. 433 :UBC 0 : pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn));
434 : :
595 alvherre@alvh.no-ip. 435 :CBC 2 : tv.tv_sec = 3;
436 : 2 : tv.tv_usec = 0;
437 : :
438 : : while (true)
439 : : {
440 [ - + ]: 2 : if (select(sock + 1, &input_mask, &output_mask, NULL, &tv) < 0)
441 : : {
595 alvherre@alvh.no-ip. 442 [ # # ]:UBC 0 : if (errno == EINTR)
443 : 0 : continue;
444 : 0 : pg_fatal("select() failed: %m");
445 : : }
595 alvherre@alvh.no-ip. 446 :CBC 2 : break;
447 : : }
448 : : }
449 [ - + ]: 2 : if (PQcancelStatus(cancelConn) != CONNECTION_OK)
595 alvherre@alvh.no-ip. 450 :UBC 0 : pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn));
55 tgl@sss.pgh.pa.us 451 :GNC 2 : consume_query_cancel(conn);
452 : :
595 alvherre@alvh.no-ip. 453 :CBC 2 : PQcancelFinish(cancelConn);
55 tgl@sss.pgh.pa.us 454 :GNC 2 : PQfinish(monitorConn);
455 : :
596 alvherre@alvh.no-ip. 456 :CBC 2 : fprintf(stderr, "ok\n");
457 : 2 : }
458 : :
459 : : static void
1688 460 : 1 : test_disallowed_in_pipeline(PGconn *conn)
461 : : {
462 : 1 : PGresult *res = NULL;
463 : :
464 : 1 : fprintf(stderr, "test error cases... ");
465 : :
466 [ - + ]: 1 : if (PQisnonblocking(conn))
1688 alvherre@alvh.no-ip. 467 :UBC 0 : pg_fatal("Expected blocking connection mode");
468 : :
1688 alvherre@alvh.no-ip. 469 [ - + ]:CBC 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 470 :UBC 0 : pg_fatal("Unable to enter pipeline mode");
471 : :
1688 alvherre@alvh.no-ip. 472 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 473 :UBC 0 : pg_fatal("Pipeline mode not activated properly");
474 : :
475 : : /* PQexec should fail in pipeline mode */
1688 alvherre@alvh.no-ip. 476 :CBC 1 : res = PQexec(conn, "SELECT 1");
477 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
1688 alvherre@alvh.no-ip. 478 :UBC 0 : pg_fatal("PQexec should fail in pipeline mode but succeeded");
1131 alvherre@alvh.no-ip. 479 [ - + ]:CBC 1 : if (strcmp(PQerrorMessage(conn),
480 : : "synchronous command execution functions are not allowed in pipeline mode\n") != 0)
1131 alvherre@alvh.no-ip. 481 :UBC 0 : pg_fatal("did not get expected error message; got: \"%s\"",
482 : : PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 483 :GNC 1 : PQclear(res);
484 : :
485 : : /* PQsendQuery should fail in pipeline mode */
1131 alvherre@alvh.no-ip. 486 [ - + ]:CBC 1 : if (PQsendQuery(conn, "SELECT 1") != 0)
1131 alvherre@alvh.no-ip. 487 :UBC 0 : pg_fatal("PQsendQuery should fail in pipeline mode but succeeded");
1131 alvherre@alvh.no-ip. 488 [ - + ]:CBC 1 : if (strcmp(PQerrorMessage(conn),
489 : : "PQsendQuery not allowed in pipeline mode\n") != 0)
1131 alvherre@alvh.no-ip. 490 :UBC 0 : pg_fatal("did not get expected error message; got: \"%s\"",
491 : : PQerrorMessage(conn));
492 : :
493 : : /* Entering pipeline mode when already in pipeline mode is OK */
1688 alvherre@alvh.no-ip. 494 [ - + ]:CBC 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 495 :UBC 0 : pg_fatal("re-entering pipeline mode should be a no-op but failed");
496 : :
1688 alvherre@alvh.no-ip. 497 [ - + ]:CBC 1 : if (PQisBusy(conn) != 0)
1688 alvherre@alvh.no-ip. 498 :UBC 0 : pg_fatal("PQisBusy should return 0 when idle in pipeline mode, returned 1");
499 : :
500 : : /* ok, back to normal command mode */
1688 alvherre@alvh.no-ip. 501 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 502 :UBC 0 : pg_fatal("couldn't exit idle empty pipeline mode");
503 : :
1688 alvherre@alvh.no-ip. 504 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 505 :UBC 0 : pg_fatal("Pipeline mode not terminated properly");
506 : :
507 : : /* exiting pipeline mode when not in pipeline mode should be a no-op */
1688 alvherre@alvh.no-ip. 508 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 509 :UBC 0 : pg_fatal("pipeline mode exit when not in pipeline mode should succeed but failed");
510 : :
511 : : /* can now PQexec again */
1688 alvherre@alvh.no-ip. 512 :CBC 1 : res = PQexec(conn, "SELECT 1");
513 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
1688 alvherre@alvh.no-ip. 514 :UBC 0 : pg_fatal("PQexec should succeed after exiting pipeline mode but failed with: %s",
515 : : PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 516 :GNC 1 : PQclear(res);
517 : :
1688 alvherre@alvh.no-ip. 518 :CBC 1 : fprintf(stderr, "ok\n");
519 : 1 : }
520 : :
521 : : static void
522 : 1 : test_multi_pipelines(PGconn *conn)
523 : : {
524 : 1 : const char *dummy_params[1] = {"1"};
525 : 1 : Oid dummy_param_oids[1] = {INT4OID};
526 : :
527 : 1 : fprintf(stderr, "multi pipeline... ");
528 : :
529 : : /*
530 : : * Queue up a couple of small pipelines and process each without returning
531 : : * to command mode first.
532 : : */
533 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 534 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
535 : :
536 : : /* first pipeline */
1688 alvherre@alvh.no-ip. 537 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
538 : : dummy_params, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 539 :UBC 0 : pg_fatal("dispatching first SELECT failed: %s", PQerrorMessage(conn));
540 : :
1688 alvherre@alvh.no-ip. 541 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 542 :UBC 0 : pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
543 : :
544 : : /* second pipeline */
1688 alvherre@alvh.no-ip. 545 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
546 : : dummy_params, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 547 :UBC 0 : pg_fatal("dispatching second SELECT failed: %s", PQerrorMessage(conn));
548 : :
549 : : /* Skip flushing once. */
651 michael@paquier.xyz 550 [ - + ]:CBC 1 : if (PQsendPipelineSync(conn) != 1)
651 michael@paquier.xyz 551 :UBC 0 : pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
552 : :
553 : : /* third pipeline */
651 michael@paquier.xyz 554 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
555 : : dummy_params, NULL, NULL, 0) != 1)
651 michael@paquier.xyz 556 :UBC 0 : pg_fatal("dispatching third SELECT failed: %s", PQerrorMessage(conn));
557 : :
1688 alvherre@alvh.no-ip. 558 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 559 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
560 : :
561 : : /* OK, start processing the results */
562 : :
563 : : /* first pipeline */
55 tgl@sss.pgh.pa.us 564 :GNC 1 : consume_result_status(conn, PGRES_TUPLES_OK);
565 : :
566 : 1 : consume_null_result(conn);
567 : :
1688 alvherre@alvh.no-ip. 568 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 0)
1688 alvherre@alvh.no-ip. 569 :UBC 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
570 : :
55 tgl@sss.pgh.pa.us 571 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
572 : :
573 : : /* second pipeline */
574 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
575 : :
576 : 1 : consume_null_result(conn);
577 : :
651 michael@paquier.xyz 578 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 0)
651 michael@paquier.xyz 579 :UBC 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
580 : :
55 tgl@sss.pgh.pa.us 581 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
582 : :
583 : : /* third pipeline */
584 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
585 : :
586 : 1 : consume_null_result(conn);
587 : :
588 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
589 : :
590 : : /* We're still in pipeline mode ... */
1688 alvherre@alvh.no-ip. 591 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 592 :UBC 0 : pg_fatal("Fell out of pipeline mode somehow");
593 : :
594 : : /* until we end it, which we can safely do now */
1688 alvherre@alvh.no-ip. 595 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 596 :UBC 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
597 : : PQerrorMessage(conn));
598 : :
1688 alvherre@alvh.no-ip. 599 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 600 :UBC 0 : pg_fatal("exiting pipeline mode didn't seem to work");
601 : :
1688 alvherre@alvh.no-ip. 602 :CBC 1 : fprintf(stderr, "ok\n");
603 : 1 : }
604 : :
605 : : /*
606 : : * Test behavior when a pipeline dispatches a number of commands that are
607 : : * not flushed by a sync point.
608 : : */
609 : : static void
1582 610 : 1 : test_nosync(PGconn *conn)
611 : : {
612 : 1 : int numqueries = 10;
613 : 1 : int results = 0;
614 : 1 : int sock = PQsocket(conn);
615 : :
616 : 1 : fprintf(stderr, "nosync... ");
617 : :
618 [ - + ]: 1 : if (sock < 0)
1582 alvherre@alvh.no-ip. 619 :UBC 0 : pg_fatal("invalid socket");
620 : :
1582 alvherre@alvh.no-ip. 621 [ - + ]:CBC 1 : if (PQenterPipelineMode(conn) != 1)
1582 alvherre@alvh.no-ip. 622 :UBC 0 : pg_fatal("could not enter pipeline mode");
1582 alvherre@alvh.no-ip. 623 [ + + ]:CBC 11 : for (int i = 0; i < numqueries; i++)
624 : : {
625 : : fd_set input_mask;
626 : : struct timeval tv;
627 : :
628 [ - + ]: 10 : if (PQsendQueryParams(conn, "SELECT repeat('xyzxz', 12)",
629 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1582 alvherre@alvh.no-ip. 630 :UBC 0 : pg_fatal("error sending select: %s", PQerrorMessage(conn));
1582 alvherre@alvh.no-ip. 631 :CBC 10 : PQflush(conn);
632 : :
633 : : /*
634 : : * If the server has written anything to us, read (some of) it now.
635 : : */
636 [ + + ]: 170 : FD_ZERO(&input_mask);
637 : 10 : FD_SET(sock, &input_mask);
638 : 10 : tv.tv_sec = 0;
639 : 10 : tv.tv_usec = 0;
640 [ - + ]: 10 : if (select(sock + 1, &input_mask, NULL, NULL, &tv) < 0)
641 : : {
595 michael@paquier.xyz 642 :UBC 0 : fprintf(stderr, "select() failed: %m\n");
1582 alvherre@alvh.no-ip. 643 : 0 : exit_nicely(conn);
644 : : }
1582 alvherre@alvh.no-ip. 645 [ - + - - ]:CBC 10 : if (FD_ISSET(sock, &input_mask) && PQconsumeInput(conn) != 1)
1582 alvherre@alvh.no-ip. 646 :UBC 0 : pg_fatal("failed to read from server: %s", PQerrorMessage(conn));
647 : : }
648 : :
649 : : /* tell server to flush its output buffer */
1582 alvherre@alvh.no-ip. 650 [ - + ]:CBC 1 : if (PQsendFlushRequest(conn) != 1)
1582 alvherre@alvh.no-ip. 651 :UBC 0 : pg_fatal("failed to send flush request");
1582 alvherre@alvh.no-ip. 652 :CBC 1 : PQflush(conn);
653 : :
654 : : /* Now read all results */
655 : : for (;;)
1582 alvherre@alvh.no-ip. 656 :ECB (9) : {
657 : : /* We expect exactly one TUPLES_OK result for each query we sent */
55 tgl@sss.pgh.pa.us 658 :GNC 10 : consume_result_status(conn, PGRES_TUPLES_OK);
659 : :
660 : : /* and one NULL result should follow each */
661 : 10 : consume_null_result(conn);
662 : :
663 : 10 : results++;
664 : :
665 : : /* if we're done, we're done */
666 [ + + ]: 10 : if (results == numqueries)
667 : 1 : break;
668 : : }
669 : :
1582 alvherre@alvh.no-ip. 670 :CBC 1 : fprintf(stderr, "ok\n");
671 : 1 : }
672 : :
673 : : /*
674 : : * When an operation in a pipeline fails the rest of the pipeline is flushed. We
675 : : * still have to get results for each pipeline item, but the item will just be
676 : : * a PGRES_PIPELINE_ABORTED code.
677 : : *
678 : : * This intentionally doesn't use a transaction to wrap the pipeline. You should
679 : : * usually use an xact, but in this case we want to observe the effects of each
680 : : * statement.
681 : : */
682 : : static void
1688 683 : 1 : test_pipeline_abort(PGconn *conn)
684 : : {
685 : 1 : PGresult *res = NULL;
686 : 1 : const char *dummy_params[1] = {"1"};
687 : 1 : Oid dummy_param_oids[1] = {INT4OID};
688 : : int i;
689 : : int gotrows;
690 : : bool goterror;
691 : :
692 : 1 : fprintf(stderr, "aborted pipeline... ");
693 : :
694 : 1 : res = PQexec(conn, drop_table_sql);
695 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1688 alvherre@alvh.no-ip. 696 :UBC 0 : pg_fatal("dispatching DROP TABLE failed: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 697 :GNC 1 : PQclear(res);
698 : :
1688 alvherre@alvh.no-ip. 699 :CBC 1 : res = PQexec(conn, create_table_sql);
700 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1688 alvherre@alvh.no-ip. 701 :UBC 0 : pg_fatal("dispatching CREATE TABLE failed: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 702 :GNC 1 : PQclear(res);
703 : :
704 : : /*
705 : : * Queue up a couple of small pipelines and process each without returning
706 : : * to command mode first. Make sure the second operation in the first
707 : : * pipeline ERRORs.
708 : : */
1688 alvherre@alvh.no-ip. 709 [ - + ]:CBC 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 710 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
711 : :
1688 alvherre@alvh.no-ip. 712 :CBC 1 : dummy_params[0] = "1";
713 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
714 : : dummy_params, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 715 :UBC 0 : pg_fatal("dispatching first insert failed: %s", PQerrorMessage(conn));
716 : :
1688 alvherre@alvh.no-ip. 717 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT no_such_function($1)",
718 : : 1, dummy_param_oids, dummy_params,
719 : : NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 720 :UBC 0 : pg_fatal("dispatching error select failed: %s", PQerrorMessage(conn));
721 : :
1688 alvherre@alvh.no-ip. 722 :CBC 1 : dummy_params[0] = "2";
723 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
724 : : dummy_params, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 725 :UBC 0 : pg_fatal("dispatching second insert failed: %s", PQerrorMessage(conn));
726 : :
1688 alvherre@alvh.no-ip. 727 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 728 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
729 : :
1688 alvherre@alvh.no-ip. 730 :CBC 1 : dummy_params[0] = "3";
731 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
732 : : dummy_params, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 733 :UBC 0 : pg_fatal("dispatching second-pipeline insert failed: %s",
734 : : PQerrorMessage(conn));
735 : :
1688 alvherre@alvh.no-ip. 736 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 737 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
738 : :
739 : : /*
740 : : * OK, start processing the pipeline results.
741 : : *
742 : : * We should get a command-ok for the first query, then a fatal error and
743 : : * a pipeline aborted message for the second insert, a pipeline-end, then
744 : : * a command-ok and a pipeline-ok for the second pipeline operation.
745 : : */
55 tgl@sss.pgh.pa.us 746 :GNC 1 : consume_result_status(conn, PGRES_COMMAND_OK);
747 : :
748 : : /* NULL result to signal end-of-results for this command */
749 : 1 : consume_null_result(conn);
750 : :
751 : : /* Second query caused error, so we expect an error next */
752 : 1 : consume_result_status(conn, PGRES_FATAL_ERROR);
753 : :
754 : : /* NULL result to signal end-of-results for this command */
755 : 1 : consume_null_result(conn);
756 : :
757 : : /*
758 : : * pipeline should now be aborted.
759 : : *
760 : : * Note that we could still queue more queries at this point if we wanted;
761 : : * they'd get added to a new third pipeline since we've already sent a
762 : : * second. The aborted flag relates only to the pipeline being received.
763 : : */
1688 alvherre@alvh.no-ip. 764 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
1688 alvherre@alvh.no-ip. 765 :UBC 0 : pg_fatal("pipeline should be flagged as aborted but isn't");
766 : :
767 : : /* third query in pipeline, the second insert */
55 tgl@sss.pgh.pa.us 768 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_ABORTED);
769 : :
770 : : /* NULL result to signal end-of-results for this command */
771 : 1 : consume_null_result(conn);
772 : :
1688 alvherre@alvh.no-ip. 773 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
1688 alvherre@alvh.no-ip. 774 :UBC 0 : pg_fatal("pipeline should be flagged as aborted but isn't");
775 : :
776 : : /* Ensure we're still in pipeline */
1688 alvherre@alvh.no-ip. 777 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 778 :UBC 0 : pg_fatal("Fell out of pipeline mode somehow");
779 : :
780 : : /*
781 : : * The end of a failed pipeline is a PGRES_PIPELINE_SYNC.
782 : : *
783 : : * (This is so clients know to start processing results normally again and
784 : : * can tell the difference between skipped commands and the sync.)
785 : : */
55 tgl@sss.pgh.pa.us 786 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
787 : :
1688 alvherre@alvh.no-ip. 788 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_ABORTED)
1688 alvherre@alvh.no-ip. 789 :UBC 0 : pg_fatal("sync should've cleared the aborted flag but didn't");
790 : :
791 : : /* We're still in pipeline mode... */
1688 alvherre@alvh.no-ip. 792 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 793 :UBC 0 : pg_fatal("Fell out of pipeline mode somehow");
794 : :
795 : : /* the insert from the second pipeline */
55 tgl@sss.pgh.pa.us 796 :GNC 1 : consume_result_status(conn, PGRES_COMMAND_OK);
797 : :
798 : : /* Read the NULL result at the end of the command */
799 : 1 : consume_null_result(conn);
800 : :
801 : : /* the second pipeline sync */
802 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
803 : :
804 : : /* Read the NULL result at the end of the command */
805 : 1 : consume_null_result(conn);
806 : :
807 : : /* Try to send two queries in one command */
1131 alvherre@alvh.no-ip. 808 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT 1; SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 809 :UBC 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 810 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 811 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 812 :CBC 1 : goterror = false;
813 [ + + ]: 2 : while ((res = PQgetResult(conn)) != NULL)
814 : : {
815 [ + - ]: 1 : switch (PQresultStatus(res))
816 : : {
817 : 1 : case PGRES_FATAL_ERROR:
818 [ - + ]: 1 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "42601") != 0)
1688 alvherre@alvh.no-ip. 819 :UBC 0 : pg_fatal("expected error about multiple commands, got %s",
820 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 821 :CBC 1 : printf("got expected %s", PQerrorMessage(conn));
822 : 1 : goterror = true;
823 : 1 : break;
1688 alvherre@alvh.no-ip. 824 :UBC 0 : default:
825 : 0 : pg_fatal("got unexpected status %s", PQresStatus(PQresultStatus(res)));
826 : : break;
827 : : }
55 tgl@sss.pgh.pa.us 828 :GNC 1 : PQclear(res);
829 : : }
1688 alvherre@alvh.no-ip. 830 [ - + ]:CBC 1 : if (!goterror)
1688 alvherre@alvh.no-ip. 831 :UBC 0 : pg_fatal("did not get cannot-insert-multiple-commands error");
832 : :
833 : : /* the second pipeline sync */
55 tgl@sss.pgh.pa.us 834 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
835 : :
1672 alvherre@alvh.no-ip. 836 :CBC 1 : fprintf(stderr, "ok\n");
837 : :
838 : : /* Test single-row mode with an error partways */
1131 839 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1.0/g FROM generate_series(3, -1, -1) g",
840 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 841 :UBC 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 842 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 843 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 844 :CBC 1 : PQsetSingleRowMode(conn);
845 : 1 : goterror = false;
1672 846 : 1 : gotrows = 0;
1688 847 [ + + ]: 5 : while ((res = PQgetResult(conn)) != NULL)
848 : : {
849 [ + + - ]: 4 : switch (PQresultStatus(res))
850 : : {
851 : 3 : case PGRES_SINGLE_TUPLE:
852 : 3 : printf("got row: %s\n", PQgetvalue(res, 0, 0));
1672 853 : 3 : gotrows++;
1688 854 : 3 : break;
855 : 1 : case PGRES_FATAL_ERROR:
856 [ - + ]: 1 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "22012") != 0)
1688 alvherre@alvh.no-ip. 857 :UBC 0 : pg_fatal("expected division-by-zero, got: %s (%s)",
858 : : PQerrorMessage(conn),
859 : : PQresultErrorField(res, PG_DIAG_SQLSTATE));
1688 alvherre@alvh.no-ip. 860 :CBC 1 : printf("got expected division-by-zero\n");
861 : 1 : goterror = true;
862 : 1 : break;
1688 alvherre@alvh.no-ip. 863 :UBC 0 : default:
864 : 0 : pg_fatal("got unexpected result %s", PQresStatus(PQresultStatus(res)));
865 : : }
1688 alvherre@alvh.no-ip. 866 :CBC 4 : PQclear(res);
867 : : }
868 [ - + ]: 1 : if (!goterror)
1688 alvherre@alvh.no-ip. 869 :UBC 0 : pg_fatal("did not get division-by-zero error");
1672 alvherre@alvh.no-ip. 870 [ - + ]:CBC 1 : if (gotrows != 3)
1672 alvherre@alvh.no-ip. 871 :UBC 0 : pg_fatal("did not get three rows");
872 : :
873 : : /* the third pipeline sync */
55 tgl@sss.pgh.pa.us 874 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
875 : :
876 : : /* We're still in pipeline mode... */
1688 alvherre@alvh.no-ip. 877 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 878 :UBC 0 : pg_fatal("Fell out of pipeline mode somehow");
879 : :
880 : : /* until we end it, which we can safely do now */
1688 alvherre@alvh.no-ip. 881 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 882 :UBC 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
883 : : PQerrorMessage(conn));
884 : :
1688 alvherre@alvh.no-ip. 885 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 886 :UBC 0 : pg_fatal("exiting pipeline mode didn't seem to work");
887 : :
888 : : /*-
889 : : * Since we fired the pipelines off without a surrounding xact, the results
890 : : * should be:
891 : : *
892 : : * - Implicit xact started by server around 1st pipeline
893 : : * - First insert applied
894 : : * - Second statement aborted xact
895 : : * - Third insert skipped
896 : : * - Sync rolled back first implicit xact
897 : : * - Implicit xact created by server around 2nd pipeline
898 : : * - insert applied from 2nd pipeline
899 : : * - Sync commits 2nd xact
900 : : *
901 : : * So we should only have the value 3 that we inserted.
902 : : */
1688 alvherre@alvh.no-ip. 903 :CBC 1 : res = PQexec(conn, "SELECT itemno FROM pq_pipeline_demo");
904 : :
905 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
1688 alvherre@alvh.no-ip. 906 :UBC 0 : pg_fatal("Expected tuples, got %s: %s",
907 : : PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 908 [ - + ]:CBC 1 : if (PQntuples(res) != 1)
1688 alvherre@alvh.no-ip. 909 :UBC 0 : pg_fatal("expected 1 result, got %d", PQntuples(res));
1688 alvherre@alvh.no-ip. 910 [ + + ]:CBC 2 : for (i = 0; i < PQntuples(res); i++)
911 : : {
912 : 1 : const char *val = PQgetvalue(res, i, 0);
913 : :
914 [ - + ]: 1 : if (strcmp(val, "3") != 0)
1688 alvherre@alvh.no-ip. 915 :UBC 0 : pg_fatal("expected only insert with value 3, got %s", val);
916 : : }
917 : :
1688 alvherre@alvh.no-ip. 918 :CBC 1 : PQclear(res);
919 : :
1211 920 : 1 : fprintf(stderr, "ok\n");
1688 921 : 1 : }
922 : :
923 : : /* State machine enum for test_pipelined_insert */
924 : : enum PipelineInsertStep
925 : : {
926 : : BI_BEGIN_TX,
927 : : BI_DROP_TABLE,
928 : : BI_CREATE_TABLE,
929 : : BI_PREPARE,
930 : : BI_INSERT_ROWS,
931 : : BI_COMMIT_TX,
932 : : BI_SYNC,
933 : : BI_DONE,
934 : : };
935 : :
936 : : static void
937 : 1 : test_pipelined_insert(PGconn *conn, int n_rows)
938 : : {
1672 939 : 1 : Oid insert_param_oids[2] = {INT4OID, INT8OID};
940 : : const char *insert_params[2];
941 : : char insert_param_0[MAXINTLEN];
942 : : char insert_param_1[MAXINT8LEN];
1688 943 : 1 : enum PipelineInsertStep send_step = BI_BEGIN_TX,
944 : 1 : recv_step = BI_BEGIN_TX;
945 : : int rows_to_send,
946 : : rows_to_receive;
947 : :
1672 948 : 1 : insert_params[0] = insert_param_0;
949 : 1 : insert_params[1] = insert_param_1;
950 : :
1688 951 : 1 : rows_to_send = rows_to_receive = n_rows;
952 : :
953 : : /*
954 : : * Do a pipelined insert into a table created at the start of the pipeline
955 : : */
956 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 957 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
958 : :
1688 alvherre@alvh.no-ip. 959 [ + + ]:CBC 4 : while (send_step != BI_PREPARE)
960 : : {
961 : : const char *sql;
962 : :
963 [ + + + - ]: 3 : switch (send_step)
964 : : {
965 : 1 : case BI_BEGIN_TX:
966 : 1 : sql = "BEGIN TRANSACTION";
967 : 1 : send_step = BI_DROP_TABLE;
968 : 1 : break;
969 : :
970 : 1 : case BI_DROP_TABLE:
971 : 1 : sql = drop_table_sql;
972 : 1 : send_step = BI_CREATE_TABLE;
973 : 1 : break;
974 : :
975 : 1 : case BI_CREATE_TABLE:
976 : 1 : sql = create_table_sql;
977 : 1 : send_step = BI_PREPARE;
978 : 1 : break;
979 : :
1688 alvherre@alvh.no-ip. 980 :UBC 0 : default:
981 : 0 : pg_fatal("invalid state");
982 : : sql = NULL; /* keep compiler quiet */
983 : : }
984 : :
985 : : pg_debug("sending: %s\n", sql);
1688 alvherre@alvh.no-ip. 986 [ - + ]:CBC 3 : if (PQsendQueryParams(conn, sql,
987 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 988 :UBC 0 : pg_fatal("dispatching %s failed: %s", sql, PQerrorMessage(conn));
989 : : }
990 : :
1688 alvherre@alvh.no-ip. 991 [ - + ]:CBC 1 : Assert(send_step == BI_PREPARE);
992 : : pg_debug("sending: %s\n", insert_sql2);
1672 993 [ - + ]: 1 : if (PQsendPrepare(conn, "my_insert", insert_sql2, 2, insert_param_oids) != 1)
1688 alvherre@alvh.no-ip. 994 :UBC 0 : pg_fatal("dispatching PREPARE failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 995 :CBC 1 : send_step = BI_INSERT_ROWS;
996 : :
997 : : /*
998 : : * Now we start inserting. We'll be sending enough data that we could fill
999 : : * our output buffer, so to avoid deadlocking we need to enter nonblocking
1000 : : * mode and consume input while we send more output. As results of each
1001 : : * query are processed we should pop them to allow processing of the next
1002 : : * query. There's no need to finish the pipeline before processing
1003 : : * results.
1004 : : */
1005 [ - + ]: 1 : if (PQsetnonblocking(conn, 1) != 0)
1688 alvherre@alvh.no-ip. 1006 :UBC 0 : pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn));
1007 : :
1688 alvherre@alvh.no-ip. 1008 [ + + ]:CBC 7952 : while (recv_step != BI_DONE)
1009 : : {
1010 : : int sock;
1011 : : fd_set input_mask;
1012 : : fd_set output_mask;
1013 : :
1014 : 7951 : sock = PQsocket(conn);
1015 : :
1016 [ - + ]: 7951 : if (sock < 0)
1688 alvherre@alvh.no-ip. 1017 :UBC 0 : break; /* shouldn't happen */
1018 : :
1688 alvherre@alvh.no-ip. 1019 [ + + ]:CBC 135167 : FD_ZERO(&input_mask);
1020 : 7951 : FD_SET(sock, &input_mask);
1021 [ + + ]: 135167 : FD_ZERO(&output_mask);
1022 : 7951 : FD_SET(sock, &output_mask);
1023 : :
1024 [ - + ]: 7951 : if (select(sock + 1, &input_mask, &output_mask, NULL, NULL) < 0)
1025 : : {
595 michael@paquier.xyz 1026 :UBC 0 : fprintf(stderr, "select() failed: %m\n");
1688 alvherre@alvh.no-ip. 1027 : 0 : exit_nicely(conn);
1028 : : }
1029 : :
1030 : : /*
1031 : : * Process any results, so we keep the server's output buffer free
1032 : : * flowing and it can continue to process input
1033 : : */
1688 alvherre@alvh.no-ip. 1034 [ + + ]:CBC 7951 : if (FD_ISSET(sock, &input_mask))
1035 : : {
1036 : 3 : PQconsumeInput(conn);
1037 : :
1038 : : /* Read until we'd block if we tried to read */
1039 [ + + + + ]: 1414 : while (!PQisBusy(conn) && recv_step < BI_DONE)
1040 : : {
1041 : : PGresult *res;
1672 tgl@sss.pgh.pa.us 1042 : 1411 : const char *cmdtag = "";
1688 alvherre@alvh.no-ip. 1043 : 1411 : const char *description = "";
1044 : : int status;
1045 : :
1046 : : /*
1047 : : * Read next result. If no more results from this query,
1048 : : * advance to the next query
1049 : : */
1050 : 1411 : res = PQgetResult(conn);
1051 [ + + ]: 1411 : if (res == NULL)
1052 : 705 : continue;
1053 : :
1054 : 706 : status = PGRES_COMMAND_OK;
1055 [ + + + + : 706 : switch (recv_step)
+ + + -
- ]
1056 : : {
1057 : 1 : case BI_BEGIN_TX:
1058 : 1 : cmdtag = "BEGIN";
1059 : 1 : recv_step++;
1060 : 1 : break;
1061 : 1 : case BI_DROP_TABLE:
1062 : 1 : cmdtag = "DROP TABLE";
1063 : 1 : recv_step++;
1064 : 1 : break;
1065 : 1 : case BI_CREATE_TABLE:
1066 : 1 : cmdtag = "CREATE TABLE";
1067 : 1 : recv_step++;
1068 : 1 : break;
1069 : 1 : case BI_PREPARE:
1070 : 1 : cmdtag = "";
1071 : 1 : description = "PREPARE";
1072 : 1 : recv_step++;
1073 : 1 : break;
1074 : 700 : case BI_INSERT_ROWS:
1075 : 700 : cmdtag = "INSERT";
1076 : 700 : rows_to_receive--;
1077 [ + + ]: 700 : if (rows_to_receive == 0)
1078 : 1 : recv_step++;
1079 : 700 : break;
1080 : 1 : case BI_COMMIT_TX:
1081 : 1 : cmdtag = "COMMIT";
1082 : 1 : recv_step++;
1083 : 1 : break;
1084 : 1 : case BI_SYNC:
1085 : 1 : cmdtag = "";
1086 : 1 : description = "SYNC";
1087 : 1 : status = PGRES_PIPELINE_SYNC;
1088 : 1 : recv_step++;
1089 : 1 : break;
1688 alvherre@alvh.no-ip. 1090 :UBC 0 : case BI_DONE:
1091 : : /* unreachable */
1682 tgl@sss.pgh.pa.us 1092 : 0 : pg_fatal("unreachable state");
1093 : : }
1094 : :
1688 alvherre@alvh.no-ip. 1095 [ - + ]:CBC 706 : if (PQresultStatus(res) != status)
1688 alvherre@alvh.no-ip. 1096 :UBC 0 : pg_fatal("%s reported status %s, expected %s\n"
1097 : : "Error message: \"%s\"",
1098 : : description, PQresStatus(PQresultStatus(res)),
1099 : : PQresStatus(status), PQerrorMessage(conn));
1100 : :
1688 alvherre@alvh.no-ip. 1101 [ - + ]:CBC 706 : if (strncmp(PQcmdStatus(res), cmdtag, strlen(cmdtag)) != 0)
1688 alvherre@alvh.no-ip. 1102 :UBC 0 : pg_fatal("%s expected command tag '%s', got '%s'",
1103 : : description, cmdtag, PQcmdStatus(res));
1104 : :
1105 : : pg_debug("Got %s OK\n", cmdtag[0] != '\0' ? cmdtag : description);
1106 : :
1688 alvherre@alvh.no-ip. 1107 :CBC 706 : PQclear(res);
1108 : : }
1109 : : }
1110 : :
1111 : : /* Write more rows and/or the end pipeline message, if needed */
1112 [ + + ]: 7951 : if (FD_ISSET(sock, &output_mask))
1113 : : {
1114 : 7949 : PQflush(conn);
1115 : :
1116 [ + + ]: 7949 : if (send_step == BI_INSERT_ROWS)
1117 : : {
1672 1118 : 700 : snprintf(insert_param_0, MAXINTLEN, "%d", rows_to_send);
1119 : : /* use up some buffer space with a wide value */
1671 1120 : 700 : snprintf(insert_param_1, MAXINT8LEN, "%lld", 1LL << 62);
1121 : :
1688 1122 [ + - ]: 700 : if (PQsendQueryPrepared(conn, "my_insert",
1123 : : 2, insert_params, NULL, NULL, 0) == 1)
1124 : : {
1125 : : pg_debug("sent row %d\n", rows_to_send);
1126 : :
1127 : 700 : rows_to_send--;
1128 [ + + ]: 700 : if (rows_to_send == 0)
1129 : 1 : send_step++;
1130 : : }
1131 : : else
1132 : : {
1133 : : /*
1134 : : * in nonblocking mode, so it's OK for an insert to fail
1135 : : * to send
1136 : : */
1688 alvherre@alvh.no-ip. 1137 :UBC 0 : fprintf(stderr, "WARNING: failed to send insert #%d: %s\n",
1138 : : rows_to_send, PQerrorMessage(conn));
1139 : : }
1140 : : }
1688 alvherre@alvh.no-ip. 1141 [ + + ]:CBC 7249 : else if (send_step == BI_COMMIT_TX)
1142 : : {
1143 [ + - ]: 1 : if (PQsendQueryParams(conn, "COMMIT",
1144 : : 0, NULL, NULL, NULL, NULL, 0) == 1)
1145 : : {
1146 : : pg_debug("sent COMMIT\n");
1147 : 1 : send_step++;
1148 : : }
1149 : : else
1150 : : {
1688 alvherre@alvh.no-ip. 1151 :UBC 0 : fprintf(stderr, "WARNING: failed to send commit: %s\n",
1152 : : PQerrorMessage(conn));
1153 : : }
1154 : : }
1688 alvherre@alvh.no-ip. 1155 [ + + ]:CBC 7248 : else if (send_step == BI_SYNC)
1156 : : {
1157 [ + - ]: 1 : if (PQpipelineSync(conn) == 1)
1158 : : {
1159 : 1 : fprintf(stdout, "pipeline sync sent\n");
1160 : 1 : send_step++;
1161 : : }
1162 : : else
1163 : : {
1688 alvherre@alvh.no-ip. 1164 :UBC 0 : fprintf(stderr, "WARNING: pipeline sync failed: %s\n",
1165 : : PQerrorMessage(conn));
1166 : : }
1167 : : }
1168 : : }
1169 : : }
1170 : :
1171 : : /* We've got the sync message and the pipeline should be done */
1688 alvherre@alvh.no-ip. 1172 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1173 :UBC 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
1174 : : PQerrorMessage(conn));
1175 : :
1688 alvherre@alvh.no-ip. 1176 [ - + ]:CBC 1 : if (PQsetnonblocking(conn, 0) != 0)
1688 alvherre@alvh.no-ip. 1177 :UBC 0 : pg_fatal("failed to clear nonblocking mode: %s", PQerrorMessage(conn));
1178 : :
1688 alvherre@alvh.no-ip. 1179 :CBC 1 : fprintf(stderr, "ok\n");
1180 : 1 : }
1181 : :
1182 : : static void
1183 : 1 : test_prepared(PGconn *conn)
1184 : : {
1185 : 1 : PGresult *res = NULL;
1186 : 1 : Oid param_oids[1] = {INT4OID};
1187 : : Oid expected_oids[4];
1188 : : Oid typ;
1189 : :
1190 : 1 : fprintf(stderr, "prepared... ");
1191 : :
1192 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1193 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1194 [ - + ]:CBC 1 : if (PQsendPrepare(conn, "select_one", "SELECT $1, '42', $1::numeric, "
1195 : : "interval '1 sec'",
1196 : : 1, param_oids) != 1)
1688 alvherre@alvh.no-ip. 1197 :UBC 0 : pg_fatal("preparing query failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1198 :CBC 1 : expected_oids[0] = INT4OID;
1199 : 1 : expected_oids[1] = TEXTOID;
1200 : 1 : expected_oids[2] = NUMERICOID;
1201 : 1 : expected_oids[3] = INTERVALOID;
1202 [ - + ]: 1 : if (PQsendDescribePrepared(conn, "select_one") != 1)
1688 alvherre@alvh.no-ip. 1203 :UBC 0 : pg_fatal("failed to send describePrepared: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1204 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1205 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1206 : :
55 tgl@sss.pgh.pa.us 1207 :GNC 1 : consume_result_status(conn, PGRES_COMMAND_OK);
1208 : :
1209 : 1 : consume_null_result(conn);
1210 : :
1211 : 1 : res = confirm_result_status(conn, PGRES_COMMAND_OK);
1688 alvherre@alvh.no-ip. 1212 [ - + ]:CBC 1 : if (PQnfields(res) != lengthof(expected_oids))
756 peter@eisentraut.org 1213 :UBC 0 : pg_fatal("expected %zu columns, got %d",
1214 : : lengthof(expected_oids), PQnfields(res));
1688 alvherre@alvh.no-ip. 1215 [ + + ]:CBC 5 : for (int i = 0; i < PQnfields(res); i++)
1216 : : {
1217 : 4 : typ = PQftype(res, i);
1218 [ - + ]: 4 : if (typ != expected_oids[i])
1688 alvherre@alvh.no-ip. 1219 :UBC 0 : pg_fatal("field %d: expected type %u, got %u",
1220 : : i, expected_oids[i], typ);
1221 : : }
1688 alvherre@alvh.no-ip. 1222 :CBC 1 : PQclear(res);
1223 : :
55 tgl@sss.pgh.pa.us 1224 :GNC 1 : consume_null_result(conn);
1225 : :
1226 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1227 : :
847 michael@paquier.xyz 1228 :CBC 1 : fprintf(stderr, "closing statement..");
1229 [ - + ]: 1 : if (PQsendClosePrepared(conn, "select_one") != 1)
847 michael@paquier.xyz 1230 :UBC 0 : pg_fatal("PQsendClosePrepared failed: %s", PQerrorMessage(conn));
847 michael@paquier.xyz 1231 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
847 michael@paquier.xyz 1232 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1233 : :
55 tgl@sss.pgh.pa.us 1234 :GNC 1 : consume_result_status(conn, PGRES_COMMAND_OK);
1235 : :
1236 : 1 : consume_null_result(conn);
1237 : :
1238 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1239 : :
1688 alvherre@alvh.no-ip. 1240 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1241 :UBC 0 : pg_fatal("could not exit pipeline mode: %s", PQerrorMessage(conn));
1242 : :
1243 : : /* Now that it's closed we should get an error when describing */
847 michael@paquier.xyz 1244 :CBC 1 : res = PQdescribePrepared(conn, "select_one");
1245 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
847 michael@paquier.xyz 1246 :UBC 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
55 tgl@sss.pgh.pa.us 1247 :GNC 1 : PQclear(res);
1248 : :
1249 : : /*
1250 : : * Also test the blocking close, this should not fail since closing a
1251 : : * non-existent prepared statement is a no-op
1252 : : */
847 michael@paquier.xyz 1253 :CBC 1 : res = PQclosePrepared(conn, "select_one");
1254 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
847 michael@paquier.xyz 1255 :UBC 0 : pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
55 tgl@sss.pgh.pa.us 1256 :GNC 1 : PQclear(res);
1257 : :
847 michael@paquier.xyz 1258 :CBC 1 : fprintf(stderr, "creating portal... ");
1259 : :
55 tgl@sss.pgh.pa.us 1260 :GNC 1 : res = PQexec(conn, "BEGIN");
1261 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
55 tgl@sss.pgh.pa.us 1262 :UNC 0 : pg_fatal("BEGIN failed: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 1263 :GNC 1 : PQclear(res);
1264 : :
1265 : 1 : res = PQexec(conn, "DECLARE cursor_one CURSOR FOR SELECT 1");
1266 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
55 tgl@sss.pgh.pa.us 1267 :UNC 0 : pg_fatal("DECLARE CURSOR failed: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 1268 :GNC 1 : PQclear(res);
1269 : :
1688 alvherre@alvh.no-ip. 1270 :CBC 1 : PQenterPipelineMode(conn);
1271 [ - + ]: 1 : if (PQsendDescribePortal(conn, "cursor_one") != 1)
1688 alvherre@alvh.no-ip. 1272 :UBC 0 : pg_fatal("PQsendDescribePortal failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1273 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1274 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1275 : :
55 tgl@sss.pgh.pa.us 1276 :GNC 1 : res = confirm_result_status(conn, PGRES_COMMAND_OK);
1688 alvherre@alvh.no-ip. 1277 :CBC 1 : typ = PQftype(res, 0);
1278 [ - + ]: 1 : if (typ != INT4OID)
1688 alvherre@alvh.no-ip. 1279 :UBC 0 : pg_fatal("portal: expected type %u, got %u",
1280 : : INT4OID, typ);
1688 alvherre@alvh.no-ip. 1281 :CBC 1 : PQclear(res);
1282 : :
55 tgl@sss.pgh.pa.us 1283 :GNC 1 : consume_null_result(conn);
1284 : :
1285 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1286 : :
847 michael@paquier.xyz 1287 :CBC 1 : fprintf(stderr, "closing portal... ");
1288 [ - + ]: 1 : if (PQsendClosePortal(conn, "cursor_one") != 1)
847 michael@paquier.xyz 1289 :UBC 0 : pg_fatal("PQsendClosePortal failed: %s", PQerrorMessage(conn));
847 michael@paquier.xyz 1290 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
847 michael@paquier.xyz 1291 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1292 : :
55 tgl@sss.pgh.pa.us 1293 :GNC 1 : consume_result_status(conn, PGRES_COMMAND_OK);
1294 : :
1295 : 1 : consume_null_result(conn);
1296 : :
1297 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1298 : :
1688 alvherre@alvh.no-ip. 1299 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1300 :UBC 0 : pg_fatal("could not exit pipeline mode: %s", PQerrorMessage(conn));
1301 : :
1302 : : /* Now that it's closed we should get an error when describing */
847 michael@paquier.xyz 1303 :CBC 1 : res = PQdescribePortal(conn, "cursor_one");
1304 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
847 michael@paquier.xyz 1305 :UBC 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
55 tgl@sss.pgh.pa.us 1306 :GNC 1 : PQclear(res);
1307 : :
1308 : : /*
1309 : : * Also test the blocking close, this should not fail since closing a
1310 : : * non-existent portal is a no-op
1311 : : */
847 michael@paquier.xyz 1312 :CBC 1 : res = PQclosePortal(conn, "cursor_one");
1313 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
847 michael@paquier.xyz 1314 :UBC 0 : pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
55 tgl@sss.pgh.pa.us 1315 :GNC 1 : PQclear(res);
1316 : :
1688 alvherre@alvh.no-ip. 1317 :CBC 1 : fprintf(stderr, "ok\n");
1318 : 1 : }
1319 : :
1320 : : /*
1321 : : * Test max_protocol_version options.
1322 : : */
1323 : : static void
209 heikki.linnakangas@i 1324 : 1 : test_protocol_version(PGconn *conn)
1325 : : {
1326 : : const char **keywords;
1327 : : const char **vals;
1328 : : int nopts;
1329 : 1 : PQconninfoOption *opts = PQconninfo(conn);
1330 : : int protocol_version;
1331 : : int max_protocol_version_index;
1332 : : int i;
1333 : :
1334 : : /*
1335 : : * Prepare keywords/vals arrays, copied from the existing connection, with
1336 : : * an extra slot for 'max_protocol_version'.
1337 : : */
1338 : 1 : nopts = 0;
1339 [ + + ]: 52 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1340 : 51 : nopts++;
1341 : 1 : nopts++; /* max_protocol_version */
1342 : 1 : nopts++; /* NULL terminator */
1343 : :
1344 : 1 : keywords = pg_malloc0(sizeof(char *) * nopts);
1345 : 1 : vals = pg_malloc0(sizeof(char *) * nopts);
1346 : :
1347 : 1 : i = 0;
1348 [ + + ]: 52 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1349 : : {
1350 [ + + ]: 51 : if (opt->val)
1351 : : {
1352 : 20 : keywords[i] = opt->keyword;
1353 : 20 : vals[i] = opt->val;
1354 : 20 : i++;
1355 : : }
1356 : : }
1357 : :
1358 : 1 : max_protocol_version_index = i;
1359 : 1 : keywords[i] = "max_protocol_version"; /* value is filled in below */
1360 : 1 : i++;
1361 : 1 : keywords[i] = vals[i] = NULL;
1362 : :
1363 : : /*
1364 : : * Test max_protocol_version=3.0
1365 : : */
1366 : 1 : vals[max_protocol_version_index] = "3.0";
1367 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1368 : :
1369 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
209 heikki.linnakangas@i 1370 :UBC 0 : pg_fatal("Connection to database failed: %s",
1371 : : PQerrorMessage(conn));
1372 : :
209 heikki.linnakangas@i 1373 :CBC 1 : protocol_version = PQfullProtocolVersion(conn);
1374 [ - + ]: 1 : if (protocol_version != 30000)
209 heikki.linnakangas@i 1375 :UBC 0 : pg_fatal("expected 30000, got %d", protocol_version);
1376 : :
209 heikki.linnakangas@i 1377 :CBC 1 : PQfinish(conn);
1378 : :
1379 : : /*
1380 : : * Test max_protocol_version=3.1. It's not valid, we went straight from
1381 : : * 3.0 to 3.2.
1382 : : */
1383 : 1 : vals[max_protocol_version_index] = "3.1";
1384 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1385 : :
1386 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_BAD)
209 heikki.linnakangas@i 1387 :UBC 0 : pg_fatal("Connecting with max_protocol_version 3.1 should have failed.");
1388 : :
209 heikki.linnakangas@i 1389 :CBC 1 : PQfinish(conn);
1390 : :
1391 : : /*
1392 : : * Test max_protocol_version=3.2
1393 : : */
1394 : 1 : vals[max_protocol_version_index] = "3.2";
1395 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1396 : :
1397 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
209 heikki.linnakangas@i 1398 :UBC 0 : pg_fatal("Connection to database failed: %s",
1399 : : PQerrorMessage(conn));
1400 : :
209 heikki.linnakangas@i 1401 :CBC 1 : protocol_version = PQfullProtocolVersion(conn);
1402 [ - + ]: 1 : if (protocol_version != 30002)
209 heikki.linnakangas@i 1403 :UBC 0 : pg_fatal("expected 30002, got %d", protocol_version);
1404 : :
209 heikki.linnakangas@i 1405 :CBC 1 : PQfinish(conn);
1406 : :
1407 : : /*
1408 : : * Test max_protocol_version=latest. 'latest' currently means '3.2'.
1409 : : */
1410 : 1 : vals[max_protocol_version_index] = "latest";
1411 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1412 : :
1413 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
209 heikki.linnakangas@i 1414 :UBC 0 : pg_fatal("Connection to database failed: %s",
1415 : : PQerrorMessage(conn));
1416 : :
209 heikki.linnakangas@i 1417 :CBC 1 : protocol_version = PQfullProtocolVersion(conn);
1418 [ - + ]: 1 : if (protocol_version != 30002)
209 heikki.linnakangas@i 1419 :UBC 0 : pg_fatal("expected 30002, got %d", protocol_version);
1420 : :
209 heikki.linnakangas@i 1421 :CBC 1 : PQfinish(conn);
1422 : :
55 tgl@sss.pgh.pa.us 1423 :GNC 1 : pfree(keywords);
1424 : 1 : pfree(vals);
1425 : 1 : PQconninfoFree(opts);
209 heikki.linnakangas@i 1426 :CBC 1 : }
1427 : :
1428 : : /* Notice processor: print notices, and count how many we got */
1429 : : static void
1211 alvherre@alvh.no-ip. 1430 : 1 : notice_processor(void *arg, const char *message)
1431 : : {
893 tgl@sss.pgh.pa.us 1432 : 1 : int *n_notices = (int *) arg;
1433 : :
1211 alvherre@alvh.no-ip. 1434 : 1 : (*n_notices)++;
1435 : 1 : fprintf(stderr, "NOTICE %d: %s", *n_notices, message);
1436 : 1 : }
1437 : :
1438 : : /* Verify behavior in "idle" state */
1439 : : static void
1440 : 1 : test_pipeline_idle(PGconn *conn)
1441 : : {
1442 : 1 : int n_notices = 0;
1443 : :
1444 : 1 : fprintf(stderr, "\npipeline idle...\n");
1445 : :
1446 : 1 : PQsetNoticeProcessor(conn, notice_processor, &n_notices);
1447 : :
1448 : : /* Try to exit pipeline mode in pipeline-idle state */
1449 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1211 alvherre@alvh.no-ip. 1450 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1131 alvherre@alvh.no-ip. 1451 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT 1", 0, NULL, NULL, NULL, NULL, 0) != 1)
1211 alvherre@alvh.no-ip. 1452 :UBC 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1211 alvherre@alvh.no-ip. 1453 :CBC 1 : PQsendFlushRequest(conn);
1454 : :
55 tgl@sss.pgh.pa.us 1455 :GNC 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1456 : :
1457 : 1 : consume_null_result(conn);
1458 : :
1131 alvherre@alvh.no-ip. 1459 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
1211 alvherre@alvh.no-ip. 1460 :UBC 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1211 alvherre@alvh.no-ip. 1461 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) == 1)
1211 alvherre@alvh.no-ip. 1462 :UBC 0 : pg_fatal("exiting pipeline succeeded when it shouldn't");
1211 alvherre@alvh.no-ip. 1463 [ - + ]:CBC 1 : if (strncmp(PQerrorMessage(conn), "cannot exit pipeline mode",
1464 : : strlen("cannot exit pipeline mode")) != 0)
1211 alvherre@alvh.no-ip. 1465 :UBC 0 : pg_fatal("did not get expected error; got: %s",
1466 : : PQerrorMessage(conn));
1211 alvherre@alvh.no-ip. 1467 :CBC 1 : PQsendFlushRequest(conn);
1468 : :
55 tgl@sss.pgh.pa.us 1469 :GNC 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1470 : :
1471 : 1 : consume_null_result(conn);
1472 : :
1211 alvherre@alvh.no-ip. 1473 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1211 alvherre@alvh.no-ip. 1474 :UBC 0 : pg_fatal("exiting pipeline failed: %s", PQerrorMessage(conn));
1475 : :
1211 alvherre@alvh.no-ip. 1476 [ - + ]:CBC 1 : if (n_notices > 0)
1211 alvherre@alvh.no-ip. 1477 :UBC 0 : pg_fatal("got %d notice(s)", n_notices);
1131 alvherre@alvh.no-ip. 1478 :CBC 1 : fprintf(stderr, "ok - 1\n");
1479 : :
1480 : : /* Have a WARNING in the middle of a resultset */
1211 1481 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1211 alvherre@alvh.no-ip. 1482 :UBC 0 : pg_fatal("entering pipeline mode failed: %s", PQerrorMessage(conn));
1131 alvherre@alvh.no-ip. 1483 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT pg_catalog.pg_advisory_unlock(1,1)", 0, NULL, NULL, NULL, NULL, 0) != 1)
1211 alvherre@alvh.no-ip. 1484 :UBC 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1211 alvherre@alvh.no-ip. 1485 :CBC 1 : PQsendFlushRequest(conn);
1486 : :
55 tgl@sss.pgh.pa.us 1487 :GNC 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1488 : :
1211 alvherre@alvh.no-ip. 1489 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1211 alvherre@alvh.no-ip. 1490 :UBC 0 : pg_fatal("failed to exit pipeline mode: %s", PQerrorMessage(conn));
1131 alvherre@alvh.no-ip. 1491 :CBC 1 : fprintf(stderr, "ok - 2\n");
1211 1492 : 1 : }
1493 : :
1494 : : static void
1688 1495 : 1 : test_simple_pipeline(PGconn *conn)
1496 : : {
1497 : 1 : const char *dummy_params[1] = {"1"};
1498 : 1 : Oid dummy_param_oids[1] = {INT4OID};
1499 : :
1500 : 1 : fprintf(stderr, "simple pipeline... ");
1501 : :
1502 : : /*
1503 : : * Enter pipeline mode and dispatch a set of operations, which we'll then
1504 : : * process the results of as they come in.
1505 : : *
1506 : : * For a simple case we should be able to do this without interim
1507 : : * processing of results since our output buffer will give us enough slush
1508 : : * to work with and we won't block on sending. So blocking mode is fine.
1509 : : */
1510 [ - + ]: 1 : if (PQisnonblocking(conn))
1688 alvherre@alvh.no-ip. 1511 :UBC 0 : pg_fatal("Expected blocking connection mode");
1512 : :
1688 alvherre@alvh.no-ip. 1513 [ - + ]:CBC 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1514 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1515 : :
1688 alvherre@alvh.no-ip. 1516 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT $1",
1517 : : 1, dummy_param_oids, dummy_params,
1518 : : NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 1519 :UBC 0 : pg_fatal("dispatching SELECT failed: %s", PQerrorMessage(conn));
1520 : :
1688 alvherre@alvh.no-ip. 1521 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 0)
1688 alvherre@alvh.no-ip. 1522 :UBC 0 : pg_fatal("exiting pipeline mode with work in progress should fail, but succeeded");
1523 : :
1688 alvherre@alvh.no-ip. 1524 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1525 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1526 : :
55 tgl@sss.pgh.pa.us 1527 :GNC 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1528 : :
1529 : 1 : consume_null_result(conn);
1530 : :
1531 : : /*
1532 : : * Even though we've processed the result there's still a sync to come and
1533 : : * we can't exit pipeline mode yet
1534 : : */
1688 alvherre@alvh.no-ip. 1535 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 0)
1688 alvherre@alvh.no-ip. 1536 :UBC 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
1537 : :
55 tgl@sss.pgh.pa.us 1538 :GNC 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1539 : :
1540 : 1 : consume_null_result(conn);
1541 : :
1542 : : /* We're still in pipeline mode... */
1688 alvherre@alvh.no-ip. 1543 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 1544 :UBC 0 : pg_fatal("Fell out of pipeline mode somehow");
1545 : :
1546 : : /* ... until we end it, which we can safely do now */
1688 alvherre@alvh.no-ip. 1547 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1548 :UBC 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
1549 : : PQerrorMessage(conn));
1550 : :
1688 alvherre@alvh.no-ip. 1551 [ - + ]:CBC 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
1688 alvherre@alvh.no-ip. 1552 :UBC 0 : pg_fatal("Exiting pipeline mode didn't seem to work");
1553 : :
1688 alvherre@alvh.no-ip. 1554 :CBC 1 : fprintf(stderr, "ok\n");
1555 : 1 : }
1556 : :
1557 : : static void
1558 : 1 : test_singlerowmode(PGconn *conn)
1559 : : {
1560 : : PGresult *res;
1561 : : int i;
1562 : 1 : bool pipeline_ended = false;
1563 : :
1564 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1565 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s",
1566 : : PQerrorMessage(conn));
1567 : :
1568 : : /* One series of three commands, using single-row mode for the first two. */
1688 alvherre@alvh.no-ip. 1569 [ + + ]:CBC 4 : for (i = 0; i < 3; i++)
1570 : : {
1571 : : char *param[1];
1572 : :
1573 : 3 : param[0] = psprintf("%d", 44 + i);
1574 : :
1575 [ - + ]: 3 : if (PQsendQueryParams(conn,
1576 : : "SELECT generate_series(42, $1)",
1577 : : 1,
1578 : : NULL,
1579 : : (const char **) param,
1580 : : NULL,
1581 : : NULL,
1582 : : 0) != 1)
1688 alvherre@alvh.no-ip. 1583 :UBC 0 : pg_fatal("failed to send query: %s",
1584 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1585 :CBC 3 : pfree(param[0]);
1586 : : }
1587 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1588 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1589 : :
1688 alvherre@alvh.no-ip. 1590 [ + + ]:CBC 5 : for (i = 0; !pipeline_ended; i++)
1591 : : {
1592 : 4 : bool first = true;
1593 : : bool saw_ending_tuplesok;
1594 : 4 : bool isSingleTuple = false;
1595 : :
1596 : : /* Set single row mode for only first 2 SELECT queries */
1597 [ + + ]: 4 : if (i < 2)
1598 : : {
1599 [ - + ]: 2 : if (PQsetSingleRowMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1600 :UBC 0 : pg_fatal("PQsetSingleRowMode() failed for i=%d", i);
1601 : : }
1602 : :
1603 : : /* Consume rows for this query */
1688 alvherre@alvh.no-ip. 1604 :CBC 4 : saw_ending_tuplesok = false;
1605 [ + + ]: 14 : while ((res = PQgetResult(conn)) != NULL)
1606 : : {
1607 : 11 : ExecStatusType est = PQresultStatus(res);
1608 : :
1609 [ + + ]: 11 : if (est == PGRES_PIPELINE_SYNC)
1610 : : {
1611 : 1 : fprintf(stderr, "end of pipeline reached\n");
1612 : 1 : pipeline_ended = true;
1613 : 1 : PQclear(res);
1614 [ - + ]: 1 : if (i != 3)
1688 alvherre@alvh.no-ip. 1615 :UBC 0 : pg_fatal("Expected three results, got %d", i);
1688 alvherre@alvh.no-ip. 1616 :CBC 1 : break;
1617 : : }
1618 : :
1619 : : /* Expect SINGLE_TUPLE for queries 0 and 1, TUPLES_OK for 2 */
1620 [ + + ]: 10 : if (first)
1621 : : {
1622 [ + + - + ]: 3 : if (i <= 1 && est != PGRES_SINGLE_TUPLE)
1688 alvherre@alvh.no-ip. 1623 :UBC 0 : pg_fatal("Expected PGRES_SINGLE_TUPLE for query %d, got %s",
1624 : : i, PQresStatus(est));
1688 alvherre@alvh.no-ip. 1625 [ + + - + ]:CBC 3 : if (i >= 2 && est != PGRES_TUPLES_OK)
1688 alvherre@alvh.no-ip. 1626 :UBC 0 : pg_fatal("Expected PGRES_TUPLES_OK for query %d, got %s",
1627 : : i, PQresStatus(est));
1688 alvherre@alvh.no-ip. 1628 :CBC 3 : first = false;
1629 : : }
1630 : :
1631 : 10 : fprintf(stderr, "Result status %s for query %d", PQresStatus(est), i);
1632 [ + + - ]: 10 : switch (est)
1633 : : {
1634 : 3 : case PGRES_TUPLES_OK:
1635 : 3 : fprintf(stderr, ", tuples: %d\n", PQntuples(res));
1636 : 3 : saw_ending_tuplesok = true;
1637 [ + + ]: 3 : if (isSingleTuple)
1638 : : {
1639 [ + - ]: 2 : if (PQntuples(res) == 0)
1640 : 2 : fprintf(stderr, "all tuples received in query %d\n", i);
1641 : : else
1688 alvherre@alvh.no-ip. 1642 :UBC 0 : pg_fatal("Expected to follow PGRES_SINGLE_TUPLE, but received PGRES_TUPLES_OK directly instead");
1643 : : }
1688 alvherre@alvh.no-ip. 1644 :CBC 3 : break;
1645 : :
1646 : 7 : case PGRES_SINGLE_TUPLE:
1647 : 7 : isSingleTuple = true;
1648 : 7 : fprintf(stderr, ", %d tuple: %s\n", PQntuples(res), PQgetvalue(res, 0, 0));
1649 : 7 : break;
1650 : :
1688 alvherre@alvh.no-ip. 1651 :UBC 0 : default:
1652 : 0 : pg_fatal("unexpected");
1653 : : }
1688 alvherre@alvh.no-ip. 1654 :CBC 10 : PQclear(res);
1655 : : }
1656 [ + + - + ]: 4 : if (!pipeline_ended && !saw_ending_tuplesok)
1688 alvherre@alvh.no-ip. 1657 :UBC 0 : pg_fatal("didn't get expected terminating TUPLES_OK");
1658 : : }
1659 : :
1660 : : /*
1661 : : * Now issue one command, get its results in with single-row mode, then
1662 : : * issue another command, and get its results in normal mode; make sure
1663 : : * the single-row mode flag is reset as expected.
1664 : : */
1110 alvherre@alvh.no-ip. 1665 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT generate_series(0, 0)",
1666 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1110 alvherre@alvh.no-ip. 1667 :UBC 0 : pg_fatal("failed to send query: %s",
1668 : : PQerrorMessage(conn));
1110 alvherre@alvh.no-ip. 1669 [ - + ]:CBC 1 : if (PQsendFlushRequest(conn) != 1)
1110 alvherre@alvh.no-ip. 1670 :UBC 0 : pg_fatal("failed to send flush request");
1110 alvherre@alvh.no-ip. 1671 [ - + ]:CBC 1 : if (PQsetSingleRowMode(conn) != 1)
1110 alvherre@alvh.no-ip. 1672 :UBC 0 : pg_fatal("PQsetSingleRowMode() failed");
1673 : :
55 tgl@sss.pgh.pa.us 1674 :GNC 1 : consume_result_status(conn, PGRES_SINGLE_TUPLE);
1675 : :
1676 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1677 : :
1678 : 1 : consume_null_result(conn);
1679 : :
1110 alvherre@alvh.no-ip. 1680 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT 1",
1681 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1110 alvherre@alvh.no-ip. 1682 :UBC 0 : pg_fatal("failed to send query: %s",
1683 : : PQerrorMessage(conn));
1110 alvherre@alvh.no-ip. 1684 [ - + ]:CBC 1 : if (PQsendFlushRequest(conn) != 1)
1110 alvherre@alvh.no-ip. 1685 :UBC 0 : pg_fatal("failed to send flush request");
1686 : :
55 tgl@sss.pgh.pa.us 1687 :GNC 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1688 : :
1689 : 1 : consume_null_result(conn);
1690 : :
1691 : : /*
1692 : : * Try chunked mode as well; make sure that it correctly delivers a
1693 : : * partial final chunk.
1694 : : */
570 tgl@sss.pgh.pa.us 1695 [ - + ]:CBC 1 : if (PQsendQueryParams(conn, "SELECT generate_series(1, 5)",
1696 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
570 tgl@sss.pgh.pa.us 1697 :UBC 0 : pg_fatal("failed to send query: %s",
1698 : : PQerrorMessage(conn));
570 tgl@sss.pgh.pa.us 1699 [ - + ]:CBC 1 : if (PQsendFlushRequest(conn) != 1)
570 tgl@sss.pgh.pa.us 1700 :UBC 0 : pg_fatal("failed to send flush request");
570 tgl@sss.pgh.pa.us 1701 [ - + ]:CBC 1 : if (PQsetChunkedRowsMode(conn, 3) != 1)
570 tgl@sss.pgh.pa.us 1702 :UBC 0 : pg_fatal("PQsetChunkedRowsMode() failed");
1703 : :
55 tgl@sss.pgh.pa.us 1704 :GNC 1 : res = confirm_result_status(conn, PGRES_TUPLES_CHUNK);
570 tgl@sss.pgh.pa.us 1705 [ - + ]:CBC 1 : if (PQntuples(res) != 3)
570 tgl@sss.pgh.pa.us 1706 :UBC 0 : pg_fatal("Expected 3 rows, got %d", PQntuples(res));
55 tgl@sss.pgh.pa.us 1707 :GNC 1 : PQclear(res);
1708 : :
1709 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_CHUNK);
570 tgl@sss.pgh.pa.us 1710 [ - + ]:CBC 1 : if (PQntuples(res) != 2)
570 tgl@sss.pgh.pa.us 1711 :UBC 0 : pg_fatal("Expected 2 rows, got %d", PQntuples(res));
55 tgl@sss.pgh.pa.us 1712 :GNC 1 : PQclear(res);
1713 : :
1714 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_OK);
570 tgl@sss.pgh.pa.us 1715 [ - + ]:CBC 1 : if (PQntuples(res) != 0)
570 tgl@sss.pgh.pa.us 1716 :UBC 0 : pg_fatal("Expected 0 rows, got %d", PQntuples(res));
55 tgl@sss.pgh.pa.us 1717 :GNC 1 : PQclear(res);
1718 : :
1719 : 1 : consume_null_result(conn);
1720 : :
1688 alvherre@alvh.no-ip. 1721 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1722 :UBC 0 : pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
1723 : :
1211 alvherre@alvh.no-ip. 1724 :CBC 1 : fprintf(stderr, "ok\n");
1688 1725 : 1 : }
1726 : :
1727 : : /*
1728 : : * Simple test to verify that a pipeline is discarded as a whole when there's
1729 : : * an error, ignoring transaction commands.
1730 : : */
1731 : : static void
1732 : 1 : test_transaction(PGconn *conn)
1733 : : {
1734 : : PGresult *res;
1735 : : bool expect_null;
1736 : 1 : int num_syncs = 0;
1737 : :
1738 : 1 : res = PQexec(conn, "DROP TABLE IF EXISTS pq_pipeline_tst;"
1739 : : "CREATE TABLE pq_pipeline_tst (id int)");
1740 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1688 alvherre@alvh.no-ip. 1741 :UBC 0 : pg_fatal("failed to create test table: %s",
1742 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1743 :CBC 1 : PQclear(res);
1744 : :
1745 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1746 :UBC 0 : pg_fatal("failed to enter pipeline mode: %s",
1747 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1748 [ - + ]:CBC 1 : if (PQsendPrepare(conn, "rollback", "ROLLBACK", 0, NULL) != 1)
1688 alvherre@alvh.no-ip. 1749 :UBC 0 : pg_fatal("could not send prepare on pipeline: %s",
1750 : : PQerrorMessage(conn));
1751 : :
1688 alvherre@alvh.no-ip. 1752 [ - + ]:CBC 1 : if (PQsendQueryParams(conn,
1753 : : "BEGIN",
1754 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 1755 :UBC 0 : pg_fatal("failed to send query: %s",
1756 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1757 [ - + ]:CBC 1 : if (PQsendQueryParams(conn,
1758 : : "SELECT 0/0",
1759 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 1760 :UBC 0 : pg_fatal("failed to send query: %s",
1761 : : PQerrorMessage(conn));
1762 : :
1763 : : /*
1764 : : * send a ROLLBACK using a prepared stmt. Doesn't work because we need to
1765 : : * get out of the pipeline-aborted state first.
1766 : : */
1688 alvherre@alvh.no-ip. 1767 [ - + ]:CBC 1 : if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
1688 alvherre@alvh.no-ip. 1768 :UBC 0 : pg_fatal("failed to execute prepared: %s",
1769 : : PQerrorMessage(conn));
1770 : :
1771 : : /* This insert fails because we're in pipeline-aborted state */
1688 alvherre@alvh.no-ip. 1772 [ - + ]:CBC 1 : if (PQsendQueryParams(conn,
1773 : : "INSERT INTO pq_pipeline_tst VALUES (1)",
1774 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 1775 :UBC 0 : pg_fatal("failed to send query: %s",
1776 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1777 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1778 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1779 :CBC 1 : num_syncs++;
1780 : :
1781 : : /*
1782 : : * This insert fails even though the pipeline got a SYNC, because we're in
1783 : : * an aborted transaction
1784 : : */
1785 [ - + ]: 1 : if (PQsendQueryParams(conn,
1786 : : "INSERT INTO pq_pipeline_tst VALUES (2)",
1787 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 1788 :UBC 0 : pg_fatal("failed to send query: %s",
1789 : : PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1790 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1791 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1792 :CBC 1 : num_syncs++;
1793 : :
1794 : : /*
1795 : : * Send ROLLBACK using prepared stmt. This one works because we just did
1796 : : * PQpipelineSync above.
1797 : : */
1798 [ - + ]: 1 : if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
1688 alvherre@alvh.no-ip. 1799 :UBC 0 : pg_fatal("failed to execute prepared: %s",
1800 : : PQerrorMessage(conn));
1801 : :
1802 : : /*
1803 : : * Now that we're out of a transaction and in pipeline-good mode, this
1804 : : * insert works
1805 : : */
1688 alvherre@alvh.no-ip. 1806 [ - + ]:CBC 1 : if (PQsendQueryParams(conn,
1807 : : "INSERT INTO pq_pipeline_tst VALUES (3)",
1808 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1688 alvherre@alvh.no-ip. 1809 :UBC 0 : pg_fatal("failed to send query: %s",
1810 : : PQerrorMessage(conn));
1811 : : /* Send two syncs now -- match up to SYNC messages below */
1688 alvherre@alvh.no-ip. 1812 [ - + ]:CBC 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1813 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1814 :CBC 1 : num_syncs++;
1815 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1688 alvherre@alvh.no-ip. 1816 :UBC 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1817 :CBC 1 : num_syncs++;
1818 : :
1819 : 1 : expect_null = false;
1820 : 1 : for (int i = 0;; i++)
1821 : 19 : {
1822 : : ExecStatusType restype;
1823 : :
1824 : 20 : res = PQgetResult(conn);
1825 [ + + ]: 20 : if (res == NULL)
1826 : : {
1827 : 8 : printf("%d: got NULL result\n", i);
1828 [ - + ]: 8 : if (!expect_null)
1688 alvherre@alvh.no-ip. 1829 :UBC 0 : pg_fatal("did not expect NULL here");
1688 alvherre@alvh.no-ip. 1830 :CBC 8 : expect_null = false;
1831 : 8 : continue;
1832 : : }
1833 : 12 : restype = PQresultStatus(res);
1834 : 12 : printf("%d: got status %s", i, PQresStatus(restype));
1835 [ - + ]: 12 : if (expect_null)
1688 alvherre@alvh.no-ip. 1836 :UBC 0 : pg_fatal("expected NULL");
1688 alvherre@alvh.no-ip. 1837 [ + + ]:CBC 12 : if (restype == PGRES_FATAL_ERROR)
1838 : 2 : printf("; error: %s", PQerrorMessage(conn));
1839 [ + + ]: 10 : else if (restype == PGRES_PIPELINE_ABORTED)
1840 : : {
1841 : 2 : printf(": command didn't run because pipeline aborted\n");
1842 : : }
1843 : : else
1844 : 8 : printf("\n");
1845 : 12 : PQclear(res);
1846 : :
1847 [ + + ]: 12 : if (restype == PGRES_PIPELINE_SYNC)
1848 : 4 : num_syncs--;
1849 : : else
1850 : 8 : expect_null = true;
1851 [ + + ]: 12 : if (num_syncs <= 0)
1852 : 1 : break;
1853 : : }
1854 : :
55 tgl@sss.pgh.pa.us 1855 :GNC 1 : consume_null_result(conn);
1856 : :
1688 alvherre@alvh.no-ip. 1857 [ - + ]:CBC 1 : if (PQexitPipelineMode(conn) != 1)
1688 alvherre@alvh.no-ip. 1858 :UBC 0 : pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
1859 : :
1860 : : /* We expect to find one tuple containing the value "3" */
1688 alvherre@alvh.no-ip. 1861 :CBC 1 : res = PQexec(conn, "SELECT * FROM pq_pipeline_tst");
1862 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
1688 alvherre@alvh.no-ip. 1863 :UBC 0 : pg_fatal("failed to obtain result: %s", PQerrorMessage(conn));
1688 alvherre@alvh.no-ip. 1864 [ - + ]:CBC 1 : if (PQntuples(res) != 1)
1688 alvherre@alvh.no-ip. 1865 :UBC 0 : pg_fatal("did not get 1 tuple");
1688 alvherre@alvh.no-ip. 1866 [ - + ]:CBC 1 : if (strcmp(PQgetvalue(res, 0, 0), "3") != 0)
1688 alvherre@alvh.no-ip. 1867 :UBC 0 : pg_fatal("did not get expected tuple");
1688 alvherre@alvh.no-ip. 1868 :CBC 1 : PQclear(res);
1869 : :
1870 : 1 : fprintf(stderr, "ok\n");
1871 : 1 : }
1872 : :
1873 : : /*
1874 : : * In this test mode we send a stream of queries, with one in the middle
1875 : : * causing an error. Verify that we can still send some more after the
1876 : : * error and have libpq work properly.
1877 : : */
1878 : : static void
1572 1879 : 1 : test_uniqviol(PGconn *conn)
1880 : : {
1881 : 1 : int sock = PQsocket(conn);
1882 : : PGresult *res;
1883 : 1 : Oid paramTypes[2] = {INT8OID, INT8OID};
1884 : : const char *paramValues[2];
1885 : : char paramValue0[MAXINT8LEN];
1886 : : char paramValue1[MAXINT8LEN];
1887 : 1 : int ctr = 0;
1888 : 1 : int numsent = 0;
1889 : 1 : int results = 0;
1890 : 1 : bool read_done = false;
1891 : 1 : bool write_done = false;
1892 : 1 : bool error_sent = false;
1893 : 1 : bool got_error = false;
1894 : 1 : int switched = 0;
1895 : 1 : int socketful = 0;
1896 : : fd_set in_fds;
1897 : : fd_set out_fds;
1898 : :
1899 : 1 : fprintf(stderr, "uniqviol ...");
1900 : :
1901 : 1 : PQsetnonblocking(conn, 1);
1902 : :
1903 : 1 : paramValues[0] = paramValue0;
1904 : 1 : paramValues[1] = paramValue1;
1905 : 1 : sprintf(paramValue1, "42");
1906 : :
1907 : 1 : res = PQexec(conn, "drop table if exists ppln_uniqviol;"
1908 : : "create table ppln_uniqviol(id bigint primary key, idata bigint)");
1909 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1572 alvherre@alvh.no-ip. 1910 :UBC 0 : pg_fatal("failed to create table: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 1911 :GNC 1 : PQclear(res);
1912 : :
1572 alvherre@alvh.no-ip. 1913 :CBC 1 : res = PQexec(conn, "begin");
1914 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1572 alvherre@alvh.no-ip. 1915 :UBC 0 : pg_fatal("failed to begin transaction: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 1916 :GNC 1 : PQclear(res);
1917 : :
1572 alvherre@alvh.no-ip. 1918 :CBC 1 : res = PQprepare(conn, "insertion",
1919 : : "insert into ppln_uniqviol values ($1, $2) returning id",
1920 : : 2, paramTypes);
55 tgl@sss.pgh.pa.us 1921 [ - + ]:GNC 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1572 alvherre@alvh.no-ip. 1922 :UBC 0 : pg_fatal("failed to prepare query: %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 1923 :GNC 1 : PQclear(res);
1924 : :
1572 alvherre@alvh.no-ip. 1925 [ - + ]:CBC 1 : if (PQenterPipelineMode(conn) != 1)
1572 alvherre@alvh.no-ip. 1926 :UBC 0 : pg_fatal("failed to enter pipeline mode");
1927 : :
1572 alvherre@alvh.no-ip. 1928 [ + - ]:CBC 7 : while (!read_done)
1929 : : {
1930 : : /*
1931 : : * Avoid deadlocks by reading everything the server has sent before
1932 : : * sending anything. (Special precaution is needed here to process
1933 : : * PQisBusy before testing the socket for read-readiness, because the
1934 : : * socket does not turn read-ready after "sending" queries in aborted
1935 : : * pipeline mode.)
1936 : : */
1937 [ + + ]: 606 : while (PQisBusy(conn) == 0)
1938 : : {
1939 : : bool new_error;
1940 : :
1941 [ + + ]: 601 : if (results >= numsent)
1942 : : {
1943 [ - + ]: 1 : if (write_done)
1572 alvherre@alvh.no-ip. 1944 :UBC 0 : read_done = true;
1572 alvherre@alvh.no-ip. 1945 :CBC 1 : break;
1946 : : }
1947 : :
1948 : 600 : res = PQgetResult(conn);
1949 : 600 : new_error = process_result(conn, res, results, numsent);
1950 [ + + - + ]: 600 : if (new_error && got_error)
1572 alvherre@alvh.no-ip. 1951 :UBC 0 : pg_fatal("got two errors");
1572 alvherre@alvh.no-ip. 1952 :CBC 600 : got_error |= new_error;
1953 [ + + ]: 600 : if (results++ >= numsent - 1)
1954 : : {
1955 [ + - ]: 1 : if (write_done)
1956 : 1 : read_done = true;
1957 : 1 : break;
1958 : : }
1959 : : }
1960 : :
1961 [ + + ]: 7 : if (read_done)
1962 : 1 : break;
1963 : :
1964 [ + + ]: 102 : FD_ZERO(&out_fds);
1965 : 6 : FD_SET(sock, &out_fds);
1966 : :
1967 [ + + ]: 102 : FD_ZERO(&in_fds);
1968 : 6 : FD_SET(sock, &in_fds);
1969 : :
1970 [ - + - + ]: 6 : if (select(sock + 1, &in_fds, write_done ? NULL : &out_fds, NULL, NULL) == -1)
1971 : : {
1572 alvherre@alvh.no-ip. 1972 [ # # ]:UBC 0 : if (errno == EINTR)
1973 : 0 : continue;
1974 : 0 : pg_fatal("select() failed: %m");
1975 : : }
1976 : :
1572 alvherre@alvh.no-ip. 1977 [ + + - + ]:CBC 6 : if (FD_ISSET(sock, &in_fds) && PQconsumeInput(conn) == 0)
1572 alvherre@alvh.no-ip. 1978 :UBC 0 : pg_fatal("PQconsumeInput failed: %s", PQerrorMessage(conn));
1979 : :
1980 : : /*
1981 : : * If the socket is writable and we haven't finished sending queries,
1982 : : * send some.
1983 : : */
1572 alvherre@alvh.no-ip. 1984 [ + - + + ]:CBC 6 : if (!write_done && FD_ISSET(sock, &out_fds))
1985 : : {
1986 : : for (;;)
1987 : 597 : {
1988 : : int flush;
1989 : :
1990 : : /*
1991 : : * provoke uniqueness violation exactly once after having
1992 : : * switched to read mode.
1993 : : */
1994 [ + + + + : 600 : if (switched >= 1 && !error_sent && ctr % socketful >= socketful / 2)
+ + ]
1995 : : {
1996 : 1 : sprintf(paramValue0, "%d", numsent / 2);
1997 : 1 : fprintf(stderr, "E");
1998 : 1 : error_sent = true;
1999 : : }
2000 : : else
2001 : : {
2002 : 599 : fprintf(stderr, ".");
2003 : 599 : sprintf(paramValue0, "%d", ctr++);
2004 : : }
2005 : :
2006 [ - + ]: 600 : if (PQsendQueryPrepared(conn, "insertion", 2, paramValues, NULL, NULL, 0) != 1)
1572 alvherre@alvh.no-ip. 2007 :UBC 0 : pg_fatal("failed to execute prepared query: %s", PQerrorMessage(conn));
1572 alvherre@alvh.no-ip. 2008 :CBC 600 : numsent++;
2009 : :
2010 : : /* Are we done writing? */
2011 [ + + + + : 600 : if (socketful != 0 && numsent % socketful == 42 && error_sent)
+ + ]
2012 : : {
2013 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1572 alvherre@alvh.no-ip. 2014 :UBC 0 : pg_fatal("failed to send flush request");
1572 alvherre@alvh.no-ip. 2015 :CBC 1 : write_done = true;
2016 : 1 : fprintf(stderr, "\ndone writing\n");
2017 : 1 : PQflush(conn);
2018 : 1 : break;
2019 : : }
2020 : :
2021 : : /* is the outgoing socket full? */
2022 : 599 : flush = PQflush(conn);
2023 [ - + ]: 599 : if (flush == -1)
1572 alvherre@alvh.no-ip. 2024 :UBC 0 : pg_fatal("failed to flush: %s", PQerrorMessage(conn));
1572 alvherre@alvh.no-ip. 2025 [ + + ]:CBC 599 : if (flush == 1)
2026 : : {
2027 [ + + ]: 2 : if (socketful == 0)
2028 : 1 : socketful = numsent;
2029 : 2 : fprintf(stderr, "\nswitch to reading\n");
2030 : 2 : switched++;
2031 : 2 : break;
2032 : : }
2033 : : }
2034 : : }
2035 : : }
2036 : :
2037 [ - + ]: 1 : if (!got_error)
1572 alvherre@alvh.no-ip. 2038 :UBC 0 : pg_fatal("did not get expected error");
2039 : :
1572 alvherre@alvh.no-ip. 2040 :CBC 1 : fprintf(stderr, "ok\n");
2041 : 1 : }
2042 : :
2043 : : /*
2044 : : * Subroutine for test_uniqviol; given a PGresult, print it out and consume
2045 : : * the expected NULL that should follow it.
2046 : : *
2047 : : * Returns true if we read a fatal error message, otherwise false.
2048 : : */
2049 : : static bool
2050 : 600 : process_result(PGconn *conn, PGresult *res, int results, int numsent)
2051 : : {
2052 : 600 : bool got_error = false;
2053 : :
2054 [ - + ]: 600 : if (res == NULL)
1572 alvherre@alvh.no-ip. 2055 :UBC 0 : pg_fatal("got unexpected NULL");
2056 : :
1572 alvherre@alvh.no-ip. 2057 [ + + + - ]:CBC 600 : switch (PQresultStatus(res))
2058 : : {
2059 : 1 : case PGRES_FATAL_ERROR:
2060 : 1 : got_error = true;
2061 : 1 : fprintf(stderr, "result %d/%d (error): %s\n", results, numsent, PQerrorMessage(conn));
2062 : 1 : PQclear(res);
55 tgl@sss.pgh.pa.us 2063 :GNC 1 : consume_null_result(conn);
1572 alvherre@alvh.no-ip. 2064 :CBC 1 : break;
2065 : :
2066 : 418 : case PGRES_TUPLES_OK:
2067 : 418 : fprintf(stderr, "result %d/%d: %s\n", results, numsent, PQgetvalue(res, 0, 0));
2068 : 418 : PQclear(res);
55 tgl@sss.pgh.pa.us 2069 :GNC 418 : consume_null_result(conn);
1572 alvherre@alvh.no-ip. 2070 :CBC 418 : break;
2071 : :
2072 : 181 : case PGRES_PIPELINE_ABORTED:
2073 : 181 : fprintf(stderr, "result %d/%d: pipeline aborted\n", results, numsent);
55 tgl@sss.pgh.pa.us 2074 :GNC 181 : PQclear(res);
2075 : 181 : consume_null_result(conn);
1572 alvherre@alvh.no-ip. 2076 :CBC 181 : break;
2077 : :
1572 alvherre@alvh.no-ip. 2078 :UBC 0 : default:
2079 : 0 : pg_fatal("got unexpected %s", PQresStatus(PQresultStatus(res)));
2080 : : }
2081 : :
1572 alvherre@alvh.no-ip. 2082 :CBC 600 : return got_error;
2083 : : }
2084 : :
2085 : :
2086 : : static void
1688 alvherre@alvh.no-ip. 2087 :UBC 0 : usage(const char *progname)
2088 : : {
2089 : 0 : fprintf(stderr, "%s tests libpq's pipeline mode.\n\n", progname);
2090 : 0 : fprintf(stderr, "Usage:\n");
1673 2091 : 0 : fprintf(stderr, " %s [OPTION] tests\n", progname);
1671 2092 : 0 : fprintf(stderr, " %s [OPTION] TESTNAME [CONNINFO]\n", progname);
1673 2093 : 0 : fprintf(stderr, "\nOptions:\n");
2094 : 0 : fprintf(stderr, " -t TRACEFILE generate a libpq trace to TRACEFILE\n");
1671 2095 : 0 : fprintf(stderr, " -r NUMROWS use NUMROWS as the test size\n");
1688 2096 : 0 : }
2097 : :
2098 : : static void
1688 alvherre@alvh.no-ip. 2099 :CBC 1 : print_test_list(void)
2100 : : {
596 2101 : 1 : printf("cancel\n");
1688 2102 : 1 : printf("disallowed_in_pipeline\n");
2103 : 1 : printf("multi_pipelines\n");
1582 2104 : 1 : printf("nosync\n");
1688 2105 : 1 : printf("pipeline_abort\n");
1211 2106 : 1 : printf("pipeline_idle\n");
1688 2107 : 1 : printf("pipelined_insert\n");
2108 : 1 : printf("prepared\n");
209 heikki.linnakangas@i 2109 : 1 : printf("protocol_version\n");
1688 alvherre@alvh.no-ip. 2110 : 1 : printf("simple_pipeline\n");
2111 : 1 : printf("singlerow\n");
2112 : 1 : printf("transaction\n");
1572 2113 : 1 : printf("uniqviol\n");
1688 2114 : 1 : }
2115 : :
2116 : : int
2117 : 15 : main(int argc, char **argv)
2118 : : {
2119 : 15 : const char *conninfo = "";
2120 : : PGconn *conn;
55 tgl@sss.pgh.pa.us 2121 :GNC 15 : FILE *trace = NULL;
2122 : : char *testname;
1688 alvherre@alvh.no-ip. 2123 :CBC 15 : int numrows = 10000;
2124 : : PGresult *res;
2125 : : int c;
2126 : :
1051 peter@eisentraut.org 2127 [ + + ]: 52 : while ((c = getopt(argc, argv, "r:t:")) != -1)
2128 : : {
1673 alvherre@alvh.no-ip. 2129 [ + + - ]: 22 : switch (c)
2130 : : {
1671 2131 : 13 : case 'r': /* numrows */
2132 : 13 : errno = 0;
2133 : 13 : numrows = strtol(optarg, NULL, 10);
2134 [ + - - + ]: 13 : if (errno != 0 || numrows <= 0)
2135 : : {
1671 alvherre@alvh.no-ip. 2136 :UBC 0 : fprintf(stderr, "couldn't parse \"%s\" as a positive integer\n",
2137 : : optarg);
2138 : 0 : exit(1);
2139 : : }
1671 alvherre@alvh.no-ip. 2140 :CBC 13 : break;
1051 peter@eisentraut.org 2141 : 9 : case 't': /* trace file */
2142 : 9 : tracefile = pg_strdup(optarg);
2143 : 9 : break;
2144 : : }
2145 : : }
2146 : :
1673 alvherre@alvh.no-ip. 2147 [ + - ]: 15 : if (optind < argc)
2148 : : {
1671 2149 : 15 : testname = pg_strdup(argv[optind]);
1673 2150 : 15 : optind++;
2151 : : }
2152 : : else
2153 : : {
1688 alvherre@alvh.no-ip. 2154 :UBC 0 : usage(argv[0]);
2155 : 0 : exit(1);
2156 : : }
2157 : :
1673 alvherre@alvh.no-ip. 2158 [ + + ]:CBC 15 : if (strcmp(testname, "tests") == 0)
2159 : : {
2160 : 1 : print_test_list();
2161 : 1 : exit(0);
2162 : : }
2163 : :
2164 [ + - ]: 14 : if (optind < argc)
2165 : : {
1671 2166 : 14 : conninfo = pg_strdup(argv[optind]);
1673 2167 : 14 : optind++;
2168 : : }
2169 : :
2170 : : /* Make a connection to the database */
1688 2171 : 14 : conn = PQconnectdb(conninfo);
2172 [ - + ]: 14 : if (PQstatus(conn) != CONNECTION_OK)
2173 : : {
1688 alvherre@alvh.no-ip. 2174 :UBC 0 : fprintf(stderr, "Connection to database failed: %s\n",
2175 : : PQerrorMessage(conn));
2176 : 0 : exit_nicely(conn);
2177 : : }
2178 : :
1672 alvherre@alvh.no-ip. 2179 :CBC 14 : res = PQexec(conn, "SET lc_messages TO \"C\"");
2180 [ - + ]: 14 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
529 peter@eisentraut.org 2181 :UBC 0 : pg_fatal("failed to set \"lc_messages\": %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 2182 :GNC 14 : PQclear(res);
986 drowley@postgresql.o 2183 :CBC 14 : res = PQexec(conn, "SET debug_parallel_query = off");
1672 alvherre@alvh.no-ip. 2184 [ - + ]: 14 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
529 peter@eisentraut.org 2185 :UBC 0 : pg_fatal("failed to set \"debug_parallel_query\": %s", PQerrorMessage(conn));
55 tgl@sss.pgh.pa.us 2186 :GNC 14 : PQclear(res);
2187 : :
2188 : : /* Set the trace file, if requested */
1673 alvherre@alvh.no-ip. 2189 [ + + ]:CBC 14 : if (tracefile != NULL)
2190 : : {
1211 2191 [ - + ]: 9 : if (strcmp(tracefile, "-") == 0)
1211 alvherre@alvh.no-ip. 2192 :UBC 0 : trace = stdout;
2193 : : else
1211 alvherre@alvh.no-ip. 2194 :CBC 9 : trace = fopen(tracefile, "w");
1673 2195 [ - + ]: 9 : if (trace == NULL)
1673 alvherre@alvh.no-ip. 2196 :UBC 0 : pg_fatal("could not open file \"%s\": %m", tracefile);
2197 : :
2198 : : /* Make it line-buffered */
1671 alvherre@alvh.no-ip. 2199 :CBC 9 : setvbuf(trace, NULL, PG_IOLBF, 0);
2200 : :
1673 2201 : 9 : PQtrace(conn, trace);
1601 noah@leadboat.com 2202 : 9 : PQsetTraceFlags(conn,
2203 : : PQTRACE_SUPPRESS_TIMESTAMPS | PQTRACE_REGRESS_MODE);
2204 : : }
2205 : :
596 alvherre@alvh.no-ip. 2206 [ + + ]: 14 : if (strcmp(testname, "cancel") == 0)
2207 : 2 : test_cancel(conn);
2208 [ + + ]: 12 : else if (strcmp(testname, "disallowed_in_pipeline") == 0)
1688 2209 : 1 : test_disallowed_in_pipeline(conn);
1673 2210 [ + + ]: 11 : else if (strcmp(testname, "multi_pipelines") == 0)
1688 2211 : 1 : test_multi_pipelines(conn);
1582 2212 [ + + ]: 10 : else if (strcmp(testname, "nosync") == 0)
2213 : 1 : test_nosync(conn);
1673 2214 [ + + ]: 9 : else if (strcmp(testname, "pipeline_abort") == 0)
1688 2215 : 1 : test_pipeline_abort(conn);
1211 2216 [ + + ]: 8 : else if (strcmp(testname, "pipeline_idle") == 0)
2217 : 1 : test_pipeline_idle(conn);
1673 2218 [ + + ]: 7 : else if (strcmp(testname, "pipelined_insert") == 0)
1688 2219 : 1 : test_pipelined_insert(conn, numrows);
1673 2220 [ + + ]: 6 : else if (strcmp(testname, "prepared") == 0)
1688 2221 : 1 : test_prepared(conn);
209 heikki.linnakangas@i 2222 [ + + ]: 5 : else if (strcmp(testname, "protocol_version") == 0)
2223 : 1 : test_protocol_version(conn);
1673 alvherre@alvh.no-ip. 2224 [ + + ]: 4 : else if (strcmp(testname, "simple_pipeline") == 0)
1688 2225 : 1 : test_simple_pipeline(conn);
1673 2226 [ + + ]: 3 : else if (strcmp(testname, "singlerow") == 0)
1688 2227 : 1 : test_singlerowmode(conn);
1673 2228 [ + + ]: 2 : else if (strcmp(testname, "transaction") == 0)
1688 2229 : 1 : test_transaction(conn);
1572 2230 [ + - ]: 1 : else if (strcmp(testname, "uniqviol") == 0)
2231 : 1 : test_uniqviol(conn);
2232 : : else
2233 : : {
1673 alvherre@alvh.no-ip. 2234 :UBC 0 : fprintf(stderr, "\"%s\" is not a recognized test name\n", testname);
1688 2235 : 0 : exit(1);
2236 : : }
2237 : :
2238 : : /* close the connection to the database and cleanup */
1688 alvherre@alvh.no-ip. 2239 :CBC 14 : PQfinish(conn);
2240 : :
55 tgl@sss.pgh.pa.us 2241 [ + + + - ]:GNC 14 : if (trace && trace != stdout)
2242 : 9 : fclose(trace);
2243 : :
1688 alvherre@alvh.no-ip. 2244 :CBC 14 : return 0;
2245 : : }
|
