chunkById method

Stream<List<TRow>> chunkById({ int size = 500, })

Keyset chunk ordered by the primary key, or by the caller's unique orderBy on a keyless table.

Implementation

Stream<List<TRow>> chunkById({int size = 500}) async* {
  if (size <= 0) {
    throw ArgumentError.value(size, 'size', 'Must be positive.');
  }
  final ordered = ensureStableOrder();
  final dialect = await _resolvedDialect();
  final orders = ordered.state.orderBy;
  MssqlKeyset.requireNamedColumns(orders, 'chunkById()');
  // One snapshot and one loader for the whole walk: two chunks of the same
  // iteration must not land on two tenants, and the caller's timeout and
  // cancellation have to reach every chunk and every include.
  final scopes = ordered.context.scopeCompiler(ordered.state.scope);
  final loader = ordered._loaderFor(session, dialect);
  List<Object?>? after;
  while (true) {
    var query = ordered._select(scopes: scopes);
    if (after != null) {
      query = query.where(MssqlKeyset.after(orders, after));
    }
    final fetched = await query
        .top(size)
        .get(
          session,
          dialect: dialect,
          timeout: state.options.timeout,
          cancellationToken: state.options.cancellationToken,
          options: state.options,
        );
    if (fetched.isEmpty) return;
    final mapped = fetched.map(binding.fromRow).toList(growable: false);
    yield await ordered._attachOn(
      session,
      mapped,
      dialect: dialect,
      loader: loader,
    );
    if (fetched.length < size) return;
    after = MssqlKeyset.keysFromRow(fetched.last, orders);
  }
}