Function FragmentToBatches

cpp/src/arrow/dataset/scanner.cc:295–332

};

Result<EnumeratedRecordBatchGenerator> FragmentToBatches(
    const Enumerated<std::shared_ptr<Fragment>>& fragment,
    const std::shared_ptr<ScanOptions>& options) {
#ifdef ARROW_WITH_OPENTELEMETRY
  util::tracing::Span span;
  START_SPAN(span, "Scanner::FragmentToBatches",
             {
                 {"arrow.dataset.fragment", fragment.value->ToString()},
                 {"arrow.dataset.fragment.index", fragment.index},
                 {"arrow.dataset.fragment.last", fragment.last},
                 {"arrow.dataset.fragment.type_name", fragment.value->type_name()},
             });
#endif
  ARROW_ASSIGN_OR_RAISE(auto batch_gen, fragment.value->ScanBatchesAsync(options));
  ArrayVector columns;
  for (const auto& field : options->dataset_schema->fields()) {
    // TODO(ARROW-7051): use helper to make empty batch
    ARROW_ASSIGN_OR_RAISE(auto array,
                          MakeArrayOfNull(field->type(), /*length=*/0, options->pool));
    columns.push_back(std::move(array));
  }
  WRAP_ASYNC_GENERATOR(batch_gen);
  batch_gen = MakeDefaultIfEmptyGenerator(
      std::move(batch_gen),
      RecordBatch::Make(options->dataset_schema, /*num_rows=*/0, std::move(columns)));
  auto enumerated_batch_gen = MakeEnumeratedGenerator(std::move(batch_gen));

  auto combine_fn = [fragment, cache_metadata = options->cache_metadata](
                        const Enumerated<std::shared_ptr<RecordBatch>>& record_batch) {
    if (!cache_metadata && record_batch.last) {
      ARROW_WARN_NOT_OK(fragment.value->ClearCachedMetadata(),
                        "Could not clear cached metadata on fragment");
    }
    return EnumeratedRecordBatch{record_batch, fragment};
  };

  return MakeMappedGenerator(enumerated_batch_gen, std::move(combine_fn));
}

Result<AsyncGenerator<EnumeratedRecordBatchGenerator>> FragmentsToBatches(
    FragmentGenerator fragment_gen, const std::shared_ptr<ScanOptions>& options) {