Method DoConsume

cpp/src/arrow/acero/scalar_aggregate_node.cc:205–225

}

Status ScalarAggregateNode::DoConsume(const ExecSpan& batch, size_t thread_index) {
  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()) + "::Consume"}});
    KernelContext batch_ctx{plan()->query_context()->exec_context()};
    DCHECK_LT(thread_index, states_[i].size());
    batch_ctx.SetState(states_[i][thread_index].get());

    std::vector<ExecValue> column_values;
    for (const int field : target_fieldsets_[i]) {
      column_values.push_back(batch.values[field]);
    }
    ExecSpan column_batch{std::move(column_values), batch.length};
    RETURN_NOT_OK(kernels_[i]->consume(&batch_ctx, column_batch));
  }
  return Status::OK();
}

Status ScalarAggregateNode::InputReceived(ExecNode* input, ExecBatch batch) {
  auto scope = TraceInputReceived(batch);