dongjoon-hyun commented on code in PR #59162:
URL: https://github.com/apache/spark/pull/59162#discussion_r4300216442
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -174,7 +189,7 @@ case class CollectFrequentItems(
val row = new UnsafeRow(2)
row.pointTo(bs, sizeOfNextRow)
// Insert the pairs into counts map.
- val key = row.get(0, child.dataType)
+ val key = toKey(row.get(0, child.dataType))
val count = row.get(1, LongType).asInstanceOf[Long]
map.update(key, count)
Review Comment:
Minor and non-blocking: now that `toKey` makes binary keys compare by
content, two serialized entries can decode to the same key, and `map.update`
keeps only the last count instead of adding them up.
A buffer written by this code never has such duplicates, but one written
before this fix can, because every row used to be its own key. One way to hit
this is a streaming query that runs `freqItems` on a binary column (for
example, a Kafka `key`) and restarts from a checkpoint after an upgrade: the
state store keeps the serialized partial buffer, and the restored state is read
back by this `deserialize`. With support 0.4 (two counters), a restored state
`[([1, 2], 1), ([1, 2], 1)]` becomes `{[1, 2] -> 1}`, and two more distinct
keys in the next batch evict it, so `[1, 2]` is missing although it is in 2 of
the 4 rows. With summing, it is kept.
Summing costs nothing for buffers without duplicates:
```suggestion
map.update(key, map.getOrElse(key, 0L) + count)
```
A test can build such a buffer directly: `serialize` a `mutable.Map[Any,
Long]` holding two separate `Array[Byte](1, 2)` keys (`fromKey` passes raw
arrays through), then `deserialize` it and check that the count is 2.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -88,6 +88,21 @@ case class CollectFrequentItems(
override def createAggregationBuffer(): mutable.Map[Any, Long] =
mutable.Map.empty[Any, Long]
+ /**
+ * Returns the map key for `value`. A BinaryType value is wrapped in a
[[BinaryKey]] so that the
+ * map compares it by content; any other value is its own key.
+ */
+ private def toKey(value: Any): Any = value match {
+ case bytes: Array[Byte] => new BinaryKey(bytes)
+ case _ => value
Review Comment:
About the description: "Other key types are not affected" doesn't hold for
every type that falls through here. Two examples:
- `MapType`: `UnsafeMapData` and `ArrayBasedMapData` don't define
`equals`/`hashCode` (on purpose, see the `MapData` scaladoc), so a map column
has the same identity problem as binary. `Seq.fill(12)(Map(1 ->
2)).toDF("m").coalesce(1).stat.freqItems(Seq("m"), 0.5)` returns `[]`. Before
SPARK-40900 (3.4.0), the keys were Scala `Map`s from `df.rdd`, so this used to
work.
- Top-level `NaN`: each row boxes a new `java.lang.Double`, and the Scala
`HashMap` compares keys with `==`, where `NaN != NaN`. So an all-`NaN` column
shows the same pattern as the binary example (`[NaN, NaN]` for 11 rows, `[]`
for 12). `collect_set` keys doubles by their normalized bits for this reason
(SPARK-57298).
I'm not suggesting fixing these in this PR. But since the description
becomes the commit message, could you narrow that sentence to `BinaryType`, and
maybe mention that other types are out of scope? The same applies to the reply
in the `collect_set` thread above, since the two approaches match only for
binary.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameStatSuite.scala:
##########
@@ -461,6 +466,39 @@ class DataFrameStatSuite extends SharedSparkSession {
assert(resultRow.get(1).asInstanceOf[scala.collection.Seq[String]].toSet
== Set("a", "b", null))
}
+ test("SPARK-59898: freqItems compares BinaryType values by content") {
+ // Every row holds the same value, each in its own byte array.
+ Seq(10, 11, 12).foreach { n =>
Review Comment:
Nit: `n = 10` passes without the fix too. Before the fix every row is its
own key, and with two counters (support 0.5) the map is emptied whenever a
third key arrives. So the result has `n % 3` copies of the value, regardless of
the partitioning. 10 rows give one copy, which is the expected result, so only
11 rows (two copies) and 12 rows (none) catch the bug. That also matches
"Without the fix it returned `[[1, 2], [1, 2]]`" in the description, which is
the 11-row case.
Maybe drop 10 and say why the sizes were chosen:
```suggestion
// Every row holds the same value, each in its own byte array. Without
the fix, the value
// came back n % 3 times: twice for 11 rows and not at all for 12.
Seq(11, 12).foreach { n =>
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -88,6 +88,21 @@ case class CollectFrequentItems(
override def createAggregationBuffer(): mutable.Map[Any, Long] =
mutable.Map.empty[Any, Long]
+ /**
+ * Returns the map key for `value`. A BinaryType value is wrapped in a
[[BinaryKey]] so that the
+ * map compares it by content; any other value is its own key.
Review Comment:
Nit: since `toKey` matches the runtime class, a value of a UDT whose
`sqlType` is `BinaryType` is also an `Array[Byte]` here (both from `child.eval`
and from `UnsafeRow.get` in `deserialize`), so it gets wrapped too. That's the
right behavior, because such a UDT had the same bug. But the scaladoc could say
so, and so could "Results for other column types are unchanged" in the
description.
```suggestion
* Returns the map key for `value`. A binary value (BinaryType, or a UDT
whose sqlType is
* BinaryType) is wrapped in a [[BinaryKey]] so that the map compares it
by content; any other
* value is its own key.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -196,3 +211,20 @@ case class CollectFrequentItems(
override protected def withNewChildInternal(newChild: Expression):
Expression =
copy(child = newChild)
}
+
+/**
+ * A [[CollectFrequentItems]] map key for a BinaryType value. Java byte arrays
use referential
+ * equality and identity hash codes, so this compares and hashes the bytes by
content instead. It
+ * wraps the array without copying it.
+ */
+private final class BinaryKey(val bytes: Array[Byte]) {
+ // Computed once: adding a key hashes it twice, to look it up and then to
update or insert it.
+ private[this] val hash = java.util.Arrays.hashCode(bytes)
Review Comment:
Optional, feel free to ignore: `java.util.Arrays.hashCode` (the
31-polynomial hash) collides a lot on structured binary keys. Kafka's
`LongSerializer`/`IntegerSerializer` write big-endian integers, and ids 0..9999
encoded that way have only 1,434 distinct hashes (for example, ids 31 and 256
hash the same). Scala's `mutable.HashMap` chains equal hashes, so with a very
small support (1e-4 gives 10,000 counters), counting an already-counted row
took about 8.7 `Arrays.equals` calls in a quick replica, versus 2 with Murmur3.
At the default support (0.01, 100 counters), it makes no difference. If you
want to match `UTF8String.hashCode`, it uses Murmur3 over its bytes:
```scala
private[this] val hash =
Murmur3_x86_32.hashUnsafeBytes(bytes, Platform.BYTE_ARRAY_OFFSET,
bytes.length, 42)
```
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameStatSuite.scala:
##########
@@ -461,6 +466,39 @@ class DataFrameStatSuite extends SharedSparkSession {
assert(resultRow.get(1).asInstanceOf[scala.collection.Seq[String]].toSet
== Set("a", "b", null))
}
+ test("SPARK-59898: freqItems compares BinaryType values by content") {
+ // Every row holds the same value, each in its own byte array.
+ Seq(10, 11, 12).foreach { n =>
+ val df = Seq.fill(n)(Array[Byte](1, 2)).toDF("b")
+ val items = df.stat.freqItems(Array("b"),
0.5).collect().head.getSeq[Array[Byte]](0)
+ assert(items.map(_.toSeq) === Seq(Seq[Byte](1, 2)))
+ }
+ // A value in half of the rows of a column spread across partitions, whose
partial results
+ // are serialized and merged.
Review Comment:
Nit: this comment reads as if this case covers the serialize/merge path, but
`.contains` can't see how that path would fail. If `deserialize` stopped
re-wrapping keys, `[1, 2]` would come back twice in about half of the partition
and iteration orders, and `.contains` would still pass. The first part of this
test (whose rows are split into two partitions) and the `merge` check in the
next test already catch that deterministically, so only the comment needs
adjusting:
```suggestion
// A value in half of the rows, among distinct values spread across
partitions, is found.
```
This is a different point from the existing comment on L481, which is about
distinct values being merged together.
--
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]
