diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index ea7e7da..ac5d4fa 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -4637,6 +4637,484 @@ int PQflush(PGconn *conn); + + Batch mode and query pipelining + + + libpq + batch mode + + + + libpq + pipelining + + + + libpq supports queueing up multiple queries into + a pipeline to be executed as a batch on the server. Batching queries allows + applications to avoid a client/server round-trip after each query to get + the results before issuing the next query. + + + + An example of batch use may be found in the source distribution in + src/test/modules/test_libpq/testlibpqbatch.c. + + + + When to use batching + + + Much like asynchronous query mode, there is no performance disadvantage to + using batching and pipelining. It somewhat increased client application + complexity and extra caution is required to prevent client/server network + deadlocks, but can offer considerable performance improvements. + + + + Batching is most useful when the server is distant, i.e. network latency + ("ping time") is high, and when many small operations are being performed in + rapid sequence. There is usually less benefit in using batches when each + query takes many multiples of the client/server round-trip time to execute. + A 100-statement operation run on a server 300ms round-trip-time away would take + 30 seconds in network latency alone without batching; with batching it may spend + as little as 0.3s waiting for results from the server. + + + + Use batches when your application does lots of small + INSERT, UPDATE and + DELETE operations that can't easily be transformed into + operations on sets or into a + COPY operation. + + + + Batching is less useful when information from one operation is required by the + client before it knows enough to send the next operation. The client must + introduce a synchronisation point and wait for a full client/server + round-trip to get the results it needs. However, it's often possible to + adjust the client design to exchange the required information server-side. + Read-modify-write cycles are especially good candidates; for example: + + BEGIN; + SELECT x FROM mytable WHERE id = 42 FOR UPDATE; + -- result: x=2 + -- client adds 1 to x: + UPDATE mytable SET x = 3 WHERE id = 42; + COMMIT; + + could be much more efficiently done with: + + UPDATE mytable SET x = x + 1 WHERE id = 42; + + + + + + The batch API was introduced in PostgreSQL 9.6, but clients using it can + use batches on server versions 8.4 and newer. Batching works on any server + that supports the v3 extended query protocol. + + + + + + + Using batch mode + + + To issue batches the application must switch + libpq into batch mode. Enter batch mode with PQbeginBatchMode(conn) or test + whether batch mode is active with PQisInBatchMode(conn). In batch mode only asynchronous operations are permitted, and + COPY is not allowed. (The restriction on COPY is an implementation + limit; the PostgreSQL protocol and server can support batched COPY). + + + + The client uses libpq's asynchronous query functions to dispatch work, + marking the end of each batch with PQsendEndBatch. + Concurrently, it uses PQgetResult and + PQgetNextQuery to get results. It may eventually exit + batch mode with PQendBatchMode once all results are + processed. + + + + + It is best to use batch mode with libpq in + non-blocking mode. If used in + blocking mode it is possible for a client/server deadlock to occur. The + client will block trying to send queries to the server, but the server will + block trying to send results from queries it's already processed to the + client. This only occurs when the client sends enough queries to fill its + output buffer and the server's receive buffer before switching to + processing input from the server, but it's hard to predict exactly when + that'll happen so it's best to always use non-blocking mode. + + + + + Issuing queries + + + After entering batch mode the application dispatches requests + using normal asynchronous libpq functions like + PQsendQueryParams, PQsendPrepare, + etc. The asynchronous requests are followed by a PQsendEndBatch(conn) call to mark + the end of the batch. The client does not need to call + PQgetResult immediately after dispatching each + operation. Result processing + is handled separately. + + + + Batched operations will be executed by the server in the order the client + sends them. The server will send the results in the order the statements + executed. The server usually begins executing the batch before all commands + in the batch are queued and the end of batch command is sent. If any + statement encounters an error the server aborts the current transaction and + skips processing the rest of the batch. Query processing resumes after the + end of the failed batch. + + + + It's fine for one operation to depend on the results of a + prior one. One query may define a table that the next query in the same + batch uses; similarly, an application may create a named prepared statement + then execute it with later statements in the same batch. + + + + + + Processing results + + + The client interleaves result + processing with sending batch queries, or for small batches may + process all results after sending the whole batch. + + + + To get the result of the first batch entry the client must call PQgetNextQuery. It must then call + PQgetResult and handle the results until + PQgetResult returns null (or would return null if + called). The result from the next batch entry may then be retrieved using + PQgetNextQuery and the cycle repeated. The + application handles individual statement results as normal. + + + + PQgetResult behaves the same as for normal asynchronous + processing except that it may contain the new PGresult types + PGRES_BATCH_END and PGRES_BATCH_ABORTED. + PGRES_BATCH_END is reported exactly once for each + PQsendEndBatch call at the corresponding point in + the result stream and at no other time. PGRES_BATCH_ABORTED + is emitted during error handling; see + error handling. + + + + PQisBusy, PQconsumeInput, etc + operate as normal when processing batch results. + + + + libpq does not provide any information to the + application about the query currently being processed. The application + must keep track of the order in which it sent queries and the expected + results. Applications will typically use a state machine or a FIFO queue + for this. + + + + + + Error handling + + + When a query in a batch causes an ERROR the server + skips processing all subsequent messages until the end-of-batch message. + The open transaction is aborted. + + + + From the client perspective, after the client gets a + PGRES_FATAL_ERROR return from + PQresultStatus the batch is flagged as aborted. + libpq will report + PGRES_BATCH_ABORTED result for each remaining queued + operation in an aborted batch. The result for + PQsendEndBatch is reported as + PGRES_BATCH_END to signal the end of the aborted batch + and resumption of normal result processing. + + + + The client must process results with + PQgetNextQuery(...) and + PQgetResult during error recovery. + + + + If the batch used an implicit transaction then operations that have + already executed are rolled back and operations that were queued for after + the failed operation are skipped entirely. The same behaviour holds if the + batch starts and commits a single explicit transaction (i.e. the first + statement is BEGIN and the last is + COMMIT) except that the session remains in an aborted + transaction state at the end of the batch. If a batch contains + multiple explicit transactions, all transactions that committed + prior to the error remain committed, the currently in-progress transaction + is aborted and all subsequent operations in the current and all later + transactions in the same batch are skipped completely. + + + + The client must not assume that work is committed when it + sends a COMMIT, only when the + corresponding result is received to confirm the commit is complete. + Because errors arrive asynchronously the application needs to be able to + restart from the last received committed change and + resend work done after that point if something goes wrong. + + + + + + Interleaving result processing and query dispatch + + + To avoid deadlocks on large batches the client should be structured around + a nonblocking I/O loop using a function like select, + poll, epoll, + WaitForMultipleObjectEx, etc. + + + + The client application should generally maintain a queue of work still to + be dispatched and a queue of work that has been dispatched but not yet had + its results processed. When the socket is writable it should dispatch more + work. When the socket is readable it should read results and process them, + matching them up to the next entry in its expected results queue. Batches + should be scoped to logical units of work, usually (but not always) one + transaction per batch. There's no need to exit batch mode and re-enter it + between batches or to wait for one batch to finish before sending the next. + + + + An example using select() and a simple state machine + to track sent and received work is in + src/test/modules/test_libpq/testlibpqbatch.c in the PostgreSQL + source distribution. + + + + + + Ending batch mode + + + Once all dispatched commands have had their results processed and the end batch + result has been consumed the application may return to non-batched mode with + PQendBatchMode(conn). + + + + + + + Functions associated with batch mode + + + + + + PQisInBatchMode + + PQisInBatchMode + + + + + + Returns 1 if a libpq connection is in batch mode, otherwise 0. + + +int PQisInBatchMode(PGconn *conn); + + + + + + + + + PQbeginBatchMode + + PQbeginBatchMode + + + + + + Causes a connection to enter batch mode if it is currently idle or + already in batch mode and returns 1 for success. Returns 0 and has no + effect if the connection is not currently idle, i.e. it has a result + ready, is waiting for more input from the server, etc. This function + does not actually send anything to the server, it just changes the + libpq connection state. + + +int PQbeginBatchMode(PGconn *conn); + + + + + + + + + PQendBatchMode + + PQendBatchMode + + + + + + Causes a connection to exit batch mode if it is currently in batch mode + with an empty queue and no pending results and returns 1 for success. + Returns 1 and takes no action if not in batch mode. If the connection has + pending batch items in the queue for reading with + PQgetNextQuery, the current statement isn't finished + processing or there are results pending for collection with + PQgetResult, returns 0 and does nothing. + + +int PQendBatchMode(PGconn *conn); + + + + + + + + + PQsendEndBatch + + PQsendEndBatch + + + + + + Delimits the end of a set of a batched commands by sending a sync message and flushing + the send buffer. The end of a batch serves as + the delimiter of an implicit transaction and + an error recovery point; see + error handling. + + +int PQsendEndBatch(PGconn *conn); + + + + + + + + + PQgetNextQuery + + PQgetNextQuery + + + + + + Causes the connection to start processing the next queued query's + results. Returns 1 if a new query was popped from the result queue + for processing. Returns 0 and has no effect if there are no query results + pending, batch mode is not enabled, or if the query currently processed + is incomplete or still has pending results. See processing results. + + +int PQgetNextQuery(PGconn *conn); + + + + + + + + + PQqueriesInBatch + + PQqueriesInBatch + + + + + + Returns the number of queries still in the queue for this batch, not + including any query that's currently having results being processsed. + This is the number of times PQgetNextQuery has to be + called before the query queue is empty again. + + +int PQqueriesInBatch(PGconn *conn); + + + + + + + + + PQbatchIsAborted + + PQbatchIsAborted + + + + + + Returns 1 if the batch curently being received on a + libpq connection in batch mode is + aborted, 0 + otherwise. The aborted flag is cleared as soon as the result of the + PQsendEndBatch at the end of the aborted batch is + processed. Clients don't usually need this function as they can tell + that the batch is aborted from PGRES_BATCH_ABORTED + result codes. + + +int PQbatchIsAborted(PGconn *conn); + + + + + + + + + + + + Retrieving Query Results Row-By-Row diff --git a/src/interfaces/libpq/.gitignore b/src/interfaces/libpq/.gitignore index cb96af7..4c0d934 100644 --- a/src/interfaces/libpq/.gitignore +++ b/src/interfaces/libpq/.gitignore @@ -22,3 +22,4 @@ /encnames.c /wchar.c /libpq.rc +/tmp_check diff --git a/src/interfaces/libpq/Makefile b/src/interfaces/libpq/Makefile index 4b1e552..8d5cf21 100644 --- a/src/interfaces/libpq/Makefile +++ b/src/interfaces/libpq/Makefile @@ -129,6 +129,11 @@ install: all installdirs install-lib installcheck: $(MAKE) -C test $@ +check: prove-check + +prove-check: + $(prove_check) + installdirs: installdirs-lib $(MKDIR_P) '$(DESTDIR)$(includedir)' '$(DESTDIR)$(includedir_internal)' '$(DESTDIR)$(datadir)' diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt index 21dd772..1b0b8c5 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -171,3 +171,10 @@ PQsslAttributeNames 168 PQsslAttribute 169 PQsetErrorContextVisibility 170 PQresultVerboseErrorMessage 171 +PQisInBatchMode 172 +PQqueriesInBatch 173 +PQbeginBatchMode 174 +PQendBatchMode 175 +PQsendEndBatch 176 +PQgetNextQuery 177 +PQbatchIsAborted 178 diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index b83af64..71418dd 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -3396,6 +3396,7 @@ static void closePGconn(PGconn *conn) { PGnotify *notify; + PGcommandQueueEntry *queue; pgParameterStatus *pstatus; sendTerminateConn(conn); @@ -3428,6 +3429,22 @@ closePGconn(PGconn *conn) free(prev); } conn->notifyHead = conn->notifyTail = NULL; + queue = conn->cmd_queue_head; + while (queue != NULL) + { + PGcommandQueueEntry *prev = queue; + queue = queue->next; + free(prev); + } + conn->cmd_queue_head = conn->cmd_queue_tail = NULL; + queue = conn->cmd_queue_recycle; + while (queue != NULL) + { + PGcommandQueueEntry *prev = queue; + queue = queue->next; + free(prev); + } + conn->cmd_queue_recycle = NULL; pstatus = conn->pstatus; while (pstatus != NULL) { diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index b551875..df48767 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -39,7 +39,9 @@ char *const pgresStatus[] = { "PGRES_NONFATAL_ERROR", "PGRES_FATAL_ERROR", "PGRES_COPY_BOTH", - "PGRES_SINGLE_TUPLE" + "PGRES_SINGLE_TUPLE", + "PGRES_BATCH_END", + "PGRES_BATCH_ABORTED" }; /* @@ -69,6 +71,9 @@ static PGresult *PQexecFinish(PGconn *conn); static int PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target); static int check_field_number(const PGresult *res, int field_num); +static PGcommandQueueEntry* PQmakePipelinedCommand(PGconn *conn); +static void PQappendPipelinedCommand(PGconn *conn, PGcommandQueueEntry *entry); +static void PQrecyclePipelinedCommand(PGconn *conn, PGcommandQueueEntry *entry); /* ---------------- @@ -1108,7 +1113,7 @@ pqRowProcessor(PGconn *conn, const char **errmsgp) conn->next_result = conn->result; conn->result = res; /* And mark the result ready to return */ - conn->asyncStatus = PGASYNC_READY; + conn->asyncStatus = PGASYNC_READY_MORE; } return 1; @@ -1131,6 +1136,13 @@ fail: int PQsendQuery(PGconn *conn, const char *query) { + if (conn->in_batch) + { + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("cannot PQsendQuery in batch mode, use PQsendQueryParams\n")); + return false; + } + if (!PQsendQueryStart(conn)) return 0; @@ -1229,9 +1241,29 @@ PQsendPrepare(PGconn *conn, const char *stmtName, const char *query, int nParams, const Oid *paramTypes) { + PGcommandQueueEntry *pipeCmd = NULL; + char **last_query; + PGQueryClass *queryclass; + if (!PQsendQueryStart(conn)) return 0; + if (conn->in_batch) + { + pipeCmd = PQmakePipelinedCommand(conn); + + if (pipeCmd == NULL) + return 0; /* error msg already set */ + + last_query = &pipeCmd->query; + queryclass = &pipeCmd->queryclass; + } + else + { + last_query = &conn->last_query; + queryclass = &conn->queryclass; + } + /* check the arguments */ if (!stmtName) { @@ -1287,18 +1319,21 @@ PQsendPrepare(PGconn *conn, goto sendFailed; /* construct the Sync message */ - if (pqPutMsgStart('S', false, conn) < 0 || - pqPutMsgEnd(conn) < 0) - goto sendFailed; + if (!conn->in_batch) + { + if (pqPutMsgStart('S', false, conn) < 0 || + pqPutMsgEnd(conn) < 0) + goto sendFailed; + } /* remember we are doing just a Parse */ - conn->queryclass = PGQUERY_PREPARE; + *queryclass = PGQUERY_PREPARE; /* and remember the query text too, if possible */ /* if insufficient memory, last_query just winds up NULL */ - if (conn->last_query) - free(conn->last_query); - conn->last_query = strdup(query); + if (*last_query) + free(*last_query); + *last_query = strdup(query); /* * Give the data a push. In nonblock mode, don't complain if we're unable @@ -1308,10 +1343,14 @@ PQsendPrepare(PGconn *conn, goto sendFailed; /* OK, it's launched! */ - conn->asyncStatus = PGASYNC_BUSY; + if (conn->in_batch) + PQappendPipelinedCommand(conn, pipeCmd); + else + conn->asyncStatus = PGASYNC_BUSY; return 1; sendFailed: + PQrecyclePipelinedCommand(conn, pipeCmd); pqHandleSendFailure(conn); return 0; } @@ -1358,6 +1397,81 @@ PQsendQueryPrepared(PGconn *conn, resultFormat); } +/* Get a new command queue entry, allocating it if required. Doesn't add it to + * the tail of the queue yet, use PQappendPipelinedCommand once the command has + * been written for that. If a command fails once it's called this, it should + * use PQrecyclePipelinedCommand to put it on the freelist or release it. + * + * If allocation fails sets the error message and returns null. + */ +static PGcommandQueueEntry* +PQmakePipelinedCommand(PGconn *conn) +{ + PGcommandQueueEntry *entry; + + if (conn->cmd_queue_recycle == NULL) + { + entry = (PGcommandQueueEntry*) malloc(sizeof(PGcommandQueueEntry)); + if (entry == NULL) + { + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("out of memory\n")); + return NULL; + } + } + else + { + entry = conn->cmd_queue_recycle; + conn->cmd_queue_recycle = entry->next; + } + entry->next = NULL; + entry->query = NULL; + + return entry; +} + +/* Append a precreated command queue entry to the queue after it's been + * sent successfully. + */ +static void +PQappendPipelinedCommand(PGconn *conn, PGcommandQueueEntry *entry) +{ + if (conn->cmd_queue_head == NULL) + conn->cmd_queue_head = entry; + else + conn->cmd_queue_tail->next = entry; + conn->cmd_queue_tail = entry; +} + +/* Push a command queue entry onto the freelist. It must be a dangling entry + * with null next pointer and not referenced by any other entry's next pointer. + */ +static void +PQrecyclePipelinedCommand(PGconn *conn, PGcommandQueueEntry *entry) +{ + if (entry == NULL) + return; + if (entry->next != NULL) + { + fprintf(stderr, "tried to recycle non-dangling command queue entry"); + abort(); + } + entry->next = conn->cmd_queue_recycle; + conn->cmd_queue_recycle = entry; +} + +/* Set up for processing a new query's results */ +static void +PQstartProcessingNewQuery(PGconn *conn) +{ + /* initialize async result-accumulation state */ + conn->result = NULL; + conn->next_result = NULL; + + /* reset single-row processing mode */ + conn->singleRowMode = false; +} + /* * Common startup code for PQsendQuery and sibling routines */ @@ -1377,19 +1491,52 @@ PQsendQueryStart(PGconn *conn) libpq_gettext("no connection to the server\n")); return false; } - /* Can't send while already busy, either. */ - if (conn->asyncStatus != PGASYNC_IDLE) + + /* Can't send while already busy, either, unless enqueuing for later */ + if (conn->asyncStatus != PGASYNC_IDLE && !conn->in_batch) { printfPQExpBuffer(&conn->errorMessage, libpq_gettext("another command is already in progress\n")); return false; } - /* initialize async result-accumulation state */ - pqClearAsyncResult(conn); - - /* reset single-row processing mode */ - conn->singleRowMode = false; + if (conn->in_batch) + { + /* When enqueuing a message we don't change much of the connection + * state since it's already in use for the current command. The + * connection state will get updated when PQgetNextQuery(...) advances + * to start processing the queued message. + * + * Just make sure we can safely enqueue given the current connection + * state. We can enqueue behind another queue item, or behind a + * non-queue command (one that sends its own sync), but we can't + * enqueue if the connection is in a copy state. + */ + switch (conn->asyncStatus) + { + case PGASYNC_QUEUED: + case PGASYNC_READY: + case PGASYNC_READY_MORE: + case PGASYNC_BUSY: + /* ok to queue */ + break; + case PGASYNC_COPY_IN: + case PGASYNC_COPY_OUT: + case PGASYNC_COPY_BOTH: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("cannot queue commands during COPY\n")); + return false; + case PGASYNC_IDLE: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, idle state in batch mode")); + break; + } + } + else + { + /* This command's results will come in immediately */ + PQstartProcessingNewQuery(conn); + } /* ready to send command message */ return true; @@ -1414,6 +1561,10 @@ PQsendQueryGuts(PGconn *conn, int resultFormat) { int i; + PGcommandQueueEntry *pipeCmd = NULL; + char **last_query; + PGQueryClass *queryclass; + /* This isn't gonna work on a 2.0 server */ if (PG_PROTOCOL_MAJOR(conn->pversion) < 3) @@ -1423,6 +1574,23 @@ PQsendQueryGuts(PGconn *conn, return 0; } + if (conn->in_batch) + { + pipeCmd = PQmakePipelinedCommand(conn); + + if (pipeCmd == NULL) + return 0; /* error msg already set */ + + last_query = &pipeCmd->query; + queryclass = &pipeCmd->queryclass; + } + else + { + last_query = &conn->last_query; + queryclass = &conn->queryclass; + } + + /* * We will send Parse (if needed), Bind, Describe Portal, Execute, Sync, * using specified statement name and the unnamed portal. @@ -1535,22 +1703,25 @@ PQsendQueryGuts(PGconn *conn, pqPutMsgEnd(conn) < 0) goto sendFailed; - /* construct the Sync message */ - if (pqPutMsgStart('S', false, conn) < 0 || - pqPutMsgEnd(conn) < 0) - goto sendFailed; + if (!conn->in_batch) + { + /* construct the Sync message */ + if (pqPutMsgStart('S', false, conn) < 0 || + pqPutMsgEnd(conn) < 0) + goto sendFailed; + } /* remember we are using extended query protocol */ - conn->queryclass = PGQUERY_EXTENDED; + *queryclass = PGQUERY_EXTENDED; /* and remember the query text too, if possible */ /* if insufficient memory, last_query just winds up NULL */ - if (conn->last_query) - free(conn->last_query); + if (*last_query) + free(*last_query); if (command) - conn->last_query = strdup(command); + *last_query = strdup(command); else - conn->last_query = NULL; + *last_query = NULL; /* * Give the data a push. In nonblock mode, don't complain if we're unable @@ -1560,10 +1731,15 @@ PQsendQueryGuts(PGconn *conn, goto sendFailed; /* OK, it's launched! */ - conn->asyncStatus = PGASYNC_BUSY; + if (conn->in_batch) + PQappendPipelinedCommand(conn, pipeCmd); + else + conn->asyncStatus = PGASYNC_BUSY; + return 1; sendFailed: + PQrecyclePipelinedCommand(conn, pipeCmd); pqHandleSendFailure(conn); return 0; } @@ -1690,6 +1866,282 @@ PQisBusy(PGconn *conn) return conn->asyncStatus == PGASYNC_BUSY; } +/* PQisInBatchMode + * Return true if currently in batch mode + */ +int +PQisInBatchMode(PGconn *conn) +{ + if (!conn) + return FALSE; + + return conn->in_batch; +} + +/* PQqueriesInBatch + * Return true if there are queries currently pending in batch mode + */ +int +PQqueriesInBatch(PGconn *conn) +{ + if (!PQisInBatchMode(conn)) + return false; + + return conn->cmd_queue_head != NULL; +} + +/* PQbatchIsAborted + * Batch being processed is aborted, results discarded until next sync + */ +int +PQbatchIsAborted(PGconn *conn) +{ + if (!PQisInBatchMode(conn)) + return false; + + return conn->batch_aborted; +} + +/* Put an idle connection in batch mode. Commands submitted after this + * can be pipelined on the connection, there's no requirement to wait for + * one to finish before the next is dispatched. + * + * COPY is not permitted in batch mode. + * + * A set of commands is terminated by a PQsendEndBatch. Multiple sets of batched + * commands may be sent while in batch mode. Batch mode can be exited by + * calling PQendBatchMode() once all results are processed. + * + * This doesn't actually send anything on the wire, it just puts libpq + * into a state where it can pipeline work. + */ +int +PQbeginBatchMode(PGconn *conn) +{ + if (!conn) + return false; + + if (conn->in_batch) + return true; + + if (conn->asyncStatus != PGASYNC_IDLE) + return false; + + conn->in_batch = true; + conn->batch_aborted = false; + conn->asyncStatus = PGASYNC_QUEUED; + + return true; +} + +/* End batch mode and return to normal command mode. + * + * Has no effect unless the client has processed all results + * from all outstanding batches and the connection is idle. + * + * Returns true if batch mode ended. + */ +int +PQendBatchMode(PGconn *conn) +{ + if (!conn) + return false; + + if (!conn->in_batch) + return true; + + switch (conn->asyncStatus) + { + case PGASYNC_IDLE: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, IDLE in batch mode")); + break; + case PGASYNC_COPY_IN: + case PGASYNC_COPY_OUT: + case PGASYNC_COPY_BOTH: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, COPY in batch mode")); + break; + case PGASYNC_READY: + case PGASYNC_READY_MORE: + case PGASYNC_BUSY: + /* can't end batch while busy */ + return false; + case PGASYNC_QUEUED: + break; + } + + /* still work to process */ + if (conn->cmd_queue_head != NULL) + return false; + + conn->in_batch = false; + conn->batch_aborted = false; + conn->asyncStatus = PGASYNC_IDLE; + + return true; +} + +/* End a batch submission by sending a protocol sync. The connection will + * remain in batch mode and unavailable for new non-batch commands until all + * results from the batch are processed by the client. + * + * It's legal to start submitting another batch immediately, without waiting + * for the results of the current batch. There's no need to end batch mode + * and start it again. + * + * If a command in a batch fails, every subsequent command up to and including + * the PQsendEndBatch command result gets set to PGRES_BATCH_ABORTED state. If the + * whole batch is processed without error, a PGresult with PGRES_BATCH_END is + * produced. + */ +int +PQsendEndBatch(PGconn *conn) +{ + PGcommandQueueEntry *entry; + + if (!conn) + return false; + + if (!conn->in_batch) + return false; + + switch (conn->asyncStatus) + { + case PGASYNC_IDLE: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, IDLE in batch mode")); + break; + case PGASYNC_COPY_IN: + case PGASYNC_COPY_OUT: + case PGASYNC_COPY_BOTH: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, COPY in batch mode")); + break; + case PGASYNC_READY: + case PGASYNC_READY_MORE: + case PGASYNC_BUSY: + case PGASYNC_QUEUED: + /* can send sync to end this batch of cmds */ + break; + } + + entry = PQmakePipelinedCommand(conn); + entry->queryclass = PGQUERY_SYNC; + entry->query = NULL; + + /* construct the Sync message */ + if (pqPutMsgStart('S', false, conn) < 0 || + pqPutMsgEnd(conn) < 0) + goto sendFailed; + + PQappendPipelinedCommand(conn, entry); + + /* Should try to flush immediately if there's room */ + PQflush(conn); + + return true; + +sendFailed: + PQrecyclePipelinedCommand(conn, entry); + pqHandleSendFailure(conn); + return false; +} + +/* PQgetNextQuery + * In batch mode, start processing the next query in the queue. + * + * Returns true if the next query was popped from the queue and can + * be processed by PQconsumeInput, PQgetResult, etc. + * + * Returns false if the current query isn't done yet, the connection + * is not in a batch, or there are no more queries to process. + */ +int +PQgetNextQuery(PGconn *conn) +{ + PGcommandQueueEntry *next_query; + + if (!conn) + return FALSE; + + if (!conn->in_batch) + return false; + + switch (conn->asyncStatus) + { + case PGASYNC_COPY_IN: + case PGASYNC_COPY_OUT: + case PGASYNC_COPY_BOTH: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, COPY in batch mode")); + break; + case PGASYNC_READY: + case PGASYNC_READY_MORE: + case PGASYNC_BUSY: + /* client still has to process current query or results */ + return false; + break; + case PGASYNC_IDLE: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext_noop("internal error, IDLE in batch mode")); + break; + case PGASYNC_QUEUED: + /* next query please */ + break; + } + + if (conn->cmd_queue_head == NULL) + { + /* In batch mode but nothing left on the queue; caller can submit + * more work or PQendBatchMode() now. */ + return false; + } + + /* Pop the next query from the queue and set up the connection state + * as if it'd just been dispatched from a non-batched call */ + next_query = conn->cmd_queue_head; + conn->cmd_queue_head = next_query->next; + next_query->next = NULL; + + PQstartProcessingNewQuery(conn); + + conn->last_query = next_query->query; + next_query->query = NULL; + conn->queryclass = next_query->queryclass; + + PQrecyclePipelinedCommand(conn, next_query); + + if (conn->batch_aborted && conn->queryclass != PGQUERY_SYNC) + { + /* + * In an aborted batch we don't get anything from the server for each + * result; we're just discarding input until we get to the next sync + * from the server. The client needs to know its queries got aborted + * so we create a fake PGresult to return immediately from PQgetResult. + */ + conn->result = PQmakeEmptyPGresult(conn, + PGRES_BATCH_ABORTED); + if (!conn->result) + { + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("out of memory")); + pqSaveErrorResult(conn); + } + conn->asyncStatus = PGASYNC_READY; + } + else + { + /* allow parsing to continue */ + conn->asyncStatus = PGASYNC_BUSY; + + /* Parse any available data */ + parseInput(conn); + } + + return true; +} + /* * PQgetResult @@ -1697,7 +2149,6 @@ PQisBusy(PGconn *conn) * query work remains or an error has occurred (e.g. out of * memory). */ - PGresult * PQgetResult(PGconn *conn) { @@ -1749,10 +2200,31 @@ PQgetResult(PGconn *conn) switch (conn->asyncStatus) { case PGASYNC_IDLE: + case PGASYNC_QUEUED: res = NULL; /* query is complete */ break; case PGASYNC_READY: res = pqPrepareAsyncResult(conn); + if (conn->in_batch) + { + /* batched queries aren't followed by a Sync to put us back in + * PGASYNC_IDLE state, and when we do get a sync we could still + * have another batch coming after this one. + * + * The connection isn't idle since we can't submit new + * nonbatched commands. It isn't also busy since the current + * command is done and we need to process a new one. + */ + conn->asyncStatus = PGASYNC_QUEUED; + } + else + { + /* Set the state back to BUSY, allowing parsing to proceed. */ + conn->asyncStatus = PGASYNC_BUSY; + } + break; + case PGASYNC_READY_MORE: + res = pqPrepareAsyncResult(conn); /* Set the state back to BUSY, allowing parsing to proceed. */ conn->asyncStatus = PGASYNC_BUSY; break; @@ -1932,6 +2404,13 @@ PQexecStart(PGconn *conn) if (!conn) return false; + if (conn->asyncStatus == PGASYNC_QUEUED || conn->in_batch) + { + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("cannot PQexec in batch mode\n")); + return false; + } + /* * Silently discard any prior query result that application didn't eat. * This is probably poor design, but it's here for backward compatibility. @@ -2126,6 +2605,9 @@ PQsendDescribePortal(PGconn *conn, const char *portal) static int PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target) { + PGcommandQueueEntry *pipeCmd = NULL; + PGQueryClass *queryclass; + /* Treat null desc_target as empty string */ if (!desc_target) desc_target = ""; @@ -2141,6 +2623,20 @@ PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target) return 0; } + if (conn->in_batch) + { + pipeCmd = PQmakePipelinedCommand(conn); + + if (pipeCmd == NULL) + return 0; /* error msg already set */ + + queryclass = &pipeCmd->queryclass; + } + else + { + queryclass = &conn->queryclass; + } + /* construct the Describe message */ if (pqPutMsgStart('D', false, conn) < 0 || pqPutc(desc_type, conn) < 0 || @@ -2149,15 +2645,18 @@ PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target) goto sendFailed; /* construct the Sync message */ - if (pqPutMsgStart('S', false, conn) < 0 || - pqPutMsgEnd(conn) < 0) - goto sendFailed; + if (!conn->in_batch) + { + if (pqPutMsgStart('S', false, conn) < 0 || + pqPutMsgEnd(conn) < 0) + goto sendFailed; + } /* remember we are doing a Describe */ - conn->queryclass = PGQUERY_DESCRIBE; + *queryclass = PGQUERY_DESCRIBE; /* reset last-query string (not relevant now) */ - if (conn->last_query) + if (conn->last_query && !conn->in_batch) { free(conn->last_query); conn->last_query = NULL; @@ -2171,10 +2670,14 @@ PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target) goto sendFailed; /* OK, it's launched! */ - conn->asyncStatus = PGASYNC_BUSY; + if (conn->in_batch) + PQappendPipelinedCommand(conn, pipeCmd); + else + conn->asyncStatus = PGASYNC_BUSY; return 1; sendFailed: + PQrecyclePipelinedCommand(conn, pipeCmd); pqHandleSendFailure(conn); return 0; } diff --git a/src/interfaces/libpq/fe-protocol2.c b/src/interfaces/libpq/fe-protocol2.c index 3b0500f..32e8c79 100644 --- a/src/interfaces/libpq/fe-protocol2.c +++ b/src/interfaces/libpq/fe-protocol2.c @@ -412,6 +412,12 @@ pqParseInput2(PGconn *conn) { char id; + if (conn->asyncStatus == PGASYNC_QUEUED || conn->in_batch) + { + fprintf(stderr, "internal error, attempt to read v2 protocol in batch mode"); + abort(); + } + /* * Loop to parse successive complete messages available in the buffer. */ diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index 53776e2..026453b 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -220,10 +220,18 @@ pqParseInput3(PGconn *conn) return; conn->asyncStatus = PGASYNC_READY; break; - case 'Z': /* backend is ready for new query */ + case 'Z': /* sync response, backend is ready for new query */ if (getReadyForQuery(conn)) return; - conn->asyncStatus = PGASYNC_IDLE; + if (conn->in_batch) + { + conn->batch_aborted = false; + conn->result = PQmakeEmptyPGresult(conn, + PGRES_BATCH_END); + conn->asyncStatus = PGASYNC_READY; + } + else + conn->asyncStatus = PGASYNC_IDLE; break; case 'I': /* empty query */ if (conn->result == NULL) @@ -305,7 +313,7 @@ pqParseInput3(PGconn *conn) * parsing until the application accepts the current * result. */ - conn->asyncStatus = PGASYNC_READY; + conn->asyncStatus = PGASYNC_READY_MORE; return; } break; @@ -880,6 +888,9 @@ pqGetErrorNotice3(PGconn *conn, bool isError) PQExpBufferData workBuf; char id; + if (isError && conn->in_batch) + conn->batch_aborted = true; + /* * Since the fields might be pretty long, we create a temporary * PQExpBuffer rather than using conn->workBuffer. workBuffer is intended diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index 1b53d0e..ffdc220 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -93,7 +93,9 @@ typedef enum PGRES_NONFATAL_ERROR, /* notice or warning message */ PGRES_FATAL_ERROR, /* query failed */ PGRES_COPY_BOTH, /* Copy In/Out data transfer in progress */ - PGRES_SINGLE_TUPLE /* single tuple from larger resultset */ + PGRES_SINGLE_TUPLE, /* single tuple from larger resultset */ + PGRES_BATCH_END, /* end of a batch of commands */ + PGRES_BATCH_ABORTED, /* Command didn't run because of an abort earlier in a batch */ } ExecStatusType; typedef enum @@ -423,6 +425,15 @@ extern PGresult *PQgetResult(PGconn *conn); extern int PQisBusy(PGconn *conn); extern int PQconsumeInput(PGconn *conn); +/* Routines for batch mode management */ +extern int PQisInBatchMode(PGconn *conn); +extern int PQbatchIsAborted(PGconn *conn); +extern int PQqueriesInBatch(PGconn *conn); +extern int PQbeginBatchMode(PGconn *conn); +extern int PQendBatchMode(PGconn *conn); +extern int PQsendEndBatch(PGconn *conn); +extern int PQgetNextQuery(PGconn *conn); + /* LISTEN/NOTIFY support */ extern PGnotify *PQnotifies(PGconn *conn); diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index e9b73a9..d1ccab0 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -217,10 +217,13 @@ typedef enum { PGASYNC_IDLE, /* nothing's happening, dude */ PGASYNC_BUSY, /* query in progress */ - PGASYNC_READY, /* result ready for PQgetResult */ + PGASYNC_READY, /* query done, waiting for client to fetch result */ + PGASYNC_READY_MORE, /* query done, waiting for client to fetch result, + More results expected from this query */ PGASYNC_COPY_IN, /* Copy In data transfer in progress */ PGASYNC_COPY_OUT, /* Copy Out data transfer in progress */ - PGASYNC_COPY_BOTH /* Copy In/Out data transfer in progress */ + PGASYNC_COPY_BOTH, /* Copy In/Out data transfer in progress */ + PGASYNC_QUEUED /* Current query done, more in queue */ } PGAsyncStatusType; /* PGQueryClass tracks which query protocol we are now executing */ @@ -229,7 +232,8 @@ typedef enum PGQUERY_SIMPLE, /* simple Query protocol (PQexec) */ PGQUERY_EXTENDED, /* full Extended protocol (PQexecParams) */ PGQUERY_PREPARE, /* Parse only (PQprepare) */ - PGQUERY_DESCRIBE /* Describe Statement or Portal */ + PGQUERY_DESCRIBE, /* Describe Statement or Portal */ + PGQUERY_SYNC /* A protocol sync to end a batch */ } PGQueryClass; /* PGSetenvStatusType defines the state of the PQSetenv state machine */ @@ -316,6 +320,22 @@ typedef struct pg_conn_host struct addrinfo *addrlist; /* list of possible backend addresses */ } pg_conn_host; +/* An entry in the pending command queue. Used by batch mode to keep track + * of the expected results of future commands we've dispatched. + * + * Note that entries in this list are reused by being zeroed and appended to + * the tail when popped off the head. The entry with null next pointer is not + * the end of the list of expected commands, that's the tail pointer in + * pg_conn. + */ +typedef struct pgCommandQueueEntry +{ + PGQueryClass queryclass; /* Query type; PGQUERY_SYNC for sync msg */ + char *query; /* SQL command, or NULL if unknown */ + struct pgCommandQueueEntry *next; +} PGcommandQueueEntry; + + /* * PGconn stores all the state data associated with a single connection * to a backend. @@ -386,6 +406,8 @@ struct pg_conn bool options_valid; /* true if OK to attempt connection */ bool nonblocking; /* whether this connection is using nonblock * sending semantics */ + bool in_batch; /* connection is in batch (pipelined) mode */ + bool batch_aborted; /* current batch is aborted, discarding until next Sync */ bool singleRowMode; /* return current query result row-by-row? */ char copy_is_binary; /* 1 = copy binary, 0 = copy text */ int copy_already_done; /* # bytes already returned in COPY @@ -393,6 +415,15 @@ struct pg_conn PGnotify *notifyHead; /* oldest unreported Notify msg */ PGnotify *notifyTail; /* newest unreported Notify msg */ + /* The command queue + * + * head is the next pending cmd, tail is where we append new commands. + * Freed entries for recycling go on the recycle linked list. + */ + PGcommandQueueEntry *cmd_queue_head; + PGcommandQueueEntry *cmd_queue_tail; + PGcommandQueueEntry *cmd_queue_recycle; + /* Support for multiple hosts in connection string */ int nconnhost; /* # of possible hosts */ int whichhost; /* host we're currently considering */ @@ -690,6 +721,8 @@ extern char *libpq_ngettext(const char *msgid, const char *msgid_plural, unsigne #define libpq_ngettext(s, p, n) ((n) == 1 ? (s) : (p)) #endif +#define libpq_gettext_noop(x) (x) + /* * These macros are needed to let error-handling code be portable between * Unix and Windows. (ugh) diff --git a/src/interfaces/libpq/t/001_libpq_async.pl b/src/interfaces/libpq/t/001_libpq_async.pl new file mode 100644 index 0000000..d8f153e --- /dev/null +++ b/src/interfaces/libpq/t/001_libpq_async.pl @@ -0,0 +1,15 @@ +use strict; +use warnings; + +use Config; +use PostgresNode; +use TestLib; +use Test::More; + +my $node = get_new_node('main'); +$node->init; +$node->start; + +my $port = $node->port; + +$node->stop('fast'); diff --git a/src/test/examples/.gitignore b/src/test/examples/.gitignore index 1957ec1..7a72420 100644 --- a/src/test/examples/.gitignore +++ b/src/test/examples/.gitignore @@ -4,3 +4,4 @@ /testlibpq4 /testlo /testlo64 +/testlibpqbatch diff --git a/src/test/examples/Makefile b/src/test/examples/Makefile index 31da210..92a6faf 100644 --- a/src/test/examples/Makefile +++ b/src/test/examples/Makefile @@ -14,7 +14,7 @@ override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS) override LDLIBS := $(libpq_pgport) $(LDLIBS) -PROGS = testlibpq testlibpq2 testlibpq3 testlibpq4 testlo testlo64 +PROGS = testlibpq testlibpq2 testlibpq3 testlibpq4 testlo testlo64 testlibpqbatch all: $(PROGS) diff --git a/src/test/modules/test_libpq/.gitignore b/src/test/modules/test_libpq/.gitignore new file mode 100644 index 0000000..11e8463 --- /dev/null +++ b/src/test/modules/test_libpq/.gitignore @@ -0,0 +1,5 @@ +# Generated subdirectories +/log/ +/results/ +/tmp_check/ +/testlibpqbatch diff --git a/src/test/modules/test_libpq/Makefile b/src/test/modules/test_libpq/Makefile new file mode 100644 index 0000000..7f1b8b0 --- /dev/null +++ b/src/test/modules/test_libpq/Makefile @@ -0,0 +1,23 @@ +# src/test/modules/test_libpq/Makefile + +OBJS = testlibpqbatch.o +PROGRAM = testlibpqbatch + +PG_CPPFLAGS = -I$(libpq_srcdir) +PG_LIBS += $(libpq) + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = src/test/modules/test_libpq +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif + +check: prove-check + +prove-check: + $(prove_check) diff --git a/src/test/modules/test_libpq/README b/src/test/modules/test_libpq/README new file mode 100644 index 0000000..d8174dd --- /dev/null +++ b/src/test/modules/test_libpq/README @@ -0,0 +1 @@ +Test programs and libraries for libpq diff --git a/src/test/modules/test_libpq/t/001_libpq_async.pl b/src/test/modules/test_libpq/t/001_libpq_async.pl new file mode 100644 index 0000000..28c3eb5 --- /dev/null +++ b/src/test/modules/test_libpq/t/001_libpq_async.pl @@ -0,0 +1,26 @@ +use strict; +use warnings; + +use Config; +use PostgresNode; +use TestLib; +use Test::More tests => 5; +use Cwd; + +my $node = get_new_node('main'); +$node->init; +$node->start; + +my $port = $node->port; + +my $numrows = 10000; +my @tests = qw(disallowed_in_batch simple_batch multi_batch batch_abort timings); + +$ENV{PATH} = "$ENV{PATH}:" . getcwd(); +for my $testname (@tests) +{ + $node->command_ok(['testlibpqbatch', 'dbname=postgres', "$numrows", "$testname"], + "testlibpqbatch $testname"); +} + +$node->stop('fast'); diff --git a/src/test/modules/test_libpq/testlibpqbatch.c b/src/test/modules/test_libpq/testlibpqbatch.c new file mode 100644 index 0000000..a3c0a33 --- /dev/null +++ b/src/test/modules/test_libpq/testlibpqbatch.c @@ -0,0 +1,1386 @@ +/* + * src/test/examples/testlibpqbatch.c + * + * + * testlibpqbatch.c + * Test of batch execution funtionality + */ + +#ifdef WIN32 +#include +#endif +#include +#include +#include +#include +#include +#include +#include "libpq-fe.h" + +static void exit_nicely(PGconn *conn); +static void simple_batch(PGconn *conn); +static void test_disallowed_in_batch(PGconn *conn); +static void batch_insert_pipelined(PGconn *conn, int n_rows); +static void batch_insert_sequential(PGconn *conn, int n_rows); +static void batch_insert_copy(PGconn *conn, int n_rows); +static void test_batch_abort(PGconn *conn); + +#ifndef VERBOSE +#define VERBOSE 0 +#endif + +static const Oid INT4OID = 23; + +static const char *const drop_table_sql += "DROP TABLE IF EXISTS batch_demo"; +static const char *const create_table_sql += "CREATE UNLOGGED TABLE batch_demo(id serial primary key, itemno integer);"; +static const char *const insert_sql += "INSERT INTO batch_demo(itemno) VALUES ($1);"; + +static void +exit_nicely(PGconn *conn) +{ + PQfinish(conn); + exit(1); +} + +static void +simple_batch(PGconn *conn) +{ + PGresult *res = NULL; + const char *dummy_params[1] = {"1"}; + Oid dummy_param_oids[1] = {INT4OID}; + + fprintf(stderr, "simple batch... "); + fflush(stderr); + + /* + * Enter batch mode and dispatch a set of operations, which we'll then + * process the results of as they come in. + * + * For a simple case we should be able to do this without interim + * processing of results since our out buffer will give us enough slush to + * work with and we won't block on sending. So blocking mode is fine. + */ + if (PQisnonblocking(conn)) + { + fprintf(stderr, "Expected blocking connection mode\n"); + goto fail; + } + + if (!PQbeginBatchMode(conn)) + { + fprintf(stderr, "failed to enter batch mode: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching SELECT failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQendBatchMode(conn)) + { + fprintf(stderr, "exiting batch mode with work in progress should fail, but succeeded\n"); + goto fail; + } + + if (!PQsendEndBatch(conn)) + { + fprintf(stderr, "Ending a batch failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + /* + * in batch mode we have to ask for the first result to be processed; + * until we do PQgetResult will return null: + */ + if (PQgetResult(conn) != NULL) + { + fprintf(stderr, "PQgetResult returned something in a batch before first PQgetNextQuery() call\n"); + goto fail; + } + + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at first batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + /* We can't PQgetNextQuery when there might still be pending results */ + if (PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() should've failed with pending results: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (res == NULL) + { + fprintf(stderr, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, "Unexpected result code %s from first batch item\n", + PQresStatus(PQresultStatus(res))); + goto fail; + } + + PQclear(res); + res = NULL; + + if (PQgetResult(conn) != NULL) + { + fprintf(stderr, "PQgetResult returned something extra after first result before PQgetNextQuery() call\n"); + goto fail; + } + + /* + * Even though we've processed the result there's still a sync to come and + * we can't exit batch mode yet + */ + if (PQendBatchMode(conn)) + { + fprintf(stderr, "exiting batch mode after query but before sync succeeded incorrectly\n"); + goto fail; + } + + /* should now get an explicit sync result */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at sync after first batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (res == NULL) + { + fprintf(stderr, "PQgetResult returned null when sync result expected: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQresultStatus(res) != PGRES_BATCH_END) + { + fprintf(stderr, "Unexpected result code %s instead of sync result, error: %s\n", + PQresStatus(PQresultStatus(res)), PQerrorMessage(conn)); + goto fail; + } + + PQclear(res); + res = NULL; + + if (PQgetResult(conn) != NULL) + { + fprintf(stderr, "PQgetResult returned something extra after end batch call\n"); + goto fail; + } + + /* We're still in a batch... */ + if (!PQisInBatchMode(conn)) + { + fprintf(stderr, "Fell out of batch mode somehow\n"); + goto fail; + } + + /* until we end it, which we can safely do now */ + if (!PQendBatchMode(conn)) + { + fprintf(stderr, "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQisInBatchMode(conn)) + { + fprintf(stderr, "exiting batch mode didn't seem to work\n"); + goto fail; + } + + fprintf(stderr, "ok\n"); + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + +static void +test_disallowed_in_batch(PGconn *conn) +{ + PGresult *res = NULL; + + fprintf(stderr, "test error cases... "); + fflush(stderr); + + if (PQisnonblocking(conn)) + { + fprintf(stderr, "Expected blocking connection mode: %u\n", __LINE__); + goto fail; + } + + if (!PQbeginBatchMode(conn)) + { + fprintf(stderr, "Unable to enter batch mode\n"); + goto fail; + } + + if (!PQisInBatchMode(conn)) + { + fprintf(stderr, "Batch mode not activated properly\n"); + goto fail; + } + + /* PQexec should fail in batch mode */ + res = PQexec(conn, "SELECT 1"); + if (PQresultStatus(res) != PGRES_FATAL_ERROR) + { + fprintf(stderr, "PQexec should fail in batch mode but succeeded\n"); + goto fail; + } + + /* So should PQsendQuery */ + if (PQsendQuery(conn, "SELECT 1") != 0) + { + fprintf(stderr, "PQsendQuery should fail in batch mode but succeeded\n"); + goto fail; + } + + /* Entering batch mode when already in batch mode is OK */ + if (!PQbeginBatchMode(conn)) + { + fprintf(stderr, "re-entering batch mode should be a no-op but failed\n"); + goto fail; + } + + if (PQisBusy(conn)) + { + fprintf(stderr, "PQisBusy should return false when idle in batch, returned true\n"); + goto fail; + } + + /* ok, back to normal command mode */ + if (!PQendBatchMode(conn)) + { + fprintf(stderr, "couldn't exit idle empty batch mode\n"); + goto fail; + } + + if (PQisInBatchMode(conn)) + { + fprintf(stderr, "Batch mode not terminated properly\n"); + goto fail; + } + + /* exiting batch mode when not in batch mode should be a no-op */ + if (!PQendBatchMode(conn)) + { + fprintf(stderr, "batch mode exit when not in batch mode should succeed but failed\n"); + goto fail; + } + + /* can now PQexec again */ + res = PQexec(conn, "SELECT 1"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, "PQexec should succeed after exiting batch mode but failed with: %s\n", + PQerrorMessage(conn)); + goto fail; + } + + fprintf(stderr, "ok\n"); + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + +static void +multi_batch(PGconn *conn) +{ + PGresult *res = NULL; + const char *dummy_params[1] = {"1"}; + Oid dummy_param_oids[1] = {INT4OID}; + + fprintf(stderr, "multi batch... "); + fflush(stderr); + + /* + * Queue up a couple of small batches and process each without returning + * to command mode first. + */ + if (!PQbeginBatchMode(conn)) + { + fprintf(stderr, "failed to enter batch mode: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching first SELECT failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendEndBatch(conn)) + { + fprintf(stderr, "Ending first batch failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching second SELECT failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendEndBatch(conn)) + { + fprintf(stderr, "Ending second batch failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + /* OK, start processing the batch results */ + if (PQgetResult(conn) != NULL) + { + fprintf(stderr, "PQgetResult returned something in a batch before first PQgetNextQuery() call\n"); + goto fail; + } + + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at first batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (res == NULL) + { + fprintf(stderr, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, "Unexpected result code %s from first batch item\n", + PQresStatus(PQresultStatus(res))); + goto fail; + } + PQclear(res); + res = NULL; + + if (PQgetResult(conn) != NULL) + { + fprintf(stderr, "PQgetResult returned something extra after first result before PQgetNextQuery() call\n"); + goto fail; + } + + if (PQendBatchMode(conn)) + { + fprintf(stderr, "exiting batch mode after query but before sync succeeded incorrectly\n"); + goto fail; + } + + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at sync after first batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (res == NULL) + { + fprintf(stderr, "PQgetResult returned null when sync result expected: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQresultStatus(res) != PGRES_BATCH_END) + { + fprintf(stderr, "Unexpected result code %s instead of sync result, error: %s (line %u)\n", + PQresStatus(PQresultStatus(res)), PQerrorMessage(conn), __LINE__); + goto fail; + } + + PQclear(res); + res = NULL; + + /* second batch */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at second batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (res == NULL) + { + fprintf(stderr, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, "Unexpected result code %s from second batch item\n", + PQresStatus(PQresultStatus(res))); + goto fail; + } + + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at second batch sync: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (res == NULL) + { + fprintf(stderr, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQresultStatus(res) != PGRES_BATCH_END) + { + fprintf(stderr, "Unexpected result code %s from second end batch\n", + PQresStatus(PQresultStatus(res))); + goto fail; + } + + /* We're still in a batch... */ + if (!PQisInBatchMode(conn)) + { + fprintf(stderr, "Fell out of batch mode somehow\n"); + goto fail; + } + + /* until we end it, which we can safely do now */ + if (!PQendBatchMode(conn)) + { + fprintf(stderr, "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQisInBatchMode(conn)) + { + fprintf(stderr, "exiting batch mode didn't seem to work\n"); + goto fail; + } + + fprintf(stderr, "ok\n"); + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + +/* + * When an operation in a batch fails the rest of the batch is flushed. We + * still have to get results for each batch item, but the item will just be + * a PGRES_BATCH_ABORTED code. + * + * This intentionally doesn't use a transaction to wrap the batch. You should + * usually use an xact, but in this case we want to observe the effects of each + * statement. + */ +static void +test_batch_abort(PGconn *conn) +{ + PGresult *res = NULL; + const char *dummy_params[1] = {"1"}; + Oid dummy_param_oids[1] = {INT4OID}; + int i; + + fprintf(stderr, "aborted batch... "); + fflush(stderr); + + res = PQexec(conn, drop_table_sql); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "dispatching DROP TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + res = PQexec(conn, create_table_sql); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "dispatching CREATE TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + + /* + * Queue up a couple of small batches and process each without returning + * to command mode first. Make sure the second operation in the first + * batch ERRORs. + */ + if (!PQbeginBatchMode(conn)) + { + fprintf(stderr, "failed to enter batch mode: %s\n", PQerrorMessage(conn)); + goto fail; + } + + dummy_params[0] = "1"; + if (!PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching first INSERT failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendQueryParams(conn, "SELECT no_such_function($1)", 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching error select failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + dummy_params[0] = "2"; + if (!PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching second insert failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendEndBatch(conn)) + { + fprintf(stderr, "Ending first batch failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + dummy_params[0] = "3"; + if (!PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids, + dummy_params, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching second-batch insert failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendEndBatch(conn)) + { + fprintf(stderr, "Ending second batch failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + + /* + * OK, start processing the batch results. + * + * We should get a tuples-ok for the first query, a fatal error, a batch + * aborted message for the second insert, a batch-end, then a command-ok + * and a batch-ok for the second batch operation. + */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at first batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (((res = PQgetResult(conn)) == NULL) || PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "Unexpected result code %s from first batch item, error='%s'\n", + res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)), + res == NULL ? PQerrorMessage(conn) : PQresultErrorMessage(res)); + goto fail; + } + PQclear(res); + res = NULL; + + /* second query, caused error */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at second batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (((res = PQgetResult(conn)) == NULL) || PQresultStatus(res) != PGRES_FATAL_ERROR) + { + fprintf(stderr, "Unexpected result code from second batch item. Wanted PGRES_FATAL_ERROR, got %s\n", + res == NULL ? "NULL" : PQresStatus(PQresultStatus(res))); + goto fail; + } + PQclear(res); + res = NULL; + + /* + * batch should now be aborted. + * + * Note that we could still queue more queries at this point if we wanted; + * they'd get added to a new third batch since we've already sent a + * second. The aborted flag relates only to the batch being received. + */ + if (!PQbatchIsAborted(conn)) + { + fprintf(stderr, "batch should be flagged as aborted but isn't\n"); + goto fail; + } + + /* third query in batch, the second insert */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at third batch entry: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (((res = PQgetResult(conn)) == NULL) || PQresultStatus(res) != PGRES_BATCH_ABORTED) + { + fprintf(stderr, "Unexpected result code from third batch item. Wanted PGRES_BATCH_ABORTED, got %s\n", + res == NULL ? "NULL" : PQresStatus(PQresultStatus(res))); + goto fail; + } + PQclear(res); + res = NULL; + + if (!PQbatchIsAborted(conn)) + { + fprintf(stderr, "batch should be flagged as aborted but isn't\n"); + goto fail; + } + + /* We're still in a batch... */ + if (!PQisInBatchMode(conn)) + { + fprintf(stderr, "Fell out of batch mode somehow\n"); + goto fail; + } + + /* the batch sync */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at first batch sync: %s\n", PQerrorMessage(conn)); + goto fail; + } + + /* + * The end of a failed batch is still a PGRES_BATCH_END so clients know to + * start processing results normally again and can tell the difference + * between skipped commands and the sync. + */ + if (((res = PQgetResult(conn)) == NULL) || PQresultStatus(res) != PGRES_BATCH_END) + { + fprintf(stderr, "Unexpected result code from first batch sync. Wanted PGRES_BATCH_END, got %s\n", + res == NULL ? "NULL" : PQresStatus(PQresultStatus(res))); + goto fail; + } + PQclear(res); + res = NULL; + + if (PQbatchIsAborted(conn)) + { + fprintf(stderr, "sync should've cleared the aborted flag but didn't\n"); + goto fail; + } + + /* We're still in a batch... */ + if (!PQisInBatchMode(conn)) + { + fprintf(stderr, "Fell out of batch mode somehow\n"); + goto fail; + } + + /* the insert from the second batch */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at first entry in second batch: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (((res = PQgetResult(conn)) == NULL) || PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "Unexpected result code %s from first item in second batch\n", + res == NULL ? "NULL" : PQresStatus(PQresultStatus(res))); + goto fail; + } + PQclear(res); + res = NULL; + + /* the second batch sync */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "PQgetNextQuery() failed at second batch sync: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (((res = PQgetResult(conn)) == NULL) || PQresultStatus(res) != PGRES_BATCH_END) + { + fprintf(stderr, "Unexpected result code %s from second batch sync\n", + res == NULL ? "NULL" : PQresStatus(PQresultStatus(res))); + goto fail; + } + PQclear(res); + res = NULL; + + /* We're still in a batch... */ + if (!PQisInBatchMode(conn)) + { + fprintf(stderr, "Fell out of batch mode somehow\n"); + goto fail; + } + + /* until we end it, which we can safely do now */ + if (!PQendBatchMode(conn)) + { + fprintf(stderr, "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQisInBatchMode(conn)) + { + fprintf(stderr, "exiting batch mode didn't seem to work\n"); + goto fail; + } + + fprintf(stderr, "ok\n"); + + /* + * Since we fired the batches off without a surrounding xact, the results + * should be: + * + * - Implicit xact started by server around 1st batch - First insert + * applied - Second statement aborted xact - Third insert skipped - Sync + * rolled back first implicit xact - Implicit xact created by server + * around 2nd batch - insert applied from 2nd batch - Sync commits 2nd + * xact + * + * So we should only have the value 3 that we inserted. + */ + res = PQexec(conn, "SELECT itemno FROM batch_demo"); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, "Expected tuples, got %s: %s", + PQresStatus(PQresultStatus(res)), PQerrorMessage(conn)); + goto fail; + } + + for (i = 0; i < PQntuples(res); i++) + { + const char *val = PQgetvalue(res, i, 0); + + if (strcmp(val, "3") != 0) + { + fprintf(stderr, "expected only insert with value 3, got %s", val); + goto fail; + } + } + + if (PQntuples(res) != 1) + { + fprintf(stderr, "expected 1 result, got %d", PQntuples(res)); + goto fail; + } + PQclear(res); + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + +/* max char length of an int32, plus sign and null terminator */ +#define MAXINTLEN 12 + +/* State machine enums for batch insert */ +typedef enum BatchInsertStep +{ + BI_BEGIN_TX, + BI_DROP_TABLE, + BI_CREATE_TABLE, + BI_PREPARE, + BI_INSERT_ROWS, + BI_COMMIT_TX, + BI_SYNC, + BI_DONE +} BatchInsertStep; + +static void +batch_insert_pipelined(PGconn *conn, int n_rows) +{ + PGresult *res = NULL; + const char *insert_params[1]; + Oid insert_param_oids[1] = {INT4OID}; + char insert_param_0[MAXINTLEN]; + BatchInsertStep send_step = BI_BEGIN_TX, + recv_step = BI_BEGIN_TX; + int rows_to_send, + rows_to_receive; + + insert_params[0] = &insert_param_0[0]; + + rows_to_send = rows_to_receive = n_rows; + + /* + * Do a batched insert into a table created at the start of the batch + */ + if (!PQbeginBatchMode(conn)) + { + fprintf(stderr, "failed to enter batch mode: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (!PQsendQueryParams(conn, "BEGIN", + 0, NULL, NULL, NULL, NULL, 0)) + { + fprintf(stderr, "xact start failed: %s\n", PQerrorMessage(conn)); + goto fail; + } +#if VERBOSE + fprintf(stdout, "sent BEGIN\n"); +#endif + send_step = BI_DROP_TABLE; + + if (!PQsendQueryParams(conn, drop_table_sql, + 0, NULL, NULL, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching DROP TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } +#if VERBOSE + fprintf(stdout, "sent DROP\n"); +#endif + send_step = BI_CREATE_TABLE; + + if (!PQsendQueryParams(conn, create_table_sql, + 0, NULL, NULL, NULL, NULL, 0)) + { + fprintf(stderr, "dispatching CREATE TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } +#if VERBOSE + fprintf(stdout, "sent CREATE\n"); +#endif + send_step = BI_PREPARE; + + if (!PQsendPrepare(conn, "my_insert", insert_sql, 1, insert_param_oids)) + { + fprintf(stderr, "dispatching PREPARE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } +#if VERBOSE + fprintf(stdout, "sent PREPARE\n"); +#endif + send_step = BI_INSERT_ROWS; + + /* + * Now we start inserting. We'll be sending enough data that we could fill + * our out buffer, so to avoid deadlocking we need to enter nonblocking + * mode and consume input while we send more output. As results of each + * query are processed we should pop them to allow processing of the next + * query. There's no need to finish the batch before processing results. + */ + if (PQsetnonblocking(conn, 1) != 0) + { + fprintf(stderr, "failed to set nonblocking mode: %s\n", PQerrorMessage(conn)); + goto fail; + } + + while (recv_step != BI_DONE) + { + int sock; + fd_set input_mask; + fd_set output_mask; + + sock = PQsocket(conn); + + if (sock < 0) + break; /* shouldn't happen */ + + FD_ZERO(&input_mask); + FD_SET(sock, &input_mask); + FD_ZERO(&output_mask); + FD_SET(sock, &output_mask); + + if (select(sock + 1, &input_mask, &output_mask, NULL, NULL) < 0) + { + fprintf(stderr, "select() failed: %s\n", strerror(errno)); + exit_nicely(conn); + } + + /* + * Process any results, so we keep the server's out buffer free + * flowing and it can continue to process input + */ + if (FD_ISSET(sock, &input_mask)) + { + PQconsumeInput(conn); + + /* Read until we'd block if we tried to read */ + while (!PQisBusy(conn) && recv_step < BI_DONE) + { + const char *cmdtag; + const char *description = NULL; + int status; + BatchInsertStep next_step; + + + res = PQgetResult(conn); + + if (res == NULL) + { + /* + * No more results from this query, advance to the next + * result + */ + if (!PQgetNextQuery(conn)) + { + fprintf(stderr, "Expected next query result but unable to dequeue: %s\n", + PQerrorMessage(conn)); + goto fail; + } +#if VERBOSE + fprintf(stdout, "next query!\n"); +#endif + continue; + } + + status = PGRES_COMMAND_OK; + next_step = recv_step + 1; + switch (recv_step) + { + case BI_BEGIN_TX: + cmdtag = "BEGIN"; + break; + case BI_DROP_TABLE: + cmdtag = "DROP TABLE"; + break; + case BI_CREATE_TABLE: + cmdtag = "CREATE TABLE"; + break; + case BI_PREPARE: + cmdtag = ""; + description = "PREPARE"; + break; + case BI_INSERT_ROWS: + cmdtag = "INSERT"; + rows_to_receive--; + if (rows_to_receive > 0) + next_step = BI_INSERT_ROWS; + break; + case BI_COMMIT_TX: + cmdtag = "COMMIT"; + break; + case BI_SYNC: + cmdtag = ""; + description = "SYNC"; + status = PGRES_BATCH_END; + break; + case BI_DONE: + /* unreachable */ + abort(); + } + if (description == NULL) + description = cmdtag; + +#if VERBOSE + fprintf(stderr, "At state %d (%s) expect tag '%s', result code %s, expect %d more rows, transition to %d\n", + recv_step, description, cmdtag, PQresStatus(status), rows_to_receive, next_step); +#endif + + if (PQresultStatus(res) != status) + { + fprintf(stderr, "%s reported status %s, expected %s. Error msg is [%s]\n", + description, PQresStatus(PQresultStatus(res)), PQresStatus(status), PQerrorMessage(conn)); + goto fail; + } + if (strncmp(PQcmdStatus(res), cmdtag, strlen(cmdtag)) != 0) + { + fprintf(stderr, "%s expected command tag '%s', got '%s'\n", + description, cmdtag, PQcmdStatus(res)); + goto fail; + } +#if VERBOSE + fprintf(stdout, "Got %s OK\n", cmdtag); +#endif + recv_step = next_step; + + PQclear(res); + res = NULL; + } + } + + /* Write more rows and/or the end batch message, if needed */ + if (FD_ISSET(sock, &output_mask)) + { + PQflush(conn); + + if (send_step == BI_INSERT_ROWS) + { + snprintf(&insert_param_0[0], MAXINTLEN, "%d", rows_to_send); + insert_param_0[MAXINTLEN - 1] = '\0'; + + if (PQsendQueryPrepared(conn, "my_insert", + 1, insert_params, NULL, NULL, 0)) + { +#if VERBOSE + fprintf(stdout, "sent row %d\n", rows_to_send); +#endif + rows_to_send--; + if (rows_to_send == 0) + send_step = BI_COMMIT_TX; + } + else + { + /* + * in nonblocking mode, so it's OK for an insert to fail + * to send + */ + fprintf(stderr, "WARNING: failed to send insert #%d: %s\n", + rows_to_send, PQerrorMessage(conn)); + } + } + else if (send_step == BI_COMMIT_TX) + { + if (PQsendQueryParams(conn, "COMMIT", + 0, NULL, NULL, NULL, NULL, 0)) + { +#if VERBOSE + fprintf(stdout, "sent COMMIT\n"); +#endif + send_step = BI_SYNC; + } + else + { + fprintf(stderr, "WARNING: failed to send commit: %s\n", + PQerrorMessage(conn)); + } + } + else if (send_step == BI_SYNC) + { + if (PQsendEndBatch(conn)) + { +#if VERBOSE + fprintf(stdout, "Dispatched end batch message\n"); +#endif + send_step = BI_DONE; + } + else + { + fprintf(stderr, "WARNING: Ending a batch failed: %s\n", + PQerrorMessage(conn)); + } + } + } + + } + + /* We've got the sync message and the batch should be done */ + if (!PQendBatchMode(conn)) + { + fprintf(stderr, "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn)); + goto fail; + } + + if (PQsetnonblocking(conn, 0) != 0) + { + fprintf(stderr, "failed to clear nonblocking mode: %s\n", PQerrorMessage(conn)); + goto fail; + } + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + + +static void +batch_insert_sequential(PGconn *conn, int nrows) +{ + PGresult *res = NULL; + const char *insert_params[1]; + Oid insert_param_oids[1] = {INT4OID}; + char insert_param_0[MAXINTLEN]; + + insert_params[0] = &insert_param_0[0]; + + res = PQexec(conn, "BEGIN"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "BEGIN failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + + res = PQexec(conn, drop_table_sql); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "DROP TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + + res = PQexec(conn, create_table_sql); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "CREATE TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + + res = PQprepare(conn, "my_insert2", insert_sql, 1, insert_param_oids); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "prepare failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + + while (nrows > 0) + { + snprintf(&insert_param_0[0], MAXINTLEN, "%d", nrows); + insert_param_0[MAXINTLEN - 1] = '\0'; + + res = PQexecPrepared(conn, "my_insert2", + 1, insert_params, NULL, NULL, 0); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "INSERT failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + nrows--; + } + + res = PQexec(conn, "COMMIT"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "COMMIT failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + +static void +batch_insert_copy(PGconn *conn, int nrows) +{ + PGresult *res = NULL; + + res = PQexec(conn, drop_table_sql); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "DROP TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + + res = PQexec(conn, create_table_sql); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "CREATE TABLE failed: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + res = NULL; + + res = PQexec(conn, "COPY batch_demo(itemno) FROM stdin"); + if (PQresultStatus(res) != PGRES_COPY_IN) + { + fprintf(stderr, "COPY: %s\n", PQerrorMessage(conn)); + goto fail; + } + PQclear(res); + res = NULL; + + while (nrows > 0) + { + char buf[MAXINTLEN+2]; + int formatted = snprintf(&buf[0], MAXINTLEN+1, "%d\n", nrows); + if (formatted >= MAXINTLEN+1) + { + fprintf(stderr, "Buffer write truncated somehow\n"); + goto fail; + } + + if (PQputCopyData(conn, buf, formatted) != 1) + { + fprintf(stderr, "Write of COPY data failed: %s\n", + PQerrorMessage(conn)); + goto fail; + } + + nrows--; + } + + if (PQputCopyEnd(conn, NULL) != 1) + { + fprintf(stderr, "Finishing COPY failed: %s", + PQerrorMessage(conn)); + goto fail; + } + + res = PQgetResult(conn); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, "COPY finished with %s: %s\n", + PQresStatus(PQresultStatus(res)), + PQresultErrorMessage(res)); + goto fail; + } + PQclear(res); + res = NULL; + + return; + +fail: + PQclear(res); + exit_nicely(conn); +} + +static void +test_timings(PGconn *conn, int number_of_rows) +{ + struct timeval start_time, + end_time, + elapsed_time; + + fprintf(stderr, "inserting %d rows batched then unbatched\n", number_of_rows); + + gettimeofday(&start_time, NULL); + batch_insert_pipelined(conn, number_of_rows); + gettimeofday(&end_time, NULL); + timersub(&end_time, &start_time, &elapsed_time); + printf("batch insert elapsed: %ld.%06lds\n", elapsed_time.tv_sec, (long)elapsed_time.tv_usec); + + gettimeofday(&start_time, NULL); + batch_insert_sequential(conn, number_of_rows); + gettimeofday(&end_time, NULL); + timersub(&end_time, &start_time, &elapsed_time); + printf("sequential insert elapsed: %ld.%06lds\n", elapsed_time.tv_sec, (long)elapsed_time.tv_usec); + + gettimeofday(&start_time, NULL); + batch_insert_copy(conn, number_of_rows); + gettimeofday(&end_time, NULL); + timersub(&end_time, &start_time, &elapsed_time); + printf("COPY elapsed: %ld.%06lds\n", elapsed_time.tv_sec, (long)elapsed_time.tv_usec); + + fprintf(stderr, "Done.\n"); +} + +static void +usage_exit(const char *progname) +{ + fprintf(stderr, "Usage: %s ['connstring' [number_of_rows [test_to_run]]]\n", progname); + fprintf(stderr, " tests: all|disallowed_in_batch|simple_batch|multi_batch|batch_abort|timings\n"); + exit(1); +} + + +int +main(int argc, char **argv) +{ + const char *conninfo; + PGconn *conn; + int number_of_rows = 10000; + + int run_disallowed_in_batch = 1, + run_simple_batch = 1, + run_multi_batch = 1, + run_batch_abort = 1, + run_timings = 1; + + /* + * If the user supplies a parameter on the command line, use it as the + * conninfo string; otherwise default to setting dbname=postgres and using + * environment variables or defaults for all other connection parameters. + */ + if (argc > 4) + { + usage_exit(argv[0]); + } + if (argc > 3) + { + if (strcmp(argv[3], "all") != 0) + { + run_disallowed_in_batch = 0; + run_simple_batch = 0; + run_multi_batch = 0; + run_batch_abort = 0; + run_timings = 0; + if (strcmp(argv[3], "disallowed_in_batch") == 0) + run_disallowed_in_batch = 1; + else if (strcmp(argv[3], "simple_batch") == 0) + run_simple_batch = 1; + else if (strcmp(argv[3], "multi_batch") == 0) + run_multi_batch = 1; + else if (strcmp(argv[3], "batch_abort") == 0) + run_batch_abort = 1; + else if (strcmp(argv[3], "timings") == 0) + run_timings = 1; + else + { + fprintf(stderr, "%s is not a recognised test name\n", argv[3]); + usage_exit(argv[0]); + } + } + } + if (argc > 2) + { + errno = 0; + number_of_rows = strtol(argv[2], NULL, 10); + if (errno) + { + fprintf(stderr, "couldn't parse '%s' as an integer or zero rows supplied: %s", argv[2], strerror(errno)); + usage_exit(argv[0]); + } + if (number_of_rows <= 0) + { + fprintf(stderr, "number_of_rows must be positive"); + usage_exit(argv[0]); + } + } + if (argc > 1) + { + conninfo = argv[1]; + } + else + { + conninfo = "dbname = postgres"; + } + + /* Make a connection to the database */ + conn = PQconnectdb(conninfo); + + /* Check to see that the backend connection was successfully made */ + if (PQstatus(conn) != CONNECTION_OK) + { + fprintf(stderr, "Connection to database failed: %s\n", + PQerrorMessage(conn)); + exit_nicely(conn); + } + + if (run_disallowed_in_batch) + test_disallowed_in_batch(conn); + + if (run_simple_batch) + simple_batch(conn); + + if (run_multi_batch) + multi_batch(conn); + + if (run_batch_abort) + test_batch_abort(conn); + + if (run_timings) + test_timings(conn, number_of_rows); + + + /* close the connection to the database and cleanup */ + PQfinish(conn); + + return 0; +}