Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * libpq_pipeline.c
4 : : * Verify libpq pipeline execution functionality
5 : : *
6 : : * Portions Copyright (c) 1996-2026, 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
59 : 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 : : 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 *
96 : 56 : confirm_result_status_impl(int line, PGconn *conn, ExecStatusType status)
97 : : {
98 : : PGresult *res;
99 : :
100 : 56 : res = PQgetResult(conn);
101 [ - + ]: 56 : if (res == NULL)
102 : 0 : pg_fatal_impl(line, "PQgetResult returned null unexpectedly: %s",
103 : : PQerrorMessage(conn));
104 [ - + ]: 56 : if (PQresultStatus(res) != status)
105 : 0 : pg_fatal_impl(line, "PQgetResult returned status %s, expected %s: %s",
106 : : PQresStatus(PQresultStatus(res)),
107 : : PQresStatus(status),
108 : : PQerrorMessage(conn));
109 : 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 : 663 : consume_null_result_impl(int line, PGconn *conn)
132 : : {
133 : : PGresult *res;
134 : :
135 : 663 : res = PQgetResult(conn);
136 [ - + ]: 663 : if (res != NULL)
137 : 0 : pg_fatal_impl(line, "expected NULL PGresult, got %s: %s",
138 : : PQresStatus(PQresultStatus(res)),
139 : : PQerrorMessage(conn));
140 : 663 : }
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);
152 [ - + ]: 12 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "57014") != 0)
153 : 0 : pg_fatal_impl(line, "query failed with a different error than cancellation: %s",
154 : : PQerrorMessage(conn));
155 : 12 : PQclear(res);
156 : :
157 [ - + ]: 12 : while (PQisBusy(conn))
158 : 0 : PQconsumeInput(conn);
159 : 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
167 : 24 : wait_for_connection_state(int line, PGconn *monitorConn, int procpid,
168 : : char *state, char *event)
169 : : {
170 : 24 : const Oid paramTypes[] = {INT4OID, TEXTOID};
171 : : const char *paramValues[2];
172 : 24 : char *pidstr = psprintf("%d", procpid);
173 : :
174 : : Assert((state == NULL) ^ (event == NULL));
175 : :
176 : 24 : paramValues[0] = pidstr;
177 [ + + ]: 24 : paramValues[1] = state ? state : event;
178 : :
179 : : while (true)
180 : 2 : {
181 : : PGresult *res;
182 : : char *value;
183 : :
184 [ + + ]: 26 : 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 : 14 : 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 : :
195 [ - + ]: 26 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
196 : 0 : pg_fatal_impl(line, "could not query pg_stat_activity: %s", PQerrorMessage(monitorConn));
197 [ - + ]: 26 : if (PQntuples(res) != 1)
198 : 0 : pg_fatal_impl(line, "unexpected number of rows received: %d", PQntuples(res));
199 [ - + ]: 26 : if (PQnfields(res) != 1)
200 : 0 : pg_fatal_impl(line, "unexpected number of columns received: %d", PQnfields(res));
201 : 26 : value = PQgetvalue(res, 0, 0);
202 [ + + ]: 26 : if (strcmp(value, "0") != 0)
203 : : {
204 : 24 : PQclear(res);
205 : 24 : break;
206 : : }
207 : 2 : PQclear(res);
208 : :
209 : : /* wait 10ms before polling again */
210 : 2 : pg_usleep(10000);
211 : : }
212 : :
213 : 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 : : */
229 : 12 : wait_for_connection_state(line, monitorConn, PQbackendPID(conn), "idle", NULL);
230 : :
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)
237 : 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 : : */
243 : 12 : wait_for_connection_state(line, monitorConn, PQbackendPID(conn), NULL, "PgSleep");
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;
256 : 2 : int nopts = 0;
257 : : int i;
258 : :
259 [ + + ]: 106 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
260 : 104 : nopts++;
261 : 2 : nopts++; /* for the NULL terminator */
262 : :
263 : 2 : keywords = pg_malloc_array(const char *, nopts);
264 : 2 : vals = pg_malloc_array(const char *, nopts);
265 : :
266 : 2 : i = 0;
267 [ + + ]: 106 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
268 : : {
269 [ + + ]: 104 : 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)
281 : 0 : pg_fatal("Connection to database failed: %s",
282 : : PQerrorMessage(copyConn));
283 : :
284 : 2 : pg_free(keywords);
285 : 2 : pg_free(vals);
286 : 2 : PQconninfoFree(opts);
287 : :
288 : 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)
305 : 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 : : */
311 : 2 : monitorConn = copy_connection(conn);
312 : : 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)))
318 : 0 : pg_fatal("failed to run PQcancel: %s", errorbuf);
319 : 2 : consume_query_cancel(conn);
320 : :
321 : : /* PGcancel object can be reused for the next query */
322 : 2 : send_cancellable_query(conn, monitorConn);
323 [ - + ]: 2 : if (!PQcancel(cancel, errorbuf, sizeof(errorbuf)))
324 : 0 : pg_fatal("failed to run PQcancel: %s", errorbuf);
325 : 2 : consume_query_cancel(conn);
326 : :
327 : 2 : PQfreeCancel(cancel);
328 : :
329 : : /* test PQrequestCancel */
330 : 2 : send_cancellable_query(conn, monitorConn);
331 [ - + ]: 2 : if (!PQrequestCancel(conn))
332 : 0 : pg_fatal("failed to run PQrequestCancel: %s", PQerrorMessage(conn));
333 : 2 : consume_query_cancel(conn);
334 : :
335 : : /* test PQcancelBlocking */
336 : 2 : send_cancellable_query(conn, monitorConn);
337 : 2 : cancelConn = PQcancelCreate(conn);
338 [ - + ]: 2 : if (!PQcancelBlocking(cancelConn))
339 : 0 : pg_fatal("failed to run PQcancelBlocking: %s", PQcancelErrorMessage(cancelConn));
340 : 2 : consume_query_cancel(conn);
341 : 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))
347 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
348 : : while (true)
349 : 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;
367 : 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 : :
375 [ - + ]: 2 : if (sock < 0)
376 : 0 : pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn));
377 : :
378 : 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 : : {
385 [ # # ]: 0 : if (errno == EINTR)
386 : 0 : continue;
387 : 0 : pg_fatal("select() failed: %m");
388 : : }
389 : 2 : break;
390 : : }
391 : : }
392 [ - + ]: 2 : if (PQcancelStatus(cancelConn) != CONNECTION_OK)
393 : 0 : pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn));
394 : 2 : consume_query_cancel(conn);
395 : :
396 : : /*
397 : : * test PQcancelReset works on the cancel connection and it can be reused
398 : : * afterwards
399 : : */
400 : 2 : PQcancelReset(cancelConn);
401 : :
402 : 2 : send_cancellable_query(conn, monitorConn);
403 [ - + ]: 2 : if (!PQcancelStart(cancelConn))
404 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
405 : : while (true)
406 : 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;
424 : 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 : :
432 [ - + ]: 2 : if (sock < 0)
433 : 0 : pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn));
434 : :
435 : 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 : : {
442 [ # # ]: 0 : if (errno == EINTR)
443 : 0 : continue;
444 : 0 : pg_fatal("select() failed: %m");
445 : : }
446 : 2 : break;
447 : : }
448 : : }
449 [ - + ]: 2 : if (PQcancelStatus(cancelConn) != CONNECTION_OK)
450 : 0 : pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn));
451 : 2 : consume_query_cancel(conn);
452 : :
453 : 2 : PQcancelFinish(cancelConn);
454 : 2 : PQfinish(monitorConn);
455 : :
456 : 2 : fprintf(stderr, "ok\n");
457 : 2 : }
458 : :
459 : : static void
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))
467 : 0 : pg_fatal("Expected blocking connection mode");
468 : :
469 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
470 : 0 : pg_fatal("Unable to enter pipeline mode");
471 : :
472 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
473 : 0 : pg_fatal("Pipeline mode not activated properly");
474 : :
475 : : /* PQexec should fail in pipeline mode */
476 : 1 : res = PQexec(conn, "SELECT 1");
477 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
478 : 0 : pg_fatal("PQexec should fail in pipeline mode but succeeded");
479 [ - + ]: 1 : if (strcmp(PQerrorMessage(conn),
480 : : "synchronous command execution functions are not allowed in pipeline mode\n") != 0)
481 : 0 : pg_fatal("did not get expected error message; got: \"%s\"",
482 : : PQerrorMessage(conn));
483 : 1 : PQclear(res);
484 : :
485 : : /* PQsendQuery should fail in pipeline mode */
486 [ - + ]: 1 : if (PQsendQuery(conn, "SELECT 1") != 0)
487 : 0 : pg_fatal("PQsendQuery should fail in pipeline mode but succeeded");
488 [ - + ]: 1 : if (strcmp(PQerrorMessage(conn),
489 : : "PQsendQuery not allowed in pipeline mode\n") != 0)
490 : 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 */
494 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
495 : 0 : pg_fatal("re-entering pipeline mode should be a no-op but failed");
496 : :
497 [ - + ]: 1 : if (PQisBusy(conn) != 0)
498 : 0 : pg_fatal("PQisBusy should return 0 when idle in pipeline mode, returned 1");
499 : :
500 : : /* ok, back to normal command mode */
501 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
502 : 0 : pg_fatal("couldn't exit idle empty pipeline mode");
503 : :
504 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
505 : 0 : pg_fatal("Pipeline mode not terminated properly");
506 : :
507 : : /* exiting pipeline mode when not in pipeline mode should be a no-op */
508 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
509 : 0 : pg_fatal("pipeline mode exit when not in pipeline mode should succeed but failed");
510 : :
511 : : /* can now PQexec again */
512 : 1 : res = PQexec(conn, "SELECT 1");
513 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
514 : 0 : pg_fatal("PQexec should succeed after exiting pipeline mode but failed with: %s",
515 : : PQerrorMessage(conn));
516 : 1 : PQclear(res);
517 : :
518 : 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)
534 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
535 : :
536 : : /* first pipeline */
537 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
538 : : dummy_params, NULL, NULL, 0) != 1)
539 : 0 : pg_fatal("dispatching first SELECT failed: %s", PQerrorMessage(conn));
540 : :
541 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
542 : 0 : pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
543 : :
544 : : /* second pipeline */
545 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
546 : : dummy_params, NULL, NULL, 0) != 1)
547 : 0 : pg_fatal("dispatching second SELECT failed: %s", PQerrorMessage(conn));
548 : :
549 : : /* Skip flushing once. */
550 [ - + ]: 1 : if (PQsendPipelineSync(conn) != 1)
551 : 0 : pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
552 : :
553 : : /* third pipeline */
554 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
555 : : dummy_params, NULL, NULL, 0) != 1)
556 : 0 : pg_fatal("dispatching third SELECT failed: %s", PQerrorMessage(conn));
557 : :
558 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
559 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
560 : :
561 : : /* OK, start processing the results */
562 : :
563 : : /* first pipeline */
564 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
565 : :
566 : 1 : consume_null_result(conn);
567 : :
568 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
569 : 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
570 : :
571 : 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 : :
578 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
579 : 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
580 : :
581 : 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 ... */
591 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
592 : 0 : pg_fatal("Fell out of pipeline mode somehow");
593 : :
594 : : /* until we end it, which we can safely do now */
595 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
596 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
597 : : PQerrorMessage(conn));
598 : :
599 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
600 : 0 : pg_fatal("exiting pipeline mode didn't seem to work");
601 : :
602 : 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
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)
619 : 0 : pg_fatal("invalid socket");
620 : :
621 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
622 : 0 : pg_fatal("could not enter pipeline mode");
623 [ + + ]: 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)
630 : 0 : pg_fatal("error sending select: %s", PQerrorMessage(conn));
631 : 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 : : {
642 : 0 : fprintf(stderr, "select() failed: %m\n");
643 : 0 : exit_nicely(conn);
644 : : }
645 [ - + - - ]: 10 : if (FD_ISSET(sock, &input_mask) && PQconsumeInput(conn) != 1)
646 : 0 : pg_fatal("failed to read from server: %s", PQerrorMessage(conn));
647 : : }
648 : :
649 : : /* tell server to flush its output buffer */
650 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
651 : 0 : pg_fatal("failed to send flush request");
652 : 1 : PQflush(conn);
653 : :
654 : : /* Now read all results */
655 : : for (;;)
656 : : {
657 : : /* We expect exactly one TUPLES_OK result for each query we sent */
658 : 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 : :
670 : 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
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)
696 : 0 : pg_fatal("dispatching DROP TABLE failed: %s", PQerrorMessage(conn));
697 : 1 : PQclear(res);
698 : :
699 : 1 : res = PQexec(conn, create_table_sql);
700 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
701 : 0 : pg_fatal("dispatching CREATE TABLE failed: %s", PQerrorMessage(conn));
702 : 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 : : */
709 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
710 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
711 : :
712 : 1 : dummy_params[0] = "1";
713 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
714 : : dummy_params, NULL, NULL, 0) != 1)
715 : 0 : pg_fatal("dispatching first insert failed: %s", PQerrorMessage(conn));
716 : :
717 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT no_such_function($1)",
718 : : 1, dummy_param_oids, dummy_params,
719 : : NULL, NULL, 0) != 1)
720 : 0 : pg_fatal("dispatching error select failed: %s", PQerrorMessage(conn));
721 : :
722 : 1 : dummy_params[0] = "2";
723 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
724 : : dummy_params, NULL, NULL, 0) != 1)
725 : 0 : pg_fatal("dispatching second insert failed: %s", PQerrorMessage(conn));
726 : :
727 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
728 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
729 : :
730 : 1 : dummy_params[0] = "3";
731 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
732 : : dummy_params, NULL, NULL, 0) != 1)
733 : 0 : pg_fatal("dispatching second-pipeline insert failed: %s",
734 : : PQerrorMessage(conn));
735 : :
736 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
737 : 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 : : */
746 : 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 : : */
764 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
765 : 0 : pg_fatal("pipeline should be flagged as aborted but isn't");
766 : :
767 : : /* third query in pipeline, the second insert */
768 : 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 : :
773 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
774 : 0 : pg_fatal("pipeline should be flagged as aborted but isn't");
775 : :
776 : : /* Ensure we're still in pipeline */
777 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
778 : 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 : : */
786 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
787 : :
788 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_ABORTED)
789 : 0 : pg_fatal("sync should've cleared the aborted flag but didn't");
790 : :
791 : : /* We're still in pipeline mode... */
792 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
793 : 0 : pg_fatal("Fell out of pipeline mode somehow");
794 : :
795 : : /* the insert from the second pipeline */
796 : 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 */
808 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1; SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
809 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
810 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
811 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
812 : 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)
819 : 0 : pg_fatal("expected error about multiple commands, got %s",
820 : : PQerrorMessage(conn));
821 : 1 : printf("got expected %s", PQerrorMessage(conn));
822 : 1 : goterror = true;
823 : 1 : break;
824 : 0 : default:
825 : 0 : pg_fatal("got unexpected status %s", PQresStatus(PQresultStatus(res)));
826 : : break;
827 : : }
828 : 1 : PQclear(res);
829 : : }
830 [ - + ]: 1 : if (!goterror)
831 : 0 : pg_fatal("did not get cannot-insert-multiple-commands error");
832 : :
833 : : /* the second pipeline sync */
834 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
835 : :
836 : 1 : fprintf(stderr, "ok\n");
837 : :
838 : : /* Test single-row mode with an error partways */
839 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1.0/g FROM generate_series(3, -1, -1) g",
840 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
841 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
842 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
843 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
844 : 1 : PQsetSingleRowMode(conn);
845 : 1 : goterror = false;
846 : 1 : gotrows = 0;
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));
853 : 3 : gotrows++;
854 : 3 : break;
855 : 1 : case PGRES_FATAL_ERROR:
856 [ - + ]: 1 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "22012") != 0)
857 : 0 : pg_fatal("expected division-by-zero, got: %s (%s)",
858 : : PQerrorMessage(conn),
859 : : PQresultErrorField(res, PG_DIAG_SQLSTATE));
860 : 1 : printf("got expected division-by-zero\n");
861 : 1 : goterror = true;
862 : 1 : break;
863 : 0 : default:
864 : 0 : pg_fatal("got unexpected result %s", PQresStatus(PQresultStatus(res)));
865 : : }
866 : 4 : PQclear(res);
867 : : }
868 [ - + ]: 1 : if (!goterror)
869 : 0 : pg_fatal("did not get division-by-zero error");
870 [ - + ]: 1 : if (gotrows != 3)
871 : 0 : pg_fatal("did not get three rows");
872 : :
873 : : /* the third pipeline sync */
874 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
875 : :
876 : : /* We're still in pipeline mode... */
877 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
878 : 0 : pg_fatal("Fell out of pipeline mode somehow");
879 : :
880 : : /* until we end it, which we can safely do now */
881 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
882 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
883 : : PQerrorMessage(conn));
884 : :
885 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
886 : 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 : : */
903 : 1 : res = PQexec(conn, "SELECT itemno FROM pq_pipeline_demo");
904 : :
905 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
906 : 0 : pg_fatal("Expected tuples, got %s: %s",
907 : : PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
908 [ - + ]: 1 : if (PQntuples(res) != 1)
909 : 0 : pg_fatal("expected 1 result, got %d", PQntuples(res));
910 [ + + ]: 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)
915 : 0 : pg_fatal("expected only insert with value 3, got %s", val);
916 : : }
917 : :
918 : 1 : PQclear(res);
919 : :
920 : 1 : fprintf(stderr, "ok\n");
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 : : {
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];
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 : :
948 : 1 : insert_params[0] = insert_param_0;
949 : 1 : insert_params[1] = insert_param_1;
950 : :
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)
957 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
958 : :
959 [ + + ]: 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 : :
980 : 0 : default:
981 : 0 : pg_fatal("invalid state");
982 : : sql = NULL; /* keep compiler quiet */
983 : : }
984 : :
985 : : pg_debug("sending: %s\n", sql);
986 [ - + ]: 3 : if (PQsendQueryParams(conn, sql,
987 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
988 : 0 : pg_fatal("dispatching %s failed: %s", sql, PQerrorMessage(conn));
989 : : }
990 : :
991 : : Assert(send_step == BI_PREPARE);
992 : : pg_debug("sending: %s\n", insert_sql2);
993 [ - + ]: 1 : if (PQsendPrepare(conn, "my_insert", insert_sql2, 2, insert_param_oids) != 1)
994 : 0 : pg_fatal("dispatching PREPARE failed: %s", PQerrorMessage(conn));
995 : 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)
1006 : 0 : pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn));
1007 : :
1008 [ + + ]: 3677 : while (recv_step != BI_DONE)
1009 : : {
1010 : : int sock;
1011 : : fd_set input_mask;
1012 : : fd_set output_mask;
1013 : :
1014 : 3676 : sock = PQsocket(conn);
1015 : :
1016 [ - + ]: 3676 : if (sock < 0)
1017 : 0 : break; /* shouldn't happen */
1018 : :
1019 [ + + ]: 62492 : FD_ZERO(&input_mask);
1020 : 3676 : FD_SET(sock, &input_mask);
1021 [ + + ]: 62492 : FD_ZERO(&output_mask);
1022 : 3676 : FD_SET(sock, &output_mask);
1023 : :
1024 [ - + ]: 3676 : if (select(sock + 1, &input_mask, &output_mask, NULL, NULL) < 0)
1025 : : {
1026 : 0 : fprintf(stderr, "select() failed: %m\n");
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 : : */
1034 [ + + ]: 3676 : 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;
1042 : 1411 : const char *cmdtag = "";
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;
1090 : 0 : case BI_DONE:
1091 : : /* unreachable */
1092 : 0 : pg_fatal("unreachable state");
1093 : : }
1094 : :
1095 [ - + ]: 706 : if (PQresultStatus(res) != status)
1096 : 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 : :
1101 [ - + ]: 706 : if (strncmp(PQcmdStatus(res), cmdtag, strlen(cmdtag)) != 0)
1102 : 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 : :
1107 : 706 : PQclear(res);
1108 : : }
1109 : : }
1110 : :
1111 : : /* Write more rows and/or the end pipeline message, if needed */
1112 [ + - ]: 3676 : if (FD_ISSET(sock, &output_mask))
1113 : : {
1114 : 3676 : PQflush(conn);
1115 : :
1116 [ + + ]: 3676 : if (send_step == BI_INSERT_ROWS)
1117 : : {
1118 : 700 : snprintf(insert_param_0, MAXINTLEN, "%d", rows_to_send);
1119 : : /* use up some buffer space with a wide value */
1120 : 700 : snprintf(insert_param_1, MAXINT8LEN, "%lld", 1LL << 62);
1121 : :
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 : : */
1137 : 0 : fprintf(stderr, "WARNING: failed to send insert #%d: %s\n",
1138 : : rows_to_send, PQerrorMessage(conn));
1139 : : }
1140 : : }
1141 [ + + ]: 2976 : 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 : : {
1151 : 0 : fprintf(stderr, "WARNING: failed to send commit: %s\n",
1152 : : PQerrorMessage(conn));
1153 : : }
1154 : : }
1155 [ + + ]: 2975 : 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 : : {
1164 : 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 */
1172 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1173 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
1174 : : PQerrorMessage(conn));
1175 : :
1176 [ - + ]: 1 : if (PQsetnonblocking(conn, 0) != 0)
1177 : 0 : pg_fatal("failed to clear nonblocking mode: %s", PQerrorMessage(conn));
1178 : :
1179 : 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)
1193 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1194 [ - + ]: 1 : if (PQsendPrepare(conn, "select_one", "SELECT $1, '42', $1::numeric, "
1195 : : "interval '1 sec'",
1196 : : 1, param_oids) != 1)
1197 : 0 : pg_fatal("preparing query failed: %s", PQerrorMessage(conn));
1198 : 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)
1203 : 0 : pg_fatal("failed to send describePrepared: %s", PQerrorMessage(conn));
1204 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1205 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1206 : :
1207 : 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);
1212 [ - + ]: 1 : if (PQnfields(res) != lengthof(expected_oids))
1213 : 0 : pg_fatal("expected %zu columns, got %d",
1214 : : lengthof(expected_oids), PQnfields(res));
1215 [ + + ]: 5 : for (int i = 0; i < PQnfields(res); i++)
1216 : : {
1217 : 4 : typ = PQftype(res, i);
1218 [ - + ]: 4 : if (typ != expected_oids[i])
1219 : 0 : pg_fatal("field %d: expected type %u, got %u",
1220 : : i, expected_oids[i], typ);
1221 : : }
1222 : 1 : PQclear(res);
1223 : :
1224 : 1 : consume_null_result(conn);
1225 : :
1226 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1227 : :
1228 : 1 : fprintf(stderr, "closing statement..");
1229 [ - + ]: 1 : if (PQsendClosePrepared(conn, "select_one") != 1)
1230 : 0 : pg_fatal("PQsendClosePrepared failed: %s", PQerrorMessage(conn));
1231 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1232 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1233 : :
1234 : 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 : :
1240 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1241 : 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 */
1244 : 1 : res = PQdescribePrepared(conn, "select_one");
1245 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
1246 : 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
1247 : 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 : : */
1253 : 1 : res = PQclosePrepared(conn, "select_one");
1254 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1255 : 0 : pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
1256 : 1 : PQclear(res);
1257 : :
1258 : 1 : fprintf(stderr, "creating portal... ");
1259 : :
1260 : 1 : res = PQexec(conn, "BEGIN");
1261 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1262 : 0 : pg_fatal("BEGIN failed: %s", PQerrorMessage(conn));
1263 : 1 : PQclear(res);
1264 : :
1265 : 1 : res = PQexec(conn, "DECLARE cursor_one CURSOR FOR SELECT 1");
1266 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1267 : 0 : pg_fatal("DECLARE CURSOR failed: %s", PQerrorMessage(conn));
1268 : 1 : PQclear(res);
1269 : :
1270 : 1 : PQenterPipelineMode(conn);
1271 [ - + ]: 1 : if (PQsendDescribePortal(conn, "cursor_one") != 1)
1272 : 0 : pg_fatal("PQsendDescribePortal failed: %s", PQerrorMessage(conn));
1273 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1274 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1275 : :
1276 : 1 : res = confirm_result_status(conn, PGRES_COMMAND_OK);
1277 : 1 : typ = PQftype(res, 0);
1278 [ - + ]: 1 : if (typ != INT4OID)
1279 : 0 : pg_fatal("portal: expected type %u, got %u",
1280 : : INT4OID, typ);
1281 : 1 : PQclear(res);
1282 : :
1283 : 1 : consume_null_result(conn);
1284 : :
1285 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1286 : :
1287 : 1 : fprintf(stderr, "closing portal... ");
1288 [ - + ]: 1 : if (PQsendClosePortal(conn, "cursor_one") != 1)
1289 : 0 : pg_fatal("PQsendClosePortal failed: %s", PQerrorMessage(conn));
1290 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1291 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1292 : :
1293 : 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 : :
1299 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1300 : 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 */
1303 : 1 : res = PQdescribePortal(conn, "cursor_one");
1304 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
1305 : 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
1306 : 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 : : */
1312 : 1 : res = PQclosePortal(conn, "cursor_one");
1313 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1314 : 0 : pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
1315 : 1 : PQclear(res);
1316 : :
1317 : 1 : fprintf(stderr, "ok\n");
1318 : 1 : }
1319 : :
1320 : : /*
1321 : : * Test max_protocol_version options.
1322 : : */
1323 : : static void
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 : 1 : int max_protocol_version_index = -1;
1332 : : int i;
1333 : :
1334 : : /* Prepare keywords/vals arrays, copied from the existing connection. */
1335 : 1 : nopts = 0;
1336 [ + + ]: 53 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1337 : 52 : nopts++;
1338 : 1 : nopts++; /* NULL terminator */
1339 : :
1340 : 1 : keywords = pg_malloc0_array(const char *, nopts);
1341 : 1 : vals = pg_malloc0_array(const char *, nopts);
1342 : :
1343 : 1 : i = 0;
1344 [ + + ]: 53 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1345 : : {
1346 : : /*
1347 : : * If the test already specified max_protocol_version, we want to
1348 : : * replace it rather than attempting to override it. This matters when
1349 : : * testing defaults, because empty option values at the end of the
1350 : : * connection string won't replace earlier settings.
1351 : : */
1352 [ + + ]: 52 : if (strcmp(opt->keyword, "max_protocol_version") == 0)
1353 : 1 : max_protocol_version_index = i;
1354 [ + + ]: 51 : else if (!opt->val)
1355 : 32 : continue;
1356 : :
1357 : 20 : keywords[i] = opt->keyword;
1358 : 20 : vals[i] = opt->val;
1359 : :
1360 : 20 : i++;
1361 : : }
1362 : :
1363 : : Assert(max_protocol_version_index >= 0);
1364 : :
1365 : : /*
1366 : : * Test default protocol_version (GREASE - should negotiate down to 3.2)
1367 : : */
1368 : 1 : vals[max_protocol_version_index] = "";
1369 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1370 : :
1371 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1372 : 0 : pg_fatal("Connection to database failed: %s",
1373 : : PQerrorMessage(conn));
1374 : :
1375 : 1 : protocol_version = PQfullProtocolVersion(conn);
1376 [ - + ]: 1 : if (protocol_version != 30002)
1377 : 0 : pg_fatal("expected 30002, got %d", protocol_version);
1378 : :
1379 : 1 : PQfinish(conn);
1380 : :
1381 : : /*
1382 : : * Test max_protocol_version=3.0
1383 : : */
1384 : 1 : vals[max_protocol_version_index] = "3.0";
1385 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1386 : :
1387 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1388 : 0 : pg_fatal("Connection to database failed: %s",
1389 : : PQerrorMessage(conn));
1390 : :
1391 : 1 : protocol_version = PQfullProtocolVersion(conn);
1392 [ - + ]: 1 : if (protocol_version != 30000)
1393 : 0 : pg_fatal("expected 30000, got %d", protocol_version);
1394 : :
1395 : 1 : PQfinish(conn);
1396 : :
1397 : : /*
1398 : : * Test max_protocol_version=3.1. It's not valid, we went straight from
1399 : : * 3.0 to 3.2.
1400 : : */
1401 : 1 : vals[max_protocol_version_index] = "3.1";
1402 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1403 : :
1404 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_BAD)
1405 : 0 : pg_fatal("Connecting with max_protocol_version 3.1 should have failed.");
1406 : :
1407 : 1 : PQfinish(conn);
1408 : :
1409 : : /*
1410 : : * Test max_protocol_version=3.2
1411 : : */
1412 : 1 : vals[max_protocol_version_index] = "3.2";
1413 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1414 : :
1415 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1416 : 0 : pg_fatal("Connection to database failed: %s",
1417 : : PQerrorMessage(conn));
1418 : :
1419 : 1 : protocol_version = PQfullProtocolVersion(conn);
1420 [ - + ]: 1 : if (protocol_version != 30002)
1421 : 0 : pg_fatal("expected 30002, got %d", protocol_version);
1422 : :
1423 : 1 : PQfinish(conn);
1424 : :
1425 : : /*
1426 : : * Test max_protocol_version=latest. 'latest' currently means '3.2'.
1427 : : */
1428 : 1 : vals[max_protocol_version_index] = "latest";
1429 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1430 : :
1431 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1432 : 0 : pg_fatal("Connection to database failed: %s",
1433 : : PQerrorMessage(conn));
1434 : :
1435 : 1 : protocol_version = PQfullProtocolVersion(conn);
1436 [ - + ]: 1 : if (protocol_version != 30002)
1437 : 0 : pg_fatal("expected 30002, got %d", protocol_version);
1438 : :
1439 : 1 : PQfinish(conn);
1440 : :
1441 : 1 : pg_free(keywords);
1442 : 1 : pg_free(vals);
1443 : 1 : PQconninfoFree(opts);
1444 : 1 : }
1445 : :
1446 : : /* Notice processor: print notices, and count how many we got */
1447 : : static void
1448 : 1 : notice_processor(void *arg, const char *message)
1449 : : {
1450 : 1 : int *n_notices = (int *) arg;
1451 : :
1452 : 1 : (*n_notices)++;
1453 : 1 : fprintf(stderr, "NOTICE %d: %s", *n_notices, message);
1454 : 1 : }
1455 : :
1456 : : /* Verify behavior in "idle" state */
1457 : : static void
1458 : 1 : test_pipeline_idle(PGconn *conn)
1459 : : {
1460 : 1 : int n_notices = 0;
1461 : :
1462 : 1 : fprintf(stderr, "\npipeline idle...\n");
1463 : :
1464 : 1 : PQsetNoticeProcessor(conn, notice_processor, &n_notices);
1465 : :
1466 : : /* Try to exit pipeline mode in pipeline-idle state */
1467 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1468 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1469 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1", 0, NULL, NULL, NULL, NULL, 0) != 1)
1470 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1471 : 1 : PQsendFlushRequest(conn);
1472 : :
1473 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1474 : :
1475 : 1 : consume_null_result(conn);
1476 : :
1477 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
1478 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1479 [ - + ]: 1 : if (PQexitPipelineMode(conn) == 1)
1480 : 0 : pg_fatal("exiting pipeline succeeded when it shouldn't");
1481 [ - + ]: 1 : if (strncmp(PQerrorMessage(conn), "cannot exit pipeline mode",
1482 : : strlen("cannot exit pipeline mode")) != 0)
1483 : 0 : pg_fatal("did not get expected error; got: %s",
1484 : : PQerrorMessage(conn));
1485 : 1 : PQsendFlushRequest(conn);
1486 : :
1487 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1488 : :
1489 : 1 : consume_null_result(conn);
1490 : :
1491 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1492 : 0 : pg_fatal("exiting pipeline failed: %s", PQerrorMessage(conn));
1493 : :
1494 [ - + ]: 1 : if (n_notices > 0)
1495 : 0 : pg_fatal("got %d notice(s)", n_notices);
1496 : 1 : fprintf(stderr, "ok - 1\n");
1497 : :
1498 : : /* Have a WARNING in the middle of a resultset */
1499 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1500 : 0 : pg_fatal("entering pipeline mode failed: %s", PQerrorMessage(conn));
1501 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT pg_catalog.pg_advisory_unlock(1,1)", 0, NULL, NULL, NULL, NULL, 0) != 1)
1502 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1503 : 1 : PQsendFlushRequest(conn);
1504 : :
1505 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1506 : :
1507 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1508 : 0 : pg_fatal("failed to exit pipeline mode: %s", PQerrorMessage(conn));
1509 : 1 : fprintf(stderr, "ok - 2\n");
1510 : 1 : }
1511 : :
1512 : : static void
1513 : 1 : test_simple_pipeline(PGconn *conn)
1514 : : {
1515 : 1 : const char *dummy_params[1] = {"1"};
1516 : 1 : Oid dummy_param_oids[1] = {INT4OID};
1517 : :
1518 : 1 : fprintf(stderr, "simple pipeline... ");
1519 : :
1520 : : /*
1521 : : * Enter pipeline mode and dispatch a set of operations, which we'll then
1522 : : * process the results of as they come in.
1523 : : *
1524 : : * For a simple case we should be able to do this without interim
1525 : : * processing of results since our output buffer will give us enough slush
1526 : : * to work with and we won't block on sending. So blocking mode is fine.
1527 : : */
1528 [ - + ]: 1 : if (PQisnonblocking(conn))
1529 : 0 : pg_fatal("Expected blocking connection mode");
1530 : :
1531 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1532 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1533 : :
1534 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1",
1535 : : 1, dummy_param_oids, dummy_params,
1536 : : NULL, NULL, 0) != 1)
1537 : 0 : pg_fatal("dispatching SELECT failed: %s", PQerrorMessage(conn));
1538 : :
1539 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
1540 : 0 : pg_fatal("exiting pipeline mode with work in progress should fail, but succeeded");
1541 : :
1542 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1543 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1544 : :
1545 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1546 : :
1547 : 1 : consume_null_result(conn);
1548 : :
1549 : : /*
1550 : : * Even though we've processed the result there's still a sync to come and
1551 : : * we can't exit pipeline mode yet
1552 : : */
1553 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
1554 : 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
1555 : :
1556 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1557 : :
1558 : 1 : consume_null_result(conn);
1559 : :
1560 : : /* We're still in pipeline mode... */
1561 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1562 : 0 : pg_fatal("Fell out of pipeline mode somehow");
1563 : :
1564 : : /* ... until we end it, which we can safely do now */
1565 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1566 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
1567 : : PQerrorMessage(conn));
1568 : :
1569 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
1570 : 0 : pg_fatal("Exiting pipeline mode didn't seem to work");
1571 : :
1572 : 1 : fprintf(stderr, "ok\n");
1573 : 1 : }
1574 : :
1575 : : static void
1576 : 1 : test_singlerowmode(PGconn *conn)
1577 : : {
1578 : : PGresult *res;
1579 : : int i;
1580 : 1 : bool pipeline_ended = false;
1581 : :
1582 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1583 : 0 : pg_fatal("failed to enter pipeline mode: %s",
1584 : : PQerrorMessage(conn));
1585 : :
1586 : : /* One series of three commands, using single-row mode for the first two. */
1587 [ + + ]: 4 : for (i = 0; i < 3; i++)
1588 : : {
1589 : : char *param[1];
1590 : :
1591 : 3 : param[0] = psprintf("%d", 44 + i);
1592 : :
1593 [ - + ]: 3 : if (PQsendQueryParams(conn,
1594 : : "SELECT generate_series(42, $1)",
1595 : : 1,
1596 : : NULL,
1597 : : (const char *const *) param,
1598 : : NULL,
1599 : : NULL,
1600 : : 0) != 1)
1601 : 0 : pg_fatal("failed to send query: %s",
1602 : : PQerrorMessage(conn));
1603 : 3 : pfree(param[0]);
1604 : : }
1605 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1606 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1607 : :
1608 [ + + ]: 5 : for (i = 0; !pipeline_ended; i++)
1609 : : {
1610 : 4 : bool first = true;
1611 : : bool saw_ending_tuplesok;
1612 : 4 : bool isSingleTuple = false;
1613 : :
1614 : : /* Set single row mode for only first 2 SELECT queries */
1615 [ + + ]: 4 : if (i < 2)
1616 : : {
1617 [ - + ]: 2 : if (PQsetSingleRowMode(conn) != 1)
1618 : 0 : pg_fatal("PQsetSingleRowMode() failed for i=%d", i);
1619 : : }
1620 : :
1621 : : /* Consume rows for this query */
1622 : 4 : saw_ending_tuplesok = false;
1623 [ + + ]: 14 : while ((res = PQgetResult(conn)) != NULL)
1624 : : {
1625 : 11 : ExecStatusType est = PQresultStatus(res);
1626 : :
1627 [ + + ]: 11 : if (est == PGRES_PIPELINE_SYNC)
1628 : : {
1629 : 1 : fprintf(stderr, "end of pipeline reached\n");
1630 : 1 : pipeline_ended = true;
1631 : 1 : PQclear(res);
1632 [ - + ]: 1 : if (i != 3)
1633 : 0 : pg_fatal("Expected three results, got %d", i);
1634 : 1 : break;
1635 : : }
1636 : :
1637 : : /* Expect SINGLE_TUPLE for queries 0 and 1, TUPLES_OK for 2 */
1638 [ + + ]: 10 : if (first)
1639 : : {
1640 [ + + - + ]: 3 : if (i <= 1 && est != PGRES_SINGLE_TUPLE)
1641 : 0 : pg_fatal("Expected PGRES_SINGLE_TUPLE for query %d, got %s",
1642 : : i, PQresStatus(est));
1643 [ + + - + ]: 3 : if (i >= 2 && est != PGRES_TUPLES_OK)
1644 : 0 : pg_fatal("Expected PGRES_TUPLES_OK for query %d, got %s",
1645 : : i, PQresStatus(est));
1646 : 3 : first = false;
1647 : : }
1648 : :
1649 : 10 : fprintf(stderr, "Result status %s for query %d", PQresStatus(est), i);
1650 [ + + - ]: 10 : switch (est)
1651 : : {
1652 : 3 : case PGRES_TUPLES_OK:
1653 : 3 : fprintf(stderr, ", tuples: %d\n", PQntuples(res));
1654 : 3 : saw_ending_tuplesok = true;
1655 [ + + ]: 3 : if (isSingleTuple)
1656 : : {
1657 [ + - ]: 2 : if (PQntuples(res) == 0)
1658 : 2 : fprintf(stderr, "all tuples received in query %d\n", i);
1659 : : else
1660 : 0 : pg_fatal("Expected to follow PGRES_SINGLE_TUPLE, but received PGRES_TUPLES_OK directly instead");
1661 : : }
1662 : 3 : break;
1663 : :
1664 : 7 : case PGRES_SINGLE_TUPLE:
1665 : 7 : isSingleTuple = true;
1666 : 7 : fprintf(stderr, ", %d tuple: %s\n", PQntuples(res), PQgetvalue(res, 0, 0));
1667 : 7 : break;
1668 : :
1669 : 0 : default:
1670 : 0 : pg_fatal("unexpected");
1671 : : }
1672 : 10 : PQclear(res);
1673 : : }
1674 [ + + - + ]: 4 : if (!pipeline_ended && !saw_ending_tuplesok)
1675 : 0 : pg_fatal("didn't get expected terminating TUPLES_OK");
1676 : : }
1677 : :
1678 : : /*
1679 : : * Now issue one command, get its results in with single-row mode, then
1680 : : * issue another command, and get its results in normal mode; make sure
1681 : : * the single-row mode flag is reset as expected.
1682 : : */
1683 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT generate_series(0, 0)",
1684 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1685 : 0 : pg_fatal("failed to send query: %s",
1686 : : PQerrorMessage(conn));
1687 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1688 : 0 : pg_fatal("failed to send flush request");
1689 [ - + ]: 1 : if (PQsetSingleRowMode(conn) != 1)
1690 : 0 : pg_fatal("PQsetSingleRowMode() failed");
1691 : :
1692 : 1 : consume_result_status(conn, PGRES_SINGLE_TUPLE);
1693 : :
1694 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1695 : :
1696 : 1 : consume_null_result(conn);
1697 : :
1698 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1",
1699 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1700 : 0 : pg_fatal("failed to send query: %s",
1701 : : PQerrorMessage(conn));
1702 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1703 : 0 : pg_fatal("failed to send flush request");
1704 : :
1705 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1706 : :
1707 : 1 : consume_null_result(conn);
1708 : :
1709 : : /*
1710 : : * Try chunked mode as well; make sure that it correctly delivers a
1711 : : * partial final chunk.
1712 : : */
1713 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT generate_series(1, 5)",
1714 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1715 : 0 : pg_fatal("failed to send query: %s",
1716 : : PQerrorMessage(conn));
1717 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1718 : 0 : pg_fatal("failed to send flush request");
1719 [ - + ]: 1 : if (PQsetChunkedRowsMode(conn, 3) != 1)
1720 : 0 : pg_fatal("PQsetChunkedRowsMode() failed");
1721 : :
1722 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_CHUNK);
1723 [ - + ]: 1 : if (PQntuples(res) != 3)
1724 : 0 : pg_fatal("Expected 3 rows, got %d", PQntuples(res));
1725 : 1 : PQclear(res);
1726 : :
1727 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_CHUNK);
1728 [ - + ]: 1 : if (PQntuples(res) != 2)
1729 : 0 : pg_fatal("Expected 2 rows, got %d", PQntuples(res));
1730 : 1 : PQclear(res);
1731 : :
1732 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_OK);
1733 [ - + ]: 1 : if (PQntuples(res) != 0)
1734 : 0 : pg_fatal("Expected 0 rows, got %d", PQntuples(res));
1735 : 1 : PQclear(res);
1736 : :
1737 : 1 : consume_null_result(conn);
1738 : :
1739 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1740 : 0 : pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
1741 : :
1742 : 1 : fprintf(stderr, "ok\n");
1743 : 1 : }
1744 : :
1745 : : /*
1746 : : * Simple test to verify that a pipeline is discarded as a whole when there's
1747 : : * an error, ignoring transaction commands.
1748 : : */
1749 : : static void
1750 : 1 : test_transaction(PGconn *conn)
1751 : : {
1752 : : PGresult *res;
1753 : : bool expect_null;
1754 : 1 : int num_syncs = 0;
1755 : :
1756 : 1 : res = PQexec(conn, "DROP TABLE IF EXISTS pq_pipeline_tst;"
1757 : : "CREATE TABLE pq_pipeline_tst (id int)");
1758 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1759 : 0 : pg_fatal("failed to create test table: %s",
1760 : : PQerrorMessage(conn));
1761 : 1 : PQclear(res);
1762 : :
1763 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1764 : 0 : pg_fatal("failed to enter pipeline mode: %s",
1765 : : PQerrorMessage(conn));
1766 [ - + ]: 1 : if (PQsendPrepare(conn, "rollback", "ROLLBACK", 0, NULL) != 1)
1767 : 0 : pg_fatal("could not send prepare on pipeline: %s",
1768 : : PQerrorMessage(conn));
1769 : :
1770 [ - + ]: 1 : if (PQsendQueryParams(conn,
1771 : : "BEGIN",
1772 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1773 : 0 : pg_fatal("failed to send query: %s",
1774 : : PQerrorMessage(conn));
1775 [ - + ]: 1 : if (PQsendQueryParams(conn,
1776 : : "SELECT 0/0",
1777 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1778 : 0 : pg_fatal("failed to send query: %s",
1779 : : PQerrorMessage(conn));
1780 : :
1781 : : /*
1782 : : * send a ROLLBACK using a prepared stmt. Doesn't work because we need to
1783 : : * get out of the pipeline-aborted state first.
1784 : : */
1785 [ - + ]: 1 : if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
1786 : 0 : pg_fatal("failed to execute prepared: %s",
1787 : : PQerrorMessage(conn));
1788 : :
1789 : : /* This insert fails because we're in pipeline-aborted state */
1790 [ - + ]: 1 : if (PQsendQueryParams(conn,
1791 : : "INSERT INTO pq_pipeline_tst VALUES (1)",
1792 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1793 : 0 : pg_fatal("failed to send query: %s",
1794 : : PQerrorMessage(conn));
1795 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1796 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1797 : 1 : num_syncs++;
1798 : :
1799 : : /*
1800 : : * This insert fails even though the pipeline got a SYNC, because we're in
1801 : : * an aborted transaction
1802 : : */
1803 [ - + ]: 1 : if (PQsendQueryParams(conn,
1804 : : "INSERT INTO pq_pipeline_tst VALUES (2)",
1805 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1806 : 0 : pg_fatal("failed to send query: %s",
1807 : : PQerrorMessage(conn));
1808 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1809 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1810 : 1 : num_syncs++;
1811 : :
1812 : : /*
1813 : : * Send ROLLBACK using prepared stmt. This one works because we just did
1814 : : * PQpipelineSync above.
1815 : : */
1816 [ - + ]: 1 : if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
1817 : 0 : pg_fatal("failed to execute prepared: %s",
1818 : : PQerrorMessage(conn));
1819 : :
1820 : : /*
1821 : : * Now that we're out of a transaction and in pipeline-good mode, this
1822 : : * insert works
1823 : : */
1824 [ - + ]: 1 : if (PQsendQueryParams(conn,
1825 : : "INSERT INTO pq_pipeline_tst VALUES (3)",
1826 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1827 : 0 : pg_fatal("failed to send query: %s",
1828 : : PQerrorMessage(conn));
1829 : : /* Send two syncs now -- match up to SYNC messages below */
1830 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1831 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1832 : 1 : num_syncs++;
1833 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1834 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1835 : 1 : num_syncs++;
1836 : :
1837 : 1 : expect_null = false;
1838 : 1 : for (int i = 0;; i++)
1839 : 19 : {
1840 : : ExecStatusType restype;
1841 : :
1842 : 20 : res = PQgetResult(conn);
1843 [ + + ]: 20 : if (res == NULL)
1844 : : {
1845 : 8 : printf("%d: got NULL result\n", i);
1846 [ - + ]: 8 : if (!expect_null)
1847 : 0 : pg_fatal("did not expect NULL here");
1848 : 8 : expect_null = false;
1849 : 8 : continue;
1850 : : }
1851 : 12 : restype = PQresultStatus(res);
1852 : 12 : printf("%d: got status %s", i, PQresStatus(restype));
1853 [ - + ]: 12 : if (expect_null)
1854 : 0 : pg_fatal("expected NULL");
1855 [ + + ]: 12 : if (restype == PGRES_FATAL_ERROR)
1856 : 2 : printf("; error: %s", PQerrorMessage(conn));
1857 [ + + ]: 10 : else if (restype == PGRES_PIPELINE_ABORTED)
1858 : : {
1859 : 2 : printf(": command didn't run because pipeline aborted\n");
1860 : : }
1861 : : else
1862 : 8 : printf("\n");
1863 : 12 : PQclear(res);
1864 : :
1865 [ + + ]: 12 : if (restype == PGRES_PIPELINE_SYNC)
1866 : 4 : num_syncs--;
1867 : : else
1868 : 8 : expect_null = true;
1869 [ + + ]: 12 : if (num_syncs <= 0)
1870 : 1 : break;
1871 : : }
1872 : :
1873 : 1 : consume_null_result(conn);
1874 : :
1875 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1876 : 0 : pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
1877 : :
1878 : : /* We expect to find one tuple containing the value "3" */
1879 : 1 : res = PQexec(conn, "SELECT * FROM pq_pipeline_tst");
1880 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
1881 : 0 : pg_fatal("failed to obtain result: %s", PQerrorMessage(conn));
1882 [ - + ]: 1 : if (PQntuples(res) != 1)
1883 : 0 : pg_fatal("did not get 1 tuple");
1884 [ - + ]: 1 : if (strcmp(PQgetvalue(res, 0, 0), "3") != 0)
1885 : 0 : pg_fatal("did not get expected tuple");
1886 : 1 : PQclear(res);
1887 : :
1888 : 1 : fprintf(stderr, "ok\n");
1889 : 1 : }
1890 : :
1891 : : /*
1892 : : * In this test mode we send a stream of queries, with one in the middle
1893 : : * causing an error. Verify that we can still send some more after the
1894 : : * error and have libpq work properly.
1895 : : */
1896 : : static void
1897 : 1 : test_uniqviol(PGconn *conn)
1898 : : {
1899 : 1 : int sock = PQsocket(conn);
1900 : : PGresult *res;
1901 : 1 : Oid paramTypes[2] = {INT8OID, INT8OID};
1902 : : const char *paramValues[2];
1903 : : char paramValue0[MAXINT8LEN];
1904 : : char paramValue1[MAXINT8LEN];
1905 : 1 : int ctr = 0;
1906 : 1 : int numsent = 0;
1907 : 1 : int results = 0;
1908 : 1 : bool read_done = false;
1909 : 1 : bool write_done = false;
1910 : 1 : bool error_sent = false;
1911 : 1 : bool got_error = false;
1912 : 1 : int switched = 0;
1913 : 1 : int socketful = 0;
1914 : : fd_set in_fds;
1915 : : fd_set out_fds;
1916 : :
1917 : 1 : fprintf(stderr, "uniqviol ...");
1918 : :
1919 : 1 : PQsetnonblocking(conn, 1);
1920 : :
1921 : 1 : paramValues[0] = paramValue0;
1922 : 1 : paramValues[1] = paramValue1;
1923 : 1 : sprintf(paramValue1, "42");
1924 : :
1925 : 1 : res = PQexec(conn, "drop table if exists ppln_uniqviol;"
1926 : : "create table ppln_uniqviol(id bigint primary key, idata bigint)");
1927 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1928 : 0 : pg_fatal("failed to create table: %s", PQerrorMessage(conn));
1929 : 1 : PQclear(res);
1930 : :
1931 : 1 : res = PQexec(conn, "begin");
1932 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1933 : 0 : pg_fatal("failed to begin transaction: %s", PQerrorMessage(conn));
1934 : 1 : PQclear(res);
1935 : :
1936 : 1 : res = PQprepare(conn, "insertion",
1937 : : "insert into ppln_uniqviol values ($1, $2) returning id",
1938 : : 2, paramTypes);
1939 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1940 : 0 : pg_fatal("failed to prepare query: %s", PQerrorMessage(conn));
1941 : 1 : PQclear(res);
1942 : :
1943 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1944 : 0 : pg_fatal("failed to enter pipeline mode");
1945 : :
1946 [ + - ]: 6 : while (!read_done)
1947 : : {
1948 : : /*
1949 : : * Avoid deadlocks by reading everything the server has sent before
1950 : : * sending anything. (Special precaution is needed here to process
1951 : : * PQisBusy before testing the socket for read-readiness, because the
1952 : : * socket does not turn read-ready after "sending" queries in aborted
1953 : : * pipeline mode.)
1954 : : */
1955 [ + + ]: 637 : while (PQisBusy(conn) == 0)
1956 : : {
1957 : : bool new_error;
1958 : :
1959 [ + + ]: 633 : if (results >= numsent)
1960 : : {
1961 [ - + ]: 1 : if (write_done)
1962 : 0 : read_done = true;
1963 : 1 : break;
1964 : : }
1965 : :
1966 : 632 : res = PQgetResult(conn);
1967 : 632 : new_error = process_result(conn, res, results, numsent);
1968 [ + + - + ]: 632 : if (new_error && got_error)
1969 : 0 : pg_fatal("got two errors");
1970 : 632 : got_error |= new_error;
1971 [ + + ]: 632 : if (results++ >= numsent - 1)
1972 : : {
1973 [ + - ]: 1 : if (write_done)
1974 : 1 : read_done = true;
1975 : 1 : break;
1976 : : }
1977 : : }
1978 : :
1979 [ + + ]: 6 : if (read_done)
1980 : 1 : break;
1981 : :
1982 [ + + ]: 85 : FD_ZERO(&out_fds);
1983 : 5 : FD_SET(sock, &out_fds);
1984 : :
1985 [ + + ]: 85 : FD_ZERO(&in_fds);
1986 : 5 : FD_SET(sock, &in_fds);
1987 : :
1988 [ - + - + ]: 5 : if (select(sock + 1, &in_fds, write_done ? NULL : &out_fds, NULL, NULL) == -1)
1989 : : {
1990 [ # # ]: 0 : if (errno == EINTR)
1991 : 0 : continue;
1992 : 0 : pg_fatal("select() failed: %m");
1993 : : }
1994 : :
1995 [ + + - + ]: 5 : if (FD_ISSET(sock, &in_fds) && PQconsumeInput(conn) == 0)
1996 : 0 : pg_fatal("PQconsumeInput failed: %s", PQerrorMessage(conn));
1997 : :
1998 : : /*
1999 : : * If the socket is writable and we haven't finished sending queries,
2000 : : * send some.
2001 : : */
2002 [ + - + + ]: 5 : if (!write_done && FD_ISSET(sock, &out_fds))
2003 : : {
2004 : : for (;;)
2005 : 629 : {
2006 : : int flush;
2007 : :
2008 : : /*
2009 : : * provoke uniqueness violation exactly once after having
2010 : : * switched to read mode.
2011 : : */
2012 [ + + + + : 632 : if (switched >= 1 && !error_sent && ctr % socketful >= socketful / 2)
+ + ]
2013 : : {
2014 : 1 : sprintf(paramValue0, "%d", numsent / 2);
2015 : 1 : fprintf(stderr, "E");
2016 : 1 : error_sent = true;
2017 : : }
2018 : : else
2019 : : {
2020 : 631 : fprintf(stderr, ".");
2021 : 631 : sprintf(paramValue0, "%d", ctr++);
2022 : : }
2023 : :
2024 [ - + ]: 632 : if (PQsendQueryPrepared(conn, "insertion", 2, paramValues, NULL, NULL, 0) != 1)
2025 : 0 : pg_fatal("failed to execute prepared query: %s", PQerrorMessage(conn));
2026 : 632 : numsent++;
2027 : :
2028 : : /* Are we done writing? */
2029 [ + + + + : 632 : if (socketful != 0 && numsent % socketful == 42 && error_sent)
+ + ]
2030 : : {
2031 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
2032 : 0 : pg_fatal("failed to send flush request");
2033 : 1 : write_done = true;
2034 : 1 : fprintf(stderr, "\ndone writing\n");
2035 : 1 : PQflush(conn);
2036 : 1 : break;
2037 : : }
2038 : :
2039 : : /* is the outgoing socket full? */
2040 : 631 : flush = PQflush(conn);
2041 [ - + ]: 631 : if (flush == -1)
2042 : 0 : pg_fatal("failed to flush: %s", PQerrorMessage(conn));
2043 [ + + ]: 631 : if (flush == 1)
2044 : : {
2045 [ + + ]: 2 : if (socketful == 0)
2046 : 1 : socketful = numsent;
2047 : 2 : fprintf(stderr, "\nswitch to reading\n");
2048 : 2 : switched++;
2049 : 2 : break;
2050 : : }
2051 : : }
2052 : : }
2053 : : }
2054 : :
2055 [ - + ]: 1 : if (!got_error)
2056 : 0 : pg_fatal("did not get expected error");
2057 : :
2058 : 1 : fprintf(stderr, "ok\n");
2059 : 1 : }
2060 : :
2061 : : /*
2062 : : * Subroutine for test_uniqviol; given a PGresult, print it out and consume
2063 : : * the expected NULL that should follow it.
2064 : : *
2065 : : * Returns true if we read a fatal error message, otherwise false.
2066 : : */
2067 : : static bool
2068 : 632 : process_result(PGconn *conn, PGresult *res, int results, int numsent)
2069 : : {
2070 : 632 : bool got_error = false;
2071 : :
2072 [ - + ]: 632 : if (res == NULL)
2073 : 0 : pg_fatal("got unexpected NULL");
2074 : :
2075 [ + + + - ]: 632 : switch (PQresultStatus(res))
2076 : : {
2077 : 1 : case PGRES_FATAL_ERROR:
2078 : 1 : got_error = true;
2079 : 1 : fprintf(stderr, "result %d/%d (error): %s\n", results, numsent, PQerrorMessage(conn));
2080 : 1 : PQclear(res);
2081 : 1 : consume_null_result(conn);
2082 : 1 : break;
2083 : :
2084 : 442 : case PGRES_TUPLES_OK:
2085 : 442 : fprintf(stderr, "result %d/%d: %s\n", results, numsent, PQgetvalue(res, 0, 0));
2086 : 442 : PQclear(res);
2087 : 442 : consume_null_result(conn);
2088 : 442 : break;
2089 : :
2090 : 189 : case PGRES_PIPELINE_ABORTED:
2091 : 189 : fprintf(stderr, "result %d/%d: pipeline aborted\n", results, numsent);
2092 : 189 : PQclear(res);
2093 : 189 : consume_null_result(conn);
2094 : 189 : break;
2095 : :
2096 : 0 : default:
2097 : 0 : pg_fatal("got unexpected %s", PQresStatus(PQresultStatus(res)));
2098 : : }
2099 : :
2100 : 632 : return got_error;
2101 : : }
2102 : :
2103 : :
2104 : : static void
2105 : 0 : usage(const char *progname)
2106 : : {
2107 : 0 : fprintf(stderr, "%s tests libpq's pipeline mode.\n\n", progname);
2108 : 0 : fprintf(stderr, "Usage:\n");
2109 : 0 : fprintf(stderr, " %s [OPTION] tests\n", progname);
2110 : 0 : fprintf(stderr, " %s [OPTION] TESTNAME [CONNINFO]\n", progname);
2111 : 0 : fprintf(stderr, "\nOptions:\n");
2112 : 0 : fprintf(stderr, " -t TRACEFILE generate a libpq trace to TRACEFILE\n");
2113 : 0 : fprintf(stderr, " -r NUMROWS use NUMROWS as the test size\n");
2114 : 0 : }
2115 : :
2116 : : static void
2117 : 1 : print_test_list(void)
2118 : : {
2119 : 1 : printf("cancel\n");
2120 : 1 : printf("disallowed_in_pipeline\n");
2121 : 1 : printf("multi_pipelines\n");
2122 : 1 : printf("nosync\n");
2123 : 1 : printf("pipeline_abort\n");
2124 : 1 : printf("pipeline_idle\n");
2125 : 1 : printf("pipelined_insert\n");
2126 : 1 : printf("prepared\n");
2127 : 1 : printf("protocol_version\n");
2128 : 1 : printf("simple_pipeline\n");
2129 : 1 : printf("singlerow\n");
2130 : 1 : printf("transaction\n");
2131 : 1 : printf("uniqviol\n");
2132 : 1 : }
2133 : :
2134 : : int
2135 : 15 : main(int argc, char **argv)
2136 : : {
2137 : 15 : const char *conninfo = "";
2138 : : PGconn *conn;
2139 : 15 : FILE *trace = NULL;
2140 : : char *testname;
2141 : 15 : int numrows = 10000;
2142 : : PGresult *res;
2143 : : int c;
2144 : :
2145 [ + + ]: 52 : while ((c = getopt(argc, argv, "r:t:")) != -1)
2146 : : {
2147 [ + + - ]: 22 : switch (c)
2148 : : {
2149 : 13 : case 'r': /* numrows */
2150 : 13 : errno = 0;
2151 : 13 : numrows = strtol(optarg, NULL, 10);
2152 [ + - - + ]: 13 : if (errno != 0 || numrows <= 0)
2153 : : {
2154 : 0 : fprintf(stderr, "couldn't parse \"%s\" as a positive integer\n",
2155 : : optarg);
2156 : 0 : exit(1);
2157 : : }
2158 : 13 : break;
2159 : 9 : case 't': /* trace file */
2160 : 9 : tracefile = pg_strdup(optarg);
2161 : 9 : break;
2162 : : }
2163 : : }
2164 : :
2165 [ + - ]: 15 : if (optind < argc)
2166 : : {
2167 : 15 : testname = pg_strdup(argv[optind]);
2168 : 15 : optind++;
2169 : : }
2170 : : else
2171 : : {
2172 : 0 : usage(argv[0]);
2173 : 0 : exit(1);
2174 : : }
2175 : :
2176 [ + + ]: 15 : if (strcmp(testname, "tests") == 0)
2177 : : {
2178 : 1 : print_test_list();
2179 : 1 : exit(0);
2180 : : }
2181 : :
2182 [ + - ]: 14 : if (optind < argc)
2183 : : {
2184 : 14 : conninfo = pg_strdup(argv[optind]);
2185 : 14 : optind++;
2186 : : }
2187 : :
2188 : : /* Make a connection to the database */
2189 : 14 : conn = PQconnectdb(conninfo);
2190 [ - + ]: 14 : if (PQstatus(conn) != CONNECTION_OK)
2191 : : {
2192 : 0 : fprintf(stderr, "Connection to database failed: %s\n",
2193 : : PQerrorMessage(conn));
2194 : 0 : exit_nicely(conn);
2195 : : }
2196 : :
2197 : 14 : res = PQexec(conn, "SET lc_messages TO \"C\"");
2198 [ - + ]: 14 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
2199 : 0 : pg_fatal("failed to set \"lc_messages\": %s", PQerrorMessage(conn));
2200 : 14 : PQclear(res);
2201 : 14 : res = PQexec(conn, "SET debug_parallel_query = off");
2202 [ - + ]: 14 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
2203 : 0 : pg_fatal("failed to set \"debug_parallel_query\": %s", PQerrorMessage(conn));
2204 : 14 : PQclear(res);
2205 : :
2206 : : /* Set the trace file, if requested */
2207 [ + + ]: 14 : if (tracefile != NULL)
2208 : : {
2209 [ - + ]: 9 : if (strcmp(tracefile, "-") == 0)
2210 : 0 : trace = stdout;
2211 : : else
2212 : 9 : trace = fopen(tracefile, "w");
2213 [ - + ]: 9 : if (trace == NULL)
2214 : 0 : pg_fatal("could not open file \"%s\": %m", tracefile);
2215 : :
2216 : : /* Make it line-buffered */
2217 : 9 : setvbuf(trace, NULL, PG_IOLBF, 0);
2218 : :
2219 : 9 : PQtrace(conn, trace);
2220 : 9 : PQsetTraceFlags(conn,
2221 : : PQTRACE_SUPPRESS_TIMESTAMPS | PQTRACE_REGRESS_MODE);
2222 : : }
2223 : :
2224 [ + + ]: 14 : if (strcmp(testname, "cancel") == 0)
2225 : 2 : test_cancel(conn);
2226 [ + + ]: 12 : else if (strcmp(testname, "disallowed_in_pipeline") == 0)
2227 : 1 : test_disallowed_in_pipeline(conn);
2228 [ + + ]: 11 : else if (strcmp(testname, "multi_pipelines") == 0)
2229 : 1 : test_multi_pipelines(conn);
2230 [ + + ]: 10 : else if (strcmp(testname, "nosync") == 0)
2231 : 1 : test_nosync(conn);
2232 [ + + ]: 9 : else if (strcmp(testname, "pipeline_abort") == 0)
2233 : 1 : test_pipeline_abort(conn);
2234 [ + + ]: 8 : else if (strcmp(testname, "pipeline_idle") == 0)
2235 : 1 : test_pipeline_idle(conn);
2236 [ + + ]: 7 : else if (strcmp(testname, "pipelined_insert") == 0)
2237 : 1 : test_pipelined_insert(conn, numrows);
2238 [ + + ]: 6 : else if (strcmp(testname, "prepared") == 0)
2239 : 1 : test_prepared(conn);
2240 [ + + ]: 5 : else if (strcmp(testname, "protocol_version") == 0)
2241 : 1 : test_protocol_version(conn);
2242 [ + + ]: 4 : else if (strcmp(testname, "simple_pipeline") == 0)
2243 : 1 : test_simple_pipeline(conn);
2244 [ + + ]: 3 : else if (strcmp(testname, "singlerow") == 0)
2245 : 1 : test_singlerowmode(conn);
2246 [ + + ]: 2 : else if (strcmp(testname, "transaction") == 0)
2247 : 1 : test_transaction(conn);
2248 [ + - ]: 1 : else if (strcmp(testname, "uniqviol") == 0)
2249 : 1 : test_uniqviol(conn);
2250 : : else
2251 : : {
2252 : 0 : fprintf(stderr, "\"%s\" is not a recognized test name\n", testname);
2253 : 0 : exit(1);
2254 : : }
2255 : :
2256 : : /* close the connection to the database and cleanup */
2257 : 14 : PQfinish(conn);
2258 : :
2259 [ + + + - ]: 14 : if (trace && trace != stdout)
2260 : 9 : fclose(trace);
2261 : :
2262 : 14 : return 0;
2263 : : }
|
