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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
