david-mollitor-db opened a new pull request, #59145:
URL: https://github.com/apache/spark/pull/59145
### What changes were proposed in this pull request?
`CollectFrequentItems` (the `collect_frequent_items` aggregate behind
`DataFrame.stat.freqItems`)
copies every evaluated input value before it looks the value up in its item
-> count map:
```scala
val key = child.eval(input)
if (key != null) {
this.add(buffer, InternalRow.copyValue(key), 1L)
} ...
```
The copy is needed for the keys the map *stores*, because the evaluated
value may point into a
reused input buffer. It is not needed for a lookup that finds an existing
entry: incrementing an
existing entry keeps the entry's stored key, so the fresh copy is simply
discarded.
This PR passes the evaluated value to `add` and copies it only where the map
stores it, at the two
insert sites:
```scala
private def add(map, key, count, copyOnInsert: Boolean) = {
def keyToInsert: Any = if (copyOnInsert) InternalRow.copyValue(key) else
key
map.get(key) match {
case Some(existing) => map(key) = existing + count // no copy
case None => ... map += keyToInsert -> count ... // copy only here
}
}
```
`update` passes `copyOnInsert = true`. `merge` passes `false`, as before,
because its keys come
from another buffer that already owns them. The null special case in
`update` is dropped, since
`InternalRow.copyValue(null)` is `null`.
### Why are the changes needed?
`update` runs once per input row. For a `UTF8String` value, `copyValue` is
`UTF8String.copy()`:
a new `byte[]`, a copy of all the bytes, and a new `UTF8String` object. For
binary, struct, array
and map values, the corresponding `copy()` is similar or more expensive. On
a hit, which is the
common case for the skewed data `freqItems` is meant for, all of that work
is garbage. After this
change a hit does no copy and allocates nothing for the key.
For other types, such as numeric and boolean values, `copyValue` returns its
input unchanged, so
nothing changes for them.
**Open for this WIP:**
- [ ] Benchmark numbers (hit-heavy and miss-heavy string columns), run via
the GitHub Actions
benchmark workflow.
- [ ] JIRA ticket.
- [ ] Possible follow-up: when the map is full, a miss with `count ==
minCount` inserts the key
only for `filterInPlace(v > minCount)` to remove it again. For
`update` (`count = 1`), a new
key never survives once the map is full. Skipping that insert, which
gives the same result,
would also avoid the copy on those misses.
### Does this PR introduce _any_ user-facing change?
No. The map ends up with the same keys and counts; only the copying of
looked-up values changes.
### How was this patch tested?
- Added `collect_frequent_items stores a copy of a value that points into a
reused buffer` to
`DataFrameStatSuite`. It inserts a value, hits it, overwrites the input
bytes as a reused buffer
would, and checks that the stored key is unaffected. It fails if the copy
on insert is removed.
- `build/sbt 'sql/testOnly org.apache.spark.sql.DataFrameStatSuite'` passes
(26 tests), and
scalastyle is clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
This pull request and its description were written by Isaac.
--
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]
