Method Finalize

cpp/src/arrow/acero/groupby_aggregate_node.cc:300–337

}

Result<ExecBatch> GroupByNode::Finalize() {
  arrow::util::tracing::Span span;
  START_COMPUTE_SPAN(span, "Finalize",
                     {{"group_by", ToStringExtra(0)}, {"node.label", label()}});

  ThreadLocalState* state = &local_states_[0];
  // If we never got any batches, then state won't have been initialized
  RETURN_NOT_OK(InitLocalStateIfNeeded(state));

  // Allocate a batch for output
  ExecBatch out_data{{}, state->grouper->num_groups()};
  out_data.values.resize(agg_kernels_.size() + key_field_ids_.size() +
                         segment_key_field_ids_.size());

  // Segment keys come first
  PlaceFields(out_data, 0, segmenter_values_);
  // Followed by keys
  ARROW_ASSIGN_OR_RAISE(ExecBatch out_keys, state->grouper->GetUniques());
  std::move(out_keys.values.begin(), out_keys.values.end(),
            out_data.values.begin() + segment_key_field_ids_.size());
  // And finally, the aggregates themselves
  std::size_t base = segment_key_field_ids_.size() + key_field_ids_.size();
  for (size_t i = 0; i < agg_kernels_.size(); ++i) {
    arrow::util::tracing::Span span_item;
    START_COMPUTE_SPAN(span_item, aggs_[i].function,
                       {{"function.name", aggs_[i].function},
                        {"function.options",
                         aggs_[i].options ? aggs_[i].options->ToString() : "<NULLPTR>"},
                        {"function.kind", std::string(kind_name()) + "::Finalize"}});
    KernelContext batch_ctx{plan_->query_context()->exec_context()};
    batch_ctx.SetState(state->agg_states[i].get());
    RETURN_NOT_OK(agg_kernels_[i]->finalize(&batch_ctx, &out_data.values[i + base]));
    state->agg_states[i].reset();
  }
  state->grouper.reset();

  return out_data;
}

Status GroupByNode::OutputNthBatch(int64_t n) {
  int64_t batch_size = output_batch_size();