Method Make

cpp/src/arrow/acero/scalar_aggregate_node.cc:163–203

}

Result<ExecNode*> ScalarAggregateNode::Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                            const ExecNodeOptions& options) {
  RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 1, "ScalarAggregateNode"));

  const auto& aggregate_options = checked_cast<const AggregateNodeOptions&>(options);
  auto aggregates = aggregate_options.aggregates;
  const auto& keys = aggregate_options.keys;
  const auto& segment_keys = aggregate_options.segment_keys;
  const auto concurrency = plan->query_context()->max_concurrency();
  // We can't use concurrency == 1 because that include I/O concurrency
  const bool is_cpu_parallel = plan->query_context()->executor()->GetCapacity() > 1;

  if (keys.size() > 0) {
    return Status::Invalid("Scalar aggregation with some key");
  }
  if (is_cpu_parallel && segment_keys.size() > 0) {
    return Status::NotImplemented("Segmented aggregation in a multi-threaded plan");
  }

  const auto& input_schema = inputs[0]->output_schema();
  auto exec_ctx = plan->query_context()->exec_context();

  ARROW_ASSIGN_OR_RAISE(
      auto args, MakeAggregateNodeArgs(input_schema, keys, segment_keys, aggregates,
                                       exec_ctx, concurrency, is_cpu_parallel));

  if (is_cpu_parallel) {
    for (auto& kernel : args.kernels) {
      if (kernel->ordered) {
        return Status::NotImplemented(
            "Using ordered aggregator in multiple threaded execution is not supported");
      }
    }
  }

  return plan->EmplaceNode<ScalarAggregateNode>(
      plan, std::move(inputs), std::move(args.output_schema), std::move(args.segmenter),
      std::move(args.segment_key_field_ids), std::move(args.target_fieldsets),
      std::move(args.aggregates), std::move(args.kernels), std::move(args.kernel_intypes),
      std::move(args.states));
}

Status ScalarAggregateNode::DoConsume(const ExecSpan& batch, size_t thread_index) {
  for (size_t i = 0; i < kernels_.size(); ++i) {