Network latency is a known drag on database throughput, and PostgreSQL 14 added a feature aimed squarely at that problem: pipeline mode, exposed through the libpq C API. The mechanism itself is not new at the protocol level, but the client-side support is what makes it practical to use.
What pipeline mode does
The extended query protocol works through a sequence of messages. A statement is handled as:
- a "Parse" message carrying the statement
- a "Bind" message carrying parameter values
- an "Execute" message requesting results
A "Sync" message closes the transaction. A call to PQexec or PQexecParams produces all of these, normally packed into one TCP packet; with a prepared statement, "Parse" is separated from "Bind" and "Execute". The client then blocks until the server answers with "ParseComplete", "BindComplete", "Data" or "NoData" depending on the statement type, "CommandComplete", and finally "ReadyForQuery", again typically in a single TCP packet.
Pipeline mode exploits the fact that nothing prevents a client from issuing several statements before sending "Sync". The server did not need to change: what PostgreSQL v14 introduced is libpq's ability to drive that flow. The new functions are:
PQenterPipelineMode— enter pipeline modePQsendFlushRequest— send a "Flush" message so the server starts returning responses to earlier requests immediately, instead of bundling them into one TCP packetPQpipelineSync— send "Sync"; this must be called explicitlyPQexitPipelineMode— leave pipeline modePQpipelineStatus— report whether libpq is currently in pipeline mode
Statements go out through the asynchronous functions PQsendQuery, PQsendQueryParams and PQsendQueryPrepared, and results come back through repeated PQgetResult calls after the sync. Because no frontend/backend protocol feature is involved, pipeline mode also works against older PostgreSQL servers.
Round trips in a money transfer
Take a transfer between two accounts:
|
1 2 3 4 5 |
CREATE TABLE account ( id bigint PRIMARY KEY, holder text NOT NULL, amount numeric(15,2) NOT NULL ); |
The transaction moves funds with:
|
1 2 3 4 |
BEGIN; UPDATE account SET amount = amount + 100 WHERE id = 42; UPDATE account SET amount = amount - 100 WHERE id = 314; COMMIT; |
Processed conventionally, that is four client-to-server round trips, so the transaction pays eight times the network latency.
Pipelining reduces the cost to twice the latency:
- the second
UPDATEcan be sent right after the first, without waiting - no explicit transaction is needed, because a pipeline is itself a single transaction
Driver code
The C snippet below runs the same transaction, using a prepared statement named stmt for the UPDATE. Connection setup and statement preparation are left out for brevity.
|
1 2 3 4 |
UPDATE account SET amount = amount + $2 WHERE id = $1 RETURNING amount; |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 |
#include <libpq-fe.h> #include <stdio.h> /* * Receive and check a statement result. * If 'res' is NULL, we expect a NULL result and * print the message if we get anything else. * If 'res' is not NULL, the result is stored there. * In that case, if the result status is different * from 'expected_status', print the message. */ static int checkResult(PGconn *conn, PGresult **res, ExecStatusType expected_status, char * const message) { PGresult *r; if (res == NULL) { if ((r = PQgetResult(conn)) == NULL) return 0; PQclear(r); fprintf(stderr, '%s: unexpected resultn', message); return 1; } if ((*res = PQgetResult(conn)) == NULL) { fprintf(stderr, '%s: missing resultn', message); return 1; } if (PQresultStatus(*res) == expected_status) return 0; fprintf(stderr, '%s: %sn', message, PQresultErrorMessage(*res)); PQclear(*res); return 1; } /* transfer 'amount' from 'from_acct' to 'to_acct' */ static int transfer(PGconn *conn, int from_acct, int to_acct, double amount) { PGresult *res; int rc; char acct[100], amt[100]; /* will fit a number */ char * const values[] = { acct, amt }; /* parameters */ /* enter pipeline mode */ if (!PQenterPipelineMode(conn)) { fprintf(stderr, 'Cannot enter pipeline mode: %sn', PQerrorMessage(conn)); return 1; } /* send query to subtract amount from the first account */ snprintf(values[0], 100, '%d', from_acct); snprintf(values[1], 100, '%.2f', -amount); if (!PQsendQueryPrepared(conn, 'stmt', /* statement name */ 2, /* parameter count */ (const char * const *) values, NULL, /* parameter lengths */ NULL, /* text parameters */ 0)) /* text result */ { fprintf(stderr, 'Error queuing first update: %sn', PQerrorMessage(conn)); rc = 1; } /* * Tell the server that it should start returning results * right now rather than wait and gather the results for * the whole pipeline in a single packet. * There is no great benefit for short statements like these, * but it can reduce the time until we get the first result. */ if (rc == 0 && PQsendFlushRequest(conn) == 0) { fprintf(stderr, 'Error queuing flush requestn'); rc = 1; } /* * Dispatch pipelined commands to the server. * There is no great benefit for short statements like these, * but it can reduce the time until we get the first result. */ if (rc == 0 && PQflush(conn) == -1) { fprintf(stderr, 'Error flushing data to the server: %sn', PQerrorMessage(conn)); rc = 1; } /* send query to add amount to the second account */ snprintf(values[0], 100, '%d', to_acct); snprintf(values[1], 100, '%.2f', amount); if (rc == 0 && !PQsendQueryPrepared(conn, 'stmt', /* statement name */ 2, /* parameter count */ (const char * const *) values, NULL, /* parameter lengths */ NULL, /* text parameters */ 0)) /* text result */ { fprintf(stderr, 'Error queuing second update: %sn', PQerrorMessage(conn)); rc = 1; } /*--- * Send a 'sync' request: * - flush the remaining statements * - end the transaction * - wait for results */ if (PQpipelineSync(conn) == 0) { fprintf(stderr, 'Error sending 'sync' request: %sn', PQerrorMessage(conn)); rc = 1; } /* consume the first statement result */ if (checkResult(conn, &res, PGRES_TUPLES_OK, 'first update')) rc = 1; else printf('Account %d now has %sn', from_acct, PQgetvalue(res, 0, 0)); if (res != NULL) PQclear(res); /* the next call must return nothing */ if (checkResult(conn, NULL, -1, 'end of first result set')) rc = 1; /* consume the second statement result */ if (checkResult(conn, &res, PGRES_TUPLES_OK, 'second update')) rc = 1; else printf('Account %d now has %sn', to_acct, PQgetvalue(res, 0, 0)); if (res != NULL) PQclear(res); /* the next call must return nothing */ if (checkResult(conn, NULL, -1, 'end of second result set')) rc = 1; /* consume the 'ReadyForQuery' response */ if (checkResult(conn, &res, PGRES_PIPELINE_SYNC, 'sync result')) rc = 1; else if (res != NULL) PQclear(res); /* exit pipeline mode */ if (PQexitPipelineMode(conn) == 0) { fprintf(stderr, 'error ending pipeline mode: %sn', PQresultErrorMessage(res)); rc = 1; } return rc; } |
Quantifying the gain
To check the improvement, the tc utility was used on Linux to add 50 milliseconds of latency to the loopback interface:
|
1 |
sudo tc qdisc add dev lo root netem delay 50ms |
It is reset with:
|
1 |
sudo tc qdisc del dev lo root netem |
The timing run compares the pipeline function against an equivalent function using an explicit transaction without pipelining:
| no pipeline (8 times network latency) | pipeline (2 times network latency) | |
|---|---|---|
| first attempt | 406 ms | 111 ms |
| second attempt | 414 ms | 104 ms |
| third attempt | 414 ms | 103 ms |
For short SQL statements such as these, the speedup is roughly proportional to the number of round trips eliminated.
Doing it in the database instead
There is a second route to comparable performance that avoids libpq and the protocol entirely: a PL/pgSQL function or procedure.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
CREATE PROCEDURE transfer( p_from_acct bigint, p_to_acct bigint, p_amount numeric ) LANGUAGE plpgsql AS $$BEGIN UPDATE account SET amount = amount - p_amount WHERE id = p_from_acct; UPDATE account SET amount = amount + p_amount WHERE id = p_to_acct; END;$$; |
This runs in a single transaction and is just as fast, for two reasons:
- the
CALLcosts a single client-server round trip - PL/pgSQL caches execution plans for the SQL statements it contains
In this scenario the PL/pgSQL procedure is probably the simpler answer, though pipeline mode gives precise control over message and data flow that a function does not expose. The one situation where the procedural route is unavailable is an environment where database functions are categorically rejected.
Takeaway
Pipeline mode arrived in the libpq C API with PostgreSQL v14 and delivers substantial speedups on laggy connections. It relies on no new frontend/backend protocol features, so old server versions are supported. Where functions are acceptable, PL/pgSQL often yields similar gains.
