stream method

Stream<TRow> stream({ int parentBatch = 100, })

Maps the native row stream. Does not buffer the whole result.

Cancelling the subscription cancels the native operation.

With no include this is the native row stream: rows are mapped as they arrive and nothing buffers the whole result.

With include it becomes a keyset walk of parentBatch parents at a time, because a session cannot run an include query while one of its own result sets is still open. That keeps memory bounded to one batch plus its children, and it means this path orders by the primary key (see chunkById) rather than leaving the order to the server.

Implementation

Stream<TRow> stream({int parentBatch = 100}) async* {
  if (parentBatch <= 0) {
    throw ArgumentError.value(
      parentBatch,
      'parentBatch',
      'Must be positive.',
    );
  }
  final token = state.options.cancellationToken;
  if (included.isEmpty) {
    yield* _streamPlain(token);
    return;
  }
  // Includes cannot be loaded from inside an open row stream. A session
  // runs one operation at a time — the native driver holds the connection
  // for as long as rows are still arriving — so an include query issued
  // between two yields would queue behind a stream that cannot finish
  // until the consumer takes the next row, and the two would wait on each
  // other forever. Keyset chunks give the same bounded memory without the
  // deadlock: each chunk is a completed query, and the includes for it run
  // while no result set is open. The cost is that this path needs a stable
  // order, which `chunkById` supplies from the primary key.
  await for (final batch in chunkById(size: parentBatch)) {
    for (final parent in batch) {
      yield parent;
    }
  }
}