Method InputReceived

cpp/src/arrow/acero/scalar_aggregate_node.cc:227–258

}

Status ScalarAggregateNode::InputReceived(ExecNode* input, ExecBatch batch) {
  auto scope = TraceInputReceived(batch);
  DCHECK_EQ(input, inputs_[0]);

  auto thread_index = plan_->query_context()->GetThreadIndex();
  auto handler = [this, thread_index](const ExecBatch& full_batch,
                                      const Segment& segment) {
    // (1) The segment is starting of a new segment group and points to
    // the beginning of the batch, then it means no data in the batch belongs
    // to the current segment group. We can output and reset kernel states.
    if (!segment.extends && segment.offset == 0)
      RETURN_NOT_OK(OutputResult(/*is_last=*/false));

    // We add segment to the current segment group aggregation
    auto exec_batch = full_batch.Slice(segment.offset, segment.length);
    RETURN_NOT_OK(DoConsume(ExecSpan(exec_batch), thread_index));
    RETURN_NOT_OK(
        ExtractSegmenterValues(&segmenter_values_, exec_batch, segment_field_ids_));

    // If the segment closes the current segment group, we can output segment group
    // aggregation.
    if (!segment.is_open) RETURN_NOT_OK(OutputResult(/*is_last=*/false));

    return Status::OK();
  };
  RETURN_NOT_OK(HandleSegments(segmenter_.get(), batch, segment_field_ids_, handler));

  if (input_counter_.Increment()) {
    RETURN_NOT_OK(OutputResult(/*is_last=*/true));
  }
  return Status::OK();
}

Status ScalarAggregateNode::InputFinished(ExecNode* input, int total_batches) {
  auto scope = TraceFinish();