You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/09/09 13:05:55 UTC
[I] `collect_list` / `collect_set` run at 0.27-0.54x of Spark despite the fastest group-by of the engines tested [datafusion-comet]
andygrove opened a new issue, #5797:
URL: https://github.com/apache/datafusion-comet/issues/5797
On the *same* grouping, Comet's ordinary aggregation is the fastest of the three engines I measured — 2.88x vs Spark, well ahead of Gluten's 1.68x. Add `collect_list` on top of that identical grouping and it goes to 0.49x, i.e. twice as slow as vanilla Spark. So this is not the group-by, the hash table, or the native shuffle; it is the collect accumulators specifically.
### Measurements
2,000,000 rows grouped by a string key with 200,000 distinct values, so every row below shuffles the same key at the same cardinality. Median of 5 iterations after 2 warmups, `noop` sink.
| query | Spark | Comet | Gluten/Velox |
|---|---|---|---|
| `SELECT k, count(*) FROM t GROUP BY k` (control) | 201.8 ms | 70.0 ms — **2.88x** | 120.3 ms — 1.68x |
| `SELECT k, collect_list(s) FROM t GROUP BY k` | 524.5 ms | 1068.7 ms — **0.49x** | 364.7 ms — 1.44x |
| `SELECT k, collect_set(s) FROM t GROUP BY k` | 544.7 ms | 1006.8 ms — **0.54x** | 287.8 ms — 1.89x |
| `SELECT k, collect_list(named_struct('a', s, 'b', l)) FROM t GROUP BY k` | 676.9 ms | 2545.5 ms — **0.27x** | 473.9 ms — 1.43x |
| `SELECT k, size(collect_list(s)) FROM t GROUP BY k` | 541.0 ms | 889.2 ms — 0.61x | 361.5 ms — 1.50x |
Taking the `count(*)` row as the control, the isolated cost of the aggregate is:
| aggregate | Comet | Velox | ratio |
|---|---|---|---|
| `collect_list` (string elements) | **999 ms** | 244 ms | 4.1x |
| `collect_set` (string elements) | **937 ms** | 168 ms | 5.6x |
| `collect_list` (struct elements) | **2476 ms** | 354 ms | 7.0x |
Two further observations:
- **Element type matters a lot in Comet and not at all in Velox.** Struct elements cost 2.5x what string elements cost on Comet; on Velox they are within noise of each other.
- **The output conversion is not the problem.** The last row reduces the array to an int before it reaches the sink, which recovers only 180 ms of the 999 ms. ~83% is the aggregate itself.
Plans are fully native in both engines; no fallback. In a larger query set, per-node SQL metrics put 5.0-8.1 s in `CometHashAggregate [Partial]` where Velox's partial aggregate took 1.0-1.6 s on the same data.
### Notes on cause
`collect_list` / `collect_set` are served by `datafusion_spark`'s `SparkCollectList` / `SparkCollectSet` (planner.rs), which wrap upstream `ArrayAggAccumulator` and `DistinctArrayAggAccumulator`. Reading that code, three things stand out, none of which I have profiled:
1. **Neither declares `groups_accumulator_supported`**, so grouped aggregation goes through DataFusion's `GroupsAccumulatorAdapter` — one boxed `Accumulator` per group plus per-batch slicing and dispatch into each. At 200k groups that is a lot of object overhead sitting on top of an otherwise vectorised aggregate, and it would explain why the gap appears only once `collect_*` replaces `count(*)` on an identical grouping. This looks like the first thing to check.
2. **`DistinctArrayAggAccumulator::merge_batch` is per-row**: it walks the state `ListArray` row by row calling `update_batch(&[val])` on 1-element arrays, so each merged row pays a fresh `RowConverter::append` + `create_hashes` + probe setup. That is `collect_set`-only and matches it being the worse of the two flat-element cases.
3. **`DistinctArrayAggAccumulator::evaluate` converts every distinct element through `ScalarValue::try_from_array` and back**, which for struct elements is a full recursive `ScalarValue::Struct` materialisation per element.
Point 3 is `collect_set`-only, so it does not explain the struct penalty I measured on `collect_list`, where `ArrayAggAccumulator` is element-type agnostic (`filter` + `concat`). I do not have an explanation for that one.
Aggregate functions were explicitly out of scope for the per-row expression scan in #4942, so I do not think any of this is tracked yet.
Measured on 1.0.0; the source pointers are from current main. Cause notes are hypotheses, not diagnoses.
### Environment
- Comet 1.0.0 (`comet-spark-spark3.5_2.12-1.0.0.jar` from Maven Central), Spark 3.5.3, Scala 2.12
- OpenJDK 17.0.20, Ubuntu 22.04 (kernel 6.8), AMD Ryzen 9 7950X3D, 124 GB RAM
- `spark.master=local[4]`, `spark.driver.memory=8g`, `spark.memory.offHeap.size=8g`,
`spark.sql.shuffle.partitions=8`, AQE on, session timezone UTC
- Comet confs: `spark.comet.enabled`, `spark.comet.exec.enabled`,
`spark.comet.exec.shuffle.enabled` all true, `CometShuffleManager`
### Method
Each query is written to a `noop` sink, 2 warmup iterations then 5 measured, median
reported. Source data is a 2,000,000-row synthetic Parquet dataset (snappy, 8 files).
The comparison numbers come from running the identical SQL, on the identical files, in
the identical JVM configuration, with the Gluten 1.6.0 Velox bundle swapped in for
Comet. This is ad-hoc measurement, not a rigorous benchmark harness — the ratios are
large and stable enough to be worth reporting, but please treat the absolute
milliseconds as indicative.
Every plan below was confirmed fully native from `explain` output — no fallback to
Spark, so these are native-vs-native numbers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `collect_list` / `collect_set` run at 0.27-0.54x of Spark despite the fastest group-by of the engines tested [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove closed issue #5797: `collect_list` / `collect_set` run at 0.27-0.54x of Spark despite the fastest group-by of the engines tested
URL: https://github.com/apache/datafusion-comet/issues/5797
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org