feat: pipeline

[cesco69]

This PR implements pipeline mode, allowing multiple queries to be sent to the PostgreSQL server without waiting for the results of previous queries. This significantly reduces network latency, especially for batch operations or high-latency connections.

Usage

Enabling Pipeline Mode

Pipeline mode is opt-in and can be enabled when creating a client or pool:

// With Client
const client = new Client({ pipelineMode: true })
await client.connect()

// With Pool - all clients will have pipeline mode enabled
const pool = new Pool({ pipelineMode: true })

When pipeline mode is enabled, you can submit multiple queries and they will be sent to the server immediately:

// All three queries are sent immediately to the server
const [result1, result2, result3] = await Promise.all([
  client.query('SELECT 1 as num'),
  client.query('SELECT 2 as num'),
  client.query('SELECT 3 as num'),
])

Backpressure

When submitting a large number of queries in pipeline mode, you may want to limit how many queries are pending at once to prevent memory issues. The client supports configurable backpressure:

const client = new Client({
  pipelineMode: true,
  pipelineMaxQueries: 100, // default is 1000
})

When the number of pending queries reaches pipelineMaxQueries, new queries are queued internally and will be sent once space becomes available. The client emits events to help you monitor backpressure:

client.on('pipelineFull', () => {
  console.log('Pipeline is full, new queries will wait')
})

client.on('pipelineDrain', () => {
  console.log('Pipeline has drained, accepting new queries')
})

You can also check the current pipeline depth:

console.log(client.pendingQueryCount) // number of queries waiting for results

The pipelineDrain event is emitted when the pending query count drops below 75% of pipelineMaxQueries (the "low water mark").

Error Isolation

If one query fails in pipeline mode, other queries continue to execute:

const results = await Promise.allSettled([
  client.query('SELECT 1 as num'),
  client.query('SELECT * FROM nonexistent_table'), // This will fail
  client.query('SELECT 3 as num'),
])

console.log(results[0].status) // 'fulfilled'
console.log(results[1].status) // 'rejected'
console.log(results[2].status) // 'fulfilled' - still succeeds!

Prepared Statements in Pipeline Mode

Prepared statements work in pipeline mode, including concurrent queries with the same statement name:

const results = await Promise.all([
  client.query({ name: 'get-user', text: 'SELECT $1::int as id', values: [1] }),
  client.query({ name: 'get-user', text: 'SELECT $1::int as id', values: [2] }),
  client.query({ name: 'get-user', text: 'SELECT $1::int as id', values: [3] }),
])

Restrictions and Limitations

• No COPY operations: COPY commands are not supported in pipeline mode
• No pg-cursor/pg-query-stream: Custom submittables are not supported in pipeline mode
• JavaScript client only: Pipeline mode is not available in pg-native, would require:
  • feat: pipeline node-libpq#118
  • feat(pg-native): pipeline (WIP) #3592)

Note: Multi-statement queries (e.g., SELECT 1; SELECT 2) are rejected by PostgreSQL when using the extended query protocol, which pipeline mode uses. This is a PostgreSQL limitation.

Transaction Behavior

Errors inside a transaction will abort subsequent queries until ROLLBACK. This is standard PostgreSQL behavior. If you send COMMIT on an aborted transaction, PostgreSQL automatically converts it to ROLLBACK:

// In a transaction, if one query fails, subsequent queries also fail
await client.query('BEGIN')
await client.query('INSERT INTO t VALUES (1)')
await client.query('INSERT INTO t VALUES (1)') // fails: duplicate key
await client.query('INSERT INTO t VALUES (2)') // also fails: transaction aborted
await client.query('COMMIT') // returns ROLLBACK command

When to Use Pipeline Mode

Pipeline mode is most beneficial when:

• You have many independent queries to execute
• Network latency is high (cloud databases, cross-region connections)
• You're doing batch operations

Related Issues

• Closes implement pipelining mode allowed by libpq 14 #2646
• Closes Extended Query: Support Batch Execution #2257
• Closes Implement Batch query package #2942

Benchmark

I have also updated the benchmark b189ed5

Query Type	Standard QPS	Pipeline QPS	Improvement
param	374	440	+18%
named	380	440	+16%
sequence	1149	1680	+46%
insert	5623	11520	+105%

Raw Output

node bench.js
============================================================
STANDARD MODE BENCHMARK
============================================================
warmup done

param queries: 1868
qps: 374

named queries: 1899
qps: 380

sequence queries: 5745
qps: 1149

insert queries: 28115
qps: 5623


============================================================
PIPELINE MODE BENCHMARK
============================================================
warmup done

param queries: 2200
qps: 440

named queries: 2200
qps: 440

sequence queries: 8400
qps: 1680

insert queries: 57600
qps: 11520


============================================================
COMPARISON: Sequential vs Pipeline (same workload)
============================================================

Running 1000 sequential queries...
Sequential: 181ms (5538 qps)
Running 1000 pipeline queries...
Pipeline:   59ms (16842 qps)

[stephenh]

Ah wow, this is amazing!

@cesco69 just curious, in your example:

// All queries are sent immediately without waiting
const [r1, r2, r3] = await Promise.all([
  client.query('SELECT 1'),
  client.query('SELECT 2'),
  client.query('SELECT 3'),
])

Afaiu if SELECT 2 breaks (invalid SQL query, an UPDATE that fails a constraint), then SELECT 3 is aborted too.

...except that you have a test showing this is not the case (nonexistent_table_xyz). 🤔

I really thought the pg docs said that SELECT 3 as num would blow up, but I guess reading again they say "If any statement encounters an error, the server aborts the current transaction and does not execute any subsequent command in the queue until the next synchronization point".

So I guess if there is no open transaction, everything is fine. 👍

Speaking of transactions, I see you have a test for it, but does the BEGIN cause the current connection to be reserved to the current caller/request? It doesn't seem like it?

Like if I have two in-flight requests, I wouldn't want user1 issuing a BEGIN to then accept pipelined statements from user2.

Or I guess that's just on me to use a Pool, and only BEGIN on a reserved-to-that-user connection ... that makes sense/seems obvious in retrospect.

This is really great! 🎉

[cesco69]

Ah wow, this is amazing!

@cesco69 just curious, in your example:

// All queries are sent immediately without waiting
const [r1, r2, r3] = await Promise.all([
  client.query('SELECT 1'),
  client.query('SELECT 2'),
  client.query('SELECT 3'),
])

Afaiu if SELECT 2 breaks (invalid SQL query, an UPDATE that fails a constraint), then SELECT 3 is aborted too.

...except that you have a test showing this is not the case (nonexistent_table_xyz). 🤔

I really thought the pg docs said that SELECT 3 as num would blow up, but I guess reading again they say "If any statement encounters an error, the server aborts the current transaction and does not execute any subsequent command in the queue until the next synchronization point".

So I guess if there is no open transaction, everything is fine. 👍

Speaking of transactions, I see you have a test for it, but does the BEGIN cause the current connection to be reserved to the current caller/request? It doesn't seem like it?

Like if I have two in-flight requests, I wouldn't want user1 issuing a BEGIN to then accept pipelined statements from user2.

Or I guess that's just on me to use a Pool, and only BEGIN on a reserved-to-that-user connection ... that makes sense/seems obvious in retrospect.

This is really great! 🎉

Thanks! 🙏

You're right about error isolation, each query gets its own sync message, so errors don't cascade outside of transactions. The postgres docs refer to aborting "the current transaction", but in autocommit mode (no explicit BEGIN), each statement is its own transaction.

And yes, for transactions you'd use pool.connect() to get a dedicated client, same as without pipeline mode. Pipeline mode doesn't change connection ownership semantics, it just optimizes the wire protocol for that connection.

I have add more tests!

[charmander]

You or @nigrosimone? And how much here is piped directly to and from AI? :/

[brianc]

I'll give this a proper & deep look tomorrow. One thing I did want to add now though before going through it, as it might inform your refactoring or touchups. I'm going to remove the queryQueue entirely pretty soon (like in a month or two) - one of the next things I plan on doing, early next week, is adding a deprecation warning if you queue a query while another query is in flight. In practice I can't see the query queue being used in "normal" code w/ async/await. I know in my own code I haven't used it in over a decade. It was a terrible decision from long ago when async patterns in node were still in their infancy. So...please don't base any design decisions here (or really anywhere, ever 🙃 ) on the query queue implementation!

[nigrosimone]

You or @nigrosimone?

Plot twist! We are the same user! :D

And how much here is piped directly to and from AI? :/

I ran into issues I couldn’t solve on my own and got help from models like Opus 4.5 and ChatGPT 5. I’d say about 60% of the code is Ai.

In the past I have tried more time to implement pipeline:

• feat: pipeline #3590
• feat: pipeline mode #3534
• POC/WIP: pipeline mode #3517

This time I decided to give it a try using AI (I’m not even a big fan myself), and I have to admit it was a huge help.

[charmander]

A twist indeed! I assume the PR description is also AI, but how about the latest comments and revisions? Some of them seemed conflicting and didn’t entirely make sense. It’s helpful to have context on this so I know how likely it is that given parts of the code exist for subtle reasons vs. being based on hallucinations, and whether my review suggestions are at risk of being applied with more bias than usual towards agreement when they might actually be wrong.

And yes, although this could go in before the query queue is removed (if it’s important that it makes it into pg 8.x…?), it might be a lot simpler without having the branching paths associated with the query queue.

[cesco69]

That’s a fair concern.

I want to clarify how I worked on this PR. I already contributed to pg in the past and I know the project reasonably well, but pipeline mode was something I never managed to implement before, because of the complexity and the many edge cases.

I did use AI, but not to just “generate the solution”. I started by writing a specification of what I wanted to achieve, listing the problematic cases that needed to be handled (error propagation, transactions, etc.), also including approaches I tried in the past and some ideas taken from other clients like Postgres.js and other old PR.

From there, I used AI to help me break this into a plan with many small tasks. Then I went through those tasks one by one and adding tests. The process was iterative and quite long, not a one-pass implementation. I was careful not to apply review suggestions automatically: I kept changes only when they made sense or test results. If some parts of the code look strange, it’s because I corrected earlier assumptions when edge cases became clearer, not because I blindly followed generated output.

About the query queue: I agree that pipeline mode would be simpler once the queue is removed. My goal here was to make it work with the current architecture.

If the goal in the future is to remove the queryQueue, this PR not make much sense, because it adds a lot of complexity and will make it harder to remove it later.
Also, without the queryQueue (and possibly without having to support callbacks, but only async/await promises), implementing pipeline mode would likely be much simpler.

Feel free to close the PR if you don't think it's valid!