From 06746cd1cfd1a83dbe3da0b643b81f816ca6b196 Mon Sep 17 00:00:00 2001 From: Stephen Haberman Date: Sun, 26 Jul 2026 12:05:07 -0500 Subject: [PATCH] pg-protocol: defer DataRow cell decoding to a lazy `fields` getter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Currently, `parseDataRowMessage` eagerly materializes a JS string for every non-null cell of every row at parse time. For a consumer that reads every cell that is exactly the right amount of work; for a consumer that reads some of them, or none of them, it is pure overhead — and on large result sets it is the dominant cost of a query. This PR makes `DataRowMessage` carry the raw payload range (`bytes` + `offset`) and decode cells lazily in a memoized `fields` getter. Existing consumers see identical values from an identical `fields` array; consumers that want to decode on their own terms can now read the payload directly instead of paying for 40 strings per row to use two of them. Two parts: 1. **`DataRowMessage` decodes lazily.** `fields` becomes a memoized getter over `bytes`/`offset`. 2. **The parser reassembles only the message that straddles a chunk, not the whole chunk.** This is what makes the lazily-decoded views safe to hold past the parse tick, and it also drops a full-chunk `memcpy` per socket read for everyone, including pg itself (that copy was how a straddling message's two halves were made contiguous: every chunk was appended to one scratch buffer and all messages were parsed out of there. Only the straddling message ever needs bytes from two chunks, though, so now only it is reassembled — every other message is already contiguous in the chunk it arrived in, and is parsed where it landed). The second part could be dropped if you'd rather land just the lazy getter, but the two together are what make the payload range useful to a consumer. ## Motivation Measured against a local PostgreSQL 17 (node v26.4.0), `select * from <40 column table>` with 100k rows: server-side execution is ~5–10 ms, but `client.query` takes ~630 ms. Roughly, that splits into ~215 ms of the parser building a string per cell (4M cell strings for this one result), ~50 ms of socket transfer, ~285 ms of work downstream of the parser — pg-types converting those strings into values (`timestamptz` into `Date`, and so on) and `Result.parseRow` building 100k 40-property row objects — and ~80 ms of accumulating the rows. Building the strings is the largest single term, and it is the only one this PR touches. Consumers that don't want those strings already exist: - ORMs that decode lazily on field access. A Joist prototype measures 2.2–2.55× faster on 100k–1M row reads — a 100k-row find that reads one column of a 40 column table goes 378 ms → 148 ms, with GC-traced heap for the held result dropping ~3.4×. - Cursor and stream consumers, which hand rows straight to a caller that may only touch a few columns. - Byte-ish workloads — see #2240, where large cell allocation was enough of a problem to be worth its own fix — and #2093, asking directly how to avoid materializing results. This layer is also demonstrably performance-sensitive and actively maintained: #2151 (parser speed improvements), #2240 (the buffer management this PR reworks), and #3533 (Dec 2025, avoiding retention of the last parse buffer). The change here is in that same vein: less copying, less allocation, no behavior change. ## Why now The whole-chunk copy is not a decision anyone made about straddling messages; it is the residue of a simpler design that was optimized twice without its premise being revisited. - #2155 (Apr 2020), the original TypeScript parser: on any read with a leftover, it allocated a fresh `leftover + chunk` buffer and copied both halves into it. Nothing about that targets straddling messages — contiguity is simply what the message loop wants to walk. - #2240 (Jun 2020): the cost being attacked was that per-read allocation. It kept the copy and removed the allocation, giving us today's single scratch buffer that doubles, is appended into, and is compacted in place when it runs out of room. Later commits in the same PR renamed the fields and extracted `mergeBuffer`; the algorithm was already what it is now. So both rounds optimized _how_ to get a contiguous buffer, not _whether_ everything had to live in one. Under eager decoding that premise costs nothing: once `parse()` returns, every message is plain JS strings, nothing points into the bytes any more, and recycling one buffer forever is strictly the best thing to do. A 64 KiB memcpy also does not stand out next to 4M string allocations in a profile. Deferring the cell decoding is what changes the calculus. The moment a message is a view over the bytes, recycling stops being free, and it becomes worth asking which bytes actually have to move. The answer — one message per read, rather than one chunk per read — was always true; it just never mattered before. ## What changed **`messages.ts`** — `DataRowMessage` stores `bytes: Buffer` and `offset: number` (the payload start, pointing at the `int16` field count), reads `fieldCount` in the constructor, and exposes `fields` as a memoized getter that walks the length-prefixed cells (`int32 len`, then `len` bytes; `len === -1` is SQL NULL), decoding each with `bytes.toString('utf-8', …)` — byte for byte what `BufferReader.string` did. Callers that read `fields` see no behavior change at all: the same array of the same values, built by the same UTF-8 decoding of the same bytes, just on first read rather than at parse time. The only difference visible to them is one property read per row. **`parser.ts`** — the `DataRow` case constructs the message from `(bytes, offset)`, which `handlePacket` already has, so `parseDataRowMessage` goes away entirely. And `mergeBuffer` is replaced: - Before: when handling split messages (very common b/c a 64 KiB read typically has many complete messages/rows — ~89 rows of the 737-byte shape in the benchmark below — and then a fraction of one more, since where messages end has nothing to do with where reads end), on main the _entire_ incoming chunk is copied into a scratch buffer, which grows by doubling and is compacted in place whenever the existing buffer can be reused, overwriting what earlier messages had been parsed out of. Those compacted-over bytes are what a lazily decoded `DataRow` still needs, meaning any "lazy parsing" would be forced to happen immediately/before the scratch buffer was reused. Our observation is that the trailing fragment is the only part of a chunk that needs bytes from later chunks: the parser consumes messages from the front, so everything ahead of that fragment is already complete where it landed. - After: the chunk is always parsed in place, and only the one message per chunk that straddles a read/chunk boundary is reassembled into a buffer of its own, which is handed to that message and never touched again. This sounds like "an extra buffer" and so a higher cost to all consumers (eager or lazy), and it is worth being precise that per read it is one. On main, the scratch buffer is allocated once — the first straddle doubles 64 KiB into 128 KiB, and every top-up after that hits the in-place compaction branch, which allocates nothing — so its cost amortizes to zero across the result, at the price of copying 64 KiB into it every read. Here, each straddling message gets its own buffer, so a 1000-read result makes 1000 of them rather than reusing one. Measured, that is +934 bytes per 64 KiB read: one message's worth, which under 4 KiB comes out of node's buffer pool (a pointer bump, ~11 messages per 8 KiB slab) and is young-generation garbage that dies with the message — for `Result.parseRow`, in the same tick. Total allocated bytes over a whole result therefore crosses main's fixed 128 KiB at ~140 reads (~9 MB of rows) and grows from there: ~1 MB for the 70 MiB benchmark below. In exchange, the 64 KiB copy per read is gone, and so is the retained buffer: between reads the parser now holds at most one message's bytes, rather than 128 KiB for the life of the query. For a consumer that reads `fields` (eager parsing), the added allocation is ~1% of the ~70 KiB of cell strings that same read already allocates — total allocation per read measures 70.0 KiB after vs 71.6 KiB before, and the benchmark below shows no change in GC counts or pause time. The one thing to know is the granularity for consumers that _do_ retain: holding one row keeps its whole ~64 KiB read alive, so a consumer holding a few scattered rows should copy out instead. Because that copy ran on essentially every read, a streaming result spent nearly its whole life in the recycled scratch buffer. Concretely, in the benchmark below the old code copies 64 KiB per read (70 MB in total); the new code copies one ~737-byte row per read. Two smaller consequences worth noting: - Every message — not just `DataRow` — is now a view over memory nothing will rewrite, so `copyData` chunks and `authenticationMD5Password` salts (both `Buffer` slices of the parse buffer today) become safe to retain as well. - Between reads the parser now retains at most one message's bytes. On main it holds the scratch buffer — 128 KiB for 64 KiB reads, more if a burst contains large messages — for as long as a response is streaming, releasing it only on a read that ends exactly on a message boundary, i.e. in practice the read that ends the response (or never, while queries are pipelined back to back). That is the same class of problem #3533 addressed. A straddling message's buffer is sized from its header, but only up to 1 MiB up front (`MAX_EAGER_MESSAGE_LENGTH`); beyond that it grows as bytes actually arrive, so a bogus length in a corrupt stream can't ask for a huge allocation. ## Benchmarks `packages/pg-protocol/src/b-data-row.ts` (new, alongside the existing `b.ts`): 100k `DataRow` frames of 40 text cells = 70.3 MiB, delivered in 1125 × 64 KiB chunks, node v26.4.0, best of 5 runs after a warm-up. | consumer | master | this PR | | ---------------------------------- | ----------------- | ---------------------- | | never reads a cell | 213 ms, 64 gc/run | **2.1 ms**, 2.6 gc/run | | reads `fields` once (what pg does) | 216 ms | **205 ms** | | reads every cell | 218 ms | **210 ms** | So: ~100× less time and ~25× fewer collections for a consumer that decodes on its own, and no regression — a small improvement, from the dropped chunk copy — for pg's own eager path. End to end against a local PostgreSQL 17, `select * from wide` (100k rows × 40 columns, 66 MB), best of 7: | consumer | master | this PR | | ----------------------------------------------- | ------ | ---------- | | `client.query`, rows accumulated as objects | 631 ms | 625 ms | | `row` events, rows dropped | 549 ms | 549 ms | | needs 2 of 40 columns, reading `fields` | 266 ms | 264 ms | | needs 2 of 40 columns, reading `bytes`/`offset` | — | **158 ms** | The last row is the point of the change: same query, same 100k rows, same checksum, 1.7× less wall time because 38 of 40 cells per row are never turned into strings. ## Compatibility - `name`, `length`, `fieldCount` and `fields` behave exactly as before for every existing consumer — pg's `Result.parseRow(msg.fields)` reads the getter once per row, one extra property read, same values. - `fields` memoizes, so repeated reads decode once. - The old `new DataRowMessage(length, fields)` constructor still works (an array second argument takes the pre-decoded path), so anything constructing these directly keeps compiling. It is marked `@deprecated` and can be dropped in a major if you'd rather not carry it. - `fields` is now typed `(string | null)[]` rather than `any[]`. - `bytes`, `offset` and `fieldCount` are public, which is what lets a consumer skip `fields` entirely. ## Tests - `inbound-parser.test.ts`: DataRow cases for multi-byte UTF-8, an empty-string cell (`len === 0`) and a NULL cell (`len === -1`); that `fieldCount` is available without decoding; that `fields` is memoized; that the payload range is readable; and that the legacy constructor still works. - New `chunked parsing` block for the buffer rework: a mixed message stream parsed whole, split at every single byte offset, in fixed-size chunks of 1/2/3/4/5/7/13/64/128 bytes (so headers split, bodies split, several messages per chunk, and messages spanning many chunks), a 3 MiB message delivered in 64 KiB chunks, and a check that messages held until after the whole stream is parsed still decode correctly. That last one fails against the old buffer management (the lazy getter reads recycled bytes and throws) and passes with the new. - One test that compared a whole `dataRow` message with `deepEqual` now asserts its shape explicitly, since `fields` is a getter. - pg-protocol 86 passing; pg 270 unit + 218 integration; pg-cursor 37; pg-query-stream 40; pg-pool 86 — all against PostgreSQL 17 on node v26. ## Follow-up Nothing here depends on it, but the natural next step is to let `fields` be decoded per cell rather than per row: with `bytes`/`offset` public, a consumer can already do that, but pg's own `Result.parseRow` still decodes all 40 cells to build a row object. Making that lazy too (a row proxy, or decoding on first property access) would push the same win into `pool.query` itself. --- packages/pg-protocol/src/b-data-row.ts | 157 +++++++++++++++++ .../pg-protocol/src/inbound-parser.test.ts | 164 +++++++++++++++++- packages/pg-protocol/src/messages.ts | 59 ++++++- packages/pg-protocol/src/parser.ts | 162 ++++++++++------- 4 files changed, 469 insertions(+), 73 deletions(-) create mode 100644 packages/pg-protocol/src/b-data-row.ts diff --git a/packages/pg-protocol/src/b-data-row.ts b/packages/pg-protocol/src/b-data-row.ts new file mode 100644 index 000000000..e3b068463 --- /dev/null +++ b/packages/pg-protocol/src/b-data-row.ts @@ -0,0 +1,157 @@ +// file for microbenchmarking DataRow parsing +// +// node dist/b-data-row.js +// +// Synthesizes a large text result set, hands it to the parser in socket-sized chunks, and times +// the three ways a consumer can treat a row: never reading a cell, reading the `fields` array +// once (what pg's `Result.parseRow` does), and touching every cell. + +import { PerformanceObserver } from 'perf_hooks' +import { BackendMessage } from './messages' +import { Parser } from './parser' + +const ROWS = 100_000 +const COLUMNS = 40 +const CHUNK_SIZE = 64 * 1024 +const RUNS = 5 +// width of the synthesized id column, so every row's frame is the same size +const ID_WIDTH = 7 + +// the parts of a DataRow message this benchmark consumes, i.e. without depending on its class +type Row = { + fieldCount: number + fields: (string | null)[] +} + +type Mode = { + name: string + /** Returns a number, so that the work of consuming a row cannot be optimized away. */ + consume: (row: Row) => number +} + +const modes: Mode[] = [ + // a consumer that never reads a cell, e.g. one decoding lazily on its own + { name: 'fieldCount only', consume: (row) => row.fieldCount }, + // one `fields` read per row, i.e. what pg itself does + { name: 'fields array', consume: (row) => row.fields.length }, + // every cell read, i.e. what building a row object ends up doing + { name: 'every cell', consume: sumCellLengths }, +] + +main() + +/** Times each consumption mode over the same synthesized chunks. */ +async function main(): Promise { + const chunks = synthesizeChunks(ROWS, COLUMNS, CHUNK_SIZE) + const bytes = chunks.reduce((total, chunk) => total + chunk.byteLength, 0) + const mib = (bytes / 1024 / 1024).toFixed(1) + console.log(`${ROWS} rows x ${COLUMNS} cells = ${mib} MiB in ${chunks.length} chunks of ${CHUNK_SIZE} bytes`) + + for (const mode of modes) { + let checksum = run(chunks, mode) // warm up, uncounted + let collections = 0 + let pause = 0 + const observer = new PerformanceObserver((list) => { + for (const entry of list.getEntries()) { + collections++ + pause += entry.duration + } + }) + observer.observe({ entryTypes: ['gc'] }) + const times: number[] = [] + for (let i = 0; i < RUNS; i++) { + const start = performance.now() + checksum += run(chunks, mode) + times.push(performance.now() - start) + // let the observer's callbacks run, so this run's collections are counted + await new Promise((resolve) => setImmediate(resolve)) + } + observer.disconnect() + times.sort((a, b) => a - b) + const best = times[0].toFixed(1) + const median = times[Math.floor(times.length / 2)].toFixed(1) + // gc counts are an allocation-pressure proxy: a scavenge means another semi-space was filled + const gc = `${(collections / RUNS).toFixed(1)} gc/run, ${(pause / RUNS).toFixed(1)}ms gc pause/run` + console.log(` ${mode.name.padEnd(16)} best ${best}ms median ${median}ms ${gc} (checksum ${checksum})`) + } +} + +/** Parses every chunk with a fresh parser, returning the consumer's checksum. */ +function run(chunks: Buffer[], mode: Mode): number { + const parser = new Parser() + let checksum = 0 + const consume = (msg: BackendMessage) => { + checksum += mode.consume(msg as unknown as Row) + } + for (const chunk of chunks) { + parser.parse(chunk, consume) + } + return checksum +} + +/** Reads every cell of a row, i.e. the most a consumer can ask of `fields`. */ +function sumCellLengths(row: Row): number { + const { fields } = row + let total = 0 + for (let i = 0; i < fields.length; i++) { + const cell = fields[i] + total += cell === null ? 0 : cell.length + } + return total +} + +/** Builds `rows` DataRow frames and slices them into standalone reads, i.e. as a socket delivers them. */ +function synthesizeChunks(rows: number, columns: number, chunkSize: number): Buffer[] { + const cells = sampleCells(columns) + const frameLength = 7 + cells.reduce((total, cell) => total + 4 + (cell === null ? 0 : Buffer.byteLength(cell)), 0) + const stream = Buffer.allocUnsafe(rows * frameLength) + let offset = 0 + for (let row = 0; row < rows; row++) { + cells[0] = String(row).padStart(ID_WIDTH, '0') + offset = writeDataRow(stream, offset, cells) + } + const chunks: Buffer[] = [] + for (let start = 0; start < offset; start += chunkSize) { + // a copy, so each chunk is its own allocation like the ones net.Socket emits + chunks.push(Buffer.from(stream.subarray(start, Math.min(start + chunkSize, offset)))) + } + return chunks +} + +/** A row's worth of representative text values, i.e. ids, timestamps, numbers, prose and NULLs. */ +function sampleCells(columns: number): (string | null)[] { + const samples = [ + '0'.repeat(ID_WIDTH), + '4c1a0e2e-6b58-4e2f-9b6f-6a1d0f7f1b23', + '2026-07-26 11:22:33.456789', + '12345', + 'true', + null, + '199.95', + 'a short description of the row', + ] + const cells: (string | null)[] = new Array(columns) + for (let i = 0; i < columns; i++) { + cells[i] = samples[i % samples.length] + } + return cells +} + +/** Writes one DataRow frame at `offset`, returning the offset just past it. */ +function writeDataRow(into: Buffer, offset: number, cells: (string | null)[]): number { + const start = offset + into[offset] = 0x44 // 'D' + offset += 5 // the int32 length is backfilled once the frame is written + offset = into.writeInt16BE(cells.length, offset) + for (const cell of cells) { + if (cell === null) { + offset = into.writeInt32BE(-1, offset) + } else { + offset = into.writeInt32BE(Buffer.byteLength(cell), offset) + offset += into.write(cell, offset, 'utf-8') + } + } + // the frame's length covers everything but the code byte + into.writeInt32BE(offset - start - 1, start + 1) + return offset +} diff --git a/packages/pg-protocol/src/inbound-parser.test.ts b/packages/pg-protocol/src/inbound-parser.test.ts index 285f4bf2b..ae07923a2 100644 --- a/packages/pg-protocol/src/inbound-parser.test.ts +++ b/packages/pg-protocol/src/inbound-parser.test.ts @@ -3,7 +3,7 @@ import BufferList from './testing/buffer-list' import { parse } from '.' import assert from 'assert' import { PassThrough } from 'stream' -import { BackendMessage } from './messages' +import { BackendMessage, DataRowMessage, Field } from './messages' import { Parser } from './parser' const authOkBuffer = buffers.authenticationOk() @@ -305,6 +305,40 @@ describe('PgPacketStream', function () { fields: ['test'], }) }) + + describe('parsing data row with an empty, a multi-byte and a null field', function () { + testForMessage(buffers.dataRow(['', 'héllo → 🐘', null, 'ok']), { + name: 'dataRow', + fieldCount: 4, + fields: ['', 'héllo → 🐘', null, 'ok'], + }) + }) + + it('decodes fields lazily, and only once', async function () { + const [message] = (await parseBuffers([buffers.dataRow(['a', null, 'b'])])) as DataRowMessage[] + // fieldCount is known without decoding any cell + assert.strictEqual(message.fieldCount, 3) + const fields = message.fields + assert.deepEqual(fields, ['a', null, 'b']) + // the decoded array is memoized, so repeated reads do not re-decode + assert.strictEqual(message.fields, fields) + }) + + it('exposes the undecoded payload', async function () { + const [message] = (await parseBuffers([buffers.dataRow(['🐘'])])) as DataRowMessage[] + // offset points at the int16 field count, i.e. past the 1 byte code and int32 length + assert.strictEqual(message.offset, 5) + assert.strictEqual(message.bytes.readInt16BE(message.offset), 1) + const len = message.bytes.readInt32BE(message.offset + 2) + assert.strictEqual(len, 4) + assert.deepEqual(message.bytes.subarray(message.offset + 6, message.offset + 6 + len), Buffer.from('🐘')) + }) + + it('accepts pre-decoded fields for backwards compatibility', function () { + const message = new DataRowMessage(11, ['a', null]) + assert.strictEqual(message.fieldCount, 2) + assert.deepEqual(message.fields, ['a', null]) + }) }) describe('notice message', function () { @@ -521,13 +555,11 @@ describe('PgPacketStream', function () { const verifyMessages = function (messages: any[]) { assert.strictEqual(messages.length, 2) - assert.deepEqual(messages[0], { - name: 'dataRow', - fieldCount: 1, - length: 11, - fields: ['!'], - }) - assert.equal(messages[0].fields[0], '!') + // dataRow decodes its cells lazily, so assert its shape rather than its own properties + assert.strictEqual(messages[0].name, 'dataRow') + assert.strictEqual(messages[0].length, 11) + assert.strictEqual(messages[0].fieldCount, 1) + assert.deepEqual(messages[0].fields, ['!']) assert.deepEqual(messages[1], { name: 'readyForQuery', length: 5, @@ -567,9 +599,125 @@ describe('PgPacketStream', function () { }) }) + // every message is parsed out of the chunk it arrived in, except the one message per chunk that + // straddles a chunk boundary, so exercise every way a stream can be cut up + describe('chunked parsing', function () { + const stream = Buffer.concat([ + buffers.rowDescription([ + { + name: 'id', + tableID: 1, + attributeNumber: 1, + dataTypeID: 23, + dataTypeSize: 4, + typeModifier: -1, + formatCode: 0, + }, + { + name: 'name', + tableID: 1, + attributeNumber: 2, + dataTypeID: 25, + dataTypeSize: -1, + typeModifier: -1, + formatCode: 0, + }, + ]), + buffers.dataRow(['one', null, 'héllo → 🐘']), + buffers.dataRow([]), + // longer than the smaller chunk sizes below, so that one message spans several chunks + buffers.dataRow(['x'.repeat(300)]), + buffers.commandComplete('SELECT 3'), + buffers.readyForQuery(), + ]) + + const expected = [ + 'rowDescription(id,name)', + 'dataRow(one,NULL,héllo → 🐘)', + 'dataRow()', + `dataRow(${'x'.repeat(300)})`, + 'commandComplete(SELECT 3)', + 'readyForQuery(5)', + ] + + it('parses the whole stream in one chunk', function () { + assert.deepEqual(parseChunks([stream]), expected) + }) + + it('parses the stream split at every offset', function () { + for (let split = 1; split < stream.byteLength; split++) { + const chunks = [copyOf(stream, 0, split), copyOf(stream, split, stream.byteLength)] + assert.deepEqual(parseChunks(chunks), expected, `split at ${split}`) + } + }) + + it('parses a message far larger than the chunks carrying it', function () { + // big enough that reassembling it has to grow its buffer several times + const cell = 'z'.repeat(3 * 1024 * 1024) + const bigStream = Buffer.concat([buffers.dataRow([cell, 'small']), buffers.readyForQuery()]) + const summaries = parseChunks(intoChunks(bigStream, 64 * 1024)) + assert.deepEqual(summaries, [`dataRow(${cell},small)`, 'readyForQuery(5)']) + }) + + it('parses the stream in fixed size chunks', function () { + for (const size of [1, 2, 3, 4, 5, 7, 13, 64, 128]) { + assert.deepEqual(parseChunks(intoChunks(stream, size)), expected, `${size} byte chunks`) + } + }) + + it('keeps messages valid after later chunks are parsed', function () { + // messages are views over the bytes they were parsed from, so hold them all and read them + // only once the whole stream has gone through the parser + const parser = new Parser() + const messages: BackendMessage[] = [] + for (const chunk of intoChunks(stream, 7)) { + parser.parse(chunk, (msg) => messages.push(msg)) + } + assert.deepEqual(messages.map(summarize), expected) + }) + }) + it('cleans up the reader after handling a packet', function () { const parser = new Parser() parser.parse(oneFieldBuf, () => {}) assert.strictEqual((parser as any).reader.buffer.byteLength, 0) }) }) + +/** Parses `chunks` with one parser, summarizing each message as it is emitted. */ +const parseChunks = function (chunks: Buffer[]): string[] { + const parser = new Parser() + const summaries: string[] = [] + for (const chunk of chunks) { + parser.parse(chunk, (msg) => summaries.push(summarize(msg))) + } + return summaries +} + +/** A message's contents as a string, i.e. `dataRow(one,NULL)`, so a whole stream can be compared at once. */ +const summarize = function (msg: any): string { + switch (msg.name) { + case 'dataRow': + return `dataRow(${msg.fields.map((field: string | null) => (field === null ? 'NULL' : field)).join(',')})` + case 'rowDescription': + return `rowDescription(${msg.fields.map((field: Field) => field.name).join(',')})` + case 'commandComplete': + return `commandComplete(${msg.text})` + default: + return `${msg.name}(${msg.length})` + } +} + +/** Cuts `buffer` into standalone `size` byte chunks, i.e. the reads a socket would deliver. */ +const intoChunks = function (buffer: Buffer, size: number): Buffer[] { + const chunks: Buffer[] = [] + for (let offset = 0; offset < buffer.byteLength; offset += size) { + chunks.push(copyOf(buffer, offset, Math.min(offset + size, buffer.byteLength))) + } + return chunks +} + +/** Copies `buffer` between the given offsets, i.e. so each chunk is its own allocation. */ +const copyOf = function (buffer: Buffer, start: number, end: number): Buffer { + return Buffer.from(buffer.subarray(start, end)) +} diff --git a/packages/pg-protocol/src/messages.ts b/packages/pg-protocol/src/messages.ts index c3fbbdd9b..37cf91c2d 100644 --- a/packages/pg-protocol/src/messages.ts +++ b/packages/pg-protocol/src/messages.ts @@ -1,5 +1,7 @@ export type Mode = 'text' | 'binary' +const emptyBuffer = Buffer.allocUnsafe(0) + export type MessageName = | 'parseComplete' | 'bindComplete' @@ -227,13 +229,64 @@ export class CommandCompleteMessage { } export class DataRowMessage { - public readonly fieldCount: number public readonly name: MessageName = 'dataRow' + public readonly fieldCount: number + /** The buffer holding this row's payload, i.e. for reading raw cell bytes without decoding them. */ + public readonly bytes: Buffer + // assigned up front, i.e. so that reading `fields` does not change the object's shape + private decodedFields: (string | null)[] | undefined = undefined + + /** Wraps the row's payload, i.e. `offset` points at the payload's `int16` field count within `bytes`. */ + constructor(length: number, bytes: Buffer, offset: number) + /** @deprecated Pass the payload range instead; pre-decoded fields are accepted for backwards compatibility. */ + constructor(length: number, fields: (string | null)[]) constructor( public length: number, - public fields: any[] + bytesOrFields: Buffer | (string | null)[], + public readonly offset: number = 0 ) { - this.fieldCount = fields.length + if (Array.isArray(bytesOrFields)) { + this.bytes = emptyBuffer + this.decodedFields = bytesOrFields + this.fieldCount = bytesOrFields.length + } else { + this.bytes = bytesOrFields + this.fieldCount = bytesOrFields.readInt16BE(offset) + } + } + + /** + * This row's cells, decoded from `bytes` on first access and then memoized. + * + * Values are identical to decoding every cell up front, with `null` for a SQL NULL. A message + * stays readable for as long as it is held: `bytes` is either the socket chunk the row arrived + * in or, when the row straddled two chunks, a buffer of its own, and the parser reuses neither. + * Holding a row therefore also holds the chunk it came from. + */ + public get fields(): (string | null)[] { + if (this.decodedFields === undefined) { + this.decodedFields = this.decodeFields() + } + return this.decodedFields + } + + /** Walks the payload's length-prefixed cells, i.e. `int32 len` (`-1` for NULL) then `len` bytes of utf8. */ + private decodeFields(): (string | null)[] { + const { bytes, fieldCount } = this + const fields: (string | null)[] = new Array(fieldCount) + // skip the int16 field count that `offset` points at + let cursor = this.offset + 2 + for (let i = 0; i < fieldCount; i++) { + const len = bytes.readInt32BE(cursor) + cursor += 4 + if (len === -1) { + fields[i] = null + } else { + fields[i] = bytes.toString('utf-8', cursor, cursor + len) + cursor += len + } + } + return fields } } diff --git a/packages/pg-protocol/src/parser.ts b/packages/pg-protocol/src/parser.ts index 3d8ce80c7..0226f3d5b 100644 --- a/packages/pg-protocol/src/parser.ts +++ b/packages/pg-protocol/src/parser.ts @@ -39,6 +39,11 @@ const HEADER_LENGTH = CODE_LENGTH + LEN_LENGTH // A placeholder for a `BackendMessage`’s length value that will be set after construction. const LATEINIT_LENGTH = -1 +// How much of a straddling message’s claimed length we are willing to allocate up front. Beyond +// this the buffer grows as the bytes actually arrive, so that a bogus length in a corrupt stream +// cannot ask for an enormous allocation. +const MAX_EAGER_MESSAGE_LENGTH = 1024 * 1024 + export type Packet = { code: number packet: Buffer @@ -77,10 +82,27 @@ const enum MessageCodes { export type MessageCallback = (msg: BackendMessage) => void +/** + * Parses the backend’s messages out of the chunks the socket delivers. + * + * A chunk is however many bytes one socket read returned, which has nothing to do with where + * messages begin and end: the backend writes a continuous stream of length-prefixed messages, so + * unless a read happens to land exactly on a message boundary it ends part way through one — which, + * for a result of any size, is nearly every read. A message can also be longer than a single read + * (a wide row, or one large cell), and then spans several chunks. Only the message at the tail of a + * chunk can be incomplete, though, so at most one message per chunk needs bytes from another one. + * + * That message is reassembled into a buffer of its own; every other message is parsed in place, as + * a view over the chunk it arrived in. Neither buffer is ever written to again, so a message stays + * valid for as long as it is held — and holding one holds the whole chunk it came from. + */ export class Parser { - private buffer: Buffer = emptyBuffer - private bufferLength: number = 0 - private bufferOffset: number = 0 + // The bytes of a message that straddles two chunks, i.e. only ever one message’s worth. Every + // other message is parsed in place, out of the chunk it arrived in. + private partial: Buffer = emptyBuffer + private partialLength: number = 0 + // The straddling message’s full length, or -1 while its own header is still incomplete. + private partialTotal: number = -1 private reader = new BufferReader() private mode: Mode @@ -91,67 +113,93 @@ export class Parser { this.mode = opts?.mode || 'text' } - public parse(buffer: Buffer, callback: MessageCallback) { - this.mergeBuffer(buffer) - const bufferFullLength = this.bufferOffset + this.bufferLength - let offset = this.bufferOffset - while (offset + HEADER_LENGTH <= bufferFullLength) { + public parse(chunk: Buffer, callback: MessageCallback) { + const chunkLength = chunk.byteLength + // finish the message the previous chunk cut in half, if any, before parsing this one + let offset = this.partialLength > 0 ? this.completePartial(chunk, callback) : 0 + while (offset + HEADER_LENGTH <= chunkLength) { // code is 1 byte long - it identifies the message type - const code = this.buffer[offset] + const code = chunk[offset] // length is 1 Uint32BE - it is the length of the message EXCLUDING the code - const length = this.buffer.readUInt32BE(offset + CODE_LENGTH) + const length = chunk.readUInt32BE(offset + CODE_LENGTH) const fullMessageLength = CODE_LENGTH + length - if (fullMessageLength + offset <= bufferFullLength) { - const message = this.handlePacket(offset + HEADER_LENGTH, code, length, this.buffer) + if (fullMessageLength + offset <= chunkLength) { + const message = this.handlePacket(offset + HEADER_LENGTH, code, length, chunk) callback(message) offset += fullMessageLength } else { break } } - if (offset === bufferFullLength) { - // No more use for the buffer - this.buffer = emptyBuffer - this.bufferLength = 0 - this.bufferOffset = 0 - } else { - // Adjust the cursors of remainingBuffer - this.bufferLength = bufferFullLength - offset - this.bufferOffset = offset + if (offset < chunkLength) { + this.startPartial(chunk, offset) } } - private mergeBuffer(buffer: Buffer): void { - if (this.bufferLength > 0) { - const newLength = this.bufferLength + buffer.byteLength - const newFullLength = newLength + this.bufferOffset - if (newFullLength > this.buffer.byteLength) { - // We can't concat the new buffer with the remaining one - let newBuffer: Buffer - if (newLength <= this.buffer.byteLength && this.bufferOffset >= this.bufferLength) { - // We can move the relevant part to the beginning of the buffer instead of allocating a new buffer - newBuffer = this.buffer - } else { - // Allocate a new larger buffer - let newBufferLength = this.buffer.byteLength * 2 - while (newLength >= newBufferLength) { - newBufferLength *= 2 - } - newBuffer = Buffer.allocUnsafe(newBufferLength) - } - // Move the remaining buffer to the new one - this.buffer.copy(newBuffer, 0, this.bufferOffset, this.bufferOffset + this.bufferLength) - this.buffer = newBuffer - this.bufferOffset = 0 + /** Copies the trailing bytes of `chunk` that do not yet form a whole message into their own buffer. */ + private startPartial(chunk: Buffer, offset: number): void { + const remaining = chunk.byteLength - offset + this.partial = emptyBuffer + this.partialLength = 0 + // the length is only known once the header is complete + this.partialTotal = remaining >= HEADER_LENGTH ? CODE_LENGTH + chunk.readUInt32BE(offset + CODE_LENGTH) : -1 + this.growPartial(remaining) + chunk.copy(this.partial, 0, offset) + this.partialLength = remaining + } + + /** + * Fills the straddling message from the head of `chunk` and emits it once whole, returning the + * offset in `chunk` where parsing continues. + */ + private completePartial(chunk: Buffer, callback: MessageCallback): number { + let consumed = 0 + if (this.partialTotal === -1) { + // the header itself was split, so complete it before its length can be read + consumed = Math.min(HEADER_LENGTH - this.partialLength, chunk.byteLength) + chunk.copy(this.partial, this.partialLength, 0, consumed) + this.partialLength += consumed + if (this.partialLength < HEADER_LENGTH) { + return consumed } - // Concat the new buffer with the remaining one - buffer.copy(this.buffer, this.bufferOffset + this.bufferLength) - this.bufferLength = newLength - } else { - this.buffer = buffer - this.bufferOffset = 0 - this.bufferLength = buffer.byteLength + this.partialTotal = CODE_LENGTH + this.partial.readUInt32BE(CODE_LENGTH) } + if (this.partialLength < this.partialTotal) { + const available = Math.min(this.partialTotal - this.partialLength, chunk.byteLength - consumed) + this.growPartial(this.partialLength + available) + chunk.copy(this.partial, this.partialLength, consumed, consumed + available) + this.partialLength += available + consumed += available + if (this.partialLength < this.partialTotal) { + return consumed + } + } + // hand the buffer off to the message and forget it, so that nothing is ever parsed twice out + // of the same bytes and the message stays valid for as long as it is held + const partial = this.partial + this.partial = emptyBuffer + this.partialLength = 0 + this.partialTotal = -1 + const message = this.handlePacket(HEADER_LENGTH, partial[0], partial.readUInt32BE(CODE_LENGTH), partial) + callback(message) + return consumed + } + + /** Sizes `partial` to hold at least `needed` bytes, i.e. the whole message when its length is known. */ + private growPartial(needed: number): void { + if (needed <= this.partial.byteLength) { + return + } + // grow geometrically, jumping straight to the message's full length when that is known and + // modest, and always keeping room for the header its length is read from + const whole = this.partialTotal === -1 ? HEADER_LENGTH : Math.min(this.partialTotal, MAX_EAGER_MESSAGE_LENGTH) + let capacity = Math.max(needed, whole, this.partial.byteLength * 2) + if (this.partialTotal !== -1 && capacity > this.partialTotal) { + capacity = this.partialTotal + } + const grown = Buffer.allocUnsafe(capacity) + this.partial.copy(grown, 0, 0, this.partialLength) + this.partial = grown } private handlePacket(offset: number, code: number, length: number, bytes: Buffer): BackendMessage { @@ -188,7 +236,8 @@ export class Parser { message = emptyQuery break case MessageCodes.DataRow: - message = parseDataRowMessage(reader) + // the cells are decoded lazily, on the first read of `message.fields` + message = new DataRowMessage(LATEINIT_LENGTH, bytes, offset) break case MessageCodes.CommandComplete: message = parseCommandCompleteMessage(reader) @@ -305,17 +354,6 @@ const parseParameterDescriptionMessage = (reader: BufferReader) => { return message } -const parseDataRowMessage = (reader: BufferReader) => { - const fieldCount = reader.int16() - const fields: any[] = new Array(fieldCount) - for (let i = 0; i < fieldCount; i++) { - const len = reader.int32() - // a -1 for length means the value of the field is null - fields[i] = len === -1 ? null : reader.string(len) - } - return new DataRowMessage(LATEINIT_LENGTH, fields) -} - const parseParameterStatusMessage = (reader: BufferReader) => { const name = reader.cstring() const value = reader.cstring()