Skip to content

Commit eb19d0f

Browse files
mcollinabrianc
andauthored
Add opt-in query pipelining (#3652)
* Add opt-in query pipelining support Allow multiple queries to be sent on the wire before waiting for responses, reducing round-trip latency. Enabled via client.pipelining = true. Each query gets its own Sync boundary so errors are isolated. Tracks in-flight named statements (submittedNamedStatements) to prevent duplicate Parse messages when pipelining queries with the same prepared statement name. Handles error/disconnect cleanup for the sent queue. * Fix pipelining edge cases and add benchmark - Clean up submittedNamedStatements on error in _handleErrorMessage to prevent stale entries from blocking future re-preparation of the same named statement after a parse failure - Guard _pulsePipelinedQueryQueue against non-queryable connections - Fix cancel() and readTimeout for sent queries: removing an already-sent query from _sentQueryQueue corrupts the pipeline response mapping since the server will still respond to it; no-op the callback instead - Add bench-pipelining.js comparing serial vs pipelined throughput * Fix pipelining activation race and add edge-case tests Gate _sentQueryQueue activation on readyForQuery=true inside _pulsePipelinedQueryQueue (and remove the redundant promotion block from _handleReadyForQuery) to eliminate the microtask/macrotask race where the next query could be activated as _activeQuery before the error's ReadyForQuery arrived, causing that RFQ to be handled by the wrong query. Also adds the error-listener fix for the query_timeout integration test so the expected stream-destroy doesn't leak as an unhandled 'error'. * Add pipelining documentation and pool-level integration - New features/pipelining.mdx documenting the opt-in flag - Client and Pool API reference updated - Pool accepts `pipelining: true` and sets it on every client it creates * Fix prettier formatting * Add pipelining support to native client via libpq pipeline mode - pg-native: handle PGRES_PIPELINE_SYNC/PGRES_PIPELINE_ABORTED in _emitResult; add pipeline() batch method using libpq 14+ pipeline mode (enterPipelineMode, pipelineSync, exitPipelineMode) - pg-native: bump libpq dependency to ^1.9.0 (has pipeline bindings) - native client: add _pulsePipelinedQueryQueue that batches all queued queries through pg-native pipeline(), delivering results per-query - native client: suppress queue length deprecation when pipelining * Fix pipeline mode to use extended query protocol and add JS vs native benchmarks Pipeline mode requires sendQueryParams (extended query protocol), not sendQuery (simple query protocol). PostgreSQL rejects PQsendQuery in pipeline mode. Benchmark script now tests all four combinations: JS serial, JS pipelined, native serial, and native pipelined. * Fix native client end() to wait for in-flight pipeline queries The native client end() was immediately terminating the connection, causing "Connection terminated" errors for queries still in the pipeline. Now waits for the drain event before closing when pipelining is active. Also fix pipeline mode to use sendQueryParams instead of sendQuery, since PostgreSQL rejects simple query protocol in pipeline mode. * Fix native pipelining: guard handleError when query.native is unset and skip JS-only tests handleError in native/query.js crashes when this.native is undefined (e.g. query_timeout fires before pipeline callback sets it). Skip named statement cleanup and query_timeout tests for native client since those features rely on JS-specific internals. * Make pipelining a constructor option named 'pipeline' Move from post-construction property (client.pipelining = true) to constructor option (new Client({ pipeline: true })). Same for the pool: new Pool({ pipeline: true }). Renames the property and internal _pipeliningInFlight to _pipelineInFlight. Tests, docs, and benchmarks updated accordingly. --------- Co-authored-by: Brian C <brian.m.carlson@gmail.com>
1 parent c5e8c9a commit eb19d0f

16 files changed

Lines changed: 982 additions & 19 deletions

File tree

docs/pages/apis/client.mdx

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,8 @@ type Config = {
2929
idle_in_transaction_session_timeout?: number, // number of milliseconds before terminating any session with an open idle transaction, default is no timeout
3030
client_encoding?: string, // specifies the character set encoding that the database uses for sending data to the client
3131
fallback_application_name?: string, // provide an application name to use if application_name is not set
32-
options?: string // command-line options to be sent to the server
32+
options?: string, // command-line options to be sent to the server
33+
pipeline?: boolean // when true, enables query pipelining. See /features/pipelining for details. Default false.
3334
}
3435
```
3536
@@ -56,6 +57,17 @@ const client = new Client()
5657
await client.connect()
5758
```
5859

60+
## client.pipeline
61+
62+
`client.pipeline: boolean` (read-only)
63+
64+
Whether this client has pipelining enabled. Set via the `pipeline` config option to the `Client` constructor. Defaults to `false`. See [Pipelining](/features/pipelining) for details and examples.
65+
66+
```js
67+
const client = new Client({ pipeline: true })
68+
await client.connect()
69+
```
70+
5971
## client.query
6072

6173
### QueryConfig

docs/pages/apis/pool.mdx

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,11 @@ type Config = {
6969
// If the function throws or returns a promise that rejects, the client is destroyed
7070
// and the error is returned to the caller requesting the connection.
7171
onConnect?: (client: Client) => void | Promise<void>
72+
73+
// When set to true, enables query pipelining on every client the pool creates.
74+
// Pipelined clients send queries to the server without waiting for previous responses.
75+
// Default is false. See /features/pipelining for details.
76+
pipeline?: boolean
7277
}
7378
```
7479

docs/pages/features/_meta.js

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
export default {
22
connecting: 'Connecting',
33
queries: 'Queries',
4+
pipelining: 'Pipelining',
45
pooling: 'Pooling',
56
transactions: 'Transactions',
67
types: 'Data Types',

docs/pages/features/pipelining.mdx

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
---
2+
title: Pipelining
3+
---
4+
5+
import { Alert } from '/components/alert.tsx'
6+
7+
## What is pipelining?
8+
9+
By default node-postgres waits for each query to complete before sending the next one. This means every query pays a full network round-trip of latency. **Query pipelining** sends multiple queries to the server without waiting for responses, and the server processes them in order. Each query still gets its own result (or error), but you avoid the idle time between them.
10+
11+
```
12+
sequential (default) pipelined
13+
───────────────────── ─────────────────────
14+
client ──Parse──▶ server client ──Parse──▶ server
15+
client ◀──Ready── server ──Parse──▶
16+
client ──Parse──▶ server ──Parse──▶
17+
client ◀──Ready── server client ◀──Ready── server
18+
client ──Parse──▶ server client ◀──Ready── server
19+
client ◀──Ready── server client ◀──Ready── server
20+
```
21+
22+
In benchmarks, pipelining typically delivers **2-3x throughput** for batches of simple queries on a local connection, with larger gains over higher-latency links.
23+
24+
## Enabling pipelining
25+
26+
Pipelining is opt-in. Pass `pipeline: true` to the `Client` constructor:
27+
28+
```js
29+
import { Client } from 'pg'
30+
31+
const client = new Client({ pipeline: true })
32+
await client.connect()
33+
34+
const [r1, r2, r3] = await Promise.all([
35+
client.query('SELECT 1 AS num'),
36+
client.query('SELECT 2 AS num'),
37+
client.query('SELECT 3 AS num'),
38+
])
39+
40+
console.log(r1.rows[0].num, r2.rows[0].num, r3.rows[0].num) // 1 2 3
41+
42+
await client.end()
43+
```
44+
45+
All query types work with pipelining: plain text, parameterized, and named prepared statements.
46+
47+
## Pipelining with a pool
48+
49+
Pass `pipeline: true` in the pool config to enable it on every client the pool creates:
50+
51+
```js
52+
import { Pool } from 'pg'
53+
54+
const pool = new Pool({ pipeline: true })
55+
56+
const client = await pool.connect()
57+
// client.pipeline is already true
58+
59+
const [users, orders] = await Promise.all([
60+
client.query('SELECT * FROM users WHERE id = $1', [1]),
61+
client.query('SELECT * FROM orders WHERE user_id = $1', [1]),
62+
])
63+
64+
client.release()
65+
```
66+
67+
<Alert>
68+
<div>
69+
<code>pool.query()</code> checks out a client for a single query and releases it immediately, so pipelining has no effect there. Use <code>pool.connect()</code> to check out a client and send multiple queries on it.
70+
</div>
71+
</Alert>
72+
73+
## Error isolation
74+
75+
Each pipelined query gets its own error boundary. A failing query in the middle of a batch does not break the other queries:
76+
77+
```js
78+
const results = await Promise.allSettled([
79+
client.query('SELECT 1 AS num'),
80+
client.query('SELECT INVALID SYNTAX'),
81+
client.query('SELECT 3 AS num'),
82+
])
83+
84+
console.log(results[0].status) // 'fulfilled'
85+
console.log(results[1].status) // 'rejected'
86+
console.log(results[2].status) // 'fulfilled'
87+
```
88+
89+
This works because node-postgres sends a `Sync` message after each query, which is how PostgreSQL delimits error boundaries in the extended query protocol.
90+
91+
## Named prepared statements
92+
93+
Named prepared statements work with pipelining. When two pipelined queries share the same statement name, node-postgres sends `Parse` only once and reuses the prepared statement for subsequent queries:
94+
95+
```js
96+
const queries = Array.from({ length: 100 }, (_, i) => ({
97+
name: 'get-user',
98+
text: 'SELECT * FROM users WHERE id = $1',
99+
values: [i],
100+
}))
101+
102+
const results = await Promise.all(queries.map(q => client.query(q)))
103+
```
104+
105+
## Graceful shutdown
106+
107+
Calling `client.end()` while pipelined queries are in flight will wait for all of them to complete before closing the connection:
108+
109+
```js
110+
const client = new Client({ pipeline: true })
111+
await client.connect()
112+
113+
const p1 = client.query('SELECT 1')
114+
const p2 = client.query('SELECT 2')
115+
const endPromise = client.end()
116+
117+
// Both queries will resolve normally
118+
const [r1, r2] = await Promise.all([p1, p2])
119+
await endPromise
120+
```
121+
122+
## When to use pipelining
123+
124+
Pipelining is most useful when you have multiple **independent** queries that don't depend on each other's results. Common use cases:
125+
126+
- Fetching data from multiple tables in parallel for a page load
127+
- Inserting or updating multiple rows simultaneously
128+
- Running a batch of analytics queries
129+
130+
<div className="alert alert-warning">
131+
Do not use pipelining inside a transaction if you need to read the result of one query before issuing the next. Pipelined queries are all sent before any responses arrive, so you cannot branch on intermediate results. For dependent queries within a transaction, use sequential <code>await</code> calls instead.
132+
</div>

packages/pg-native/index.js

Lines changed: 156 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -199,6 +199,10 @@ Client.prototype._emitResult = function (pq) {
199199
break
200200
}
201201

202+
case 'PGRES_PIPELINE_SYNC':
203+
case 'PGRES_PIPELINE_ABORTED':
204+
break
205+
202206
default:
203207
this._readError('unrecognized command status: ' + status)
204208
break
@@ -314,6 +318,158 @@ Client.prototype._onResult = function (result) {
314318
this._resultCount++
315319
}
316320

321+
// Send a batch of queries in pipeline mode and collect results in order.
322+
// Each entry in `queries` is {text, values?, name?}.
323+
// `cb(err, results)` where results is an array, one per query,
324+
// of {err, rows, result} objects.
325+
Client.prototype.pipeline = function (queries, cb) {
326+
const pq = this.pq
327+
328+
if (!pq.pipelineModeSupported || !pq.pipelineModeSupported()) {
329+
return cb(new Error('Pipeline mode is not supported. Requires PostgreSQL 14+ client libraries.'))
330+
}
331+
332+
if (!pq.enterPipelineMode()) {
333+
return cb(new Error(pq.errorMessage() || 'Failed to enter pipeline mode'))
334+
}
335+
336+
pq.setNonBlocking(true)
337+
338+
// Send all queries, each followed by a sync
339+
for (let i = 0; i < queries.length; i++) {
340+
const q = queries[i]
341+
let sent
342+
if (q.name) {
343+
if (q._alreadyPrepared) {
344+
sent = pq.sendQueryPrepared(q.name, q.values || [])
345+
} else {
346+
// send prepare then execute in same pipeline batch
347+
sent = pq.sendPrepare(q.name, q.text, (q.values || []).length)
348+
if (sent) {
349+
sent = pq.sendQueryPrepared(q.name, q.values || [])
350+
}
351+
}
352+
} else {
353+
// In pipeline mode, simple query protocol (sendQuery) is not allowed.
354+
// Always use extended query protocol (sendQueryParams).
355+
sent = pq.sendQueryParams(q.text, q.values || [])
356+
}
357+
358+
if (!sent) {
359+
const err = new Error(pq.errorMessage() || 'Failed to send pipelined query')
360+
pq.exitPipelineMode()
361+
return cb(err)
362+
}
363+
364+
pq.pipelineSync()
365+
}
366+
367+
// Flush all queued data to the socket
368+
this._waitForDrain(pq, (err) => {
369+
if (err) {
370+
pq.exitPipelineMode()
371+
return cb(err)
372+
}
373+
this._readPipelineResults(queries, cb)
374+
})
375+
}
376+
377+
// Read pipeline results for `queries.length` sync points.
378+
// Calls cb(null, results) when all syncs have been received.
379+
Client.prototype._readPipelineResults = function (queries, cb) {
380+
const pq = this.pq
381+
const self = this
382+
const results = []
383+
let queryIndex = 0
384+
let currentResult = null
385+
let currentError = null
386+
387+
const processResults = function () {
388+
if (!pq.consumeInput()) {
389+
pq.exitPipelineMode()
390+
return cb(new Error(pq.errorMessage() || 'Failed to consume input'))
391+
}
392+
393+
while (!pq.isBusy()) {
394+
if (!pq.getResult()) {
395+
// null between result groups in pipeline — try again
396+
if (pq.isBusy()) return // more data needed
397+
if (!pq.getResult()) {
398+
// truly no more results — should not happen before all syncs
399+
break
400+
}
401+
}
402+
403+
const status = pq.resultStatus()
404+
405+
if (status === 'PGRES_PIPELINE_SYNC') {
406+
// End of one query's results + sync
407+
if (currentError) {
408+
results.push({ err: currentError, rows: null, result: null })
409+
} else if (currentResult) {
410+
results.push({ err: null, rows: currentResult.rows, result: currentResult })
411+
} else {
412+
results.push({ err: null, rows: [], result: null })
413+
}
414+
currentResult = null
415+
currentError = null
416+
queryIndex++
417+
418+
if (queryIndex >= queries.length) {
419+
// All queries processed
420+
pq.exitPipelineMode()
421+
return cb(null, results)
422+
}
423+
continue
424+
}
425+
426+
if (status === 'PGRES_FATAL_ERROR') {
427+
currentError = new Error(pq.resultErrorMessage())
428+
// Extract error fields
429+
const fields = pq.resultErrorFields()
430+
if (fields) {
431+
for (const key in fields) {
432+
currentError[key] = fields[key]
433+
}
434+
}
435+
continue
436+
}
437+
438+
if (status === 'PGRES_PIPELINE_ABORTED') {
439+
// Query skipped due to previous error in same sync group
440+
continue
441+
}
442+
443+
if (status === 'PGRES_TUPLES_OK' || status === 'PGRES_COMMAND_OK' || status === 'PGRES_EMPTY_QUERY') {
444+
currentResult = self._consumeQueryResults(pq)
445+
continue
446+
}
447+
}
448+
449+
// Still waiting for more data — will be called again when readable
450+
}
451+
452+
// Use the libuv readable watcher
453+
this._stopReading()
454+
let done = false
455+
const origCb = cb
456+
cb = function (err, results) {
457+
if (done) return
458+
done = true
459+
pq.removeListener('readable', onReadable)
460+
self._stopReading()
461+
origCb(err, results)
462+
}
463+
const onReadable = function () {
464+
processResults()
465+
}
466+
pq.on('readable', onReadable)
467+
pq.startReader()
468+
469+
// Try an initial read in case data is already available
470+
processResults()
471+
}
472+
317473
Client.prototype._onReadyForQuery = function () {
318474
// remove instance callback
319475
const cb = this._queryCallback

packages/pg-pool/test/index.js

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,33 @@ describe('pool', function () {
203203
})
204204
})
205205

206+
it('enables pipeline on clients when configured', async function () {
207+
const pool = new Pool({ pipeline: true })
208+
const client = await pool.connect()
209+
expect(client.pipeline).to.be(true)
210+
211+
const [r1, r2, r3] = await Promise.all([
212+
client.query('SELECT 1 AS num'),
213+
client.query('SELECT 2 AS num'),
214+
client.query('SELECT 3 AS num'),
215+
])
216+
217+
expect(r1.rows[0].num).to.eql(1)
218+
expect(r2.rows[0].num).to.eql(2)
219+
expect(r3.rows[0].num).to.eql(3)
220+
221+
client.release()
222+
return pool.end()
223+
})
224+
225+
it('does not enable pipeline by default', async function () {
226+
const pool = new Pool()
227+
const client = await pool.connect()
228+
expect(client.pipeline).to.be(false)
229+
client.release()
230+
return pool.end()
231+
})
232+
206233
it('recovers from query errors', function () {
207234
const pool = new Pool()
208235

0 commit comments

Comments
 (0)