diff --git a/README.md b/README.md
index ec8e955e..4ca77c87 100644
--- a/README.md
+++ b/README.md
@@ -430,6 +430,12 @@ function processRows() {
 }
 ```
 
+Nothing is emitted while the request is paused: `row`, `recordset`, `info`, `error`, `rowsaffected` and `done`
+ events wait until it is resumed. The msnodesqlv8 driver reads rows 50 at a time and only stops reading at the end
+ of a full batch, so the rows it reads before then are held until the request is resumed. A paused request that is
+ cancelled does not wait to be resumed: it drops the rows it holds, emits the other events it holds, and completes
+ with an `error` (`ECANCEL`) and `done` (see [cancel](#cancel)).
+
 ## Connection Pools
 
 An important concept to understand when using this library is [Connection Pooling](https://en.wikipedia.org/wiki/Connection_pool) as this library uses connection pooling extensively. As one Node JS process is able to handle multiple requests at once, we can take advantage of this long running process to create a pool of database connections for reuse; this saves overhead of connecting to the database for each request
@@ -1271,7 +1277,7 @@ request.cancel()
 
 A cancelled stream emits an `error` event with code `ECANCEL`, followed by `done`.
 
-With the `msnodesqlv8` driver, a request that is already running can only be cancelled from msnodesqlv8 5.6. Earlier versions can cancel a paused stream, but let any other running request run to the end and complete as if `cancel()` had not been called.
+With the `msnodesqlv8` driver, a request that is already running can only be cancelled from msnodesqlv8 5.6. Earlier versions can cancel a paused stream, but let any other running request run to the end and complete as if `cancel()` had not been called. A paused stream is cancelled once the driver stops reading, at the end of a batch of 50 rows or the start of the next result set; a query that ends first, or times out, completes as cancelled.
 
 ## Transaction
 
diff --git a/lib/msnodesqlv8/request.js b/lib/msnodesqlv8/request.js
index c6711c1f..c2a6ae71 100644
--- a/lib/msnodesqlv8/request.js
+++ b/lib/msnodesqlv8/request.js
@@ -14,6 +14,12 @@ const JSON_COLUMN_ID = 'JSON_F52E2B61-18A1-11d1-B105-00805F49916B'
 const XML_COLUMN_ID = 'XML_F52E2B61-18A1-11d1-B105-00805F49916B'
 const EMPTY_BUFFER = Buffer.alloc(0)
 
+// msnodesqlv8 fetches a result set's rows this many at a time. It is only safe to pause once
+// it has handed over every row of the batch it fetched: paused part way through a result set's
+// last, short batch, it moves on to the next result set, or ends the query, with the rest of
+// that batch still to come.
+const DRIVER_ROW_BATCH_SIZE = 50
+
 // String() converts a Symbol, which only fails to convert implicitly; an object without a
 // working toString falls back to its tag
 const thrownMessage = function (value) {
@@ -347,6 +353,39 @@ class Request extends BaseRequest {
       // the query can go on emitting after it, so the listeners ignore anything that follows
       let finished = false
 
+      // A stream goes on receiving rows after it is paused, until the driver reaches the end of a
+      // full batch (see DRIVER_ROW_BATCH_SIZE). What the stream emits waits here while it is
+      // paused, and is emitted in the order it arrived once it is resumed.
+      const pending = []
+      let rowsInRecordset = 0
+      let driverPaused = false
+      // a cancel of a paused stream waits for the driver to reach a point where it can be paused
+      let cancelDeferred = false
+      // msnodesqlv8 has freed the statement, which it does without 'done' after an error ends the query
+      let freed = false
+      let resuming = false
+      // a refused cancel undoes only the latest one
+      let cancelAttempts = 0
+
+      const emitInOrder = (isRow, emit) => {
+        if (this.stream && (this.paused || pending.length)) {
+          pending.push({ isRow, emit })
+        } else {
+          emit()
+        }
+      }
+
+      // once a cancel is through, what is waiting is emitted at once, but for the rows of a
+      // paused stream, which are dropped
+      const emitPendingOnCancel = () => {
+        // decided once, so that a listener that resumes the stream part way through cannot let
+        // the rows of the cancelled stream through
+        const dropRows = this.paused
+        for (const { isRow, emit } of pending.splice(0)) {
+          if (!isRow || !dropRows) emit()
+        }
+      }
+
       const handleError = (req, connection, info, moreErrors) => {
         // a cancel is reported from 'done'. msnodesqlv8 >= 5.6 also sends an 'Operation canceled'
         // error after a cancel, and one that arrives before 'done' is dropped here
@@ -359,7 +398,7 @@ class Request extends BaseRequest {
         err.code = 'EREQUEST'
 
         if (this.stream) {
-          this.emit('error', err)
+          emitInOrder(false, () => this.emit('error', err))
         } else {
           if (doReturn && !hasReturned) {
             if (req) {
@@ -493,30 +532,108 @@ class Request extends BaseRequest {
 
         this._setCurrentRequest(req)
 
-        this._cancel = () => {
-          debug('request(%d): cancel', IDS.get(this))
-          // newer msnodesqlv8 versions drop a repeat cancel without calling back, so the
-          // outcome of the first one stands
-          if (typeof req.getCancelSent === 'function' && req.getCancelSent()) {
-            return
-          }
-          cancelRequested = true
+        const cancelDriver = (attempt = cancelAttempts) => {
           req.cancelQuery(err => {
             if (err) {
               debug('request(%d): failed to cancel', IDS.get(this), err)
               // msnodesqlv8 < 5.6 refuses to cancel a query that is neither paused nor polling,
-              // and the query runs to the end
-              if (!hasReturned) cancelRequested = false
+              // and the query runs to the end; if it reached the end of a batch meanwhile, a
+              // stream that is no longer paused resumes it
+              if (!hasReturned && attempt === cancelAttempts) {
+                cancelRequested = false
+                this._resumeStream()
+              }
             }
-            // this fixes an issue where paused connections don't emit a done event
+            // this fixes an issue where paused connections don't emit a done event; a query whose
+            // cancel was refused goes on, though it may have paused at the end of a batch since
             try {
-              if (req.isPaused()) req.emit('done')
+              if (!err && req.isPaused()) req.emit('done')
             } catch (err) {
               // do nothing
             }
           })
         }
 
+        // every version of msnodesqlv8 cancels a paused query; it is paused here only where it stops
+        // for a pause, with nothing of its own under way
+        const pauseAndCancelDriver = () => {
+          cancelDeferred = false
+          // msnodesqlv8 5.0 to 5.5 send a query one cancel, which a cancel made before the query
+          // started has used up and they refused, so the query runs on
+          if (typeof req.getCancelSent === 'function' && req.getCancelSent()) {
+            cancelRequested = false
+            return false
+          }
+          driverPaused = true
+          req.pauseQuery()
+          cancelDriver()
+          return true
+        }
+
+        // called with each row once it is complete
+        const pauseDriverAtEndOfBatch = () => {
+          if (driverPaused || rowsInRecordset % DRIVER_ROW_BATCH_SIZE !== 0) return
+          if (cancelDeferred && pauseAndCancelDriver()) return
+          if (this.stream && this.paused) {
+            driverPaused = true
+            req.pauseQuery()
+          }
+        }
+
+        // until the first result set the driver has read nothing, so a stream paused before then,
+        // or before its query starts, holds it back at once
+        this._pauseStream = () => {
+          if (!columns && !driverPaused && !finished) {
+            driverPaused = true
+            req.pauseQuery()
+          }
+        }
+        if (this.stream && this.paused) this._pauseStream()
+
+        this._resumeStream = () => {
+          // a listener that resumes the stream while it is emitting what was waiting leaves the
+          // rest to this loop, so that every listener gets the events in order
+          if (resuming) return
+          resuming = true
+          try {
+            while (pending.length && !this.paused) {
+              pending.shift().emit()
+            }
+          } finally {
+            resuming = false
+          }
+          // cancelled while it emitted what was waiting after the query ended, and then paused, a
+          // stream completes as cancelled straight away, as it does when it is paused first
+          if (finished && cancelRequested && this.paused && pending.length) emitPendingOnCancel()
+          // a query that is being cancelled stays paused, so that the cancel can go through
+          if (!pending.length && !this.paused && driverPaused && !finished && !cancelRequested) {
+            driverPaused = false
+            req.resumeQuery()
+          }
+        }
+
+        this._cancel = () => {
+          debug('request(%d): cancel', IDS.get(this))
+          // newer msnodesqlv8 versions drop a repeat cancel without calling back, so the
+          // outcome of the first one stands
+          if (cancelDeferred || (typeof req.getCancelSent === 'function' && req.getCancelSent())) {
+            return
+          }
+          cancelRequested = true
+          cancelAttempts++
+          if (!this.stream || !this.paused || driverPaused) {
+            return cancelDriver()
+          }
+          // A paused stream holds the driver back only once it reaches the end of a batch. Until
+          // then the driver goes on reading, and pausing or cancelling it part way through a fetch,
+          // or as it moves to the next result set or ends the query, can leave its queue broken.
+          // So it is paused and cancelled when it next stops, at the end of a batch or the start
+          // of a result set, and a query that ends first, at the latest on the request timeout,
+          // is reported as cancelled.
+          if (freed) return req.emit('done')
+          cancelDeferred = true
+        }
+
         req.on('meta', metadata => {
           if (finished) return
 
@@ -536,7 +653,7 @@ class Request extends BaseRequest {
                   const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON')
 
                   if (this.stream) {
-                    this.emit('error', ex2)
+                    emitInOrder(false, () => this.emit('error', ex2))
                   } else {
                     console.error(ex2)
                   }
@@ -548,12 +665,14 @@ class Request extends BaseRequest {
               chunksBuffer = null
               if (row && row.___return___ == null) {
                 // row with ___return___ col is the last row
-                if (this.stream && !this.paused) this.emit('row', row)
+                const last = row
+                if (this.stream) emitInOrder(true, () => this.emit('row', last))
               }
             }
           }
 
           row = null
+          rowsInRecordset = 0
           columns = metadata
           recordset = []
 
@@ -588,16 +707,23 @@ class Request extends BaseRequest {
           }
           if (this.stream) {
             if (!hasReturnColumn) {
-              this.emit('recordset', recordset.columns)
+              const recordsetColumns = recordset.columns
+              emitInOrder(false, () => this.emit('recordset', recordsetColumns))
             }
           } else {
             recordsets.push(recordset)
           }
           if (this.arrayRowMode) recordsetcolumns.push(recordset.columns)
+
+          // the driver fetches nothing of a result set it is paused at the start of
+          if (cancelDeferred) pauseAndCancelDriver()
         })
 
         req.on('row', rownumber => {
-          if (finished || (row && isChunkedRecordset)) return
+          if (finished) return
+
+          rowsInRecordset++
+          if (row && isChunkedRecordset) return
 
           if (this.arrayRowMode) {
             row = []
@@ -642,11 +768,14 @@ class Request extends BaseRequest {
               }
             }
             if (!hasReturnColumn) {
-              if (this.stream && !this.paused && idx === columns.length - 1) {
-                this.emit('row', row)
+              if (this.stream && idx === columns.length - 1) {
+                const complete = row
+                emitInOrder(true, () => this.emit('row', complete))
               }
             }
           }
+
+          if (idx === columns.length - 1) pauseDriverAtEndOfBatch()
         })
 
         req.on('rowcount', rowCount => {
@@ -654,7 +783,7 @@ class Request extends BaseRequest {
 
           rowsAffected.push(rowCount)
           if (this.stream) {
-            this.emit('rowsaffected', rowCount)
+            emitInOrder(false, () => this.emit('rowsaffected', rowCount))
           }
         })
 
@@ -666,7 +795,7 @@ class Request extends BaseRequest {
             msg.message = parsedMessage[1]
           }
 
-          this.emit('info', {
+          const info = {
             message: msg.message,
             number: msg.code,
             state: msg.sqlstate,
@@ -674,7 +803,8 @@ class Request extends BaseRequest {
             lineNumber: msg.lineNumber || 0,
             serverName: msg.serverName,
             procName: msg.procName
-          })
+          }
+          emitInOrder(false, () => this.emit('info', info))
 
           // query terminated
           if (msg.code === 3621 && !hasReturned) {
@@ -706,6 +836,14 @@ class Request extends BaseRequest {
           handleError(req, connection, info, moreErrors)
         })
 
+        req.on('free', () => {
+          freed = true
+          if (cancelDeferred && !finished) {
+            cancelDeferred = false
+            req.emit('done')
+          }
+        })
+
         req.once('done', () => {
           finished = true
 
@@ -729,6 +867,9 @@ class Request extends BaseRequest {
 
             const err = new RequestError('Canceled.', 'ECANCEL')
             if (this.stream) {
+              delete this._resumeStream
+              delete this._pauseStream
+              emitPendingOnCancel()
               // as with tedious, a cancelled stream reports the cancel as an error and still
               // completes with what was read up to it
               this.emit('error', err)
@@ -770,7 +911,7 @@ class Request extends BaseRequest {
                     const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON')
 
                     if (this.stream) {
-                      this.emit('error', ex2)
+                      emitInOrder(false, () => this.emit('error', ex2))
                     } else {
                       console.error(ex2)
                     }
@@ -782,7 +923,8 @@ class Request extends BaseRequest {
                 chunksBuffer = null
                 if (row && row.___return___ == null) {
                   // row with ___return___ col is the last row
-                  if (this.stream && !this.paused) { this.emit('row', row) }
+                  const last = row
+                  if (this.stream) emitInOrder(true, () => this.emit('row', last))
                 }
               }
             }
@@ -803,14 +945,32 @@ class Request extends BaseRequest {
             }
           }
 
-          delete this._cancel
           this.parent.release(connection)
 
           debug('request(%d): completed', IDS.get(this))
 
           if (this.stream) {
-            callback(null, this._nested ? row : null, output, rowsAffected, recordsetcolumns)
+            // the driver is through, but a paused stream completes once it has emitted what is
+            // still waiting. Cancelled before then, it completes as cancelled: at once if it is
+            // paused, dropping its rows, or else once it has emitted the rest
+            this._cancel = () => {
+              cancelRequested = true
+              if (this.paused) emitPendingOnCancel()
+            }
+            const last = this._nested ? row : null
+            emitInOrder(false, () => {
+              delete this._cancel
+              delete this._resumeStream
+              delete this._pauseStream
+              if (cancelRequested) {
+                debug('request(%d): canceled', IDS.get(this))
+                this.emit('error', new RequestError('Canceled.', 'ECANCEL'))
+                return callback(null, null, output, rowsAffected, recordsetcolumns)
+              }
+              callback(null, last, output, rowsAffected, recordsetcolumns)
+            })
           } else {
+            delete this._cancel
             callback(null, recordsets, output, rowsAffected, recordsetcolumns)
           }
         })
@@ -964,17 +1124,18 @@ class Request extends BaseRequest {
     })
   }
 
+  // the query pauses the driver itself, mostly at the end of a batch (see DRIVER_ROW_BATCH_SIZE)
   _pause () {
     super._pause()
-    if (this._currentRequest) {
-      this._currentRequest.pauseQuery()
+    if (this._pauseStream) {
+      this._pauseStream()
     }
   }
 
   _resume () {
     super._resume()
-    if (this._currentRequest) {
-      this._currentRequest.resumeQuery()
+    if (this._resumeStream) {
+      this._resumeStream()
     }
   }
 }
diff --git a/test/common/tests.js b/test/common/tests.js
index a586acf8..eb542ee9 100644
--- a/test/common/tests.js
+++ b/test/common/tests.js
@@ -670,6 +670,39 @@ module.exports = (sql, driver) => {
       req.pipe(blockingTransformStream).pipe(wstream)
     },
 
+    'query with pipe and back pressure across partial row batches' (done) {
+      // msnodesqlv8 fetches rows 50 at a time, so each of these result sets ends with a short
+      // batch, or a batch of one row, that the stream is paused part way through or at the end of
+      const recordsets = [['a', 75], ['b', 51], ['c', 120]]
+      const expected = recordsets.flatMap(([column, count]) => Array.from({ length: count }, (_, i) => `${column}=${i + 1}`))
+      const rows = []
+      // takes one row at a time, and more slowly than the driver fetches a batch, so the request
+      // is paused throughout, including when a batch of one row arrives
+      const wstream = new stream.Writable({
+        objectMode: true,
+        highWaterMark: 1,
+        write (row, encoding, cb) {
+          rows.push(Object.entries(row).map(([column, value]) => `${column}=${value}`).join())
+          setTimeout(cb, 10)
+        }
+      })
+      wstream.on('finish', () => {
+        try {
+          assert.deepStrictEqual(rows, expected, 'every row should arrive once, in order, under its own column')
+          done()
+        } catch (err) {
+          done(err)
+        }
+      })
+      wstream.on('error', done)
+
+      const req = new sql.Request()
+      req.query(recordsets.map(([column, count]) =>
+        `select top (${count}) cast(row_number() over (order by object_id, column_id) as int) as ${column} from sys.all_columns order by object_id, column_id`
+      ).join(';'))
+      req.pipe(wstream)
+    },
+
     'query with pipe and an error' (endsTheBatch, done) {
       // pipe() returns the destination, not the stream it puts in front of it, so the
       // destination is the only place a request's error can be handled
diff --git a/test/common/unit.js b/test/common/unit.js
index 1cbae175..3ddfe670 100644
--- a/test/common/unit.js
+++ b/test/common/unit.js
@@ -2301,13 +2301,16 @@ describe('msnodesqlv8 query events', () => {
     assert.deepStrictEqual(recordset, [{ a: 1 }], 'the query should complete')
   })
 
-  // every msnodesqlv8 version cancels a paused query, and ends it with 'done' before calling back
+  // every msnodesqlv8 version cancels a paused query, and ends it with 'done' before calling back;
+  // a stream paused part way through a result set is cancelled once the driver has handed over
+  // the batch, so this query, whose batch is its last, ends first
   it('should stream a cancel of a paused query once, then complete with what was read', async () => {
     let paused = false
     const { request, pool } = setup(notify => {
       result('a', 1)(notify)
       notify.emit('rowcount', 1)
       result('b', 2)(notify)
+      setImmediate(() => notify.emit('done'))
     }, {
       isPaused: () => paused,
       pauseQuery () { paused = true },
@@ -2375,4 +2378,627 @@ describe('msnodesqlv8 query events', () => {
     await tick()
     assert.strictEqual(connection.hasError, true, 'a link failure after a cancel should mark the connection as broken')
   })
+
+  describe('stream back pressure', () => {
+    const { Writable, pipeline } = require('node:stream')
+    const BATCH = 50
+
+    // Plays `recordsets` ([column, row count] each) the way msnodesqlv8's reader does: it
+    // fetches a result set's rows BATCH at a time and hands them over until it is paused. What
+    // it does after a pause part way through a batch is what differs: up to 4.x it goes on as if
+    // the batch was handed over; 5.0 to 5.5 do too, and when resumed after the last batch of a
+    // result set, move on to the next one a second time; 5.6 waits for the rest of the batch,
+    // unless it was paused on the batch's last row. `timeOutAtFetch` names a fetch that does not
+    // come back before the request times out, and `failAtEnd` ends the query on an error; a query
+    // that ends on an error emits 'free' and no 'done', as msnodesqlv8 does. `beginDelay` holds
+    // back the end of the first statement, and `stallBegin` keeps it from ending.
+    const reader = (version, recordsets, { timeOutAtFetch, failAtEnd, beginDelay, stallBegin } = {}) => {
+      const pausedAt = []
+      const calls = []
+      let notify = null
+      let queried = false
+      let handle = false
+      let token = null
+      let cancelSent = false
+      let fetches = 0
+      let set = 0
+      let read = 0
+      let batch = null
+      let index = 0
+      let paused = false
+      let running = true
+
+      const meta = () => [{ name: recordsets[set][0], sqlType: 'int', size: 4, nullable: false }]
+
+      const fetch = () => {
+        const data = []
+        while (data.length < BATCH && read < recordsets[set][1]) data.push(++read)
+        // a fetch that comes back short has reached the end of the result set
+        return { data, endOfRows: data.length < BATCH }
+      }
+
+      const isPaused = () => paused
+
+      // handing over a row can pause the query, which stops the loop
+      const dispatchRows = () => {
+        if (!batch) return
+        while (!isPaused() && index < batch.data.length) {
+          notify.emit('row', index)
+          notify.emit('column', 0, batch.data[index++], false)
+        }
+      }
+
+      const nextResult = () => setImmediate(() => {
+        if (!running) return
+        set++
+        read = 0
+        if (set < recordsets.length) {
+          notify.emit('meta', meta())
+          dispatch()
+        } else if (failAtEnd) {
+          fail(Object.assign(new Error('Something went wrong'), { code: 50000, sqlstate: '42000' }))
+        } else {
+          running = false
+          notify.emit('done')
+          notify.emit('free')
+        }
+      })
+
+      const fail = err => {
+        running = false
+        notify.fail(err, false)
+        setImmediate(() => notify.emit('free'))
+      }
+
+      const dispatch = () => {
+        if (!running || paused) return
+        if (++fetches === timeOutAtFetch) {
+          return setTimeout(fail, 10, Object.assign(new Error('Query timeout expired'), { code: 0, sqlstate: 'HYT00' }))
+        }
+        setImmediate(() => {
+          if (!running) return
+          batch = fetch()
+          index = 0
+          dispatchRows()
+          if (version === '5.6' && paused && index < batch.data.length) return
+          if (batch.endOfRows) {
+            nextResult()
+          } else {
+            dispatch()
+          }
+        })
+      }
+
+      // every version cancels a paused query, and ends it with 'done' before calling back;
+      // before 5.6 a query that is neither paused nor polling is not cancelled
+      const sendCancel = callback => {
+        if (version !== '5.6' && !paused) {
+          calls.push('cancel refused')
+          return setImmediate(callback, new Error('Error: [msnodesql] cancel only supported for statements where polling is enabled.'))
+        }
+        calls.push('cancel')
+        running = false
+        setImmediate(() => {
+          notify.emit('done')
+          callback(null)
+        })
+      }
+
+      const query = {
+        isPaused,
+        pauseQuery () {
+          calls.push('pause')
+          pausedAt.push(batch ? `${index} of ${batch.data.length}` : 'before the first batch')
+          paused = true
+        },
+        resumeQuery () {
+          calls.push('resume')
+          paused = false
+          dispatchRows()
+          if (paused) return
+          if (version !== '2.x to 4.x' && batch && batch.endOfRows && index >= batch.data.length) {
+            nextResult()
+          } else {
+            dispatch()
+          }
+        },
+        // from 5.0 a query is sent one cancel: one made before the query has a statement waits
+        // for it, and a repeat is dropped without calling back
+        cancelQuery (callback) {
+          if (version !== '2.x to 4.x') {
+            if (cancelSent) return calls.push('cancel dropped')
+            if (!handle) {
+              token = callback
+              return
+            }
+            cancelSent = true
+          }
+          sendCancel(callback)
+        }
+      }
+      if (version !== '2.x to 4.x') query.getCancelSent = () => cancelSent
+
+      return {
+        pausedAt,
+        calls,
+        fetches: () => fetches,
+        queried: () => queried,
+        query,
+        play (emitter) {
+          notify = emitter
+          queried = true
+          const begin = () => {
+            handle = true
+            if (token) {
+              const callback = token
+              token = null
+              cancelSent = true
+              sendCancel(callback)
+            }
+            if (stallBegin || !running) return
+            notify.emit('meta', meta())
+            dispatch()
+          }
+          if (beginDelay) {
+            setTimeout(begin, beginDelay)
+          } else {
+            begin()
+          }
+        }
+      }
+    }
+
+    const setupReader = (version, recordsets, options) => {
+      const query = reader(version, recordsets, options)
+      return { ...setup(query.play, query.query), pausedAt: query.pausedAt, calls: query.calls, fetches: query.fetches, queried: query.queried }
+    }
+
+    const rowsOf = recordsets => recordsets.flatMap(([column, count]) => Array.from({ length: count }, (_, i) => `${column}=${i + 1}`))
+
+    // the rows the stream hands to a writable that takes one at a time, so that it is paused
+    // throughout
+    const pipeToSlowWritable = request => new Promise((resolve, reject) => {
+      const rows = []
+      const writable = new Writable({
+        objectMode: true,
+        highWaterMark: 1,
+        write (row, encoding, callback) {
+          rows.push(Object.entries(row).map(([column, value]) => `${column}=${value}`).join())
+          setImmediate(callback)
+        }
+      })
+      writable.on('finish', () => resolve(rows))
+      request.on('error', reject)
+      request.pipe(writable)
+    })
+
+    // a short last batch in the first result set, a last batch of one row in the next, and a
+    // short last batch in the last
+    const RECORDSETS = [['a', 75], ['b', 51], ['c', 120]]
+
+    for (const version of ['2.x to 4.x', '5.0 to 5.5', '5.6']) {
+      it(`should pipe every row of every result set with msnodesqlv8 ${version}`, async () => {
+        const { request, pausedAt } = setupReader(version, RECORDSETS)
+        const completed = request.query('select a; select b; select c')
+        const rows = await pipeToSlowWritable(request)
+        await completed
+        assert.deepStrictEqual(rows, rowsOf(RECORDSETS), 'every row should arrive once, in order, under its own column')
+        assert.ok(pausedAt.length > 0, 'the driver should have been paused')
+        assert.deepStrictEqual(pausedAt.filter(at => at !== `${BATCH} of ${BATCH}`), [], 'the driver should only be paused once it has handed over a whole batch')
+      })
+    }
+
+    it('should emit nothing while a stream is paused, and complete once it is resumed', async () => {
+      const { request, notify } = setupReader('5.6', [['a', 75]])
+      request.stream = true
+      const rows = []
+      let completed = false
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 60) request.pause()
+      })
+      request.on('done', () => { completed = true })
+      const finished = new Promise(resolve => notify.once('done', resolve))
+      const result = request.query('select a')
+
+      await finished
+      await tick()
+      assert.strictEqual(rows.length, 60, 'no row should be emitted while the stream is paused')
+      assert.strictEqual(completed, false, 'the stream should not complete while rows are waiting')
+
+      request.resume()
+      await result
+      assert.deepStrictEqual(rows, rowsOf([['a', 75]]).map(row => Number(row.slice(2))), 'the waiting rows should be emitted once the stream is resumed')
+      assert.strictEqual(completed, true, 'the stream should complete once it has emitted the waiting rows')
+    })
+
+    it('should stream a cancel of a paused stream without emitting the rows still waiting', async () => {
+      const { request, notify, pool } = setupReader('5.6', [['a', 75]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 10) request.pause()
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      // the driver goes on to the end of its batch
+      while (!notify.isPaused()) await tick()
+      request.cancel()
+      await completed
+      assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+      assert.strictEqual(pool.released, 1, 'the connection should be released once')
+    })
+
+    it('should cancel a paused stream once the driver has handed over its batch', async () => {
+      const { request, pausedAt, calls } = setupReader('5.6', [['a', 75]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 10) {
+          request.pause()
+          request.cancel()
+        }
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      await completed
+      assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+      assert.deepStrictEqual(pausedAt, ['50 of 50'], 'the driver should be paused at the end of its batch')
+      assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be cancelled once it is paused')
+    })
+
+    it('should cancel a paused stream at the start of the next result set', async () => {
+      const { request, calls } = setupReader('5.6', [['a', 30], ['b', 30]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row)
+        if (rows.length === 10) {
+          request.pause()
+          request.cancel()
+        }
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a; select b')
+
+      await completed
+      assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+      assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused and cancelled once, before it reads the next result set')
+    })
+
+    it('should not pause a driver that is ending the query when a paused stream is cancelled', async () => {
+      const { request, notify, calls } = setupReader('5.6', [['a', 75]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 60) request.pause()
+      })
+      // once the driver has handed over its last, short batch it goes on to end the query
+      notify.on('column', (index, value) => {
+        if (value === 75) setImmediate(() => request.cancel())
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      await completed
+      assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'a stream that drops rows should report the cancel')
+      assert.deepStrictEqual(calls, [], 'the driver should be left to end the query')
+    })
+
+    it('should stream a cancel of a paused stream whose query has ended with rows still waiting', async () => {
+      const { request, notify, pool } = setupReader('5.6', [['a', 75]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 60) request.pause()
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      const finished = new Promise(resolve => notify.once('done', resolve))
+      request.query('select a')
+
+      await finished
+      await tick()
+      request.cancel()
+      await completed
+      assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'a stream that drops rows should report the cancel')
+      assert.strictEqual(pool.released, 1, 'the connection should be released once')
+    })
+
+    it('should keep the driver paused for a cancel made while a stream emits its waiting rows', async () => {
+      const { request, notify, calls } = setupReader('5.0 to 5.5', [['a', 200]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      let completedAfter = null
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        if (row.a === 10) request.pause()
+        if (row.a === 30) request.cancel()
+      })
+      // a listener after the one that cancels
+      request.on('row', row => rows.push(row.a))
+      const completed = new Promise(resolve => request.on('done', () => {
+        completedAfter = rows.length
+        resolve()
+      }))
+      request.query('select a')
+
+      // the driver goes on to the end of its batch, then the stream is resumed
+      while (!notify.isPaused()) await tick()
+      request.resume()
+      await completed
+      assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should not be resumed under the cancel')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be the only error')
+      assert.deepStrictEqual(rows, Array.from({ length: 50 }, (_, i) => i + 1), 'the rows handed over before the cancel was through should be emitted in order')
+      assert.strictEqual(completedAfter, 50, 'the stream should complete after its last row')
+    })
+
+    it('should go on piping every row when msnodesqlv8 refuses a cancel', async () => {
+      const { request, calls } = setupReader('2.x to 4.x', [['a', 75], ['b', 51]])
+      request.on('row', row => {
+        if (row.a === 5) request.cancel()
+      })
+      const completed = request.query('select a; select b')
+      const rows = await pipeToSlowWritable(request)
+      await completed
+      assert.strictEqual(calls[0], 'cancel refused', 'the cancel should have been refused')
+      assert.deepStrictEqual(rows, rowsOf([['a', 75], ['b', 51]]), 'a query whose cancel is refused should run to the end')
+    })
+
+    it('should complete a cancelled paused stream once a fetch under way times out', async () => {
+      const { request, calls } = setupReader('5.0 to 5.5', [['a', 200]], { timeOutAtFetch: 3 })
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => rows.push(row.a))
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      // the third fetch does not come back
+      while (rows.length < 100) await tick()
+      await tick()
+      request.pause()
+      request.cancel()
+      await completed
+      assert.deepStrictEqual(calls, [], 'the driver should not be paused or cancelled part way through a fetch')
+      assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the timeout and then the cancel should be streamed')
+      assert.strictEqual(rows.length, 100, 'no row should be emitted after the stream was paused')
+    })
+
+    it('should report a cancel of a paused stream although an earlier cancel was refused', async () => {
+      const { request, calls } = setupReader('2.x to 4.x', [['a', 75]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        // refused, as the driver is not paused, but only once the next cancel is under way
+        if (rows.length === 10) request.cancel()
+        if (rows.length === 13) {
+          request.pause()
+          request.cancel()
+        }
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      await completed
+      assert.deepStrictEqual(calls, ['cancel refused', 'pause', 'cancel'], 'the second cancel should go through at the end of the batch')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel that went through should be streamed')
+      assert.strictEqual(rows.length, 13, 'no row should be emitted after the stream was paused')
+    })
+
+    it('should hold the driver back from the start for a stream paused before its query', async () => {
+      const { request, notify, calls, fetches } = setupReader('5.0 to 5.5', [['a', 75]])
+      request.stream = true
+      request.pause()
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => rows.push(row.a))
+      const started = new Promise(resolve => notify.once('meta', resolve))
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      await started
+      await tick()
+      await tick()
+      assert.strictEqual(fetches(), 0, 'the driver should fetch nothing while the stream is paused')
+      request.cancel()
+      await completed
+      assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused before it fetches, and cancelled there')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+      assert.strictEqual(rows.length, 0, 'no row should be emitted')
+    })
+
+    it('should complete a cancelled paused stream whose query ends on an error', async () => {
+      const { request, notify, pool } = setupReader('5.6', [['a', 75]], { failAtEnd: true })
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 60) request.pause()
+      })
+      notify.on('column', (index, value) => {
+        if (value === 75) setImmediate(() => request.cancel())
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      await completed
+      assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused')
+      assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the error and then the cancel should be streamed')
+      assert.strictEqual(pool.released, 1, 'the connection should be released once')
+    })
+
+    it('should emit what was waiting in order to every listener when one resumes the stream', async () => {
+      const { request } = setupReader('5.6', [['a', 120]])
+      request.stream = true
+      const rows = []
+      request.on('row', row => {
+        if (row.a === 10) {
+          request.pause()
+          setTimeout(() => request.resume(), 5)
+        }
+        // a stream that is not paused can be resumed all the same
+        if (row.a > 10 && row.a < 15) request.resume()
+      })
+      // a listener after the one that resumes
+      request.on('row', row => rows.push(row.a))
+      await request.query('select a')
+      assert.deepStrictEqual(rows, Array.from({ length: 120 }, (_, i) => i + 1), 'every listener should get the rows in order')
+    })
+
+    it('should complete as cancelled a stream cancelled and paused again while it emits what was waiting', async () => {
+      const { request, notify } = setupReader('5.6', [['a', 30]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row.a)
+        if (rows.length === 5) request.pause()
+        if (rows.length === 10) {
+          request.cancel()
+          request.pause()
+        }
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      const finished = new Promise(resolve => notify.once('done', resolve))
+      request.query('select a')
+
+      // the query ends while the stream is paused, and the stream is resumed after it
+      await finished
+      await tick()
+      request.resume()
+      await completed
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+      assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused')
+    })
+
+    it('should cancel a paused stream whose first statement does not end', async () => {
+      const { request, calls, queried } = setupReader('5.6', [['a', 75]], { stallBegin: true })
+      request.stream = true
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      while (!queried()) await tick()
+      request.pause()
+      request.cancel()
+      await completed
+      assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused and cancelled at once')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+    })
+
+    it('should cancel a stream paused before its first statement ends, after a cancel made while it was flowing', async () => {
+      const { request, calls, queried } = setupReader('5.0 to 5.5', [['a', 75]], { beginDelay: 10 })
+      request.stream = true
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      const completed = new Promise(resolve => request.on('done', resolve))
+      request.query('select a')
+
+      while (!queried()) await tick()
+      // msnodesqlv8 holds this cancel until the query has a statement, and refuses it then
+      request.cancel()
+      request.pause()
+      request.cancel()
+      await completed
+      assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused at once, so that the cancel is accepted')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+    })
+
+    it('should not emit the rows of a cancelled paused stream that a listener resumes', async () => {
+      const { request, notify } = setupReader('5.6', [['a', 75], ['b', 30]])
+      request.stream = true
+      const rows = []
+      const errors = []
+      request.on('error', err => errors.push(err.code))
+      request.on('row', row => {
+        rows.push(row)
+        if (rows.length === 60) request.pause()
+      })
+      // the next result set's 'recordset' waits ahead of its rows, and its listener resumes the stream
+      request.on('recordset', () => request.resume())
+      const completed = new Promise(resolve => request.on('done', resolve))
+      const finished = new Promise(resolve => notify.once('done', resolve))
+      request.query('select a; select b')
+
+      await finished
+      await tick()
+      request.cancel()
+      await completed
+      assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused and cancelled')
+      assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error')
+    })
+
+    it('should run a stream that is held back to the end once nothing reads it', async () => {
+      const { request, pool, calls } = setupReader('5.6', [['a', 75], ['b', 120]])
+      const failure = new Error('the destination failed')
+      let written = 0
+      const destination = new Writable({
+        objectMode: true,
+        highWaterMark: 1,
+        write (row, encoding, callback) {
+          setImmediate(callback, ++written === 20 ? failure : null)
+        }
+      })
+      const completed = new Promise(resolve => request.on('done', resolve))
+      const failed = new Promise(resolve => pipeline(request.toReadableStream(), destination, resolve))
+      request.query('select a; select b')
+
+      assert.strictEqual(await failed, failure, 'the pipeline should fail with the destination\'s error')
+      await completed
+      assert.deepStrictEqual(calls, ['pause', 'resume'], 'the driver, held back at the end of its batch, should be resumed to run to the end')
+      assert.strictEqual(pool.released, 1, 'the connection should be released once')
+    })
+
+    it('should leave a cancel of a request that is no longer streaming to msnodesqlv8', async () => {
+      const { request, notify, calls } = setupReader('2.x to 4.x', [['a', 120]])
+      request.stream = true
+      request.pause()
+      request.stream = false
+      notify.once('column', () => setImmediate(() => request.cancel()))
+      const result = await request.query('select a')
+      assert.deepStrictEqual(calls, ['cancel refused'], 'the driver should not be paused for the cancel')
+      assert.strictEqual(result.recordset.length, 120, 'a query whose cancel is refused should run to the end')
+    })
+
+    it('should not pause the driver for a request that is no longer streaming', async () => {
+      const { request, calls } = setupReader('5.6', [['a', 75]])
+      request.stream = true
+      request.pause()
+      request.stream = false
+      const result = await request.query('select a')
+      assert.strictEqual(result.recordset.length, 75, 'every row should be read')
+      assert.deepStrictEqual(calls, [], 'the driver should not be paused')
+    })
+  })
 })
diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js
index a0ebd515..e67f85f4 100644
--- a/test/msnodesqlv8/msnodesqlv8.js
+++ b/test/msnodesqlv8/msnodesqlv8.js
@@ -118,6 +118,7 @@ describe('msnodesqlv8', function () {
     it('query with toReadableStream', done => TESTS['query with toReadableStream'](done))
     it('query with pipe', done => TESTS['query with pipe'](done))
     it('query with pipe and back pressure', (done) => TESTS['query with pipe and back pressure'](done))
+    it('query with pipe and back pressure across partial row batches', done => TESTS['query with pipe and back pressure across partial row batches'](done))
     it.skip('query with pipe and an error that ends the batch (msnodesqlv8 streams do not complete after one)', done => TESTS['query with pipe and an error'](true, done))
     if (continuesAfterErrors) {
       it('query with pipe and an error that does not end the batch', done => TESTS['query with pipe and an error'](false, done))
diff --git a/test/tedious/tedious.js b/test/tedious/tedious.js
index dd2ea9d7..68bdefef 100644
--- a/test/tedious/tedious.js
+++ b/test/tedious/tedious.js
@@ -107,6 +107,7 @@ describe('tedious', () => {
     it('query with toReadableStream', done => TESTS['query with toReadableStream'](done))
     it('query with pipe', done => TESTS['query with pipe'](done))
     it('query with pipe and back pressure', (done) => TESTS['query with pipe and back pressure'](done))
+    it('query with pipe and back pressure across partial row batches', done => TESTS['query with pipe and back pressure across partial row batches'](done))
     it('query with pipe and an error that ends the batch', done => TESTS['query with pipe and an error'](true, done))
     it('query with pipe and an error that does not end the batch', done => TESTS['query with pipe and an error'](false, done))
     it('query with toReadableStream and a failing destination', done => TESTS['query with toReadableStream and a failing destination'](done))