dongjoon-hyun opened a new pull request, #59567:
URL: https://github.com/apache/spark/pull/59567
### What changes were proposed in this pull request?
This PR fixes `mode()` (and the internal `pandas_mode()` that backs
`Series.mode()` and `DataFrame.mode()` in the pandas API on Spark) returning a
wrong result for `BINARY` input.
`Mode` and `PandasMode` count value frequencies in an `OpenHashMap[AnyRef,
Long]` keyed by the input value. A `BINARY` value is a Java `byte[]`, whose
`equals`/`hashCode` are reference-based, and each row reads its value into a
new array. As a result, every occurrence of a binary value got its own buffer
entry with count 1. Merging partial buffers had the same problem, because
`deserialize` also creates a new array for each key.
This PR keys binary values by content in
`TypedAggregateWithHashMapAsBuffer`, the base class that owns this buffer and
its serialization.
- `toBufferKey` wraps a `byte[]` in a new private `BinaryKey` class, whose
`equals`/`hashCode` compare the bytes by content without copying the array.
`fromBufferKey` unwraps it.
- `deserialize` wraps the keys that it reads, and `serialize` unwraps the
keys that it writes.
- `Mode` and `PandasMode` wrap the key in `update`, and unwrap the keys once
at the start of `eval`. So the tie-break of `mode() WITHIN GROUP (ORDER BY
...)` and `mode(col, true)` still compares the original bytes with the existing
binary ordering.
Since the check is on the value, a UDT whose `sqlType` is `BinaryType` is
fixed as well. A struct or array key that contains binary fields was already
compared by content in the buffer, because its row and array representations
compare the bytes. `collect_set` had the same issue and fixed it in SPARK-31500.
### Why are the changes needed?
To fix a correctness bug. `X'02'` is the most frequent value below, but
every row was counted separately, so `mode()` returned an arbitrary input value
and `mode() WITHIN GROUP` returned the smallest or the largest value.
```sql
SELECT
mode(b), -- Expected X'02', but an arbitrary
value (changes between runs)
mode() WITHIN GROUP (ORDER BY b), -- Expected X'02', but X'01'
mode() WITHIN GROUP (ORDER BY b DESC) -- Expected X'02', but X'03'
FROM VALUES (X'02'), (X'01'), (X'01'), (X'02'), (X'02'), (X'03') AS t(b);
```
### Does this PR introduce _any_ user-facing change?
Yes. `mode()` returns the most frequent value for `BINARY` input.
Previously, it returned an arbitrary input value, or the smallest or the
largest value with `WITHIN GROUP` or `deterministic = true`. This bug exists
since `mode()` was added in Apache Spark 3.4.0
([v3.4.0](https://github.com/apache/spark/releases/tag/v3.4.0), 2023-04-07).
### How was this patch tested?
Pass the CIs with the newly added test case in `DataFrameAggregateSuite`.
The test fails without this fix. It covers `mode`, `mode() WITHIN GROUP` (the
same aggregate as `mode(col, true)`) and `pandas_mode` on every aggregation
path: the `Complete` mode, adjacent `Partial` and `Final` aggregates
(serialize, deserialize and merge), and a merge across three partitions. It
also covers a tie, where binary values compare as unsigned bytes.
In addition, I checked a UDT whose `sqlType` is `BinaryType` manually, with
a temporary test UDT that is not part of this PR.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
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]
