streamQuery method

Implementation

Stream<Result<QueryResult>> streamQuery(
  String connectionId,
  String sql, {
  int fetchSize = 1000,
  int? chunkSize,
}) async* {
  final nativeId = state.connectionIds[connectionId];
  if (nativeId == null) {
    yield const Failure<QueryResult, OdbcError>(
      ValidationError(message: 'Invalid connection ID'),
    );
    return;
  }
  final effectiveChunk = resolveStreamChunkSizeBytes(
    chunkSize: chunkSize,
    options: state.optionsFor(connectionId),
  );
  final opts = state.optionsFor(connectionId);
  final maxBytes = opts?.maxResultBufferBytes;
  final queryTimeout = opts?.queryTimeout;
  final lazyStrings = opts?.lazyStrings ?? false;
  // Row-shaped APIs always use row-major wire. Server profiles that default
  // to columnar would otherwise rematerialize typed → QueryResult rows.
  // Prefer [streamQueryColumnar] for columnar end-to-end.

  Stream<Result<QueryResult>> createSource() async* {
    try {
      await for (final chunk in streamNativeQueryWithFallback(
        nativeId,
        sql,
        maxBufferBytes: maxBytes,
        lazyStrings: lazyStrings,
        fetchSize: fetchSize,
        chunkSize: effectiveChunk,
      )) {
        yield Success(parser.toQueryResult(chunk));
      }
    } on Exception catch (e) {
      yield await _errors.streamingFailureFromException(e);
    }
  }

  final source = createSource();

  yield* streamWithQueryTimeout(
    source: source,
    queryTimeout: queryTimeout,
    onTimeoutItem: const Failure<QueryResult, OdbcError>(
      QueryError(message: odbcQueryTimedOutMessage),
    ),
  );
}