fix(pg-protocol): keep the bytes of binary columns instead of decoding them as text

packages/pg-protocol/src/buffer-reader.ts

@@ -51,7 +51,9 @@ export class BufferReader {
   }
 
   public bytes(length: number): Buffer {
-    const result = this.buffer.slice(this.offset, this.offset + length)
+    // a copy, not a view: the parser reuses its buffer for the next chunk, and a view would
+    // change under a message that was already delivered
+    const result = Buffer.from(this.buffer.subarray(this.offset, this.offset + length))
     this.offset += length
     return result
   }

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

@@ -577,6 +577,35 @@ describe('PgPacketStream', function () {
     })
   })
 
+  // the row description says which columns are binary, and their values are bytes rather than
+  // text: decoding them as utf8 would lose every byte it cannot carry
+  it('keeps the bytes of a binary column', async function () {
+    const description = buffers.rowDescription([
+      { name: 'n', dataTypeID: 23, formatCode: 1 },
+      { name: 't', dataTypeID: 25, formatCode: 0 },
+    ])
+    const row = new BufferList()
+      .addInt16(2)
+      .addInt32(4)
+      .add(Buffer.from([0, 0, 0x03, 0xe8]))
+      .addInt32(2)
+      .add(Buffer.from('é', 'utf8'))
+      .join(true, 'D')
+    const messages = await parseBuffers([description, row])
+    assert.deepStrictEqual((messages[1] as any).fields, [Buffer.from([0, 0, 0x03, 0xe8]), 'é'])
+  })
+
+  // the parser moves what is left of a chunk to the front of its buffer before reading the next
+  // one, so a message that kept a view into that buffer would change after being delivered
+  it('keeps a copyData chunk intact after the parser reuses its buffer', async function () {
+    const copyData = buffers.copyData(Buffer.alloc(64, 0xaa))
+    const commandComplete = buffers.commandComplete('COPY 1')
+    const fullBuffer = Buffer.concat([copyData, commandComplete])
+    const messages = await parseBuffers([fullBuffer.subarray(0, fullBuffer.length - 1), fullBuffer.subarray(-1)])
+    assert.strictEqual(messages.length, 2)
+    assert.deepEqual((messages[0] as any).chunk, Buffer.alloc(64, 0xaa))
+  })
+
   it('cleans up the reader after handling a packet', function () {
     const parser = new Parser()
     parser.parse(oneFieldBuf, () => {})

packages/pg-protocol/src/parser.ts

@@ -83,6 +83,9 @@ export class Parser {
   private bufferOffset: number = 0
   private reader = new BufferReader()
   private mode: Mode
+  // which columns of the rows to come are in the binary format, from the last row description:
+  // a binary value is bytes, and decoding it as text would lose every byte utf8 cannot carry
+  private binaryColumns: boolean[] | null = null
 
   constructor(opts?: StreamOptions) {
     if (opts?.mode === 'binary') {
@@ -188,7 +191,7 @@ export class Parser {
         message = emptyQuery
         break
       case MessageCodes.DataRow:
-        message = parseDataRowMessage(reader)
+        message = parseDataRowMessage(reader, this.binaryColumns)
         break
       case MessageCodes.CommandComplete:
         message = parseCommandCompleteMessage(reader)
@@ -216,6 +219,7 @@ export class Parser {
         break
       case MessageCodes.RowDescriptionMessage:
         message = parseRowDescriptionMessage(reader)
+        this.binaryColumns = binaryColumnsOf(message as RowDescriptionMessage)
         break
       case MessageCodes.ParameterDescriptionMessage:
         message = parseParameterDescriptionMessage(reader)
@@ -276,6 +280,12 @@ const parseNotificationMessage = (reader: BufferReader) => {
   return new NotificationResponseMessage(LATEINIT_LENGTH, processId, channel, payload)
 }
 
+// null when every column is text, which is nearly always, so the row parser has one check to make
+const binaryColumnsOf = (message: RowDescriptionMessage): boolean[] | null => {
+  const formats = message.fields.map((field) => field.format === 'binary')
+  return formats.includes(true) ? formats : null
+}
+
 const parseRowDescriptionMessage = (reader: BufferReader) => {
   const fieldCount = reader.int16()
   const message = new RowDescriptionMessage(LATEINIT_LENGTH, fieldCount)
@@ -306,13 +316,13 @@ const parseParameterDescriptionMessage = (reader: BufferReader) => {
   return message
 }
 
-const parseDataRowMessage = (reader: BufferReader) => {
+const parseDataRowMessage = (reader: BufferReader, binaryColumns: boolean[] | null) => {
   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)
+    fields[i] = len === -1 ? null : binaryColumns && binaryColumns[i] ? reader.bytes(len) : reader.string(len)
   }
   return new DataRowMessage(LATEINIT_LENGTH, fields)
 }

packages/pg/test/integration/client/binary-results-tests.js

@@ -0,0 +1,25 @@
+'use strict'
+const helper = require('./test-helper')
+const assert = require('assert')
+const suite = new helper.Suite()
+
+const Client = helper.Client
+
+// a binary value is bytes, and any byte utf8 cannot carry used to be lost between the parser
+// and the type parsers: 1000 came back as 1007
+suite.test('binary results keep every byte of a value', async function () {
+  const client = new Client(helper.config)
+  await client.connect()
+  try {
+    const result = await client.query({
+      text: 'SELECT $1::int4 AS n, $2::float8 AS f, $3::text AS t',
+      values: [1000, -2.5, 'é'],
+      binary: true,
+    })
+    assert.strictEqual(result.rows[0].n, 1000)
+    assert.strictEqual(result.rows[0].f, -2.5)
+    assert.strictEqual(result.rows[0].t, 'é')
+  } finally {
+    await client.end()
+  }
+})