Method MakeAggregateNodeArgs

cpp/src/arrow/acero/groupby_aggregate_node.cc:69–178

}

Result<AggregateNodeArgs<HashAggregateKernel>> GroupByNode::MakeAggregateNodeArgs(
    const std::shared_ptr<Schema>& input_schema, const std::vector<FieldRef>& keys,
    const std::vector<FieldRef>& segment_keys, const std::vector<Aggregate>& aggs,
    ExecContext* ctx, const bool is_cpu_parallel) {
  // Find input field indices for key fields
  std::vector<int> key_field_ids(keys.size());
  for (size_t i = 0; i < keys.size(); ++i) {
    ARROW_ASSIGN_OR_RAISE(auto match, keys[i].FindOne(*input_schema));
    key_field_ids[i] = match[0];
  }

  // Find input field indices for segment key fields
  std::vector<int> segment_key_field_ids(segment_keys.size());
  for (size_t i = 0; i < segment_keys.size(); ++i) {
    ARROW_ASSIGN_OR_RAISE(auto match, segment_keys[i].FindOne(*input_schema));
    segment_key_field_ids[i] = match[0];
  }

  // Check key fields and segment key fields are disjoint
  std::unordered_set<int> key_field_id_set(key_field_ids.begin(), key_field_ids.end());
  for (const auto& segment_key_field_id : segment_key_field_ids) {
    if (key_field_id_set.find(segment_key_field_id) != key_field_id_set.end()) {
      return Status::Invalid("Group-by aggregation with field '",
                             input_schema->field(segment_key_field_id)->name(),
                             "' as both key and segment key");
    }
  }

  // Find input field indices for aggregates
  std::vector<std::vector<int>> agg_src_fieldsets(aggs.size());
  for (size_t i = 0; i < aggs.size(); ++i) {
    const auto& target_fieldset = aggs[i].target;
    for (const auto& target : target_fieldset) {
      ARROW_ASSIGN_OR_RAISE(auto match, target.FindOne(*input_schema));
      agg_src_fieldsets[i].push_back(match[0]);
    }
  }

  // Build vector of aggregate source field data types
  std::vector<std::vector<TypeHolder>> agg_src_types(aggs.size());
  for (size_t i = 0; i < aggs.size(); ++i) {
    for (const auto& agg_src_field_id : agg_src_fieldsets[i]) {
      agg_src_types[i].push_back(input_schema->field(agg_src_field_id)->type().get());
    }
  }

  // Build vector of segment key field data types
  std::vector<TypeHolder> segment_key_types(segment_keys.size());
  for (size_t i = 0; i < segment_keys.size(); ++i) {
    auto segment_key_field_id = segment_key_field_ids[i];
    segment_key_types[i] = input_schema->field(segment_key_field_id)->type().get();
  }

  ARROW_ASSIGN_OR_RAISE(auto segmenter, RowSegmenter::Make(std::move(segment_key_types),
                                                           /*nullable_keys=*/false, ctx));

  // Construct aggregates
  ARROW_ASSIGN_OR_RAISE(auto agg_kernels, GetKernels(ctx, aggs, agg_src_types));