diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml
index 92556c7ce0..15d0c03c89 100644
--- a/doc/src/sgml/libpq.sgml
+++ b/doc/src/sgml/libpq.sgml
@@ -4829,6 +4829,465 @@ int PQflush(PGconn *conn);
+
+ Batch mode and query pipelining
+
+
+ libpq
+ batch mode
+
+
+
+ libpq
+ pipelining
+
+
+
+ libpq supports queueing up 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.
+
+
+
+ When to use batching
+
+
+ Much like asynchronous query mode, there is no performance disadvantage to
+ using batching and pipelining. It increases client application complexity
+ and extra caution is required to prevent client/server deadlocks but
+ can sometimes 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 not 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 10.0, but clients using PostgresSQL 10.0 version of libpq can
+ use batches on server versions 7.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
+ a connection into batch mode. Enter batch mode with PQenterBatchMode(conn) or test
+ whether batch mode is active with PQbatchStatus(conn). In batch mode only asynchronous operations are permitted, and
+ COPY is not recommended as it most likely will trigger failure in batch processing.
+ Using any synchronous command execution functions such as PQfn,
+ PQexec or one of its sibling functions are error conditions.
+ Functions allowed in batch mode are described in .
+
+
+
+ The client uses libpq's asynchronous query functions to dispatch work,
+ marking the end of each batch with PQbatchSendQueue.
+ And to get results, it uses PQgetResult. It may eventually exit
+ batch mode with PQexitBatchMode 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 has 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.
+ Batch mode consumes more memory when send/recv is not done as required even in non-blocking mode.
+
+
+
+
+ Issuing queries
+
+
+ After entering batch mode the application dispatches requests
+ using normal asynchronous libpq functions such as
+ PQsendQueryParams, PQsendPrepare,
+ PQsendQueryPrepared, PQsendDescribePortal,
+ PQsendDescribePrepared.
+ The asynchronous requests are followed by a PQbatchSendQueue(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 may begin 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 user must call
+ PQgetResult and handle the results until
+ PQgetResult returns null. The result from the next batch entry
+ may then be retrieved using PQgetResult again and the cycle repeated. The
+ application handles individual statement results as normal.
+
+
+
+ To enter single-row mode, call PQsetSingleRowMode before retrieving
+ results with PQgetResult. This mode selection is effective
+ only for the query currently being processed. For more information on the use of PQsetSingleRowMode
+ , refer to .
+
+
+
+
+ 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
+ PQbatchSendQueue 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 (exept that PQgetResult
+ return null to indicate that we start returning the results of next query). 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
+ PQbatchSendQueue 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
+ 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.
+ Based on available memory, results from socket should be read frequently and
+ there's no need to wait till the batch end to read the results. 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
+ PQexitBatchMode(conn).
+
+
+
+
+
+
+ Functions associated with batch mode
+
+
+
+
+
+ PQbatchStatus
+
+ PQbatchStatus
+
+
+
+
+
+ Returns current batch mode status of the libpq connection.
+
+int PQbatchStatus(PGconn *conn);
+
+
+
+
+
+ PQBATCH_MODE_ON
+
+
+
+
+ Returns PQBATCH_MODE_ON if libpq connection is in batch mode.
+
+
+
+
+
+
+ PQBATCH_MODE_OFF
+
+
+
+
+ Returns PQBATCH_MODE_OFF if libpq connection is not in batch mode.
+
+
+
+
+
+
+ PQBATCH_MODE_ABORTED
+
+
+
+ Returns PQBATCH_MODE_ABORTED if libpq connection is in
+ aborted status. The aborted flag is cleared as soon as the result of the
+ PQbatchSendQueue at the end of the aborted batch is
+ processed. Clients don't usually need this function to verify aborted status
+ as they can tell that the batch is aborted from PGRES_BATCH_ABORTED
+ result codes.
+
+
+
+
+
+
+
+
+
+
+
+ PQenterBatchMode
+
+ PQenterBatchMode
+
+
+
+
+
+ Causes a connection to enter batch mode if it is currently idle or
+ already in batch mode.
+
+
+int PQenterBatchMode(PGconn *conn);
+
+
+
+
+ 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.
+
+
+
+
+
+
+
+ PQexitBatchMode
+
+ PQexitBatchMode
+
+
+
+
+
+ Causes a connection to exit batch mode if it is currently in batch mode
+ with an empty queue and no pending results.
+
+int PQexitBatchMode(PGconn *conn);
+
+
+ Returns 1 for success.
+ Returns 1 and takes no action if not in batch mode. If the current statement isn't finished
+ processing or there are results pending for collection with
+ PQgetResult, returns 0 and does nothing.
+
+
+
+
+
+
+
+ PQbatchSendQueue
+
+ PQbatchSendQueue
+
+
+
+
+
+ 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 PQbatchSendQueue(PGconn *conn);
+
+
+ Returns 1 for success. Returns 0 if the connection is not in batch mode
+ or sending a sync message is failed.
+
+
+
+
+
+
+
+
+
+
+
Retrieving Query Results Row-by-Row
@@ -4869,6 +5328,15 @@ int PQflush(PGconn *conn);
Each object should be freed with as usual.
+
+
+ When using batch mode, activate the single row mode on the current batch query by
+ calling PQsetSingleRowMode
+ before retrieving results with PQgetResult.
+ See for more information.
+
+
+
diff --git a/doc/src/sgml/lobj.sgml b/doc/src/sgml/lobj.sgml
index cf4653fe0f..a1b46c83ed 100644
--- a/doc/src/sgml/lobj.sgml
+++ b/doc/src/sgml/lobj.sgml
@@ -130,6 +130,10 @@
libpq library.
+
+ Client applications cannot use these functions while libpq connection is in batch mode.
+
+
Creating a Large Object
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 8afa5a29b4..69bf757e85 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -1022,6 +1022,9 @@ libpqrcv_exec(WalReceiverConn *conn, const char *query,
walres->status = WALRCV_ERROR;
walres->err = pchomp(PQerrorMessage(conn->streamConn));
break;
+ default:
+ /* This is just to keep compiler quiet */
+ break;
}
PQclear(pgres);
diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt
index bbc1f90481..ffb18fcf47 100644
--- a/src/interfaces/libpq/exports.txt
+++ b/src/interfaces/libpq/exports.txt
@@ -179,3 +179,8 @@ PQgetgssctx 176
PQsetSSLKeyPassHook_OpenSSL 177
PQgetSSLKeyPassHook_OpenSSL 178
PQdefaultSSLKeyPassHook_OpenSSL 179
+PQenterBatchMode 180
+PQexitBatchMode 181
+PQbatchSendQueue 182
+PQbatchProcessQueue 183
+PQbatchStatus 184
diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index 7bee9dd201..25c79cb7b4 100644
--- a/src/interfaces/libpq/fe-connect.c
+++ b/src/interfaces/libpq/fe-connect.c
@@ -535,6 +535,24 @@ pqDropConnection(PGconn *conn, bool flushInput)
}
}
+/*
+ * PQfreeCommandQueue
+ * Free all the entries of PGcommandQueueEntry queue passed.
+ */
+static void
+PQfreeCommandQueue(PGcommandQueueEntry *queue)
+{
+
+ while (queue != NULL)
+ {
+ PGcommandQueueEntry *prev = queue;
+
+ queue = queue->next;
+ if (prev->query)
+ free(prev->query);
+ free(prev);
+ }
+}
/*
* pqDropServerData
@@ -554,6 +572,7 @@ pqDropServerData(PGconn *conn)
{
PGnotify *notify;
pgParameterStatus *pstatus;
+ PGcommandQueueEntry *queue;
/* Forget pending notifies */
notify = conn->notifyHead;
@@ -566,6 +585,14 @@ pqDropServerData(PGconn *conn)
}
conn->notifyHead = conn->notifyTail = NULL;
+ queue = conn->cmd_queue_head;
+ PQfreeCommandQueue(queue);
+ conn->cmd_queue_head = conn->cmd_queue_tail = NULL;
+
+ queue = conn->cmd_queue_recycle;
+ PQfreeCommandQueue(queue);
+ conn->cmd_queue_recycle = NULL;
+
/* Reset ParameterStatus data, as well as variables deduced from it */
pstatus = conn->pstatus;
while (pstatus != NULL)
diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index eea0237c3a..ff9a6ca253 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"
};
/*
@@ -70,7 +72,10 @@ 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);
+static int pqBatchFlush(PGconn *conn);
/* ----------------
* Space management for PGresult.
@@ -1210,7 +1215,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;
@@ -1233,6 +1238,13 @@ fail:
int
PQsendQuery(PGconn *conn, const char *query)
{
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("cannot PQsendQuery in batch mode, use PQsendQueryParams\n"));
+ return false;
+ }
+
if (!PQsendQueryStart(conn))
return 0;
@@ -1331,6 +1343,10 @@ 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;
@@ -1389,31 +1405,51 @@ PQsendPrepare(PGconn *conn,
goto sendFailed;
/* construct the Sync message */
- if (pqPutMsgStart('S', false, conn) < 0 ||
- pqPutMsgEnd(conn) < 0)
- goto sendFailed;
+ if (conn->batch_status == PQBATCH_MODE_OFF)
+ {
+ if (pqPutMsgStart('S', false, conn) < 0 ||
+ pqPutMsgEnd(conn) < 0)
+ goto sendFailed;
+
+ last_query = &conn->last_query;
+ queryclass = &conn->queryclass;
+ }
+ else
+ {
+ pipeCmd = PQmakePipelinedCommand(conn);
+
+ if (pipeCmd == NULL)
+ return 0; /* error msg already set */
+
+ last_query = &pipeCmd->query;
+ queryclass = &pipeCmd->queryclass;
+ }
/* 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
* to send it all; PQgetResult() will do any additional flushing needed.
*/
- if (pqFlush(conn) < 0)
+ if (pqBatchFlush(conn) < 0)
goto sendFailed;
/* OK, it's launched! */
- conn->asyncStatus = PGASYNC_BUSY;
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ PQappendPipelinedCommand(conn, pipeCmd);
+ else
+ conn->asyncStatus = PGASYNC_BUSY;
return 1;
sendFailed:
+ PQrecyclePipelinedCommand(conn, pipeCmd);
/* error message should be set up already */
return 0;
}
@@ -1461,7 +1497,80 @@ PQsendQueryPrepared(PGconn *conn,
}
/*
- * Common startup code for PQsendQuery and sibling routines
+ * PQmakePipelinedCommand
+ * 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;
+}
+
+/*
+ * PQappendPipelinedCommand
+ * 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;
+}
+
+/*
+ * PQrecyclePipelinedCommand
+ * Push a command queue entry onto the freelist. It must be an 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, libpq_gettext("tried to recycle non-dangling command queue entry"));
+ abort();
+ }
+ if (entry->query)
+ free(entry->query);
+
+ entry->next = conn->cmd_queue_recycle;
+ conn->cmd_queue_recycle = entry;
+}
+
+/*
+ * PQsendQueryStart
+ * Common startup code for PQsendQuery and sibling routines
*/
static bool
PQsendQueryStart(PGconn *conn)
@@ -1479,20 +1588,60 @@ 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->batch_status == PQBATCH_MODE_OFF)
{
printfPQExpBuffer(&conn->errorMessage,
libpq_gettext("another command is already in progress\n"));
return false;
}
- /* initialize async result-accumulation state */
- pqClearAsyncResult(conn);
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ /*
+ * 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 PQbatchQueueProcess(...)
+ * 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;
+ break;
+ 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.
+ * Initialize async result-accumulation state
+ */
+ pqClearAsyncResult(conn);
- /* reset single-row processing mode */
- conn->singleRowMode = false;
+ /* reset single-row processing mode */
+ conn->singleRowMode = false;
+ }
/* ready to send command message */
return true;
}
@@ -1516,6 +1665,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)
@@ -1525,6 +1678,23 @@ PQsendQueryGuts(PGconn *conn,
return 0;
}
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ 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.
@@ -1637,35 +1807,42 @@ 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->batch_status == PQBATCH_MODE_OFF)
+ {
+ /* 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
* to send it all; PQgetResult() will do any additional flushing needed.
*/
- if (pqFlush(conn) < 0)
+ if (pqBatchFlush(conn) < 0)
goto sendFailed;
/* OK, it's launched! */
- conn->asyncStatus = PGASYNC_BUSY;
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ PQappendPipelinedCommand(conn, pipeCmd);
+ else
+ conn->asyncStatus = PGASYNC_BUSY;
return 1;
sendFailed:
+ PQrecyclePipelinedCommand(conn, pipeCmd);
/* error message should be set up already */
return 0;
}
@@ -1766,14 +1943,308 @@ PQisBusy(PGconn *conn)
return conn->asyncStatus == PGASYNC_BUSY || conn->write_failed;
}
+/*
+ * PQbatchStatus
+ * Returns current batch mode status
+ */
+int
+PQbatchStatus(PGconn *conn)
+{
+ if (!conn)
+ return false;
+
+ return conn->batch_status;
+}
+
+/*
+ * PQenterBatchMode
+ * 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.
+ *
+ * Queuing of a new query or syncing during COPY is not allowed.
+ *
+ * A set of commands is terminated by a PQbatchQueueSync. Multiple sets of batched
+ * commands may be sent while in batch mode. Batch mode can be exited by
+ * calling PQbatchEnd() 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
+PQenterBatchMode(PGconn *conn)
+{
+ if (!conn)
+ return 0;
+
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ return 1;
+
+ if (conn->asyncStatus != PGASYNC_IDLE)
+ return 0;
+
+ conn->batch_status = PQBATCH_MODE_ON;
+ conn->asyncStatus = PGASYNC_QUEUED;
+
+ return 1;
+}
+
+/*
+ * PQexitBatchMode
+ * 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 1 if batch mode ended.
+ */
+int
+PQexitBatchMode(PGconn *conn)
+{
+ if (!conn)
+ goto exitFailed;
+
+ if (conn->batch_status == PQBATCH_MODE_OFF)
+ return 1;
+
+ switch (conn->asyncStatus)
+ {
+ case PGASYNC_READY:
+ case PGASYNC_READY_MORE:
+ case PGASYNC_BUSY:
+ /* can't end batch while busy */
+ goto exitFailed;
+ default:
+ break;
+ }
+
+ /* still work to process */
+ if (conn->cmd_queue_head != NULL)
+ goto exitFailed;
+
+ conn->batch_status = PQBATCH_MODE_OFF;
+ conn->asyncStatus = PGASYNC_IDLE;
+
+ /* Flush any pending data in out buffer */
+ if (pqFlush(conn) < 0)
+ goto sendFailed;
+ return 1;
+
+sendFailed:
+ /* error message should be set up already */
+
+exitFailed:
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext_noop("internal error, Failed to exit batch mode"));
+ return 0;
+}
+
+
+/*
+ * internal function pqBatchProcessQueue
+ *
+ * In batch mode, start processing the next query in the queue.
+ *
+ * Returns 1 if the next query was popped from the queue and can
+ * be processed by PQconsumeInput, PQgetResult, etc.
+ *
+ * Returns 0 if the current query isn't done yet, the connection
+ * is not in a batch, or there are no more queries to process.
+ */
+static int
+pqBatchProcessQueue(PGconn *conn)
+{
+ PGcommandQueueEntry *next_query;
+
+ if (!conn)
+ return 0;
+
+ if (conn->batch_status == PQBATCH_MODE_OFF)
+ return 0;
+
+ 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 0;
+ 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 PQbatchEnd() now.
+ */
+ return 0;
+ }
+
+ /*
+ * 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;
+
+ /* Initialize async result-accumulation state */
+ pqClearAsyncResult(conn);
+
+ /* reset single-row processing mode */
+ conn->singleRowMode = false;
+
+
+ conn->last_query = next_query->query;
+ next_query->query = NULL;
+ conn->queryclass = next_query->queryclass;
+
+ PQrecyclePipelinedCommand(conn, next_query);
+
+ if (conn->batch_status == PQBATCH_MODE_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);
+ return 0;
+ }
+ conn->asyncStatus = PGASYNC_READY;
+ }
+ else
+ {
+ /* allow parsing to continue */
+ conn->asyncStatus = PGASYNC_BUSY;
+ }
+
+ return 1;
+}
+
+/*
+ * PQbatchSendQueue
+ *
+ * End a batch submission.
+ *
+ * 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 PQbatchQueueSync 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.
+ *
+ * Queries can already have been sent before PQbatchSendQueue is called, but
+ * PQbatchSendQueue need to be called before retrieving command results.
+ *
+ * The connection will remain in batch mode and unavailable for new synchronous
+ * command execution functions until all results from the batch are processed
+ * by the client.
+ */
+int
+PQbatchSendQueue(PGconn *conn)
+{
+ PGcommandQueueEntry *entry;
+
+ if (!conn)
+ return 0;
+
+ if (conn->batch_status == PQBATCH_MODE_OFF)
+ return 0;
+
+ switch (conn->asyncStatus)
+ {
+ case PGASYNC_IDLE:
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext_noop("internal error, IDLE in batch mode"));
+ return 0;
+ 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"));
+ return 0;
+ 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);
+ if (entry == NULL)
+ return 0; /* error msg already set */
+
+ 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);
+
+ /*
+ * Give the data a push. In nonblock mode, don't complain if we're unable
+ * to send it all; PQgetResult() will do any additional flushing needed.
+ */
+ if (PQflush(conn) < 0)
+ goto sendFailed;
+
+ /*
+ * Call pqBatchProcessQueue so the user can call start calling getResult.
+ */
+ pqBatchProcessQueue(conn);
+
+ return 1;
+
+sendFailed:
+ PQrecyclePipelinedCommand(conn, entry);
+ /* error message should be set up already */
+ return 0;
+}
/*
* PQgetResult
* Get the next PGresult produced by a query. Returns NULL if no
* query work remains or an error has occurred (e.g. out of
* memory).
+ *
+ * In batch mode, once all the result of a query have been
+ * returned, PQgetResult returns NULL to let the user know that
+ * the next batched query is being processed.
+ *
+ * At the end of the batch, returns a end-of-batch result with
+ * PQresultStatus(result) == PGRES_BATCH_END.
*/
-
PGresult *
PQgetResult(PGconn *conn)
{
@@ -1842,9 +2313,38 @@ PQgetResult(PGconn *conn)
switch (conn->asyncStatus)
{
case PGASYNC_IDLE:
+ case PGASYNC_QUEUED:
res = NULL; /* query is complete */
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ /*
+ * In batch mode, we prepare the processing of the results
+ * of the next query.
+ */
+ pqBatchProcessQueue(conn);
+ }
break;
case PGASYNC_READY:
+ res = pqPrepareAsyncResult(conn);
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ /*
+ * In batch mode, query execution state cannot be IDLE as there
+ * can be other queries or results waiting in the queue
+ *
+ * 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;
@@ -2025,6 +2525,13 @@ PQexecStart(PGconn *conn)
if (!conn)
return false;
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("Synchronous command execution functions are not allowed 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.
@@ -2219,6 +2726,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 = "";
@@ -2234,6 +2744,20 @@ PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target)
return 0;
}
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ 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 ||
@@ -2242,15 +2766,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->batch_status == PQBATCH_MODE_OFF)
+ {
+ 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)
+ /* reset last-query string (not relevant now) */
+ if (conn->last_query && conn->batch_status != PQBATCH_MODE_OFF)
{
free(conn->last_query);
conn->last_query = NULL;
@@ -2260,14 +2787,18 @@ PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target)
* Give the data a push. In nonblock mode, don't complain if we're unable
* to send it all; PQgetResult() will do any additional flushing needed.
*/
- if (pqFlush(conn) < 0)
+ if (pqBatchFlush(conn) < 0)
goto sendFailed;
/* OK, it's launched! */
- conn->asyncStatus = PGASYNC_BUSY;
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ PQappendPipelinedCommand(conn, pipeCmd);
+ else
+ conn->asyncStatus = PGASYNC_BUSY;
return 1;
sendFailed:
+ PQrecyclePipelinedCommand(conn, pipeCmd);
/* error message should be set up already */
return 0;
}
@@ -2665,6 +3196,13 @@ PQfn(PGconn *conn,
/* clear the error string */
resetPQExpBuffer(&conn->errorMessage);
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("Synchronous command execution functions are not allowed in batch mode\n"));
+ return NULL;
+ }
+
if (conn->sock == PGINVALID_SOCKET || conn->asyncStatus != PGASYNC_IDLE ||
conn->result != NULL)
{
@@ -3858,3 +4396,15 @@ PQunescapeBytea(const unsigned char *strtext, size_t *retbuflen)
*retbuflen = buflen;
return tmpbuf;
}
+/* pqBatchFlush
+ * In batch mode, data will be flushed only when the out buffer reaches the threshold value.
+ * In non-batch mode, data will be flushed all the time.
+ */
+static int
+pqBatchFlush(PGconn *conn)
+{
+ if ((conn->batch_status == PQBATCH_MODE_OFF) ||
+ (conn->outCount >= OUTBUFFER_THRESHOLD))
+ return(pqFlush(conn));
+ return 0; /* Just to keep compiler quiet */
+}
diff --git a/src/interfaces/libpq/fe-protocol2.c b/src/interfaces/libpq/fe-protocol2.c
index 9360c541be..2ff3fa4883 100644
--- a/src/interfaces/libpq/fe-protocol2.c
+++ b/src/interfaces/libpq/fe-protocol2.c
@@ -406,6 +406,12 @@ pqParseInput2(PGconn *conn)
{
char id;
+ if (conn->batch_status != PQBATCH_MODE_OFF)
+ {
+ 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 1696525475..9c37a9c4ff 100644
--- a/src/interfaces/libpq/fe-protocol3.c
+++ b/src/interfaces/libpq/fe-protocol3.c
@@ -217,10 +217,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->batch_status != PQBATCH_MODE_OFF)
+ {
+ conn->batch_status = PQBATCH_MODE_ON;
+ 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)
@@ -875,6 +883,9 @@ pqGetErrorNotice3(PGconn *conn, bool isError)
PQExpBufferData workBuf;
char id;
+ if (isError && conn->batch_status != PQBATCH_MODE_OFF)
+ conn->batch_status = PQBATCH_MODE_ABORTED;
+
/*
* If this is an error message, pre-emptively clear any incomplete query
* result we may have. We'd just throw it away below anyway, and
diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h
index 3b6a9fbce3..b6e3cde539 100644
--- a/src/interfaces/libpq/libpq-fe.h
+++ b/src/interfaces/libpq/libpq-fe.h
@@ -15,6 +15,10 @@
#ifndef LIBPQ_FE_H
#define LIBPQ_FE_H
+#ifdef WIN32
+#include
+#endif
+
#ifdef __cplusplus
extern "C"
{
@@ -97,7 +101,10 @@ 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
@@ -137,6 +144,17 @@ typedef enum
PQPING_NO_ATTEMPT /* connection not attempted (bad params) */
} PGPing;
+/*
+ * PQBatchStatus - Current status of batch mode
+ */
+
+typedef enum
+{
+ PQBATCH_MODE_OFF,
+ PQBATCH_MODE_ON,
+ PQBATCH_MODE_ABORTED
+} PQBatchStatus;
+
/* PGconn encapsulates a connection to the backend.
* The contents of this struct are not supposed to be known to applications.
*/
@@ -435,6 +453,12 @@ extern PGresult *PQgetResult(PGconn *conn);
extern int PQisBusy(PGconn *conn);
extern int PQconsumeInput(PGconn *conn);
+/* Routines for batch mode management */
+extern int PQbatchStatus(PGconn *conn);
+extern int PQenterBatchMode(PGconn *conn);
+extern int PQexitBatchMode(PGconn *conn);
+extern int PQbatchSendQueue(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 1de91ae295..dee4c85382 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -217,10 +217,15 @@ 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 +234,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 */
@@ -301,6 +307,22 @@ typedef enum pg_conn_host_type
CHT_UNIX_SOCKET
} pg_conn_host_type;
+/* 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;
+
+
/*
* pg_conn_host stores all information about each of possibly several hosts
* mentioned in the connection string. Most fields are derived by splitting
@@ -394,6 +416,7 @@ struct pg_conn
bool options_valid; /* true if OK to attempt connection */
bool nonblocking; /* whether this connection is using nonblock
* sending semantics */
+ PQBatchStatus batch_status; /* Batch(pipelining) mode status of connection */
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 OUT */
@@ -406,6 +429,16 @@ struct pg_conn
pg_conn_host *connhost; /* details about each named host */
char *connip; /* IP address for current network connection */
+ /*
+ * 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;
+
/* Connection data */
pgsocket sock; /* FD for socket, PGINVALID_SOCKET if
* unconnected */
@@ -798,6 +831,11 @@ extern ssize_t pg_GSS_read(PGconn *conn, void *ptr, size_t len);
*/
#define pqIsnonblocking(conn) ((conn)->nonblocking)
+/*
+ * Connection's outbuffer threshold.
+ */
+#define OUTBUFFER_THRESHOLD 65536
+
#ifdef ENABLE_NLS
extern char *libpq_gettext(const char *msgid) pg_attribute_format_arg(1);
extern char *libpq_ngettext(const char *msgid, const char *msgid_plural, unsigned long n) pg_attribute_format_arg(1) pg_attribute_format_arg(2);
@@ -806,6 +844,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/test/modules/Makefile b/src/test/modules/Makefile
index a6d2ffbf9e..287214c544 100644
--- a/src/test/modules/Makefile
+++ b/src/test/modules/Makefile
@@ -17,6 +17,7 @@ SUBDIRS = \
test_extensions \
test_ginpostinglist \
test_integerset \
+ test_libpq \
test_misc \
test_parser \
test_pg_dump \
diff --git a/src/test/modules/test_libpq/.gitignore b/src/test/modules/test_libpq/.gitignore
new file mode 100644
index 0000000000..11e8463984
--- /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 0000000000..d907063f65
--- /dev/null
+++ b/src/test/modules/test_libpq/Makefile
@@ -0,0 +1,25 @@
+# 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
+
+testlibpqbatch.o: testlibpqbatch.c
+testlibpqbatch: testlibpqbatch.o
+check: testlibpqbatch 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 0000000000..d8174dd579
--- /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 0000000000..2c361880fa
--- /dev/null
+++ b/src/test/modules/test_libpq/t/001_libpq_async.pl
@@ -0,0 +1,27 @@
+use strict;
+use warnings;
+
+use Config;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 6;
+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 singlerowmode);
+$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 0000000000..b82634171b
--- /dev/null
+++ b/src/test/modules/test_libpq/testlibpqbatch.c
@@ -0,0 +1,1003 @@
+/*
+ * src/test/modules/test_libpq/testlibpqbatch.c
+ *
+ *
+ * testlibpqbatch.c
+ * Test of batch execution functionality
+ */
+
+#include
+#include
+#include
+#include
+#include
+#include
+#include "c.h"
+#include "libpq-fe.h"
+#include "portability/instr_time.h"
+
+#define EXPECT(condition, ...) \
+ if (0 == (condition)) \
+ { \
+ fprintf(stderr, __VA_ARGS__); \
+ goto fail; \
+ }
+
+
+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);
+static void test_singlerowmode(PGconn *conn);
+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);";
+
+/* max char length of an int32, plus sign and null terminator */
+#define MAXINTLEN 12
+
+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... ");
+
+ /*
+ * 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.
+ */
+ EXPECT(!PQisnonblocking(conn), "Expected blocking connection mode\n");
+
+ EXPECT(PQenterBatchMode(conn), "failed to enter batch mode: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0),
+ "dispatching SELECT failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(!PQexitBatchMode(conn), "exiting batch mode with work in progress should fail, but succeeded\n");
+
+ EXPECT(PQbatchSendQueue(conn), "Ending a batch failed: %s\n", PQerrorMessage(conn));
+
+ res = PQgetResult(conn);
+ EXPECT(res != NULL, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQresultStatus(res) == PGRES_TUPLES_OK, "Unexpected result code %s from first batch item\n", PQresStatus(PQresultStatus(res)));
+
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "PQgetResult returned something extra after first query result.\n");
+
+ /*
+ * Even though we've processed the result there's still a sync to come and
+ * we can't exit batch mode yet
+ */
+ EXPECT(!PQexitBatchMode(conn), "exiting batch mode after query but before sync succeeded incorrectly\n");
+
+ res = PQgetResult(conn);
+ EXPECT(res != NULL, "PQgetResult returned null when sync result PGRES_BATCH_END expected: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQresultStatus(res) == PGRES_BATCH_END, "Unexpected result code %s instead of sync result, error: %s\n", PQresStatus(PQresultStatus(res)), PQerrorMessage(conn))
+
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "PQgetResult returned something extra after end batch call\n");
+
+ /* We're still in a batch... */
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_OFF, "Fell out of batch mode somehow\n");
+
+ /* until we end it, which we can safely do now */
+ EXPECT(PQexitBatchMode(conn), "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchStatus(conn) == PQBATCH_MODE_OFF, "exiting batch mode didn't seem to work\n");
+
+ 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... ");
+
+ EXPECT(!PQisnonblocking(conn), "Expected blocking connection mode: %u\n", __LINE__);
+
+ EXPECT(PQenterBatchMode(conn), "Unable to enter batch mode\n");
+
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_OFF, "Batch mode not activated properly\n");
+
+ /* PQexec should fail in batch mode */
+ res = PQexec(conn, "SELECT 1");
+ EXPECT(PQresultStatus(res) == PGRES_FATAL_ERROR, "PQexec should fail in batch mode but succeeded\n");
+
+ /* So should PQsendQuery */
+ EXPECT(PQsendQuery(conn, "SELECT 1") == 0, "PQsendQuery should fail in batch mode but succeeded\n");
+
+ /* Entering batch mode when already in batch mode is OK */
+ EXPECT(PQenterBatchMode(conn), "re-entering batch mode should be a no-op but failed\n");
+
+ EXPECT(!PQisBusy(conn), "PQisBusy should return false when idle in batch, returned true\n");
+
+ /* ok, back to normal command mode */
+ EXPECT(PQexitBatchMode(conn), "couldn't exit idle empty batch mode\n");
+
+ EXPECT(PQbatchStatus(conn) == PQBATCH_MODE_OFF, "Batch mode not terminated properly\n");
+
+ /* exiting batch mode when not in batch mode should be a no-op */
+ EXPECT(PQexitBatchMode(conn), "batch mode exit when not in batch mode should succeed but failed\n");
+
+ /* can now PQexec again */
+ res = PQexec(conn, "SELECT 1");
+ EXPECT(PQresultStatus(res) == PGRES_TUPLES_OK, "PQexec should succeed after exiting batch mode but failed with: %s\n", PQerrorMessage(conn));
+
+ 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... ");
+
+ /*
+ * Queue up a couple of small batches and process each without returning
+ * to command mode first.
+ */
+ EXPECT(PQenterBatchMode(conn), "failed to enter batch mode: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0), "dispatching first SELECT failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchSendQueue(conn), "Ending first batch failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0), "dispatching second SELECT failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchSendQueue(conn), "Ending second batch failed: %s\n", PQerrorMessage(conn));
+
+ /* OK, start processing the batch results */
+ res = PQgetResult(conn);
+ EXPECT(res != NULL, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQresultStatus(res) == PGRES_TUPLES_OK, "Unexpected result code %s from first batch item\n", PQresStatus(PQresultStatus(res)));
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "PQgetResult returned something extra after first result\n");
+
+ EXPECT(!PQexitBatchMode(conn), "exiting batch mode after query but before sync succeeded incorrectly\n");
+
+ res = PQgetResult(conn);
+ EXPECT(res != NULL, "PQgetResult returned null when sync result expected: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQresultStatus(res) == PGRES_BATCH_END, "Unexpected result code %s instead of sync result, error: %s (line %u)\n", PQresStatus(PQresultStatus(res)), PQerrorMessage(conn), __LINE__);
+
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ /* second batch */
+
+ res = PQgetResult(conn);
+ EXPECT(res != NULL, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQresultStatus(res) == PGRES_TUPLES_OK, "Unexpected result code %s from second batch item\n", PQresStatus(PQresultStatus(res)));
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ res = PQgetResult(conn);
+ EXPECT(res != NULL, "PQgetResult returned null when there's a batch item: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQresultStatus(res) == PGRES_BATCH_END, "Unexpected result code %s from second end batch\n", PQresStatus(PQresultStatus(res)));
+
+ /* We're still in a batch... */
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_OFF, "Fell out of batch mode somehow\n");
+
+ /* until we end it, which we can safely do now */
+ EXPECT(PQexitBatchMode(conn), "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchStatus(conn) == PQBATCH_MODE_OFF, "exiting batch mode didn't seem to work\n");
+
+ 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... ");
+
+ res = PQexec(conn, drop_table_sql);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "dispatching DROP TABLE failed: %s\n", PQerrorMessage(conn));
+
+ res = PQexec(conn, create_table_sql);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "dispatching CREATE TABLE failed: %s\n", PQerrorMessage(conn));
+
+
+ /*
+ * 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.
+ */
+ EXPECT(PQenterBatchMode(conn), "failed to enter batch mode: %s\n", PQerrorMessage(conn));
+
+ dummy_params[0] = "1";
+ EXPECT(PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0), "dispatching first INSERT failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQsendQueryParams(conn, "SELECT no_such_function($1)", 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0), "dispatching error select failed: %s\n", PQerrorMessage(conn));
+
+ dummy_params[0] = "2";
+ EXPECT(PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0), "dispatching second insert failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchSendQueue(conn), "Ending first batch failed: %s\n", PQerrorMessage(conn));
+
+ dummy_params[0] = "3";
+ EXPECT(PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
+ dummy_params, NULL, NULL, 0), "dispatching second-batch insert failed: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchSendQueue(conn), "Ending second batch failed: %s\n", PQerrorMessage(conn));
+
+ /*
+ * 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.
+ */
+
+ EXPECT(((res = PQgetResult(conn)) != NULL) && PQresultStatus(res) == PGRES_COMMAND_OK,
+ "Unexpected result code %s from first batch item, error='%s'\n",
+ res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)),
+ res == NULL ? PQerrorMessage(conn) : PQresultErrorMessage(res));
+
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ /* second query, caused error */
+
+ EXPECT(((res = PQgetResult(conn)) != NULL) && PQresultStatus(res) == PGRES_FATAL_ERROR,
+ "Unexpected result code from second batch item. Wanted PGRES_FATAL_ERROR, got %s\n",
+ res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)));
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ /*
+ * 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.
+ */
+ EXPECT(PQbatchStatus(conn) == PQBATCH_MODE_ABORTED, "batch should be flagged as aborted but isn't\n");
+
+ /* third query in batch, the second insert */
+
+ EXPECT(((res = PQgetResult(conn)) != NULL) && PQresultStatus(res) == PGRES_BATCH_ABORTED,
+ "Unexpected result code from third batch item. Wanted PGRES_BATCH_ABORTED, got %s\n",
+ res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)));
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ EXPECT(PQbatchStatus(conn) == PQBATCH_MODE_ABORTED, "batch should be flagged as aborted but isn't\n");
+
+ /* We're still in a batch... */
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_OFF, "Fell out of batch mode somehow\n");
+
+ /*
+ * 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.
+ */
+ EXPECT(((res = PQgetResult(conn)) != NULL) && PQresultStatus(res) == PGRES_BATCH_END,
+ "Unexpected result code from first batch sync. Wanted PGRES_BATCH_END, got %s\n",
+ res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)));
+
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_ABORTED, "sync should've cleared the aborted flag but didn't\n");
+
+ /* We're still in a batch... */
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_OFF, "Fell out of batch mode somehow\n");
+
+ /* the insert from the second batch */
+ EXPECT(((res = PQgetResult(conn)) != NULL) && PQresultStatus(res) == PGRES_COMMAND_OK, "Unexpected result code %s from first item in second batch\n", res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)));
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ /* the second batch sync */
+ EXPECT(((res = PQgetResult(conn)) != NULL) && PQresultStatus(res) == PGRES_BATCH_END, "Unexpected result code %s from second batch sync\n", res == NULL ? "NULL" : PQresStatus(PQresultStatus(res)));
+ PQclear(res);
+ res = NULL;
+
+ EXPECT(PQgetResult(conn) == NULL, "Expected null result, got %s\n", PQresStatus(PQresultStatus(res)));
+
+ /* We're still in a batch... */
+ EXPECT(PQbatchStatus(conn) != PQBATCH_MODE_OFF, "Fell out of batch mode somehow\n");
+
+ /* until we end it, which we can safely do now */
+ EXPECT(PQexitBatchMode(conn), "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQbatchStatus(conn) == PQBATCH_MODE_OFF, "exiting batch mode didn't seem to work\n");
+
+ 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");
+
+ EXPECT(PQresultStatus(res) == PGRES_TUPLES_OK, "Expected tuples, got %s: %s", PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
+
+ for (i = 0; i < PQntuples(res); i++)
+ {
+ const char *val = PQgetvalue(res, i, 0);
+
+ EXPECT(strcmp(val, "3") == 0, "expected only insert with value 3, got %s", val);
+ }
+
+ EXPECT(PQntuples(res) == 1, "expected 1 result, got %d", PQntuples(res));
+ PQclear(res);
+
+ return;
+
+fail:
+ PQclear(res);
+ exit_nicely(conn);
+}
+
+
+/* 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
+ */
+ EXPECT(PQenterBatchMode(conn), "failed to enter batch mode: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQsendQueryParams(conn, "BEGIN",
+ 0, NULL, NULL, NULL, NULL, 0), "xact start failed: %s\n", PQerrorMessage(conn));
+
+ fprintf(stdout, "sent BEGIN\n");
+
+ send_step = BI_DROP_TABLE;
+
+ EXPECT(PQsendQueryParams(conn, drop_table_sql,
+ 0, NULL, NULL, NULL, NULL, 0), "dispatching DROP TABLE failed: %s\n", PQerrorMessage(conn));
+
+ fprintf(stdout, "sent DROP\n");
+
+ send_step = BI_CREATE_TABLE;
+
+ EXPECT(PQsendQueryParams(conn, create_table_sql,
+ 0, NULL, NULL, NULL, NULL, 0), "dispatching CREATE TABLE failed: %s\n", PQerrorMessage(conn));
+
+ fprintf(stdout, "sent CREATE\n");
+
+ send_step = BI_PREPARE;
+
+ EXPECT(PQsendPrepare(conn, "my_insert", insert_sql, 1, insert_param_oids), "dispatching PREPARE failed: %s\n", PQerrorMessage(conn));
+
+ fprintf(stdout, "sent PREPARE\n");
+
+ 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.
+ */
+ EXPECT(PQsetnonblocking(conn, 1) == 0, "failed to set nonblocking mode: %s\n", PQerrorMessage(conn));
+
+ 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
+ */
+
+ fprintf(stdout, "next query!\n");
+ 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;
+
+ 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);
+
+ EXPECT(PQresultStatus(res) == status, "%s reported status %s, expected %s. Error msg is [%s]\n",
+ description, PQresStatus(PQresultStatus(res)), PQresStatus(status), PQerrorMessage(conn));
+
+ EXPECT(strncmp(PQcmdStatus(res), cmdtag, strlen(cmdtag)) == 0, "%s expected command tag '%s', got '%s'\n",
+ description, cmdtag, PQcmdStatus(res));
+
+ fprintf(stdout, "Got %s OK\n", cmdtag);
+
+ 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))
+ {
+ fprintf(stdout, "sent row %d\n", rows_to_send);
+
+ 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))
+ {
+ fprintf(stdout, "sent COMMIT\n");
+ send_step = BI_SYNC;
+ }
+ else
+ {
+ fprintf(stderr, "WARNING: failed to send commit: %s\n",
+ PQerrorMessage(conn));
+ }
+ }
+ else if (send_step == BI_SYNC)
+ {
+ if (PQbatchSendQueue(conn))
+ {
+ fprintf(stdout, "Dispatched end batch message\n");
+ 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 */
+ EXPECT(PQexitBatchMode(conn), "attempt to exit batch mode failed when it should've succeeded: %s\n", PQerrorMessage(conn));
+
+ EXPECT(PQsetnonblocking(conn, 0) == 0, "failed to clear nonblocking mode: %s\n", PQerrorMessage(conn));
+
+ 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");
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "BEGIN failed: %s\n", PQerrorMessage(conn));
+ PQclear(res);
+
+ res = PQexec(conn, drop_table_sql);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "DROP TABLE failed: %s\n", PQerrorMessage(conn));
+ PQclear(res);
+
+ res = PQexec(conn, create_table_sql);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "CREATE TABLE failed: %s\n", PQerrorMessage(conn));
+ PQclear(res);
+
+ res = PQprepare(conn, "my_insert2", insert_sql, 1, insert_param_oids);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "prepare failed: %s\n", PQerrorMessage(conn));
+ 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);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "INSERT failed: %s\n", PQerrorMessage(conn));
+
+ PQclear(res);
+ nrows--;
+ }
+
+ res = PQexec(conn, "COMMIT");
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "COMMIT failed: %s\n", PQerrorMessage(conn));
+ 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);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "DROP TABLE failed: %s\n", PQerrorMessage(conn));
+ PQclear(res);
+
+ res = PQexec(conn, create_table_sql);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "CREATE TABLE failed: %s\n", PQerrorMessage(conn));
+ PQclear(res);
+ res = NULL;
+
+ res = PQexec(conn, "COPY batch_demo(itemno) FROM stdin");
+ EXPECT(PQresultStatus(res) == PGRES_COPY_IN, "COPY: %s\n", PQerrorMessage(conn));
+ PQclear(res);
+ res = NULL;
+
+ while (nrows > 0)
+ {
+ char buf[12 + 2];
+ int formatted = snprintf(&buf[0], 12 + 1, "%d\n", nrows);
+
+ EXPECT(formatted < 12 + 1, "Buffer write truncated somehow\n");
+
+ EXPECT(PQputCopyData(conn, buf, formatted) == 1, "Write of COPY data failed: %s\n",
+ PQerrorMessage(conn));
+
+ nrows--;
+ }
+
+ EXPECT(PQputCopyEnd(conn, NULL) == 1, "Finishing COPY failed: %s",
+ PQerrorMessage(conn));
+
+ res = PQgetResult(conn);
+ EXPECT(PQresultStatus(res) == PGRES_COMMAND_OK, "COPY finished with %s: %s\n",
+ PQresStatus(PQresultStatus(res)),
+ PQresultErrorMessage(res));
+ PQclear(res);
+ res = NULL;
+
+ return;
+
+fail:
+ PQclear(res);
+ exit_nicely(conn);
+}
+
+static void
+test_timings(PGconn *conn, int number_of_rows)
+{
+ instr_time start_time, end_time;
+
+ fprintf(stderr, "inserting %d rows batched then unbatched\n", number_of_rows);
+
+ INSTR_TIME_SET_CURRENT(start_time);
+ batch_insert_pipelined(conn, number_of_rows);
+ INSTR_TIME_SET_CURRENT(end_time);
+ INSTR_TIME_SUBTRACT(end_time, start_time);
+
+ printf("batch insert elapsed: %.8f ms\n",
+ INSTR_TIME_GET_MILLISEC(end_time));
+
+ INSTR_TIME_SET_CURRENT(start_time);
+ batch_insert_sequential(conn, number_of_rows);
+ INSTR_TIME_SET_CURRENT(end_time);
+ INSTR_TIME_SUBTRACT(end_time, start_time);
+ printf("sequential insert elapsed: %.8f ms\n",
+ INSTR_TIME_GET_MILLISEC(end_time));
+
+ INSTR_TIME_SET_CURRENT(start_time);
+ batch_insert_copy(conn, number_of_rows);
+ INSTR_TIME_SET_CURRENT(end_time);
+ INSTR_TIME_SUBTRACT(end_time, start_time);
+ printf("COPY elapsed: %.8f ms\n",
+ INSTR_TIME_GET_MILLISEC(end_time));
+
+ 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|singlerowmode\n");
+ exit(1);
+}
+
+static void
+test_singlerowmode(PGconn *conn)
+{
+ PGresult *res;
+ int i,r;
+
+ /* 1 batch, 3 queries in it */
+ r = PQenterBatchMode(conn);
+
+ for (i=0; i < 3; i++) {
+ r = PQsendQueryParams(conn,
+ "SELECT 1",
+ 0,
+ NULL,
+ NULL,
+ NULL,
+ NULL,
+ 0);
+ }
+ PQbatchSendQueue(conn);
+
+ i=0;
+ while (true)
+ {
+ int isSingleTuple = 0;
+ ExecStatusType last_est = 0;
+
+ /* Set single row mode for only first 3 SELECT queries */
+ if(i < 3)
+ {
+ r = PQsetSingleRowMode(conn);
+ if (r!=1)
+ {
+ fprintf(stderr, "PQsetSingleRowMode() failed for i=%d\n", i);
+ }
+ }
+ while ((res = PQgetResult(conn)) != NULL)
+ {
+ ExecStatusType est = PQresultStatus(res);
+ fprintf(stderr, "Result status: %d (%s) for i=%d", est, PQresStatus(est), i);
+ if (est == PGRES_TUPLES_OK)
+ {
+ fprintf(stderr, ", tuples: %d\n", PQntuples(res));
+ EXPECT(isSingleTuple, " Expected to follow PGREG_SINGLE_TUPLE, but received PGRES_TUPLES_OK directly instead\n");
+ isSingleTuple=0;
+ }
+ else if (est == PGRES_SINGLE_TUPLE)
+ {
+ isSingleTuple = 1;
+ fprintf(stderr, ", single tuple: %d\n", PQntuples(res));
+ }
+ else if (est == PGRES_BATCH_END)
+ {
+ fprintf(stderr, ", end of batch reached\n");
+ }
+ else if (est != PGRES_COMMAND_OK)
+ {
+ fprintf(stderr, ", error: %s\n", PQresultErrorMessage(res));
+ }
+ PQclear(res);
+ last_est = est;
+ }
+ i++;
+ if (last_est == PGRES_BATCH_END) break;
+ }
+ PQexitBatchMode(conn);
+ PQclear(res);
+ res = NULL;
+ return;
+fail:
+ PQclear(res);
+ exit_nicely(conn);
+
+}
+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,
+ run_singlerowmode = 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;
+ run_singlerowmode = 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 if (strcmp(argv[3], "singlerowmode") == 0)
+ run_singlerowmode = 1;
+ else
+ {
+ fprintf(stderr, "%s is not a recognized 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);
+
+ if(run_singlerowmode)
+ test_singlerowmode(conn);
+ /* close the connection to the database and cleanup */
+ PQfinish(conn);
+
+ return 0;
+}
diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm
index 20da7985c1..b131afaa1a 100644
--- a/src/tools/msvc/Mkvcbuild.pm
+++ b/src/tools/msvc/Mkvcbuild.pm
@@ -50,7 +50,8 @@ my @contrib_excludes = (
'pgcrypto', 'sepgsql',
'brin', 'test_extensions',
'test_misc', 'test_pg_dump',
- 'snapshot_too_old', 'unsafe_tests');
+ 'snapshot_too_old', 'unsafe_tests',
+ 'test_libpq');
# Set of variables for frontend modules
my $frontend_defines = { 'initdb' => 'FRONTEND' };