diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 2607f6d0e56b9..4df51a2e53984 100644 --- a/docs/sql-migration-guide.md +++ b/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 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala index eb7b1d269cdbe..fcc4cdd65347e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala +++ b/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 = diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastWithAnsiOffSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastWithAnsiOffSuite.scala index 9f9a6f275a3fd..a00b2bffae11d 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastWithAnsiOffSuite.scala +++ b/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) + } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastWithAnsiOnSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastWithAnsiOnSuite.scala index ef3e1e6eca9bc..8eced9c6eca80 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastWithAnsiOnSuite.scala +++ b/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\"" + )) + } }