Streaming results

findAll() builds every row into memory before it returns. For a result set
that does not fit — an export, a backfill, a migration of ten million rows —
stream it instead. There are two layers:

Model.iterate() / Model.stream() page through findAll() with a
keyset cursor. They take ordinary FindOptions, need no SQL, and work on
every dialect. Start here.

dialect.queryStream(sql) takes a SQL string and returns a Node
Readable in object mode, using a native server-side cursor where the driver
has one.

The MapTransform / FilterTransform / BatchTransform / StreamIterator
transforms compose on top of either.

The examples use the shop schema from Querying (users,
orders, products). The SQL shown was captured from the ORM.

Streaming from a model

Model.iterate() walks every matching row without loading them all, and takes
the same options as findAll():

import { createWriteStream } from 'node:fs';

const out = createWriteStream('paid-orders.csv');
out.write('id,userId,total\n');

for await (const o of Order.iterate({ where: { status: 'paid' } }, { batchSize: 500 })) {
  out.write(`${o.id},${o.userId},${o.total}\n`);
}
out.end();

Each page is one findAll(). The first page carries your where; every later
page adds the last primary key seen:

A page shorter than batchSize is the last one, so the walk ends without an
extra empty query. The default batchSize is 100.

limit and offset apply to the whole walk, not to each page. With
limit: 3 and batchSize: 2, the second page asks for only the one row still
owed:

for await (const o of Order.iterate({ where: { status: 'paid' }, limit: 3 }, { batchSize: 2 })) {
  // three orders in total
}

Model.stream() gives the same rows as a Node Readable in object mode, for
pipeline() and the transforms below:

import { pipeline } from 'node:stream/promises';
import { BatchTransform } from 'ts-prorm-orm';

await pipeline(
  User.stream({ where: { status: 'active' } }, { batchSize: 500 }),
  new BatchTransform({ batchSize: 500 }),
  async function* (batches) {
    for await (const batch of batches) yield await search.bulkIndex(batch);
  },
);

Both page under a keyset cursor, exactly as findInBatches() does, which is
what makes them work on all 27 dialects with no server-side cursor. Paging by the
last id seen also cannot skip or repeat a row when the table is written to
mid-scan, and stays fast at depth because the database never counts past rows.
They inherit scopes, paranoid filtering, value casting and hooks from
findAll().

[!NOTE]
The keyset only works in primary-key order, so a caller-supplied order is
replaced with ORDER BY "id" ASC rather than silently producing wrong
pages. If you need another order, sort after the walk, or use
dialect.queryStream() with your own ORDER BY.

Backpressure

stream() is Readable.from() over the iterate() generator, so the next page
is fetched only when the consumer has taken the previous one:

A consumer that abandons the stream after two rows causes exactly one query:

let n = 0;
for await (const o of Order.stream({}, { batchSize: 3 })) {
  if (++n === 2) break;
}
// SELECT * FROM "orders" ORDER BY "id" ASC LIMIT 3     (and nothing else)

[!TIP]
Every transform stage in a pipeline() buffers up to its highWaterMark
objects (default 1000) before it stops pulling, so a pipeline reads ahead
by roughly that many rows per stage. To bound memory for large rows, pass a
smaller highWaterMark to the transforms.

Handler style

findInBatches() and findEach() remain for handler-style iteration:

await Order.findInBatches({ where: { status: 'paid' } }, async (batch) => {
  await ship(batch);
}, { batchSize: 500 });

Streaming a SQL string

dialect.queryStream() takes SQL and returns a Node Readable in object mode:

const dialect = prorm.getDialectInstance()!;
const stream = dialect.queryStream('SELECT id, sku, stock FROM products ORDER BY id', {
  batchSize: 1000,
  highWaterMark: 500,
});

for await (const row of stream) {
  await ship(row);
}

Because it is a standard Readable, pipeline(), pipe(), for await,
backpressure and destroy() all behave the way you expect from Node streams. A
failing query destroys the stream with the driver's error, so the for await
loop rejects rather than ending silently.

[!NOTE]
Prefer the model methods: they are typed, they respect scopes, paranoid
filtering and hooks, and they require no SQL. Reach for queryStream() when
you want a native server-side cursor on PostgreSQL or CockroachDB, or an order
other than the primary key, and are willing to write the query.

Three implementations

Dialect	Mechanism
PostgreSQL, CockroachDB	A server-side cursor (`DECLARE … CURSOR`, then `FETCH batchSize`).
MySQL, MariaDB	The driver's row stream: one execution on a connection from `mysql2`'s callback pool, released when the stream ends or fails.
SQL Server	The `mssql` request's row events: one execution, paused while the consumer is behind.
SQLite	`better-sqlite3`'s `Statement#iterate()`.
Everything else	A paginated fallback: repeated pages of the query, one round trip per page.

On PostgreSQL the cursor needs a transaction. Outside one, the stream checks a
connection out of the pool and wraps the cursor in its own BEGIN / COMMIT;
inside prorm.transaction() (pass transaction), it declares the cursor on
that transaction's connection and leaves the commit to you:

[!IMPORTANT]
A stream holds its connection until it ends or is destroyed. Always consume it
to the end, or call destroy(); an abandoned cursor stream keeps a pooled
connection and an open transaction. See
Connection pooling.

The paginated fallback

The fallback wraps your statement and appends the dialect's own pagination
clause. Each page is fetched only after the previous one has been drained by
the consumer — real backpressure, not read-ahead buffering. It is slower than a
cursor for very large sets, since every page is a fresh query, but it never
loads the whole result at once. Oracle supplies its own page builder for its
pagination syntax while reusing the same loop.

The third page of SELECT id, email FROM users ORDER BY id with
batchSize: 1000:

[!WARNING]
The fallback pages with OFFSET, so the query must have a stable order or
rows can be skipped or repeated between pages.

[!NOTE]
Before 4.9.2, MySQL's queryStream() called pool.createQueryStream(),
which the mysql2/promise pool does not have, so it always failed; and SQL
Server used the paginated fallback with ORDER BY (SELECT NULL), which is no
order at all, so rows could be skipped or repeated between pages. Both now
stream one execution of the statement.

Options

Option	Default	Effect
`batchSize`	`1000`	Rows fetched per page / cursor `FETCH`.
`highWaterMark`	`1000`	Stream buffer size.
`model` + `mapToModel`	—	Map each row to a model instance instead of a plain object.
`transaction`	—	Run the stream inside a transaction.
`logging`	—	Per-stream logging override.

The transforms

All four are object-mode Transform streams with a default highWaterMark of
1000.

`MapTransform`

import { MapTransform } from 'ts-prorm-orm';

stream.pipe(new MapTransform({
  mapFn: async (row) => ({ ...row, region: await lookupRegion(row.ip) }),
}));

mapFn may be sync or async; a throw destroys the stream with that error.

`FilterTransform`

import { FilterTransform } from 'ts-prorm-orm';

stream.pipe(new FilterTransform({ filterFn: (row) => row.total > 100 }));

[!TIP]
Filtering in the database is cheaper: every row a FilterTransform drops was
still read, sent and parsed. Use it only for predicates SQL cannot express, or
that depend on something outside the database.

`BatchTransform`

Regroups a row stream into arrays, so downstream work happens per chunk rather
than per row. The final partial batch is flushed at end-of-stream.

import { BatchTransform } from 'ts-prorm-orm';

stream
  .pipe(new BatchTransform({ batchSize: 500 }))
  .on('data', async (rows) => { await bulkIndex(rows); });

`StreamIterator`

Turns any readable into an async iterator, for stream sources that do not
already support for await:

import { StreamIterator } from 'ts-prorm-orm';

const it = someStream.pipe(new StreamIterator());
for await (const row of it) { /* … */ }

A worked pipeline

Re-index every paid order over 12.00 in a search engine, 500 at a time:

import { pipeline } from 'node:stream/promises';
import { MapTransform, FilterTransform, BatchTransform } from 'ts-prorm-orm';

await pipeline(
  Order.stream({ where: { status: 'paid' } }, { batchSize: 2000 }),
  new FilterTransform({ filterFn: (o) => Number(o.total) > 12 }),
  new MapTransform({ mapFn: (o) => ({ id: o.id, total: Number(o.total) }) }),
  new BatchTransform({ batchSize: 500 }),
  async function* (batches) {
    for await (const batch of batches) {
      await search.bulkIndex(batch);
      yield `${batch.length}\n`;
    }
  },
  process.stdout,
);

pipeline() propagates errors and destroys every stage, which is what you want
for a long-running job. (The total > 12 filter belongs in the where in real
code; it is here to show the transform.)

Choosing an approach

Situation	Use
Result fits comfortably in memory	`findAll()`
Walking a large result	`Model.iterate()`
Piping it somewhere, with transforms	`Model.stream()`
Handler-style, a batch at a time	`findInBatches()` / `findEach()`
A native cursor or a custom order, and you will write the SQL	`dialect.queryStream()`

Querying — findInBatches keyset iteration

Bulk operations

Raw queries

Connection pooling — a long stream holds a connection

Streaming results

Streaming from a model#

Backpressure#

Handler style#

Streaming a SQL string#

Three implementations#

The paginated fallback#

Options#

The transforms#

MapTransform#

FilterTransform#

BatchTransform#

StreamIterator#

A worked pipeline#

Choosing an approach#

Related reading#