perf(pg): read integers, booleans and text straight from the row bytes

packages/pg-protocol/src/inbound-parser.test.ts

@@ -577,6 +577,24 @@ describe('PgPacketStream', function () {
     })
   })
 
+  it('reads the fields of a lazy row', function () {
+    const values = ['a', null, '', 'è€😀']
+    const msgs: any[] = []
+    new Parser({ lazyRows: true }).parse(buffers.dataRow(values), (msg: any) => {
+      msgs.push({ name: msg.name, length: msg.length, fieldCount: msg.fieldCount, fields: msg.fields })
+    })
+    new Parser().parse(buffers.dataRow(values), (msg: any) => {
+      msgs.push({ name: msg.name, length: msg.length, fieldCount: msg.fieldCount, fields: msg.fields })
+    })
+    assert.deepStrictEqual(msgs[0], msgs[1])
+    assert.deepStrictEqual(msgs[0].fields, values)
+  })
+
+  it('keeps rows eager by default', async function () {
+    const [msg] = await parseBuffers([oneFieldBuf])
+    assert.strictEqual('buffer' in msg, false)
+  })
+
   it('cleans up the reader after handling a packet', function () {
     const parser = new Parser()
     parser.parse(oneFieldBuf, () => {})

packages/pg-protocol/src/index.ts

@@ -2,8 +2,12 @@ import { DatabaseError } from './messages'
 import { serialize } from './serializer'
 import { Parser, MessageCallback } from './parser'
 
-export function parse(stream: NodeJS.ReadableStream, callback: MessageCallback): Promise<void> {
-  const parser = new Parser()
+export function parse(
+  stream: NodeJS.ReadableStream,
+  callback: MessageCallback,
+  options?: { lazyRows?: boolean }
+): Promise<void> {
+  const parser = new Parser(options)
   stream.on('data', (buffer: Buffer) => parser.parse(buffer, callback))
   return new Promise((resolve) => stream.on('end', () => resolve()))
 }

packages/pg-protocol/src/messages.ts

@@ -260,3 +260,34 @@ export class NoticeMessage implements BackendMessage, NoticeOrError {
   public line: string | undefined
   public routine: string | undefined
 }
+
+// A data row still in the socket buffer, it is valid only while the parser callback runs.
+export class RawDataRowMessage {
+  public readonly name: MessageName = 'dataRow'
+  private decoded: any[] | null = null
+  constructor(
+    public length: number,
+    public readonly fieldCount: number,
+    public readonly buffer: Buffer,
+    public readonly offset: number
+  ) {}
+
+  public get fields(): any[] {
+    if (this.decoded === null) {
+      const fields = new Array(this.fieldCount)
+      let offset = this.offset
+      for (let i = 0; i < this.fieldCount; i++) {
+        const len = this.buffer.readInt32BE(offset)
+        offset += 4
+        if (len === -1) {
+          fields[i] = null
+        } else {
+          fields[i] = this.buffer.toString('utf-8', offset, offset + len)
+          offset += len
+        }
+      }
+      this.decoded = fields
+    }
+    return this.decoded
+  }
+}

packages/pg-protocol/src/parser.ts

@@ -18,6 +18,7 @@ import {
   ParameterDescriptionMessage,
   Field,
   DataRowMessage,
+  RawDataRowMessage,
   ParameterStatusMessage,
   BackendKeyDataMessage,
   DatabaseError,
@@ -47,7 +48,8 @@ export type Packet = {
 const emptyBuffer = Buffer.allocUnsafe(0)
 
 type StreamOptions = TransformOptions & {
-  mode: Mode
+  mode?: Mode
+  lazyRows?: boolean
 }
 
 const enum MessageCodes {
@@ -83,12 +85,14 @@ export class Parser {
   private bufferOffset: number = 0
   private reader = new BufferReader()
   private mode: Mode
+  private lazyRows: boolean
 
   constructor(opts?: StreamOptions) {
     if (opts?.mode === 'binary') {
       throw new Error('Binary mode not supported yet')
     }
     this.mode = opts?.mode || 'text'
+    this.lazyRows = opts?.lazyRows === true
   }
 
   public parse(buffer: Buffer, callback: MessageCallback) {
@@ -188,7 +192,9 @@ export class Parser {
         message = emptyQuery
         break
       case MessageCodes.DataRow:
-        message = parseDataRowMessage(reader)
+        message = this.lazyRows
+          ? new RawDataRowMessage(LATEINIT_LENGTH, bytes.readInt16BE(offset), bytes, offset + 2)
+          : parseDataRowMessage(reader)
         break
       case MessageCodes.CommandComplete:
         message = parseCommandCompleteMessage(reader)

packages/pg/lib/client.js

@@ -95,6 +95,7 @@ class Client extends EventEmitter {
         keepAlive: c.keepAlive || false,
         keepAliveInitialDelayMillis: c.keepAliveInitialDelayMillis || 0,
         encoding: this.connectionParameters.client_encoding || 'utf8',
+        lazyRows: true,
       })
     this._queryQueue = []
     this._sentQueryQueue = []

packages/pg/lib/connection.js

@@ -29,6 +29,7 @@ class Connection extends EventEmitter {
     this.sslNegotiation = config.sslNegotiation || 'postgres'
     this._ending = false
     this._emitMessage = false
+    this.lazyRows = config.lazyRows === true
     const self = this
     this.on('newListener', function (eventName) {
       if (eventName === 'message') {
@@ -135,13 +136,21 @@ class Connection extends EventEmitter {
   }
 
   attachListeners(stream) {
-    parse(stream, (msg) => {
-      const eventName = msg.name === 'error' ? 'errorMessage' : msg.name
-      if (this._emitMessage) {
-        this.emit('message', msg)
-      }
-      this.emit(eventName, msg)
-    })
+    parse(
+      stream,
+      (msg) => {
+        // The client reads a lazy row at once, other listeners may keep it after the buffer is reused.
+        if (this.lazyRows && msg.name === 'dataRow' && (this._emitMessage || this.listenerCount('dataRow') > 1)) {
+          void msg.fields
+        }
+        const eventName = msg.name === 'error' ? 'errorMessage' : msg.name
+        if (this._emitMessage) {
+          this.emit('message', msg)
+        }
+        this.emit(eventName, msg)
+      },
+      { lazyRows: this.lazyRows }
+    )
   }
 
   requestSsl() {

packages/pg/lib/query.js

@@ -87,7 +87,7 @@ class Query extends EventEmitter {
     }
 
     try {
-      row = this._result.parseRow(msg.fields)
+      row = this._result.parseDataRow(msg)
     } catch (err) {
       this._canceledDueToError = err
       return

packages/pg/lib/result.js

@@ -1,9 +1,44 @@
 'use strict'
 
 const types = require('pg-types')
+const textParsers = require('pg-types/lib/textParsers')
 
 const matchRegexp = /^([A-Za-z]+)(?: (\d+))?(?: (\d+))?/
 
+// How a text column is read from the row bytes. Only the built-in parsers take a shortcut,
+// a parser set with setTypeParser gets the string as always.
+const OTHER = 0
+const INT = 1
+const BOOL = 2
+const TEXT = 3
+
+// From the source and not from getTypeParser, which also returns the overrides.
+const builtinParsers = new Map()
+textParsers.init((oid, parser) => builtinParsers.set(oid, parser))
+const noParse = types.getTypeParser(25, 'unknown format')
+
+const kindOf = (oid, parser) => {
+  if (parser === noParse) return TEXT
+  if (parser !== builtinParsers.get(oid)) return OTHER
+  if (oid === 21 || oid === 23 || oid === 26) return INT // int2, int4, oid
+  return oid === 16 ? BOOL : OTHER
+}
+
+const text =
+  typeof Buffer.prototype.utf8Slice === 'function'
+    ? (buffer, start, end) => buffer.utf8Slice(start, end)
+    : (buffer, start, end) => buffer.toString('utf8', start, end)
+
+// The server writes these as digits with an optional minus, and they always fit a double.
+const int = (buffer, start, end) => {
+  const negative = buffer[start] === 45
+  let value = 0
+  for (let i = negative ? start + 1 : start; i < end; i++) {
+    value = value * 10 + buffer[i] - 48
+  }
+  return negative ? -value : value
+}
+
 // result object returned from query
 // in the 'end' event and also
 // passed as second argument to provided callback
@@ -15,6 +50,9 @@ class Result {
     this.rows = []
     this.fields = []
     this._parsers = undefined
+    this._names = []
+    this._kinds = []
+    this._hasBinary = false
     this._types = types
     this.RowCtor = null
     this.rowAsArray = rowMode === 'array'
@@ -75,6 +113,43 @@ class Result {
     return row
   }
 
+  // Same result as parseRow(msg.fields), without a string for every value.
+  parseDataRow(msg) {
+    const names = this._names
+    if (msg.buffer === undefined || this.rowAsArray || this._hasBinary || msg.fieldCount !== names.length) {
+      return this.parseRow(msg.fields)
+    }
+    const buffer = msg.buffer
+    const kinds = this._kinds
+    const parsers = this._parsers
+    const row = { ...this._prebuiltEmptyResultObject }
+    let offset = msg.offset
+    for (let i = 0, len = names.length; i < len; i++) {
+      const size = (buffer[offset] << 24) | (buffer[offset + 1] << 16) | (buffer[offset + 2] << 8) | buffer[offset + 3]
+      offset += 4
+      if (size === -1) {
+        row[names[i]] = null
+        continue
+      }
+      const end = offset + size
+      switch (kinds[i]) {
+        case INT:
+          row[names[i]] = int(buffer, offset, end)
+          break
+        case BOOL:
+          row[names[i]] = buffer[offset] === 116
+          break
+        case TEXT:
+          row[names[i]] = text(buffer, offset, end)
+          break
+        default:
+          row[names[i]] = parsers[i](text(buffer, offset, end))
+      }
+      offset = end
+    }
+    return row
+  }
+
   addRow(row) {
     this.rows.push(row)
   }
@@ -91,15 +166,22 @@ class Result {
 
     const row = Object.create(null)
 
+    this._names = new Array(fieldDescriptions.length)
+    this._kinds = new Array(fieldDescriptions.length)
+    this._hasBinary = false
+
     for (let i = 0; i < fieldDescriptions.length; i++) {
       const desc = fieldDescriptions[i]
       row[desc.name] = null
+      this._names[i] = desc.name
+      this._hasBinary ||= desc.format === 'binary'
 
       if (this._types) {
         this._parsers[i] = this._types.getTypeParser(desc.dataTypeID, desc.format || 'text')
       } else {
         this._parsers[i] = types.getTypeParser(desc.dataTypeID, desc.format || 'text')
       }
+      this._kinds[i] = kindOf(desc.dataTypeID, this._parsers[i])
     }
 
     this._prebuiltEmptyResultObject = { ...row }

packages/pg/test/integration/client/rows-from-bytes-tests.js

@@ -0,0 +1,73 @@
+'use strict'
+const helper = require('./test-helper')
+const assert = require('assert')
+const Result = require('../../../lib/result')
+
+const pg = helper.pg
+const suite = new helper.Suite()
+
+const query = `
+  select 7::int a, (-2147483648)::int b, (-32768)::int2 c, 4294967295::oid d,
+    true e, false f, 'è€😀'::text g, ''::text h, 'x'::varchar i, null::int j, 1.5::float8 k`
+
+const expected = {
+  a: 7,
+  b: -2147483648,
+  c: -32768,
+  d: 4294967295,
+  e: true,
+  f: false,
+  g: 'è€😀',
+  h: '',
+  i: 'x',
+  j: null,
+  k: 1.5,
+}
+
+// Without this the rows are correct but go through parseRow, and the shortcut is dead code.
+const countParseRow = async (fn) => {
+  const parseRow = Result.prototype.parseRow
+  let calls = 0
+  Result.prototype.parseRow = function (rowData) {
+    calls++
+    return parseRow.call(this, rowData)
+  }
+  try {
+    await fn()
+  } finally {
+    Result.prototype.parseRow = parseRow
+  }
+  return calls
+}
+
+suite.test('a client reads the rows of a query from the bytes', async function () {
+  if (helper.args.native || pg.defaults.binary) {
+    return
+  }
+  const client = new pg.Client()
+  await client.connect()
+  try {
+    const calls = await countParseRow(async () => {
+      assert.deepStrictEqual((await client.query(query)).rows, [expected])
+      assert.deepStrictEqual((await client.query({ text: query, name: 'bytes' })).rows, [expected])
+    })
+    assert.strictEqual(calls, 0)
+  } finally {
+    await client.end()
+  }
+})
+
+suite.test('a pooled client reads the rows of a query from the bytes', async function () {
+  if (helper.args.native || pg.defaults.binary) {
+    return
+  }
+  const pool = new pg.Pool()
+  try {
+    const calls = await countParseRow(async () => {
+      assert.deepStrictEqual((await pool.query(query)).rows, [expected])
+    })
+    assert.strictEqual(calls, 0)
+  } finally {
+    await pool.end()
+  }
+})