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()