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

Author: mcollina

packages/pg-native/index.js

@@ -191,6 +191,10 @@ Client.prototype._emitResult = function (pq) {
       break
     }
 
+    case 'PGRES_PIPELINE_SYNC':
+    case 'PGRES_PIPELINE_ABORTED':
+      break
+
     default:
       this._readError('unrecognized command status: ' + status)
       break
@@ -306,6 +310,158 @@ Client.prototype._onResult = function (result) {
   this._resultCount++
 }
 
+// Send a batch of queries in pipeline mode and collect results in order.
+// Each entry in `queries` is {text, values?, name?}.
+// `cb(err, results)` where results is an array, one per query,
+// of {err, rows, result} objects.
+Client.prototype.pipeline = function (queries, cb) {
+  const pq = this.pq
+
+  if (!pq.pipelineModeSupported || !pq.pipelineModeSupported()) {
+    return cb(new Error('Pipeline mode is not supported. Requires PostgreSQL 14+ client libraries.'))
+  }
+
+  if (!pq.enterPipelineMode()) {
+    return cb(new Error(pq.errorMessage() || 'Failed to enter pipeline mode'))
+  }
+
+  pq.setNonBlocking(true)
+
+  // Send all queries, each followed by a sync
+  for (let i = 0; i < queries.length; i++) {
+    const q = queries[i]
+    let sent
+    if (q.name) {
+      if (q._alreadyPrepared) {
+        sent = pq.sendQueryPrepared(q.name, q.values || [])
+      } else {
+        // send prepare then execute in same pipeline batch
+        sent = pq.sendPrepare(q.name, q.text, (q.values || []).length)
+        if (sent) {
+          sent = pq.sendQueryPrepared(q.name, q.values || [])
+        }
+      }
+    } else if (q.values) {
+      sent = pq.sendQueryParams(q.text, q.values)
+    } else {
+      sent = pq.sendQuery(q.text)
+    }
+
+    if (!sent) {
+      const err = new Error(pq.errorMessage() || 'Failed to send pipelined query')
+      pq.exitPipelineMode()
+      return cb(err)
+    }
+
+    pq.pipelineSync()
+  }
+
+  // Flush all queued data to the socket
+  this._waitForDrain(pq, (err) => {
+    if (err) {
+      pq.exitPipelineMode()
+      return cb(err)
+    }
+    this._readPipelineResults(queries, cb)
+  })
+}
+
+// Read pipeline results for `queries.length` sync points.
+// Calls cb(null, results) when all syncs have been received.
+Client.prototype._readPipelineResults = function (queries, cb) {
+  const pq = this.pq
+  const self = this
+  const results = []
+  let queryIndex = 0
+  let currentResult = null
+  let currentError = null
+
+  const processResults = function () {
+    if (!pq.consumeInput()) {
+      pq.exitPipelineMode()
+      return cb(new Error(pq.errorMessage() || 'Failed to consume input'))
+    }
+
+    while (!pq.isBusy()) {
+      if (!pq.getResult()) {
+        // null between result groups in pipeline — try again
+        if (pq.isBusy()) return // more data needed
+        if (!pq.getResult()) {
+          // truly no more results — should not happen before all syncs
+          break
+        }
+      }
+
+      const status = pq.resultStatus()
+
+      if (status === 'PGRES_PIPELINE_SYNC') {
+        // End of one query's results + sync
+        if (currentError) {
+          results.push({ err: currentError, rows: null, result: null })
+        } else if (currentResult) {
+          results.push({ err: null, rows: currentResult.rows, result: currentResult })
+        } else {
+          results.push({ err: null, rows: [], result: null })
+        }
+        currentResult = null
+        currentError = null
+        queryIndex++
+
+        if (queryIndex >= queries.length) {
+          // All queries processed
+          pq.exitPipelineMode()
+          return cb(null, results)
+        }
+        continue
+      }
+
+      if (status === 'PGRES_FATAL_ERROR') {
+        currentError = new Error(pq.resultErrorMessage())
+        // Extract error fields
+        const fields = pq.resultErrorFields()
+        if (fields) {
+          for (const key in fields) {
+            currentError[key] = fields[key]
+          }
+        }
+        continue
+      }
+
+      if (status === 'PGRES_PIPELINE_ABORTED') {
+        // Query skipped due to previous error in same sync group
+        continue
+      }
+
+      if (status === 'PGRES_TUPLES_OK' || status === 'PGRES_COMMAND_OK' || status === 'PGRES_EMPTY_QUERY') {
+        currentResult = self._consumeQueryResults(pq)
+        continue
+      }
+    }
+
+    // Still waiting for more data — will be called again when readable
+  }
+
+  // Use the libuv readable watcher
+  this._stopReading()
+  let done = false
+  const origCb = cb
+  cb = function (err, results) {
+    if (done) return
+    done = true
+    pq.removeListener('readable', onReadable)
+    self._stopReading()
+    origCb(err, results)
+  }
+  const onReadable = function () {
+    processResults()
+  }
+  pq.on('readable', onReadable)
+  pq.startReader()
+
+  // Try an initial read in case data is already available
+  processResults()
+}
+
 Client.prototype._onReadyForQuery = function () {
   // remove instance callback
   const cb = this._queryCallback

packages/pg-native/package.json

@@ -34,7 +34,7 @@
   },
   "homepage": "https://github.com/brianc/node-postgres/tree/master/packages/pg-native",
   "dependencies": {
-    "libpq": "^1.8.15",
+    "libpq": "^1.9.0",
     "pg-types": "2.2.0"
   },
   "devDependencies": {

packages/pg/lib/native/client.js

@@ -36,6 +36,8 @@ const Client = (module.exports = function (config) {
   this._connecting = false
   this._connected = false
   this._queryable = true
+  this.pipelining = false
+  this._pipeliningInFlight = false
 
   // keep these on the object for legacy reasons
   // for the time being. TODO: deprecate all this jazz
@@ -234,7 +236,7 @@ Client.prototype.query = function (config, values, callback) {
     return result
   }
 
-  if (this._queryQueue.length > 0) {
+  if (this._queryQueue.length > 0 && !this.pipelining) {
     queryQueueLengthDeprecationNotice()
   }
 
@@ -280,6 +282,9 @@ Client.prototype._pulseQueryQueue = function (initialConnection) {
   if (!this._connected) {
     return
   }
+  if (this.pipelining && !initialConnection) {
+    return this._pulsePipelinedQueryQueue()
+  }
   if (this._hasActiveQuery()) {
     return
   }
@@ -298,6 +303,83 @@ Client.prototype._pulseQueryQueue = function (initialConnection) {
   })
 }
 
+Client.prototype._pulsePipelinedQueryQueue = function () {
+  if (!this._connected || this._pipeliningInFlight) {
+    return
+  }
+  if (this._queryQueue.length === 0) {
+    if (this.hasExecuted) {
+      this.emit('drain')
+    }
+    return
+  }
+
+  this._pipeliningInFlight = true
+  const self = this
+  const queries = []
+  const nativeQueries = []
+  const utils = require('../utils')
+
+  while (this._queryQueue.length > 0) {
+    const query = this._queryQueue.shift()
+    this.hasExecuted = true
+    nativeQueries.push(query)
+
+    const values = query.values ? query.values.map(utils.prepareValue) : null
+    const pipelineEntry = { text: query.text, name: query.name }
+    if (values) {
+      pipelineEntry.values = values
+    }
+    if (query.name && this.namedQueries[query.name]) {
+      pipelineEntry._alreadyPrepared = true
+    }
+    queries.push(pipelineEntry)
+  }
+
+  this.native.pipeline(queries, function (err, results) {
+    self._pipeliningInFlight = false
+
+    if (err) {
+      // Total pipeline failure — error all queries
+      for (let i = 0; i < nativeQueries.length; i++) {
+        const q = nativeQueries[i]
+        q.native = self.native
+        q.handleError(err)
+      }
+      self._pulsePipelinedQueryQueue()
+      return
+    }
+
+    // Deliver results to each query
+    for (let i = 0; i < nativeQueries.length; i++) {
+      const q = nativeQueries[i]
+      const r = results[i]
+      q.native = self.native
+
+      if (r.err) {
+        q.handleError(r.err)
+      } else {
+        // Track named queries on success
+        if (q.name) {
+          self.namedQueries[q.name] = q.text
+        }
+        q.state = 'end'
+        q.emit('end', r.result)
+        if (q.callback) {
+          q.callback(null, r.result)
+        }
+      }
+
+      setImmediate(function () {
+        q.emit('_done')
+      })
+    }
+
+    // Process any queries that arrived while we were reading
+    self._pulsePipelinedQueryQueue()
+  })
+}
+
 // attempt to cancel an in-progress query
 Client.prototype.cancel = function (query) {
   if (this._activeQuery === query) {