streamQueryColumnarBatched method

Stream<TypedColumnarResult> streamQueryColumnarBatched(int connectionId, String sql, { int fetchSize = 1000, int chunkSize = 64 * 1024, bool lazyStrings = false, ResultEncoding resultEncoding = ResultEncoding.columnar, })

Like streamQueryBatched but decodes columnar v2 frames directly to TypedColumnarResult without a row-major intermediate.

Implementation

Stream<TypedColumnarResult> streamQueryColumnarBatched(
  int connectionId,
  String sql, {
  int fetchSize = 1000,
  int chunkSize = 64 * 1024,
  bool lazyStrings = false,
  ResultEncoding resultEncoding = ResultEncoding.columnar,
}) async* {
  final streamId = streamStartBatched(
    connectionId,
    sql,
    fetchSize: fetchSize,
    chunkSize: chunkSize,
    resultEncodingWire: resultEncoding.wireCode,
  );

  if (streamId == 0) {
    throw Exception('Failed to start batched stream: ${_native.getError()}');
  }

  final pending = BinaryFrameAccumulator();
  var completed = false;
  OdbcError? primary;
  try {
    while (true) {
      final result = streamFetch(streamId, bufferSize: chunkSize);

      if (!result.success) {
        throw QueryError(
          message: 'Stream fetch failed: ${_native.getError()}',
        );
      }

      final data = result.data;
      if (data == null || data.isEmpty) {
        break;
      }
      pending.add(data);

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

      if (!result.hasMore) break;
    }

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