streamQueryBatched method

Stream<ParsedRowBuffer> streamQueryBatched(int connectionId, String sql, { int fetchSize = 1000, int chunkSize = 64 * 1024, int? maxBufferBytes, int resultEncodingWire = 0, bool lazyStrings = false, Uint8List? paramsBuffer, })

Runs sql in the worker using native batched streaming.

This path uses odbc_stream_start_batched + odbc_stream_fetch, yielding chunks progressively. maxBufferBytes caps internal pending bytes for message framing.

Implementation

Stream<ParsedRowBuffer> streamQueryBatched(
  int connectionId,
  String sql, {
  int fetchSize = 1000,
  int chunkSize = 64 * 1024,
  int? maxBufferBytes,
  int resultEncodingWire = 0,
  bool lazyStrings = false,
  Uint8List? paramsBuffer,
}) async* {
  final streamId = await streamStartBatched(
    connectionId,
    sql,
    fetchSize: fetchSize,
    chunkSize: chunkSize,
    resultEncodingWire: resultEncodingWire,
    paramsBuffer: paramsBuffer,
  );
  if (streamId == 0) {
    final workerError = await _safeGetWorkerError();
    throw AsyncError(
      code: AsyncErrorCode.queryFailed,
      message: workerError ?? 'Failed to start batched stream',
    );
  }

  final pending = BinaryFrameAccumulator();
  final limit = maxBufferBytes;
  var completed = false;
  OdbcError? primary;
  Object? thrown;
  StackTrace? thrownStack;
  try {
    while (true) {
      final fetched = await streamFetch(streamId, bufferSize: chunkSize);
      if (!fetched.success) {
        final workerError = fetched.error ?? await _safeGetWorkerError();
        throw AsyncError(
          code: AsyncErrorCode.queryFailed,
          message: workerError ?? 'Batched stream fetch failed',
        );
      }

      final data = fetched.data;
      if (data != null && data.isNotEmpty) {
        pending.add(data);
        if (limit != null && pending.length > limit) {
          throw const AsyncError(
            code: AsyncErrorCode.resourceExhausted,
            message: 'Streaming buffer exceeded maxBufferBytes',
          );
        }

        for (final msg in pending.drainFrames()) {
          yield decodeBatchedStreamFrame(
            msg,
            lazyStrings: lazyStrings,
          );
        }
      }

      if (!fetched.hasMore) {
        break;
      }
    }

    if (pending.length > 0) {
      throw const FormatException(
        'Leftover bytes after stream; expected complete protocol messages',
      );
    }
    completed = true;
  } on Object catch (error, stack) {
    thrown = error;
    thrownStack = stack;
    primary = translateOdbcError(
      error,
      operation: 'streamQuery',
      stackTrace: stack,
    );
  } finally {
    final failure = await cleanupNativeResources(
      [
        if (!completed)
          (operation: 'streamCancel', action: () => streamCancel(streamId)),
        (operation: 'streamClose', action: () => streamClose(streamId)),
      ],
      primary: primary,
    );
    if (failure != null) {
      if (identical(failure, primary) && thrown != null) {
        NativeCallContext.record(failure);
        Error.throwWithStackTrace(thrown, thrownStack!);
      }
      Error.throwWithStackTrace(
        failure,
        failure.details.stackTrace ?? StackTrace.current,
      );
    }
  }
}