Function DeclarationToRecordBatchGenerator

cpp/src/arrow/acero/exec_plan.cc:1007–1029

Convert a `Declaration` using `QueryOptions` to a `RecordBatch` generator. Additional outputs: `out_schema` is the schema for the generated record batches `out_plan` is the backing `ExecPlan`, which may be stopped to cancel the generation

// * `out_schema` is the schema for the generated record batches
// * `out_plan` is the backing `ExecPlan`, which may be stopped to cancel the generation
Result<AsyncGenerator<std::shared_ptr<RecordBatch>>> DeclarationToRecordBatchGenerator(
    Declaration declaration, QueryOptions options,
    ::arrow::internal::Executor* cpu_executor, std::shared_ptr<Schema>* out_schema,
    std::shared_ptr<ExecPlan>* out_plan) {
  if (out_plan == nullptr) {
    std::shared_ptr<ExecPlan> tmp_plan;
    return DeclarationToRecordBatchGenerator(std::move(declaration), std::move(options),
                                             cpu_executor, out_schema, &tmp_plan);
  }
  auto converter = std::make_shared<BatchConverter>();
  ExecContext exec_ctx(options.memory_pool, cpu_executor, options.function_registry);
  std::shared_ptr<ExecPlan>& plan = *out_plan;
  ARROW_ASSIGN_OR_RAISE(plan, ExecPlan::Make(options, exec_ctx));
  Declaration with_sink = Declaration::Sequence(
      {declaration,
       {"sink", SinkNodeOptions(&converter->exec_batch_gen, &converter->schema)}});
  ARROW_RETURN_NOT_OK(with_sink.AddToPlan(plan.get()));
  ARROW_RETURN_NOT_OK(plan->Validate());
  plan->StartProducing();
  converter->exec_plan = plan;
  ARROW_ASSIGN_OR_RAISE(*out_schema, converter->InitializeSchema(options.field_names));
  return [conv = std::move(converter)] { return (*conv)(); };
}

}  // namespace