Method OutputResult

cpp/src/arrow/acero/scalar_aggregate_node.cc:290–320

}

Status ScalarAggregateNode::OutputResult(bool is_last) {
  ExecBatch batch{{}, 1};
  batch.values.resize(kernels_.size() + segment_field_ids_.size());

  // First, insert segment keys
  PlaceFields(batch, /*base=*/0, segmenter_values_);

  // Followed by aggregate values
  std::size_t base = segment_field_ids_.size();
  for (size_t i = 0; i < kernels_.size(); ++i) {
    arrow::util::tracing::Span span;
    START_COMPUTE_SPAN(span, 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 ctx{plan()->query_context()->exec_context()};
    ARROW_ASSIGN_OR_RAISE(auto merged, ScalarAggregateKernel::MergeAll(
                                           kernels_[i], &ctx, std::move(states_[i])));
    RETURN_NOT_OK(kernels_[i]->finalize(&ctx, &batch.values[base + i]));
  }

  ARROW_RETURN_NOT_OK(output_->InputReceived(this, std::move(batch)));
  total_output_batches_++;
  if (is_last) {
    ARROW_RETURN_NOT_OK(output_->InputFinished(this, total_output_batches_));
  } else {
    ARROW_RETURN_NOT_OK(ResetKernelStates());
  }
  return Status::OK();
}

}  // namespace aggregate
}  // namespace acero