Method ScanBatchesAsync

cpp/src/arrow/dataset/file_ipc.cc:139–181

}

Result<RecordBatchGenerator> IpcFileFormat::ScanBatchesAsync(
    const std::shared_ptr<ScanOptions>& options,
    const std::shared_ptr<FileFragment>& file) const {
  auto self = shared_from_this();
  auto source = file->source();
  auto open_reader = OpenReaderAsync(source);
  auto reopen_reader = [self, options,
                        source](std::shared_ptr<ipc::RecordBatchFileReader> reader)
      -> Future<std::shared_ptr<ipc::RecordBatchFileReader>> {
    ARROW_ASSIGN_OR_RAISE(auto options,
                          GetReadOptions(*reader->schema(), *self, *options));
    return OpenReader(source, options);
  };
  auto readahead_level = options->batch_readahead;
  auto default_fragment_scan_options = this->default_fragment_scan_options;
  auto open_generator = [=](const std::shared_ptr<ipc::RecordBatchFileReader>& reader)
      -> Result<RecordBatchGenerator> {
    ARROW_ASSIGN_OR_RAISE(
        auto ipc_scan_options,
        GetFragmentScanOptions<IpcFragmentScanOptions>(kIpcTypeName, options.get(),
                                                       default_fragment_scan_options));

    RecordBatchGenerator generator;
    if (ipc_scan_options->cache_options) {
      // Transferring helps performance when coalescing
      ARROW_ASSIGN_OR_RAISE(generator, reader->GetRecordBatchGenerator(
                                           /*coalesce=*/true, options->io_context,
                                           *ipc_scan_options->cache_options,
                                           ::arrow::internal::GetCpuThreadPool()));
    } else {
      ARROW_ASSIGN_OR_RAISE(generator, reader->GetRecordBatchGenerator(
                                           /*coalesce=*/false, options->io_context));
    }
    WRAP_ASYNC_GENERATOR_WITH_CHILD_SPAN(
        generator, "arrow::dataset::IpcFileFormat::ScanBatchesAsync::Next");
    if (readahead_level == 0) {
      return MakeChunkedBatchGenerator(std::move(generator), options->batch_size);
    }
    auto batch_generator = MakeReadaheadGenerator(std::move(generator), readahead_level);
    return MakeChunkedBatchGenerator(std::move(batch_generator), options->batch_size);
  };
  return MakeFromFuture(open_reader.Then(reopen_reader).Then(open_generator));
}

Future<std::optional<int64_t>> IpcFileFormat::CountRows(
    const std::shared_ptr<FileFragment>& file, compute::Expression predicate,