[SPARK-48656][Core] Throw better exception message when CartesianRDD partitions overflow

[timlee0119]

What changes were proposed in this pull request?

Throws a better exception message when integer overflow happens during calculating the number of partitions for CartesianRDD. Before this PR, the following commands

val rdd1 = sc.parallelize(Seq(1, 2, 3), numSlices = 65536)
val rdd2 = sc.parallelize(Seq(1, 2, 3), numSlices = 65536)
rdd2.cartesian(rdd1).partitions

throw this error message:

java.lang.ArrayIndexOutOfBoundsException: Index 0 out of bounds for length 0
  at org.apache.spark.rdd.CartesianRDD.$anonfun$getPartitions$2(CartesianRDD.scala:69)
  at org.apache.spark.rdd.CartesianRDD.$anonfun$getPartitions$2$adapted(CartesianRDD.scala:67)
...

Now it throws

java.lang.ArithmeticException: Integer overflows when calculating the number of partitions for CartesianRDD (rdd1 ID = 1 has 65536 partitions; rdd2 ID = 0 has 65536 partitions). Please reduce the number of partitions in the children RDD.
  at org.apache.spark.rdd.CartesianRDD.getPartitions(CartesianRDD.scala:73)
  at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:340)
...

Throwing overflow exception upfront is clearer than silently wrapping the multiplication result which might or might not trigger the OOB.

Why are the changes needed?

So number of partitions overflowing is easier to debug.

Does this PR introduce any user-facing change?

Yes, see the above console output.

How was this patch tested?

1. Existing tests
2. Manually making sure the error message is updated

Was this patch authored or co-authored using generative AI tooling?

No

[JoshRosen]

If you want to add a unit test case for this, I think we could do it in RDDSuite near the existing unit tests for RDD .cartesian() at

spark/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala

Lines 895 to 915 in d067fc6

	test("cartesian on empty RDD") {
	val a = sc.emptyRDD[Int]
	val b = sc.parallelize(1 to 3)
	val cartesian_result = Array.empty[(Int, Int)]
	assert(a.cartesian(a).collect().toList === cartesian_result)
	assert(a.cartesian(b).collect().toList === cartesian_result)
	assert(b.cartesian(a).collect().toList === cartesian_result)
	}
	test("cartesian on non-empty RDDs") {
	val a = sc.parallelize(1 to 3)
	val b = sc.parallelize(2 to 4)
	val c = sc.parallelize(1 to 1)
	val a_cartesian_b =
	Array((1, 2), (1, 3), (1, 4), (2, 2), (2, 3), (2, 4), (3, 2), (3, 3), (3, 4))
	val a_cartesian_c = Array((1, 1), (2, 1), (3, 1))
	val c_cartesian_a = Array((1, 1), (1, 2), (1, 3))
	assert(a.cartesian[Int](b).collect().toList.sorted === a_cartesian_b)
	assert(a.cartesian[Int](c).collect().toList.sorted === a_cartesian_c)
	assert(c.cartesian[Int](a).collect().toList.sorted === c_cartesian_a)
	}

and in the new test case we could do something like

test("SPARK-48656: cartesian partition count overflow") {
  val e = intercept[ArithmeticException] {
    val rdd1 = sc.parallelize(Seq(1, 2, 3), numSlices = 65536)
    val rdd2 = sc.parallelize(Seq(1, 2, 3), numSlices = 65536)
    rdd2.cartesian(rdd1).partitions
  }
  assert(e.getMessage.contains("CartesianRDD") && e.getMessage.contains("overflow"))
}

or something like that.

[JoshRosen]

FYI, @wayneguow has also opened a PR for this at #47019. Compared to this PR, that other PR has more test cases and also integrates with the new error framework, so I incline towards adopting its approach intead.

[timlee0119]

@JoshRosen Thanks for the pointers for unit testing! Let's try to merge #47019 and I'll close this once that's merged