This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 4c9287e343b9 [SPARK-59427][SQL] Avoid a redundant map lookup in
CollectFrequentItems.add
4c9287e343b9 is described below
commit 4c9287e343b991861f5dd590038ca0f7d4d0674f
Author: david-mollitor-db <[email protected]>
AuthorDate: Tue Sep 29 20:23:58 2026 +0000
[SPARK-59427][SQL] Avoid a redundant map lookup in CollectFrequentItems.add
### What changes were proposed in this pull request?
`CollectFrequentItems` (the `collect_frequent_items` aggregate behind
`DataFrame.stat.freqItems`)
maintains an item → count buffer (`mutable.Map[Any, Long]`) and updates it
once per input value in
the private `add` method (from both `update` and `merge`). Its "already
tracked" path was:
```scala
if (map.contains(key)) {
map(key) += count
}
```
On a hit this does **three** hash lookups on the same key — `contains`,
then `apply` and `update`
(the desugaring of `map(key) += count`). This PR looks the key up once and
branches on the result,
dropping the hit path to **two** lookups (`get` + `update`):
```scala
map.get(key) match {
case Some(existing) => map(key) = existing + count
case None => // unchanged: insert-if-room, else the bounded-counter
eviction
...
}
```
The miss path is unchanged (one `get` in place of one `contains`), and the
eviction logic in the
`None` branch is untouched.
### Why are the changes needed?
Scala's `HashMap` re-hashes the key on every operation, so the old hit path
recomputed
`key.hashCode()` three times per increment. For a `UTF8String` key — the
common `freqItems` case of
high-cardinality string columns — `hashCode` is an uncached Murmur3 pass
over the whole value (and
`equals` on a collision is a full byte compare), so the redundant lookup is
O(key length). `add`
runs per input value, and once the counter map saturates (its steady state)
the hit path dominates.
Two lookups is the floor for incrementing an existing key whose value is an
immutable `Long` (a read
plus a write), so this removes the one avoidable lookup without changing
the algorithm.
### Does this PR introduce _any_ user-facing change?
No. `map(key) = existing + count` is exactly the previous `map(key) +=
count`; results are identical
for all inputs.
### How was this patch tested?
Existing `DataFrameStatSuite` tests pass (including "Frequent Items" and
"Frequent Items 2");
scalastyle is clean. This is a behavior-preserving internal refactor, so no
new test was added.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #58726 from david-mollitor-db/freqitems-single-map-lookup.
Authored-by: david-mollitor-db <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 09ecd0b9aae921c6aa0034164357059e97909c96)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../spark/sql/execution/stat/FrequentItems.scala | 31 +++++++++++-----------
1 file changed, 16 insertions(+), 15 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala
index c6c0e1fbd60e..214d52da80ce 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala
@@ -89,23 +89,24 @@ case class CollectFrequentItems(
mutable.Map.empty[Any, Long]
private def add(map: mutable.Map[Any, Long], key: Any, count: Long):
mutable.Map[Any, Long] = {
- if (map.contains(key)) {
- map(key) += count
- } else {
- if (map.size < size) {
- map += key -> count
- } else {
- // Non-empty here: this branch runs only when map.size >= size, and
size > 0.
- val minCount = map.values.min
- val remainder = count - minCount
- if (remainder >= 0) {
- map += key -> count // something will get kicked out, so we can add
this
- map.filterInPlace((k, v) => v > minCount)
- map.mapValuesInPlace((k, v) => v - minCount)
+ map.get(key) match {
+ case Some(existing) =>
+ map(key) = existing + count
+ case None =>
+ if (map.size < size) {
+ map += key -> count
} else {
- map.mapValuesInPlace((k, v) => v - count)
+ // Non-empty here: this branch runs only when map.size >= size, and
size > 0.
+ val minCount = map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can
add this
+ map.filterInPlace((k, v) => v > minCount)
+ map.mapValuesInPlace((k, v) => v - minCount)
+ } else {
+ map.mapValuesInPlace((k, v) => v - count)
+ }
}
- }
}
map
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
