david-mollitor-db opened a new pull request, #59162:
URL: https://github.com/apache/spark/pull/59162
### What changes were proposed in this pull request?
`CollectFrequentItems`, the aggregate behind `DataFrame.stat.freqItems`,
keeps its counters in a
`mutable.Map[Any, Long]` keyed by the column value. A `BinaryType` value is
an `Array[Byte]`, and
Java arrays use referential equality and identity hash codes. Each row's
value is a separate
array, so equal binary values never match.
This PR keys binary values by a small private `BinaryKey` wrapper instead.
The wrapper compares and
hashes the bytes by content (`java.util.Arrays.equals` / `hashCode`, the
hash computed once), and
it wraps the array without copying it:
```scala
private final class BinaryKey(val bytes: Array[Byte]) {
private[this] val hash = java.util.Arrays.hashCode(bytes)
override def hashCode(): Int = hash
override def equals(other: Any): Boolean = other match {
case that: BinaryKey => java.util.Arrays.equals(bytes, that.bytes)
case _ => false
}
}
```
- `update` wraps each binary value as it is added:
`toKey(InternalRow.copyValue(key))`.
- `deserialize` wraps binary keys again as it reads them.
- `eval` and `serialize` unwrap them, so the output type and the serialized
buffer format are
unchanged.
- Every other value is its own key, as before.
Not copying the array is as safe as the current code, which already stores
the array itself:
`InternalRow.copyValue` returns an `Array[Byte]` unchanged.
`CollectSet` solves the same problem by keying binary values with an
`UnsafeArrayData` copy of
their bytes. Here that would add a copy of every input row's value; the
wrapper adds only one
small object per row.
### Why are the changes needed?
It is a correctness bug. Because equal binary values never match, their
counts never exceed 1, and
once the map is full the eviction step keeps clearing it. The result depends
on the number and
order of rows rather than on how frequent a value is. A value can come back
several times, or be
missing even when it is in every row. That breaks `freqItems`' documented
guarantee: false
positives are possible, but false negatives are not.
```scala
Seq(11, 12).foreach { n =>
val df = Seq.fill(n)(Array[Byte](1, 2)).toDF("b").coalesce(1) // every
row is the same value
println(df.stat.freqItems(Seq("b"),
0.5).collect().head.getSeq[Array[Byte]](0).map(_.toSeq))
}
// before: n=11 -> ArraySeq(ArraySeq(1, 2), ArraySeq(1, 2)), n=12 ->
ArraySeq()
// after: ArraySeq(ArraySeq(1, 2)) for both
```
Binary columns come up in practice. The `key` and `value` columns of a batch
Kafka read are
`BinaryType`, so looking for hot keys with `freqItems` hits exactly this
case. Parquet files from
systems that don't annotate strings are read as binary unless
`spark.sql.parquet.binaryAsString`
is set. Hash digests and UUIDs are often stored as binary too.
The bug dates back to the original implementation (SPARK-7242, 1.4.0), which
keyed a
`Map[Any, Long]` by `row.get(i)`. The rewrite as an aggregate (SPARK-40900,
3.4.0) kept the same
keying.
Other key types are not affected. `UTF8String`, `BinaryView` (used by
`GEOMETRY` / `GEOGRAPHY`),
`UnsafeRow`, `UnsafeArrayData` and `GenericArrayData` all compare by
content, including binary
values nested in a struct or array.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix. `freqItems` on a `BinaryType` column now counts equal values
together, so it
returns each frequent value once and no longer misses frequent values.
Results for other column
types are unchanged.
### How was this patch tested?
New tests in `DataFrameStatSuite`, both of which fail without the fix:
- `SPARK-59898: freqItems compares BinaryType values by content`, which is
end to end:
- A column where every row is the same binary value (10, 11 and 12 rows)
returns that value
exactly once. Without the fix it returned `[[1, 2], [1, 2]]`.
- A binary value in half of the rows of a column spread over 4 partitions
is found. This path
also serializes and merges partial results.
- `SPARK-59898: collect_frequent_items keys BinaryType values by content`,
which calls the
aggregate directly. Three updates with the same bytes plus a null give two
entries, before and
after a `serialize`/`deserialize` round trip and a `merge`. Without the
fix there were three
separate `[1, 2]` entries.
`build/sbt 'sql/testOnly org.apache.spark.sql.DataFrameStatSuite'` passes
(27 tests), and
scalastyle is clean.
This overlaps #59145 (SPARK-59896), which changes the same lines in
`update`; whichever merges
second needs a small rebase. When they are combined, binary values still
have to be wrapped before
the lookup, since the lookup itself needs content equality.
### 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]
