TPC-DS Q75 regression in DF 55

[gruuya]

Describe the bug

We noticed a significant regression in our benchmarks, notably for TPC-DS Q75.

To Reproduce

The essence of the regression can be captured by this minimal example

> SELECT VERSION();
+--------------------------------------------+
| version()                                  |
+--------------------------------------------+
| Apache DataFusion 55.0.0, aarch64 on macos |
+--------------------------------------------+
1 row(s) fetched.
Elapsed 0.013 seconds.

> COPY (
  SELECT (v % 5)::int      AS d_year,
         (v % 1000)::int   AS brand,
         (v % 100)::int    AS class,
         (v % 10)::int     AS cat,
         (v % 997)::int    AS manu,
         (v % 97)::bigint  AS cnt,
         arrow_cast(v::double / 100.0, 'Decimal128(31, 15)') AS amt
  FROM generate_series(1, 10000000) AS s(v)
  ORDER BY d_year
) TO '/tmp/sorted.parquet' STORED AS PARQUET;

> CREATE EXTERNAL TABLE src STORED AS PARQUET LOCATION '/tmp/sorted.parquet'
  WITH ORDER (d_year ASC);
+----------+
| count    |
+----------+
| 10000000 |
+----------+
1 row(s) fetched.
Elapsed 0.743 seconds.

0 row(s) fetched.
Elapsed 0.003 seconds.

> SELECT count(*) FROM (
  SELECT DISTINCT d_year, brand, class, cat, manu, cnt, amt FROM src
);
+----------+
| count(*) |
+----------+
| 10000000 |
+----------+
1 row(s) fetched.
Elapsed 4.025 seconds.

Crucially the regression can be toggled away by disabling the enable_migration_aggregate config

> SET datafusion.execution.enable_migration_aggregate = false;
0 row(s) fetched.
Elapsed 0.001 seconds.

> SELECT count(*) FROM (
  SELECT DISTINCT d_year, brand, class, cat, manu, cnt, amt FROM src
);
+----------+
| count(*) |
+----------+
| 10000000 |
+----------+
1 row(s) fetched.
Elapsed 0.470 seconds.

Expected behavior

Compare to default behavior on DF54 (prior to enable_migration_aggregate config)

> SELECT VERSION();
+--------------------------------------------+
| version()                                  |
+--------------------------------------------+
| Apache DataFusion 54.1.0, aarch64 on macos |
+--------------------------------------------+
1 row(s) fetched.
Elapsed 0.013 seconds.

> CREATE EXTERNAL TABLE src STORED AS PARQUET LOCATION '/tmp/sorted.parquet'
  WITH ORDER (d_year ASC);
0 row(s) fetched.
Elapsed 0.003 seconds.

> SELECT count(*) FROM (
  SELECT DISTINCT d_year, brand, class, cat, manu, cnt, amt FROM src
);
+----------+
| count(*) |
+----------+
| 10000000 |
+----------+
1 row(s) fetched.
Elapsed 0.513 seconds.

Additional context

Potentially related to #24980

[alamb]

Likely related

• [EPIC] Split Aggregation Logic into Dedicated Streams #22710