andygrove opened a new issue, #5795:
URL: https://github.com/apache/datafusion-comet/issues/5795
`element_at` on a **map** column is the one map operation where Comet loses
to Spark, and it is ~35x more expensive than any other map kernel I measured.
Everything else about Comet's map support is fast — reading and materialising a
whole map column is 1.88x faster than Spark.
### Measurements
2,000,000 rows, `attrs map<string, string>` (10% null maps, 0-6 entries
each, 60 distinct keys, 400 distinct values). Median of 5 iterations after 2
warmups, `noop` sink. Speedup is vs vanilla Spark; higher is better.
| query | Spark | Comet | Gluten/Velox |
|---|---|---|---|
| `SELECT attrs FROM t` (whole map, no kernel) | 169.9 ms | 90.3 ms —
**1.88x** | 52.8 ms — 3.22x |
| `SELECT size(attrs) FROM t` | 73.4 ms | 66.5 ms — 1.10x | 52.6 ms — 1.40x |
| `SELECT size(map_keys(attrs)) FROM t` | 72.2 ms | 70.7 ms — 1.02x | 50.9
ms — 1.42x |
| `SELECT size(map_values(attrs)) FROM t` | 70.4 ms | 69.4 ms — 1.01x | 47.3
ms — 1.49x |
| `SELECT element_at(attrs, 'a1') FROM t` | 97.9 ms | 288.6 ms — **0.34x** |
58.6 ms — 1.67x |
Reading `size(attrs)` as the control (same scan, same column, one kernel,
scalar output), the isolated cost of the lookup is:
- **Comet: +222 ms**
- **Velox: +6 ms**
The plan is fully native in both engines (`CometNativeScan -> CometProject
-> CometColumnarToRow`), so this is not a fallback.
This also fully accounts for a larger query I was benchmarking — `SELECT
event_type, element_at(attrs,'a1'), size(attrs), size(map_keys(attrs)) FROM t`
runs at 0.37x on Comet, and the `element_at` is essentially all of it.
### Likely cause
Reading the source, `GetMapValue` and `element_at(map, key)` both serialise
to the DataFusion `map_extract` scalar UDF (`serde/maps.scala`,
`serde/arrays.scala`; the planner then unwraps the 1-element list in
`planner.rs`). There is no Comet-native map lookup kernel —
`spark-expr/src/map_funcs/` only exports `map_sort`.
Upstream `general_map_extract_inner` in `datafusion-functions-nested` looks
like O(rows x entries-per-row):
```rust
let query_key = query_keys_array.slice(row_index, 1);
let value_index = (0..len).find(|&i| keys.slice(start + i, 1).as_ref() ==
query_key.as_ref());
```
so for every row it re-slices the query key and then linear-scans that row's
entries, allocating two `ArrayRef` slices per candidate and comparing them
through `dyn Array` equality — no hashing, no vectorised compare. The
constant-literal key is not hoisted out of the row loop, and
`make_scalar_function` expands the scalar key to a full-length array first.
That shape matches the measurement (flat per-row cost, independent of the
other map functions), but I have not profiled it to confirm, and I measured
1.0.0 rather than current main — worth verifying before acting on the diagnosis.
If it holds, the constant-key case looks like it could become: encode the
map's keys once, build a gather-index array by hashing or comparing against the
single key, and issue one `take`. Related to the `list_extract` item under
#4942, though that one is the array path, not the map path.
### 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]
