stream method

Stream<R> stream({ int fetchSize = 500, })

Bounded keyset batches within an explicit transaction. Non-primary-key ordering and offsets are rejected; the cursor never buffers all results.

Implementation

Stream<R> stream({int fetchSize = 500}) async* {
  if (!session.inTransaction) {
    throw StateError('Streaming requires a transaction');
  }
  if (fetchSize < 1 || _offset != null || _orders.isNotEmpty) {
    throw ArgumentError('Streaming uses primary-key order without offsets');
  }
  final primary = _primary;
  if (primary.type != ScalarType.integer && primary.type != ScalarType.text) {
    throw UnsupportedError('Streaming supports integer/text primary keys');
  }
  Object? cursor;
  var remaining = _limit;
  while (remaining == null || remaining > 0) {
    final count = remaining == null || remaining > fetchSize
        ? fetchSize
        : remaining;
    var page = TableQuery<R>._(
      session,
      definition,
      decode,
      _filters,
      [(primary.field, asc)],
      count,
      null,
    );
    if (cursor != null) {
      page = page.whereFields({primary.field: gt<Object?>(cursor)});
    }
    final bindings = page._bindings();
    final result = await session.run(
      'SELECT $_columns FROM $_onlyTable${page._readSuffix(bindings)}',
      parameters: bindings.values,
    );
    for (final cells in result.rows) {
      yield decode(cells);
    }
    if (result.rows.length < count) break;
    cursor = result.rows.last[definition.columns.indexOf(primary)];
    if (remaining != null) remaining -= result.rows.length;
  }
}