Method CountRowsAsync

cpp/src/arrow/dataset/scanner.cc:735–788

}

Future<int64_t> AsyncScanner::CountRowsAsync(Executor* executor) {
  ARROW_ASSIGN_OR_RAISE(auto fragment_gen, GetFragments());

  compute::ExecContext exec_context(scan_options_->pool, executor);

  ARROW_ASSIGN_OR_RAISE(auto plan, acero::ExecPlan::Make(exec_context));
  // Drop projection since we only need to count rows
  const auto options = std::make_shared<ScanOptions>(*scan_options_);
  ARROW_ASSIGN_OR_RAISE(auto empty_projection,
                        ProjectionDescr::FromNames(std::vector<std::string>(),
                                                   *scan_options_->dataset_schema,
                                                   scan_options_->add_augmented_fields));
  SetProjection(options.get(), empty_projection);

  auto total = std::make_shared<std::atomic<int64_t>>(0);

  fragment_gen = MakeMappedGenerator(
      std::move(fragment_gen),
      [options, total](const std::shared_ptr<Fragment>& fragment) {
        return fragment->CountRows(options->filter, options)
            .Then([options, total, fragment](std::optional<int64_t> fast_count) mutable
                  -> std::shared_ptr<Fragment> {
              if (fast_count) {
                // fast path: got row count directly; skip scanning this fragment
                (*total) += *fast_count;
                return std::make_shared<InMemoryFragment>(options->dataset_schema,
                                                          RecordBatchVector{});
              }

              // slow path: actually filter this fragment's batches
              return std::move(fragment);
            });
      });

  acero::Declaration count_plan = acero::Declaration::Sequence(
      {{"scan",
        ScanNodeOptions{std::make_shared<FragmentDataset>(scan_options_->dataset_schema,
                                                          std::move(fragment_gen)),
                        options}},
       {"project", acero::ProjectNodeOptions{{options->filter}, {"mask"}}},
       {"aggregate", acero::AggregateNodeOptions{{compute::Aggregate{
                         "sum", nullptr, "mask", "selected_count"}}}}});

  return acero::DeclarationToBatchesAsync(std::move(count_plan), exec_context)
      .Then([total](const RecordBatchVector& batches) -> Result<int64_t> {
        DCHECK_EQ(1, batches.size());
        ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Scalar> count_scalar,
                              batches[0]->column(0)->GetScalar(0));
        return total->load() +
               static_cast<int64_t>(
                   ::arrow::internal::checked_pointer_cast<UInt64Scalar>(count_scalar)
                       ->value);
      });
}

Future<int64_t> AsyncScanner::CountRowsAsync() {
  return CountRowsAsync(scan_options_->cpu_executor