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

[asfimport]

The following code produces non-deterministic result for getting the minimum value of a sequence of 1e5 and 1e6 integers.

sapply(1:100, function(x) {
  # create parquet file with a val column with numbers 1 to 100,000
  arrow::write_parquet(
    data.frame(val = 1:1e5), "test.parquet")
  # find minimum value
  arrow::open_dataset("test.parquet") %>%
    dplyr::summarise(min_val = min(val)) %>%
    dplyr::collect() %>% dplyr::pull(min_val)
}) %>% table()

sapply(1:100, function(x) {
  # create parquet file with a val column with numbers 1 to 1,000,000
  arrow::write_parquet(
    data.frame(val = 1:1e6), "test.parquet")
  # find minimum value
  arrow::open_dataset("test.parquet") %>%
    dplyr::summarise(min_val = min(val)) %>%
    dplyr::collect() %>% dplyr::pull(min_val)
}) %>% table()

The first 100 simulations using numbers 1 to 1e5 is able to find the minimum number (1) all 100 times.

The second 100 simulations using numbers 1 to 1e6 only finds the minimum number (1) 65 out of 100 times. It finds near multiples of 131073, 25, 8, and 2 times respectively.

. 1
100 
. 1 131073 262145 393217 
 65     25      8      2 

 

Environment: $ lsb_release -a
No LSB modules are available.
Distributor ID: Ubuntu
Description: Ubuntu 20.04.4 LTS
Release: 20.04
Codename: focal
Reporter: Robert On
Assignee: Aldrin Montana / @drin

Related issues:

• [C++] count_distinct aggregates incorrectly across row groups (relates to)

PRs and other links:

• GitHub Pull Request #13509

_{Note: This issue was originally created as ARROW-16904. Please see the migration documentation for further details.}

[asfimport]

Neal Richardson / @nealrichardson:
That's not good. I am pretty sure this is the same issue as ARROW-16807. I can reproduce it with the starwars data example used there:

> replicate(100, ds %>% summarize(min(height, na.rm = TRUE)) %>% pull())
  [1] 79 79 66 66 79 79 66 79 66 66 66 66 66 79 66 66 66 66 79 94 88 79 66 79 79
 [26] 66 66 79 66 66 66 66 66 66 66 79 79 66 79 79 66 88 79 66 66 94 66 66 66 79
 [51] 66 66 66 66 79 66 66 66 79 66 94 79 79 79 66 79 66 79 79 66 79 66 79 66 88
 [76] 88 66 66 66 66 66 66 66 66 66 66 66 79 79 66 66 79 66 66 66 66 66 66 66 66

[asfimport]

Aldrin Montana / @drin:
I'll pick this up for now and update based on progress on ARROW-16807. It certainly sounds likely to be the same issue, but I'll update when I find out.

[asfimport]

Jeroen van Straten / @jvanstraten:
I probably fixed this as part of https://issues.apache.org/jira/projects/ARROW/issues/ARROW-16700 / #13509. min/max wasn't working correctly when multiple Consume calls would be chained for the same ScalarAggregator instance; only the last call would affect the state. I'm not deep enough into Acero to understand under what circumstances it follows this pattern (which was broken and isn't tested) and under what circumstances it will only call Consume once per instance and then Merge the instances (which works correctly and is tested), though.

[asfimport]

Aldrin Montana / @drin:
Created a PR to add to @jvanstraten 's PR:
    jvanstraten#5

I mainly add tests, but I am also trying to include some minor re-organization of the kernels themselves where the fixes were added. Currently, the additional tests only capture this specific bug, and don't exercise the other fixes in the same source file (aggregate_basic_internal.h).

[asfimport]

Aldrin Montana / @drin:
previous PR was merged into Jeroen's branch, so just waiting on PR#13509's approval.

[asfimport]

David Li / @lidavidm:
Issue resolved by pull request 13509
#13509

[asfimport]

Robert On:
Confirmed that nightly build of arrow (v8.0.0.20220712) computes the correct minimums 100 out of 100 times:

> sapply(1:100, function {
+ # create parquet file with a single column with numbers 1 to 1,000,000
+   arrow::write_parquet(
+     data.frame(val = 1:1e6), "test.parquet")
+ 
+   arrow::open_dataset("test.parquet") %>%
+   dplyr::summarise(min_val = min(val)) %>%
+   dplyr::collect() %>% dplyr::pull(min_val)
+ }) %>% table().
1 
100