diff --git a/packages/pg-protocol/src/inbound-parser.test.ts b/packages/pg-protocol/src/inbound-parser.test.ts
index 8687194c3..cfd8562d3 100644
--- a/packages/pg-protocol/src/inbound-parser.test.ts
+++ b/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, () => {})
diff --git a/packages/pg-protocol/src/index.ts b/packages/pg-protocol/src/index.ts
index 703ff2e49..e384417d9 100644
--- a/packages/pg-protocol/src/index.ts
+++ b/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()))
 }
diff --git a/packages/pg-protocol/src/messages.ts b/packages/pg-protocol/src/messages.ts
index c3fbbdd9b..b7b6a7ddb 100644
--- a/packages/pg-protocol/src/messages.ts
+++ b/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
+  }
+}
diff --git a/packages/pg-protocol/src/parser.ts b/packages/pg-protocol/src/parser.ts
index df48ca4a1..2a503d1e0 100644
--- a/packages/pg-protocol/src/parser.ts
+++ b/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)
diff --git a/packages/pg/lib/client.js b/packages/pg/lib/client.js
index 2b13c1de7..4d1979ab9 100644
--- a/packages/pg/lib/client.js
+++ b/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 = []
diff --git a/packages/pg/lib/connection.js b/packages/pg/lib/connection.js
index 85f3113a1..86abe46a0 100644
--- a/packages/pg/lib/connection.js
+++ b/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() {
diff --git a/packages/pg/lib/query.js b/packages/pg/lib/query.js
index 3b93cb5ba..f59dcab37 100644
--- a/packages/pg/lib/query.js
+++ b/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
diff --git a/packages/pg/lib/result.js b/packages/pg/lib/result.js
index 329fbf9fc..833e294df 100644
--- a/packages/pg/lib/result.js
+++ b/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 }
diff --git a/packages/pg/test/integration/client/rows-from-bytes-tests.js b/packages/pg/test/integration/client/rows-from-bytes-tests.js
new file mode 100644
index 000000000..3a527d760
--- /dev/null
+++ b/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()
+  }
+})
diff --git a/packages/pg/test/unit/client/parse-data-row-tests.js b/packages/pg/test/unit/client/parse-data-row-tests.js
new file mode 100644
index 000000000..203bc5bc6
--- /dev/null
+++ b/packages/pg/test/unit/client/parse-data-row-tests.js
@@ -0,0 +1,146 @@
+'use strict'
+const helper = require('./test-helper')
+const assert = require('assert')
+const EventEmitter = require('events')
+const types = require('pg-types')
+const { Parser } = require('pg-protocol/dist/parser')
+const Connection = require('../../../lib/connection')
+const Result = require('../../../lib/result')
+const buffers = require('../../test-buffers')
+
+const suite = new helper.Suite()
+const test = suite.test.bind(suite)
+
+const INT2 = 21
+const INT4 = 23
+const OID = 26
+const BOOL = 16
+const TEXT = 25
+const FLOAT8 = 701
+
+const makeResult = (oids, rowMode, typeOverrides, ResultClass = Result) => {
+  const result = new ResultClass(rowMode, typeOverrides)
+  result.addFields(oids.map((oid, i) => ({ name: 'c' + i, dataTypeID: oid, format: 'text' })))
+  return result
+}
+
+const parseWire = (result, values) => {
+  let row
+  new Parser({ lazyRows: true }).parse(buffers.dataRow(values), (msg) => {
+    row = result.parseDataRow(msg)
+  })
+  return row
+}
+
+const assertSameAsParseRow = (oids, values) => {
+  const result = makeResult(oids)
+  assert.deepStrictEqual(parseWire(result, values), result.parseRow(values))
+}
+
+test('integers read from the bytes are the same as parsed from the text', function () {
+  const values = ['0', '7', '-1', '2147483647', '-2147483648']
+  assertSameAsParseRow(
+    values.map(() => INT4),
+    values
+  )
+  assertSameAsParseRow([INT2, INT2], ['32767', '-32768'])
+  assertSameAsParseRow([OID, OID], ['0', '4294967295'])
+})
+
+test('booleans read from the bytes are the same as parsed from the text', function () {
+  assertSameAsParseRow([BOOL, BOOL], ['t', 'f'])
+})
+
+test('text read from the bytes is the same as parsed from the text', function () {
+  assertSameAsParseRow([TEXT, TEXT, TEXT], ['', 'wat', 'è€😀'])
+})
+
+test('null is null for every kind of column', function () {
+  assertSameAsParseRow([INT4, BOOL, TEXT, FLOAT8], [null, null, null, null])
+})
+
+test('types with no shortcut use their parser', function () {
+  assertSameAsParseRow([FLOAT8, INT4, TEXT], ['1.5', '2', 'x'])
+})
+
+test('a parser from setTypeParser gets the string', function () {
+  const original = types.getTypeParser(INT4, 'text')
+  const seen = []
+  types.setTypeParser(INT4, (value) => {
+    seen.push(value)
+    return 'parsed'
+  })
+  try {
+    assert.deepStrictEqual(parseWire(makeResult([INT4]), ['12']), { c0: 'parsed' })
+    assert.deepStrictEqual(seen, ['12'])
+  } finally {
+    types.setTypeParser(INT4, original)
+  }
+})
+
+test('a parser set before pg is loaded gets the string', function () {
+  const path = require.resolve('../../../lib/result')
+  const cached = require.cache[path]
+  const original = types.getTypeParser(BOOL, 'text')
+  types.setTypeParser(BOOL, (value) => 'parsed ' + value)
+  delete require.cache[path]
+  try {
+    const row = parseWire(makeResult([BOOL], undefined, undefined, require(path)), ['t'])
+    assert.deepStrictEqual(row, { c0: 'parsed t' })
+  } finally {
+    types.setTypeParser(BOOL, original)
+    require.cache[path] = cached
+  }
+})
+
+test('a text parser set before pg is loaded gets the string', function () {
+  const path = require.resolve('../../../lib/result')
+  const cached = require.cache[path]
+  types.setTypeParser(TEXT, (value) => 'parsed ' + value)
+  delete require.cache[path]
+  try {
+    const row = parseWire(makeResult([TEXT], undefined, undefined, require(path)), ['x'])
+    assert.deepStrictEqual(row, { c0: 'parsed x' })
+  } finally {
+    types.setTypeParser(TEXT, types.getTypeParser(TEXT, 'unknown format'))
+    require.cache[path] = cached
+  }
+})
+
+test('parsers of a query override are used', function () {
+  const overrides = { getTypeParser: (oid) => (oid === INT4 ? (value) => 'parsed ' + value : String) }
+  assert.deepStrictEqual(parseWire(makeResult([INT4], undefined, overrides), ['12']), { c0: 'parsed 12' })
+})
+
+test('array mode returns an array', function () {
+  assert.deepStrictEqual(parseWire(makeResult([INT4, TEXT], 'array'), ['12', 'x']), [12, 'x'])
+})
+
+test('a row kept by another listener does not depend on the socket buffer', function () {
+  const stream = new EventEmitter()
+  const con = new Connection({ stream, lazyRows: true })
+  con.on('dataRow', () => {}) // The client
+  let kept
+  con.on('dataRow', (msg) => {
+    kept = msg
+  })
+  con.attachListeners(stream)
+  const buffer = buffers.dataRow(['kept'])
+  stream.emit('data', buffer)
+  buffer.fill(0)
+  assert.deepStrictEqual(kept.fields, ['kept'])
+})
+
+test('a connection without a client keeps eager rows', function () {
+  const stream = new EventEmitter()
+  const con = new Connection({ stream })
+  let kept
+  con.on('dataRow', (msg) => {
+    kept = msg
+  })
+  con.attachListeners(stream)
+  const buffer = buffers.dataRow(['kept'])
+  stream.emit('data', buffer)
+  buffer.fill(0)
+  assert.deepStrictEqual(kept.fields, ['kept'])
+})