Method InitLocalStateIfNeeded

cpp/src/arrow/acero/groupby_aggregate_node.cc:419–449

}

Status GroupByNode::InitLocalStateIfNeeded(ThreadLocalState* state) {
  // Get input schema
  auto input_schema = inputs_[0]->output_schema();

  if (state->grouper != nullptr) return Status::OK();

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

  // Construct grouper
  ARROW_ASSIGN_OR_RAISE(state->grouper,
                        Grouper::Make(key_types, plan_->query_context()->exec_context()));

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

  ARROW_ASSIGN_OR_RAISE(state->agg_states,
                        InitKernels(agg_kernels_, plan_->query_context()->exec_context(),
                                    aggs_, agg_src_types));

  return Status::OK();
}

}  // namespace aggregate
}  // namespace acero