[SPARK-60119][SQL] Skip rescaling in Decimal.changePrecision when the result is known from the magnitude

sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala

@@ -31,6 +31,10 @@ import org.apache.spark.unsafe.types.UTF8String
  * into [[CompilationErrors]].
  */
 private[sql] object DataTypeErrors extends DataTypeErrorsBase {
+  // Above this many trailing zeros, a decimal value in an error message is rendered in
+  // scientific notation instead of plain notation.
+  private val MAX_PLAIN_STRING_TRAILING_ZEROS = 1000
+
   def unsupportedOperationExceptionError(): SparkUnsupportedOperationException = {
     new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_2225")
   }
@@ -210,7 +214,10 @@ private[sql] object DataTypeErrors extends DataTypeErrorsBase {
     new SparkArithmeticException(
       errorClass = "NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION",
       messageParameters = Map(
-        "value" -> value.toPlainString,
+        // A huge negative scale (legacy mode only) makes the plain string too long to build,
+        // e.g. for 1E+2147483647.
+        "value" -> (if (value.scale < -MAX_PLAIN_STRING_TRAILING_ZEROS) value.toString
+          else value.toPlainString),
         "precision" -> decimalPrecision.toString,
         "scale" -> decimalScale.toString,
         "config" -> toSQLConf("spark.sql.ansi.enabled")),

sql/api/src/main/scala/org/apache/spark/sql/types/Decimal.scala

@@ -458,6 +458,19 @@ final class Decimal extends Ordered[Decimal] with Serializable {
     if (dv.ne(null)) {
       // We get here if either we started with a BigDecimal, or we switched to one because we would
       // have overflowed our Long; in either case we must rescale dv to the new scale.
+      if (math.abs(dv.scale.toLong - scale) > DecimalType.MAX_PRECISION) {
+        // setScale takes time proportional to the scale change and fails once the change
+        // exceeds the Int range, e.g. for 1e-2147483647. Skip it when the result is known.
+        val numIntegralDigits = dv.precision.toLong - dv.scale
+        if (dv.signum != 0 && numIntegralDigits > precision.toLong - scale) {
+          return false
+        }
+        if (dv.signum == 0 || numIntegralDigits < -scale.toLong - 1) {
+          // |dv| is below 0.01 ulp of the new scale, so the result depends only on the sign
+          // and the rounding mode. Round a value of the same sign and 0.01 ulp instead.
+          dv = BigDecimal(dv.signum, scale + 2)
+        }
+      }
       dv = dv.setScale(scale, roundMode)
       if (dv.precision > precision) {
         return false

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/CastUtils.java

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions;
 
 import org.apache.spark.QueryContext;
+import org.apache.spark.sql.errors.DataTypeErrors;
 import org.apache.spark.sql.errors.QueryExecutionErrors;
 import org.apache.spark.sql.types.DataType;
 import org.apache.spark.sql.types.DataTypes;
@@ -107,7 +108,7 @@ public static short doubleToShortExact(double v) {
   public static Decimal changePrecisionExact(
       Decimal d, int precision, int scale, QueryContext context) {
     if (d.changePrecision(precision, scale)) return d;
-    throw QueryExecutionErrors.cannotChangeDecimalPrecisionError(d, precision, scale, context);
+    throw DataTypeErrors.cannotChangeDecimalPrecisionError(d, precision, scale, context);
   }
 
   public static Decimal changePrecisionOrNull(Decimal d, int precision, int scale) {

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -103,22 +103,6 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
     )
   }
 
-  def cannotChangeDecimalPrecisionError(
-      value: Decimal,
-      decimalPrecision: Int,
-      decimalScale: Int,
-      context: QueryContext = null): ArithmeticException = {
-    new SparkArithmeticException(
-      errorClass = "NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION",
-      messageParameters = Map(
-        "value" -> value.toPlainString,
-        "precision" -> decimalPrecision.toString,
-        "scale" -> decimalScale.toString,
-        "config" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
-      context = getQueryContext(context),
-      summary = getSummary(context))
-  }
-
   def invalidInputSyntaxForBooleanError(
       s: UTF8String,
       context: QueryContext): SparkRuntimeException = {

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

@@ -2144,6 +2144,12 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
       }
   }
 
+  test("SPARK-60119: cast string with a huge negative exponent to decimal") {
+    Seq("1e-2147483647", "-1e-2147483647", "1e-100000000", "0e-2147483647").foreach { str =>
+      checkEvaluation(cast(str, DecimalType(10, 2)), Decimal("0.00"))
+    }
+  }
+
   test("cast ANSI intervals to/from decimals") {
     Seq(
       (Duration.ZERO, DayTimeIntervalType(DAY), DecimalType(10, 3)) -> Decimal(0, 10, 3),

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

@@ -563,6 +563,15 @@ class CastWithAnsiOffSuite extends CastSuiteBase {
     checkEvaluation(cast("abcd", DecimalType(38, 1)), null)
   }
 
+  test("SPARK-60119: cast string with a huge exponent to decimal with negative scale allowed") {
+    withSQLConf(SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED.key -> "true") {
+      Seq("1e2147483647", "-1e2147483646", "12e2147483647", "1e100000000").foreach { str =>
+        checkEvaluation(cast(str, DecimalType(10, 2)), null)
+      }
+      checkEvaluation(cast("0e2147483647", DecimalType(10, 2)), Decimal("0.00"))
+    }
+  }
+
   test("data type casting II") {
     checkEvaluation(
       cast(cast(cast(cast(cast(cast("5", ByteType), TimestampType),

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

@@ -481,6 +481,36 @@ class CastWithAnsiOnSuite extends CastSuiteBase with QueryErrorsBase {
       castErrMsg("abcd", DecimalType(38, 1)))
   }
 
+  test("SPARK-60119: cast string with a huge exponent to decimal with negative scale allowed") {
+    withSQLConf(SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED.key -> "true") {
+      Seq("1e2147483647", "-1e2147483646", "12e2147483647", "1e100000000").foreach { str =>
+        checkExceptionInExpression[ArithmeticException](
+          cast(str, DecimalType(10, 2)),
+          "cannot be represented as Decimal(10, 2)")
+      }
+      checkEvaluation(cast("0e2147483647", DecimalType(10, 2)), Decimal("0.00"))
+
+      // The value is shown in plain notation unless its scale is hugely negative.
+      Seq("1e2147483647" -> "1E+2147483647", "1e40" -> ("1" + "0" * 40)).foreach {
+        case (str, value) =>
+          if (!isTryCast) {
+            checkError(
+              exception = intercept[SparkArithmeticException](
+                cast(str, DecimalType(10, 2)).eval()),
+              condition = "NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION",
+              parameters = Map(
+                "value" -> value,
+                "precision" -> "10",
+                "scale" -> "2",
+                "config" -> """"spark.sql.ansi.enabled""""),
+              queryContext = Array(ExpectedContext(fragment = "", start = -1, stop = -1)))
+          } else {
+            checkEvaluation(cast(str, DecimalType(10, 2)), null)
+          }
+      }
+    }
+  }
+
   protected def checkCastToBooleanError(l: Literal, to: DataType, tryCastResult: Any): Unit = {
     checkExceptionInExpression[SparkRuntimeException](
       cast(l, to), """cannot be cast to "BOOLEAN"""")

sql/catalyst/src/test/scala/org/apache/spark/sql/types/DecimalSuite.scala

@@ -492,4 +492,57 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester with SQLHelper
         s"unscaled: $unscaled, scaleFrom: $scaleFrom, scaleTo: $scaleTo, mode: $roundMode")
     }
   }
+
+  test("SPARK-60119: changePrecision with a source scale far from the target scale") {
+    // A value below 0.01 ulp of the target scale rounds to 0 or +/-1 ulp depending only on its
+    // sign and the rounding mode, so it must round like a small value of the same sign.
+    Seq("1e-2147483647", "-1e-2147483647", "9.99e-100000000", "-1e-100000000",
+      "0e-2147483647").foreach { str =>
+      val bd = new java.math.BigDecimal(str)
+      allSupportedRoundModes.foreach { mode =>
+        val expected = BigDecimal(bd.signum, 50).setScale(2, mode)
+        val d = Decimal(bd)
+        assert(d.changePrecision(10, 2, mode), s"$str, $mode")
+        assert(d.toString === expected.toString, s"$str, $mode")
+      }
+    }
+
+    withSQLConf(SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED.key -> "true") {
+      Seq("1e2147483647", "-1e2147483646", "12e2147483647", "1e100000000").foreach { str =>
+        allSupportedRoundModes.foreach { mode =>
+          val d = Decimal(new java.math.BigDecimal(str))
+          assert(d.toPrecision(10, 2, mode) === null, s"$str, $mode")
+        }
+      }
+      val zero = Decimal(new java.math.BigDecimal("0e2147483647"))
+      assert(zero.changePrecision(10, 2))
+      assert(zero.toString === "0.00")
+    }
+
+    // Values around the bounds of the shortcuts must round as BigDecimal.setScale does.
+    Seq(false, true).foreach { allowNegativeScale =>
+      withSQLConf(
+        SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED.key -> allowNegativeScale.toString) {
+        val targets =
+          Seq((10, 2), (3, 3), (5, 0)) ++ (if (allowNegativeScale) Seq((5, -3)) else Nil)
+        for {
+          // The long mantissas make the scale change large enough to take the shortcuts.
+          mantissa <- Seq("1", "4.9", "5", "5.1", "9.99", "9.995",
+            "1." + "0" * 79 + "1", "4." + "9" * 80, "5." + "0" * 80, "5." + "0" * 79 + "1",
+            "9." + "9" * 80)
+          exponent <- -8 to 12
+          sign <- Seq("", "-")
+          mode <- allSupportedRoundModes
+          (precision, scale) <- targets
+        } {
+          val bd = new java.math.BigDecimal(s"$sign${mantissa}e$exponent")
+          val rounded = BigDecimal(bd).setScale(scale, mode)
+          val expected = if (rounded.precision > precision) null else rounded.toString
+          val d = Decimal(bd).toPrecision(precision, scale, mode)
+          assert(Option(d).map(_.toString).orNull === expected,
+            s"$bd, $mode, ($precision, $scale)")
+        }
+      }
+    }
+  }
 }