GH-45788: [C++][Acero] Fix data race in aggregate node (#45789)

### Rationale for this change

Data race described in #45788 .

### What changes are included in this PR?

Put the racing member `segmenter_values` in thread local state.

### Are these changes tested?

Yes. UT added.

### Are there any user-facing changes?

None.

* GitHub Issue: #45788

Authored-by: Rossi Sun <zanmato1984@gmail.com>
Signed-off-by: Antoine Pitrou <antoine@python.org>

Author: zanmato1984

cpp/src/arrow/acero/aggregate_internal.cc

@@ -182,9 +182,8 @@ Status ExtractSegmenterValues(std::vector<Datum>* values_ptr,
                               const std::vector<int>& field_ids) {
   DCHECK_GT(input_batch.length, 0);
   std::vector<Datum>& values = *values_ptr;
+  DCHECK_EQ(values.size(), field_ids.size());
   int64_t row = input_batch.length - 1;
-  values.clear();
-  values.resize(field_ids.size());
   for (size_t i = 0; i < field_ids.size(); i++) {
     const Datum& value = input_batch.values[field_ids[i]];
     if (value.is_scalar()) {

cpp/src/arrow/acero/aggregate_internal.h

@@ -171,6 +171,7 @@ class ScalarAggregateNode : public ExecNode, public TracedNode {
         TracedNode(this),
         segmenter_(std::move(segmenter)),
         segment_field_ids_(std::move(segment_field_ids)),
+        segmenter_values_(segment_field_ids_.size()),
         target_fieldsets_(std::move(target_fieldsets)),
         aggs_(std::move(aggs)),
         kernels_(std::move(kernels)),
@@ -249,6 +250,7 @@ class GroupByNode : public ExecNode, public TracedNode {
       : ExecNode(input->plan(), {input}, {"groupby"}, std::move(output_schema)),
         TracedNode(this),
         segmenter_(std::move(segmenter)),
+        segmenter_values_(segment_key_field_ids.size()),
         key_field_ids_(std::move(key_field_ids)),
         segment_key_field_ids_(std::move(segment_key_field_ids)),
         agg_src_types_(std::move(agg_src_types)),

cpp/src/arrow/acero/aggregate_node_test.cc

@@ -213,6 +213,26 @@ TEST(GroupByNode, NoSkipNulls) {
   AssertExecBatchesEqualIgnoringOrder(out_schema, {expected_batch}, out_batches.batches);
 }
 
+TEST(GroupByNode, BasicParallel) {
+  const int64_t num_batches = 8;
+
+  std::vector<ExecBatch> batches(num_batches, ExecBatchFromJSON({int32()}, "[[42]]"));
+
+  Declaration plan = Declaration::Sequence(
+      {{"exec_batch_source",
+        ExecBatchSourceNodeOptions(schema({field("key", int32())}), batches)},
+       {"aggregate", AggregateNodeOptions{/*aggregates=*/{{"hash_count_all", "count(*)"}},
+                                          /*keys=*/{"key"}}}});
+
+  ASSERT_OK_AND_ASSIGN(BatchesWithCommonSchema out_batches,
+                       DeclarationToExecBatches(plan));
+
+  ExecBatch expected_batch = ExecBatchFromJSON(
+      {int32(), int64()}, "[[42, " + std::to_string(num_batches) + "]]");
+  AssertExecBatchesEqualIgnoringOrder(out_batches.schema, {expected_batch},
+                                      out_batches.batches);
+}
+
 TEST(ScalarAggregateNode, AnyAll) {
   // GH-43768: boolean_any and boolean_all with constant input should work well
   // when min_count != 0.
@@ -265,5 +285,24 @@ TEST(ScalarAggregateNode, AnyAll) {
   }
 }
 
+TEST(ScalarAggregateNode, BasicParallel) {
+  const int64_t num_batches = 8;
+
+  std::vector<ExecBatch> batches(num_batches, ExecBatchFromJSON({int32()}, "[[42]]"));
+
+  Declaration plan = Declaration::Sequence(
+      {{"exec_batch_source",
+        ExecBatchSourceNodeOptions(schema({field("", int32())}), batches)},
+       {"aggregate", AggregateNodeOptions{/*aggregates=*/{{"count_all", "count(*)"}}}}});
+
+  ASSERT_OK_AND_ASSIGN(BatchesWithCommonSchema out_batches,
+                       DeclarationToExecBatches(plan));
+
+  ExecBatch expected_batch =
+      ExecBatchFromJSON({int64()}, "[[" + std::to_string(num_batches) + "]]");
+  AssertExecBatchesEqualIgnoringOrder(out_batches.schema, {expected_batch},
+                                      out_batches.batches);
+}
+
 }  // namespace acero
 }  // namespace arrow