[SPARK-58217][SQL] CAST(DECIMAL AS TIMESTAMP) silently produces garbage timestamps on overflow

docs/sql-migration-guide.md

@@ -30,6 +30,7 @@ license: |
 - Since Spark 4.3, the adaptive execution rule `org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` has been renamed to `DemoteBroadcastHashJoin`, which now only demotes broadcast hash joins (emitting `NO_BROADCAST_HASH`). Its selection of shuffled hash join over sort merge join has moved to a new physical rule gated by `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default `true`). If you previously disabled the shuffled-hash-join preference by listing `org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` in `spark.sql.adaptive.optimizer.excludedRules`, that name no longer matches any rule (unknown names are silently ignored); set `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` to `false` instead.
 - Since Spark 4.3, the exact `percentile`, `percentile_cont`, and `median` aggregate functions (including their `WITHIN GROUP (ORDER BY ...)` forms) compute the linear interpolation between two neighboring values as `lower + fraction * (higher - lower)` instead of `(1 - fraction) * lower + fraction * higher`. The two are equal in exact arithmetic, but the new form is monotonically non-decreasing in the requested percentage and avoids a rounding error the old form could introduce. As a result these functions may return a value that differs from earlier releases in the last ULP. `percentile_disc` and `percentile_approx` are unaffected.
 - Since Spark 4.3, the new `COMMENT ON COLUMN ... IS NULL` syntax removes a column comment by passing a `null` comment to `TableChange.updateColumnComment(String[], String)`, so a `UpdateColumnComment` table change may now carry a `null` `newComment()`. Previously `newComment()` was always non-null. DataSource V2 catalogs that handle `UpdateColumnComment` should null-check `newComment()` and treat `null` as "remove the column comment" (mirroring how `UpdateColumnDefaultValue` already carries a `null` value to drop a default).
+- Since Spark 4.3, when casting a decimal value to timestamp overflows Long after scaling, Spark returns null (non-ANSI mode) or raises `CAST_OVERFLOW` error (ANSI mode) instead of a wrapping value.
 
 ## Upgrading from Spark SQL 4.1 to 4.2
 

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

@@ -557,6 +557,7 @@ object Cast extends QueryErrorsBase {
     case (TimestampType, ByteType | ShortType | IntegerType) => true
     case (_: TimeType, ByteType | ShortType) => true
     case (FloatType | DoubleType, TimestampType) => true
+    case (_: DecimalType, TimestampType) => true
     case (TimestampType, DateType) => false
     case (_: TimestampLTZNanosType, DateType) => false
     case (_: TimestampNTZNanosType, DateType) => false
@@ -679,6 +680,11 @@ case class Cast(
   with ToStringBase
   with SupportQueryContext
   with QueryErrorsBase {
+
+  private val MICROS_PER_SECOND_BD = java.math.BigDecimal.valueOf(MICROS_PER_SECOND)
+  private val LONG_MAX_BD = java.math.BigDecimal.valueOf(Long.MaxValue)
+  private val LONG_MIN_BD = java.math.BigDecimal.valueOf(Long.MinValue)
+
   override def nullIntolerant: Boolean = true
 
   def this(child: Expression, dataType: DataType, timeZoneId: Option[String]) =
@@ -1009,9 +1015,15 @@ case class Cast(
         DateTimeUtils.makeTimestampNTZNanos(currentDate(zoneId), nanos, precision))
   }
 
-  private[this] def decimalToTimestamp(d: Decimal): Long = {
-    (d.toBigDecimal * MICROS_PER_SECOND).longValue
+  private[this] def decimalToTimestamp(d: Decimal): Any = {
+    val result = d.toJavaBigDecimal.multiply(MICROS_PER_SECOND_BD)
+    if (result.compareTo(LONG_MAX_BD) > 0 || result.compareTo(LONG_MIN_BD) < 0) {
+      errorOrNull(d, DecimalType(d.precision, d.scale), TimestampType)
+    } else {
+      result.longValue()
+    }
   }
+
   private[this] def doubleToTimestamp(d: Double): Any = {
     if (d.isNaN || d.isInfinite) null else (d * MICROS_PER_SECOND).toLong
   }
@@ -2003,7 +2015,31 @@ case class Cast(
       (c, evPrim, evNull) =>
         code"$evPrim = $dateTimeUtilsCls.convertTz($c.epochMicros, $zid, java.time.ZoneOffset.UTC);"
     case DecimalType() =>
-      (c, evPrim, evNull) => code"$evPrim = ${decimalToTimestampCode(c)};"
+      val fromDt = ctx.addReferenceObj("from", from, from.getClass.getName)
+      val toDt = ctx.addReferenceObj("to", TimestampType, TimestampType.getClass.getName)
+      val microsBD = ctx.addReferenceObj("microsBD",
+        MICROS_PER_SECOND_BD, classOf[java.math.BigDecimal].getName)
+      val maxBD = ctx.addReferenceObj("maxBD",
+        LONG_MAX_BD, classOf[java.math.BigDecimal].getName)
+      val minBD = ctx.addReferenceObj("minBD",
+        LONG_MIN_BD, classOf[java.math.BigDecimal].getName)
+      (c, evPrim, evNull) =>
+        val result = ctx.freshName("decResult")
+        val overflow = if (ansiEnabled) {
+          code"""throw QueryExecutionErrors.castingCauseOverflowError($c, $fromDt, $toDt);"""
+        } else {
+          code"$evNull = true;"
+        }
+        code"""
+          java.math.BigDecimal $result = $c.toJavaBigDecimal()
+              .multiply($microsBD);
+          if ($result.compareTo($maxBD) > 0 ||
+              $result.compareTo($minBD) < 0) {
+            $overflow
+          } else {
+            $evPrim = $result.longValue();
+          }
+        """
     case DoubleType =>
       (c, evPrim, evNull) =>
         if (ansiEnabled) {
@@ -2267,10 +2303,6 @@ case class Cast(
         """
   }
 
-  private[this] def decimalToTimestampCode(d: ExprValue): Block = {
-    val block = inline"new java.math.BigDecimal($MICROS_PER_SECOND)"
-    code"($d.toBigDecimal().bigDecimal().multiply($block)).longValue()"
-  }
   private[this] def longToTimeStampCode(l: ExprValue): Block =
     code"java.util.concurrent.TimeUnit.SECONDS.toMicros($l)"
   private[this] def timestampToLongCode(ts: ExprValue): Block =

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

@@ -942,4 +942,48 @@ class CastWithAnsiOffSuite extends CastSuiteBase {
     checkEvaluation(cast(largeTime1, ShortType), null)
     checkEvaluation(cast(largeTime1, ByteType), null)
   }
+
+  test("SPARK-58217: cast large decimal to timestamp with overflow with ansi off") {
+    // Create a Decimal large enough that multiplying by MICROS_PER_SECOND overflows Long
+    val largeDecimal = Literal(Decimal(
+      new java.math.BigDecimal("99999999999999999999"), 38, 0))
+    checkEvaluation(
+      cast(largeDecimal, TimestampType, UTC_OPT), null)
+
+    // Negative overflow should also return null
+    val negativeDecimal = Literal(Decimal(
+      new java.math.BigDecimal("-99999999999999999999"), 38, 0))
+    checkEvaluation(
+      cast(negativeDecimal, TimestampType, UTC_OPT), null)
+
+    // Decimal with non-zero scale whose integer part overflows Long when scaled
+    val scaledDecimal = Literal(Decimal(
+      new java.math.BigDecimal("99999999999999999999.999999"), 38, 6))
+    checkEvaluation(
+      cast(scaledDecimal, TimestampType, UTC_OPT), null)
+
+    // A normal Decimal value should still work (1 second after epoch)
+    val normalDecimal = Literal(Decimal(1L, 10, 0))
+    checkEvaluation(
+      cast(normalDecimal, TimestampType, UTC_OPT), MICROS_PER_SECOND)
+
+    // Zero value should return epoch
+    val zeroDecimal = Literal(Decimal(0L, 10, 0))
+    checkEvaluation(
+      cast(zeroDecimal, TimestampType, UTC_OPT), 0L)
+
+    // Boundary: Long.MAX_VALUE / MICROS_PER_SECOND = 9223372036854
+    // 9223372036854 * 1000000 = 9223372036854000000 < Long.MAX_VALUE -> should pass
+    val justUnder = Literal(Decimal(
+      new java.math.BigDecimal("9223372036854"), 20, 0))
+    checkEvaluation(
+      cast(justUnder, TimestampType, UTC_OPT),
+      new java.math.BigDecimal("9223372036854").longValue * MICROS_PER_SECOND)
+
+    // 9223372036855 * 1000000 = 9223372036855000000 > Long.MAX_VALUE -> should overflow
+    val justOver = Literal(Decimal(
+      new java.math.BigDecimal("9223372036855"), 20, 0))
+    checkEvaluation(
+      cast(justOver, TimestampType, UTC_OPT), null)
+  }
 }

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

@@ -927,4 +927,31 @@ class CastWithAnsiOnSuite extends CastSuiteBase with QueryErrorsBase {
       )
     }
   }
+
+  test("SPARK-58217: cast large decimal to timestamp overflow with ansi on") {
+    val largeDecimal = Literal(Decimal(
+      new java.math.BigDecimal("99999999999999999999"), 38, 0))
+    checkErrorInExpression[SparkArithmeticException](
+      cast(largeDecimal, TimestampType),
+      "CAST_OVERFLOW",
+      Map(
+        "value" -> "99999999999999999999BD",
+        "sourceType" -> "\"DECIMAL(38,0)\"",
+        "targetType" -> "\"TIMESTAMP\"",
+        "ansiConfig" -> "\"spark.sql.ansi.enabled\""
+      ))
+
+    // Negative overflow should also throw
+    val negativeDecimal = Literal(Decimal(
+      new java.math.BigDecimal("-99999999999999999999"), 38, 0))
+    checkErrorInExpression[SparkArithmeticException](
+      cast(negativeDecimal, TimestampType),
+      "CAST_OVERFLOW",
+      Map(
+        "value" -> "-99999999999999999999BD",
+        "sourceType" -> "\"DECIMAL(38,0)\"",
+        "targetType" -> "\"TIMESTAMP\"",
+        "ansiConfig" -> "\"spark.sql.ansi.enabled\""
+      ))
+  }
 }