`collect_list` / `collect_set` run at 0.27-0.54x of Spark despite the fastest group-by of the engines tested

[andygrove]

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.