Function ExecNode_Aggregate

r/src/compute-exec.cpp:382–408

[[acero::export]]

// [[acero::export]]
std::shared_ptr<acero::ExecNode> ExecNode_Aggregate(
    const std::shared_ptr<acero::ExecNode>& input, cpp11::list options,
    std::vector<std::string> key_names) {
  std::vector<arrow::compute::Aggregate> aggregates;

  for (cpp11::list name_opts : options) {
    auto function = cpp11::as_cpp<std::string>(name_opts["fun"]);
    auto opts = make_compute_options(function, name_opts["options"]);
    auto target_names = cpp11::as_cpp<std::vector<std::string>>(name_opts["targets"]);
    auto name = cpp11::as_cpp<std::string>(name_opts["name"]);

    std::vector<arrow::FieldRef> targets;
    for (auto&& target : target_names) {
      targets.emplace_back(std::move(target));
    }
    aggregates.push_back(arrow::compute::Aggregate{std::move(function), opts,
                                                   std::move(targets), std::move(name)});
  }

  std::vector<arrow::FieldRef> keys;
  for (auto&& name : key_names) {
    keys.emplace_back(std::move(name));
  }
  return MakeExecNodeOrStop(
      "aggregate", input->plan(), {input.get()},
      acero::AggregateNodeOptions{std::move(aggregates), std::move(keys)});
}

// [[acero::export]]
std::shared_ptr<acero::ExecNode> ExecNode_Join(