[SPARK-59524][CORE] Enforce the array-key / HashPartitioner guard in subtractByKey and the Java fakeClassTag key path

[zahed1994]

What changes were proposed in this pull request?

In 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 ClassTags 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.

[zahed1994]

Hi @dongjoon-hyun and @uros-b,

Friendly ping when you have a moment.

This PR addresses the follow-up items discussed in #58758 (SPARK-59524):

1. Adds failOnHashPartitionerWithArrayKey(p) in subtractByKey.
2. Resolves actual key/value ClassTags in JavaSparkContext.parallelizePairs from the in-memory driver collection.
3. Adds @Since("4.4.0") typed fromJavaRDD(rdd, keyClass) overloads in JavaPairRDD to handle JVM type erasure.

Thank you!

[uros-b]

This is a correct, well-scoped follow-up that closes the two gaps identified on the parent PR; guard placement, @SInCE version, and error/test wiring are all right. Before merge, we should weigh one edge: resolving the value-side ClassTag from the first element is unnecessary for the key-only guard and needlessly regresses heterogeneous collections to ArrayStoreException on keys()/values().collect() - prefer leaving ctagV = fakeClassTag.

[uros-b]

Resolving ClassTag[K] AND ClassTag[V] from the first non-null element's runtime class (vs the prior fakeClassTag -> Object) stores a leaf ClassTag. For a heterogeneous collection (e.g. List<Tuple2<Number, V>> with keys [Integer(1), Double(2.0)], legal Java) a later keys().collect() / values().collect() allocates Array[K]/Array[V] via that specific tag (rdd.map(_._1) -> iter.toArray) and can throw ArrayStoreException where Object[] previously succeeded - a newly introduced behavior change in a widely used public API, uncovered by the tests.

[zahed1994]

Thanks for the sharp catch, @uros-b - that is a subtle and critical edge case.

I have updated the PR accordingly:

1. Value side (ctagV): Reverted unconditionally to fakeClassTag. Because HashPartitioner is fundamentally a key-only guard, inferring value types was unnecessary and risked ArrayStoreException when collecting heterogeneous values.
2. Key side (ctagK): Gated the specialization strictly to array types (isArray). For any non-array or heterogeneous keys (e.g. Number with Integer and Double), it falls back to fakeClassTag, ensuring keys().collect() continues allocating Object[] as before without ArrayStoreException.
3. Tests: Added a regression test in JavaAPISuite covering heterogeneous keys and values to ensure neither throws ArrayStoreException, while array keys under HashPartitioner continue to be blocked.