ARROW-16904: [C++] min/max not deterministic if Parquet files have multiple row groups

cpp/src/arrow/compute/exec/options.h

@@ -109,6 +109,10 @@ class ARROW_EXPORT ProjectNodeOptions : public ExecNodeOptions {
 };
 
 /// \brief Make a node which aggregates input batches, optionally grouped by keys.
+///
+/// If the keys attribute is a non-empty vector, then each aggregate in `aggregates` is
+/// expected to be a HashAggregate function. If the keys attribute is an empty vector,
+/// then each aggregate is assumed to be a ScalarAggregate function.
 class ARROW_EXPORT AggregateNodeOptions : public ExecNodeOptions {
  public:
   explicit AggregateNodeOptions(std::vector<Aggregate> aggregates,

cpp/src/arrow/compute/exec/plan_test.cc

@@ -933,6 +933,38 @@ TEST(ExecPlanExecution, SourceGroupedSum) {
   }
 }
 
+TEST(ExecPlanExecution, SourceMinMaxScalar) {
+  for (bool parallel : { false, true }) {
+    SCOPED_TRACE(parallel ? "parallel/merged" : "serial");
+
+    auto input = MakeGroupableBatches(/*multiplicity=*/parallel ? 100 : 1);
+    auto minmax_opts = std::make_shared<ScalarAggregateOptions>();
+    auto expected_value = StructScalar::Make(
+        ScalarVector{ScalarFromJSON(int32(), R"(-8)"), ScalarFromJSON(int32(), R"(12)")},
+        {"min","max"});
+    auto expected_result = ExecBatch::Make({*expected_value});
+
+    ASSERT_OK_AND_ASSIGN(auto plan, ExecPlan::Make());
+    AsyncGenerator<util::optional<ExecBatch>> sink_gen;
+
+    // NOTE: Test `ScalarAggregateNode` by omitting `keys` attribute
+    ASSERT_OK(
+      Declaration::Sequence({
+        {"source",
+         SourceNodeOptions {input.schema, input.gen(parallel, /*slow=*/false)}},
+        {"aggregate", AggregateNodeOptions{/*aggregates=*/{{"min_max",
+                                                           std::move(minmax_opts),
+                                                           "i32", "min_max"}},
+                                           /*keys=*/{}}},
+        {"sink", SinkNodeOptions{&sink_gen}}
+      })
+      .AddToPlan(plan.get()));
+
+    ASSERT_THAT(StartAndCollect(plan.get(), sink_gen),
+                Finishes(ResultWith(UnorderedElementsAreArray({*expected_result}))));
+  }
+}
+
 TEST(ExecPlanExecution, NestedSourceFilter) {
   for (bool parallel : {false, true}) {
     SCOPED_TRACE(parallel ? "parallel/merged" : "serial");

cpp/src/arrow/compute/kernels/aggregate_basic_internal.h

@@ -440,13 +440,11 @@ struct MinMaxImpl : public ScalarAggregator {
     local.has_nulls = !scalar.is_valid;
     this->count += scalar.is_valid;
 
-    if (local.has_nulls && !options.skip_nulls) {
-      this->state = local;
-      return Status::OK();
+    if (!local.has_nulls || options.skip_nulls) {
+      local.MergeOne(internal::UnboxScalar<ArrowType>::Unbox(scalar));
     }
 
-    local.MergeOne(internal::UnboxScalar<ArrowType>::Unbox(scalar));
-    this->state = local;
+    this->state += local;
     return Status::OK();
   }
 
@@ -457,19 +455,17 @@ struct MinMaxImpl : public ScalarAggregator {
     local.has_nulls = null_count > 0;
     this->count += arr.length() - null_count;
 
-    if (local.has_nulls && !options.skip_nulls) {
-      this->state = local;
-      return Status::OK();
-    }
-
-    if (local.has_nulls) {
-      local += ConsumeWithNulls(arr);
-    } else {  // All true values
+    if (!local.has_nulls) {
       for (int64_t i = 0; i < arr.length(); i++) {
         local.MergeOne(arr.GetView(i));
       }
     }
-    this->state = local;
+
+    else if (local.has_nulls && options.skip_nulls) {
+      local += ConsumeWithNulls(arr);
+    }
+
+    this->state += local;
     return Status::OK();
   }
 
@@ -585,17 +581,14 @@ struct BooleanMinMaxImpl : public MinMaxImpl<BooleanType, SimdLevel> {
 
     local.has_nulls = null_count > 0;
     this->count += valid_count;
-    if (local.has_nulls && !options.skip_nulls) {
-      this->state = local;
-      return Status::OK();
+    if (!local.has_nulls || options.skip_nulls) {
+      const auto true_count = arr.true_count();
+      const auto false_count = valid_count - true_count;
+      local.max = true_count > 0;
+      local.min = false_count == 0;
     }
 
-    const auto true_count = arr.true_count();
-    const auto false_count = valid_count - true_count;
-    local.max = true_count > 0;
-    local.min = false_count == 0;
-
-    this->state = local;
+    this->state += local;
     return Status::OK();
   }
 
@@ -604,17 +597,14 @@ struct BooleanMinMaxImpl : public MinMaxImpl<BooleanType, SimdLevel> {
 
     local.has_nulls = !scalar.is_valid;
     this->count += scalar.is_valid;
-    if (local.has_nulls && !options.skip_nulls) {
-      this->state = local;
-      return Status::OK();
+    if (!local.has_nulls || options.skip_nulls) {
+      const int true_count = scalar.is_valid && scalar.value;
+      const int false_count = scalar.is_valid && !scalar.value;
+      local.max = true_count > 0;
+      local.min = false_count == 0;
     }
 
-    const int true_count = scalar.is_valid && scalar.value;
-    const int false_count = scalar.is_valid && !scalar.value;
-    local.max = true_count > 0;
-    local.min = false_count == 0;
-
-    this->state = local;
+    this->state += local;
     return Status::OK();
   }
 };

cpp/src/arrow/compute/kernels/aggregate_test.cc

@@ -1563,7 +1563,7 @@ TEST_F(TestBooleanMinMaxKernel, Basics) {
 TYPED_TEST_SUITE(TestIntegerMinMaxKernel, PhysicalIntegralArrowTypes);
 TYPED_TEST(TestIntegerMinMaxKernel, Basics) {
   ScalarAggregateOptions options;
-  std::vector<std::string> chunked_input1 = {"[5, 1, 2, 3, 4]", "[9, 1, null, 3, 4]"};
+  std::vector<std::string> chunked_input1 = {"[5, 1, 2, 3, 4]", "[9, 8, null, 3, 4]"};
   std::vector<std::string> chunked_input2 = {"[5, null, 2, 3, 4]", "[9, 1, 2, 3, 4]"};
   std::vector<std::string> chunked_input3 = {"[5, 1, 2, 3, null]", "[9, 1, null, 3, 4]"};
   auto item_ty = default_type_instance<TypeParam>();

r/tests/testthat/test-dataset.R

@@ -618,6 +618,31 @@ test_that("UnionDataset handles InMemoryDatasets", {
   expect_equal(actual, expected)
 })
 
+test_that("scalar aggregates with many batches", {
+  test_data <- data.frame(val=1:1e7)
+  expected_result_distr <- (
+    sapply(1:100, function (iter_ndx) {
+      test_data                              %>%
+        dplyr::summarise(min_val = min(val)) %>%
+        dplyr::collect()                     %>%
+        dplyr::pull(min_val)
+    }) %>% table()
+  )
+
+  ds_tmpfile <- tempfile('test-aggregate', fileext='.parquet')
+  arrow::write_parquet(test_data, ds_tmpfile)
+  actual_result_distr <- (
+    sapply(1:100, function (iter_ndx) {
+      arrow::open_dataset(ds_tmpfile)        %>%
+        dplyr::summarise(min_val = min(val)) %>%
+        dplyr::collect()                     %>%
+        dplyr::pull(min_val)
+    }) %>% table()
+  )
+
+  expect_equal(actual_result_distr, expected_result_distr)
+})
+
 test_that("map_batches", {
   ds <- open_dataset(dataset_dir, partitioning = "part")