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 631 : consume_null_result_impl(int line, PGconn *conn)
132 : {
133 : PGresult *res;
134 :
135 631 : res = PQgetResult(conn);
136 631 : if (res != NULL)
137 0 : pg_fatal_impl(line, "expected NULL PGresult, got %s: %s",
138 : PQresStatus(PQresultStatus(res)),
139 : PQerrorMessage(conn));
140 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);
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 1 : {
181 : PGresult *res;
182 : char *value;
183 :
184 25 : 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 13 : 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 25 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
196 0 : pg_fatal_impl(line, "could not query pg_stat_activity: %s", PQerrorMessage(monitorConn));
197 25 : if (PQntuples(res) != 1)
198 0 : pg_fatal_impl(line, "unexpected number of rows received: %d", PQntuples(res));
199 25 : if (PQnfields(res) != 1)
200 0 : pg_fatal_impl(line, "unexpected number of columns received: %d", PQnfields(res));
201 25 : value = PQgetvalue(res, 0, 0);
202 25 : if (strcmp(value, "0") != 0)
203 : {
204 24 : PQclear(res);
205 24 : break;
206 : }
207 1 : PQclear(res);
208 :
209 : /* wait 10ms before polling again */
210 1 : 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 104 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
260 102 : 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 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)
281 0 : pg_fatal("Connection to database failed: %s",
282 : PQerrorMessage(copyConn));
283 :
284 2 : pfree(keywords);
285 2 : pfree(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 5730 : while (recv_step != BI_DONE)
1009 : {
1010 : int sock;
1011 : fd_set input_mask;
1012 : fd_set output_mask;
1013 :
1014 5729 : sock = PQsocket(conn);
1015 :
1016 5729 : if (sock < 0)
1017 0 : break; /* shouldn't happen */
1018 :
1019 97393 : FD_ZERO(&input_mask);
1020 5729 : FD_SET(sock, &input_mask);
1021 97393 : FD_ZERO(&output_mask);
1022 5729 : FD_SET(sock, &output_mask);
1023 :
1024 5729 : 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 5729 : if (FD_ISSET(sock, &input_mask))
1035 : {
1036 2 : PQconsumeInput(conn);
1037 :
1038 : /* Read until we'd block if we tried to read */
1039 1413 : 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 5729 : if (FD_ISSET(sock, &output_mask))
1113 : {
1114 5729 : PQflush(conn);
1115 :
1116 5729 : 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 5029 : 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 5028 : 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 52 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1337 51 : 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 52 : 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 51 : if (strcmp(opt->keyword, "max_protocol_version") == 0)
1353 1 : max_protocol_version_index = i;
1354 50 : else if (!opt->val)
1355 31 : 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 : pfree(keywords);
1442 1 : pfree(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 5 : 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 604 : while (PQisBusy(conn) == 0)
1956 : {
1957 : bool new_error;
1958 :
1959 601 : if (results >= numsent)
1960 : {
1961 1 : if (write_done)
1962 0 : read_done = true;
1963 1 : break;
1964 : }
1965 :
1966 600 : res = PQgetResult(conn);
1967 600 : new_error = process_result(conn, res, results, numsent);
1968 600 : if (new_error && got_error)
1969 0 : pg_fatal("got two errors");
1970 600 : got_error |= new_error;
1971 600 : if (results++ >= numsent - 1)
1972 : {
1973 1 : if (write_done)
1974 1 : read_done = true;
1975 1 : break;
1976 : }
1977 : }
1978 :
1979 5 : if (read_done)
1980 1 : break;
1981 :
1982 68 : FD_ZERO(&out_fds);
1983 4 : FD_SET(sock, &out_fds);
1984 :
1985 68 : FD_ZERO(&in_fds);
1986 4 : FD_SET(sock, &in_fds);
1987 :
1988 4 : 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 4 : 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 4 : if (!write_done && FD_ISSET(sock, &out_fds))
2003 : {
2004 : for (;;)
2005 597 : {
2006 : int flush;
2007 :
2008 : /*
2009 : * provoke uniqueness violation exactly once after having
2010 : * switched to read mode.
2011 : */
2012 600 : 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 599 : fprintf(stderr, ".");
2021 599 : sprintf(paramValue0, "%d", ctr++);
2022 : }
2023 :
2024 600 : if (PQsendQueryPrepared(conn, "insertion", 2, paramValues, NULL, NULL, 0) != 1)
2025 0 : pg_fatal("failed to execute prepared query: %s", PQerrorMessage(conn));
2026 600 : numsent++;
2027 :
2028 : /* Are we done writing? */
2029 600 : 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 599 : flush = PQflush(conn);
2041 599 : if (flush == -1)
2042 0 : pg_fatal("failed to flush: %s", PQerrorMessage(conn));
2043 599 : 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 600 : process_result(PGconn *conn, PGresult *res, int results, int numsent)
2069 : {
2070 600 : bool got_error = false;
2071 :
2072 600 : if (res == NULL)
2073 0 : pg_fatal("got unexpected NULL");
2074 :
2075 600 : 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 418 : case PGRES_TUPLES_OK:
2085 418 : fprintf(stderr, "result %d/%d: %s\n", results, numsent, PQgetvalue(res, 0, 0));
2086 418 : PQclear(res);
2087 418 : consume_null_result(conn);
2088 418 : break;
2089 :
2090 181 : case PGRES_PIPELINE_ABORTED:
2091 181 : fprintf(stderr, "result %d/%d: pipeline aborted\n", results, numsent);
2092 181 : PQclear(res);
2093 181 : consume_null_result(conn);
2094 181 : break;
2095 :
2096 0 : default:
2097 0 : pg_fatal("got unexpected %s", PQresStatus(PQresultStatus(res)));
2098 : }
2099 :
2100 600 : 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 : }
|
