zahed1994 opened a new pull request, #59095: URL: https://github.com/apache/spark/pull/59095
### What changes were proposed in this pull request? In [SPARK-58758](https://issues.apache.org/jira/browse/SPARK-58758) (PR #58758), array keys were prohibited in `HashPartitioner` and operations relying on array hashing (`partitionBy`, `combineByKeyWithClassTag`, `aggregateByKey`, `foldByKey`, `cogroup`). However, two gaps remained: 1. **`subtractByKey`**: `PairRDDFunctions.subtractByKey` creates a `SubtractedRDD` directly and defaults to `HashPartitioner`, but never called `failOnHashPartitionerWithArrayKey(p)`. Because `SubtractedRDD` matches array keys by reference identity, `arrPairs.subtractByKey(arrPairs)` silently returned every element instead of subtracting matching keys. This PR adds `failOnHashPartitionerWithArrayKey(p)` in `subtractByKey`. 2. **Java API `fakeClassTag`**: In the Java API, `JavaSparkContext.parallelizePairs` and `JavaPairRDD.fromJavaRDD` defaulted to `fakeClassTag` (`ClassTag.AnyRef`, whose runtime class is `Object`). Consequently, `keyClass.isArray` evaluated to `false`, and `byte[]` keys bypassed the array-key guard in `JavaPairRDD.partitionBy`, `join`, `cogroup`, and `subtractByKey`. - In `JavaSparkContext.parallelizePairs`, the key/value `ClassTag`s are now dynamically resolved from non-null elements in the driver-side in-memory collection. - In `JavaPairRDD`, overloaded `fromJavaRDD` methods accepting `keyClass: Class[K]` / `valueClass: Class[V]` are added with `@Since("4.4.0")` to allow Java callers to explicitly provide key type information across JVM generic type erasure. ### Why are the changes needed? - Array hashing in Java produces reference-based hash codes rather than content-based hash codes, causing incorrect partitioning, grouping, and subtraction. - Without these fixes, array keys in `subtractByKey` silently produce incorrect results, and Java pair RDDs with array keys completely evade the safety guard introduced in SPARK-58758. ### Does this PR introduce _any_ user-facing change? Yes. Calling `subtractByKey` or Java pair operations (`partitionBy`, `join`, `cogroup`, `subtractByKey`) on an RDD with array keys and a `HashPartitioner` will now fail fast with `SparkException: [UNSUPPORTED_ARRAY_KEY.HASH_PARTITIONER]` rather than silently returning incorrect results. ### How was this patch tested? - Added `subtractByKey` test cases in `core/src/test/scala/org/apache/spark/PartitioningSuite.scala`. - Added `testArrayKeyUnderHashPartitionerFails` in `core/src/test/java/test/org/apache/spark/JavaAPISuite.java`. - Ran: - `build/sbt "core/testOnly org.apache.spark.PartitioningSuite"` - `build/sbt "core/testOnly test.org.apache.spark.JavaAPISuite"` - `build/sbt core/scalastyle` ### Was this patch authored or co-authored using generative AI tooling? No. -- 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]
