Map `element_at` is ~35x more expensive than every other map kernel, and slower than Spark

[andygrove]

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):

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.