[SPARK-59686][SQL] Clamp the rounding scale in round and bround

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala

@@ -1687,6 +1687,13 @@ case class Logarithm(left: Expression, right: Expression)
  * @param mode rounding mode (e.g. HALF_UP, HALF_EVEN)
  * @param modeStr rounding mode string name (e.g. "ROUND_HALF_UP", "ROUND_HALF_EVEN")
  */
+object RoundBase {
+  // Beyond this magnitude a rounding scale cannot change the result for any Spark numeric type,
+  // since 1074 exceeds both the fractional digits of the smallest subnormal double and the
+  // integral digits of the largest finite double.
+  val MAX_SCALE: Int = 1100
+}
+
 abstract class RoundBase(child: Expression, scale: Expression,
     mode: BigDecimal.RoundingMode.Value, modeStr: String)
   extends BinaryExpression with Serializable with ImplicitCastInputTypes with SupportQueryContext {
@@ -1743,7 +1750,14 @@ abstract class RoundBase(child: Expression, scale: Expression,
   // avoid unnecessary `child` evaluation in both codegen and non-codegen eval
   // by checking if scaleV == null as well.
   private lazy val scaleV: Any = scale.eval(EmptyRow)
-  protected lazy val _scale: Int = scaleV.asInstanceOf[Int]
+  // The requested scale is clamped before it reaches `BigDecimal.setScale`, which throws for
+  // magnitudes beyond roughly 1e9. Clamping cannot change a result: no finite value has more
+  // than 309 integral digits, so every scale at or below `-MAX_SCALE` rounds to zero, and the
+  // exact decimal expansion of a finite double needs at most 1074 fractional digits, so every
+  // scale at or above `MAX_SCALE` leaves the value unchanged. It also keeps `-_scale` in
+  // `dataType` from overflowing for a scale of `Int.MinValue`.
+  protected lazy val _scale: Int =
+    math.max(-RoundBase.MAX_SCALE, math.min(RoundBase.MAX_SCALE, scaleV.asInstanceOf[Int]))
 
   override def initQueryContext(): Option[QueryContext] = {
     if (ansiEnabled) {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MathExpressionsSuite.scala

@@ -748,6 +748,33 @@ class MathExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(Truncate(Literal(1.23), Literal.create(null, IntegerType)), null)
   }
 
+  test("round/bround with an extreme scale") {
+    // `BigDecimal.setScale` throws for scales beyond roughly 1e9, so before clamping these
+    // raised a raw java.lang.ArithmeticException, or a spurious ARITHMETIC_OVERFLOW on the
+    // ANSI integral path, even though neighbouring scales return the result below.
+    Seq(Int.MinValue, Int.MinValue + 1, -1000000000, -10000000).foreach { scale =>
+      checkEvaluation(Round(Literal(1.5d), Literal(scale)), 0.0d)
+      checkEvaluation(BRound(Literal(1.5d), Literal(scale)), 0.0d)
+      checkEvaluation(Round(Literal(1.5f), Literal(scale)), 0.0f)
+      checkEvaluation(Round(Literal(1L), Literal(scale)), 0L)
+      checkEvaluation(Round(Literal(1, IntegerType), Literal(scale)), 0)
+      checkEvaluation(Round(Literal(1L), Literal(scale), ansiEnabled = true), 0L)
+    }
+    // A scale past the digits a value carries leaves it unchanged.
+    Seq(Int.MaxValue, Int.MaxValue - 1, 1000000000, 10000000).foreach { scale =>
+      checkEvaluation(Round(Literal(1.5d), Literal(scale)), 1.5d)
+      checkEvaluation(BRound(Literal(1.5d), Literal(scale)), 1.5d)
+      checkEvaluation(Round(Literal(1.5f), Literal(scale)), 1.5f)
+      checkEvaluation(Round(Literal(1L), Literal(scale)), 1L)
+      checkEvaluation(Round(Literal(1L), Literal(scale), ansiEnabled = true), 1L)
+    }
+    // The smallest subnormal has the longest exact decimal expansion, so it is the value most
+    // at risk from clamping the scale.
+    checkEvaluation(Round(Literal(Double.MinPositiveValue), Literal(Int.MaxValue)),
+      Double.MinPositiveValue)
+    checkEvaluation(Round(Literal(Double.MaxValue), Literal(Int.MinValue)), 0.0d)
+  }
+
   test("round/bround/floor/ceil") {
     val scales = -6 to 6
     val doublePi: Double = math.Pi