diff --git a/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md b/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md index 9d44b603017..205144dc253 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md +++ b/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md @@ -163,6 +163,25 @@ Spark's behavior of using Java `BigDecimal.toString()` semantics, which produces notation (e.g. a value of 12300 stored as `Decimal(7,-2)` with unscaled value 123 is rendered as `"1.23E+4"`). +## Decimal with Negative Scale to Integer, Floating Point, and Timestamp + +Casting a `DecimalType` with a negative scale to `ByteType`, `ShortType`, `IntegerType`, +`LongType`, `FloatType`, `DoubleType`, or `TimestampType` has no native implementation, and +neither does casting any of the integer types to a negative-scale `DecimalType`. Comet reports +these as unsupported, which routes them through the JVM codegen dispatcher (Spark's own +generated code) inside the Comet pipeline, so the results match Spark exactly. + +The native kernels are unusable here for two separate reasons. The integer and timestamp paths +align scale by computing `10^|scale|` in fixed-point arithmetic, which overflows when the scale +is negative. The floating-point paths divide by `10^scale`, which is not exactly representable +for a negative scale, so the result silently diverges from Spark's exact +`BigDecimal.doubleValue()` (a `Decimal(20,-5)` holding `1000000` comes back as +`999999.9999999999`). See [#5013](https://github.com/apache/datafusion-comet/issues/5013). + +All other cast directions to and from negative-scale decimals run natively: `StringType` in +both directions (subject to the config gate described above), `BooleanType`, and widening to +another `DecimalType`. + ## Legacy Mode diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 5cfe74aa412..4012c32052c 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -1075,14 +1075,20 @@ impl PhysicalPlanner { left.data_type(&input_schema), right.data_type(&input_schema), ) { + // The width test is done in `i16` rather than `u8`: a negative scale makes + // `s as u8` wrap to a huge value and `p - s as u8` underflow, which panics in debug + // builds and silently selects the wrong branch in release (`overflow-checks = false`). + // Precision and scale are both bounded well inside `i16`, so this cannot overflow. ( DataFusionOperator::Plus | DataFusionOperator::Minus | DataFusionOperator::Multiply, Ok(DataType::Decimal128(p1, s1)), Ok(DataType::Decimal128(p2, s2)), ) if ((op == DataFusionOperator::Plus || op == DataFusionOperator::Minus) - && max(s1, s2) as u8 + max(p1 - s1 as u8, p2 - s2 as u8) - >= DECIMAL128_MAX_PRECISION) - || (op == DataFusionOperator::Multiply && p1 + p2 >= DECIMAL128_MAX_PRECISION) => + && max(s1 as i16, s2 as i16) + + max(p1 as i16 - s1 as i16, p2 as i16 - s2 as i16) + >= DECIMAL128_MAX_PRECISION as i16) + || (op == DataFusionOperator::Multiply + && p1 as i16 + p2 as i16 >= DECIMAL128_MAX_PRECISION as i16) => { let data_type = return_type.map(to_arrow_datatype).unwrap(); let (p_out, s_out) = match &data_type { diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala index d29ef7cd3b7..9f067848d96 100644 --- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala +++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala @@ -41,6 +41,17 @@ object CometCast private[comet] val negativeScaleDecimalToStringReason: String = "Negative-scale decimal requires spark.sql.legacy.allowNegativeScaleOfDecimal=true" + // Casts between negative-scale decimals and integer / floating-point / timestamp types have + // no usable native path. Integer and timestamp scale-align by multiplying by 10^|scale|, + // which overflows in debug builds (aborts) and silently wraps in release builds (wrong + // results). Float and double divide by 10^scale, which is not exactly representable for a + // negative scale, so the result silently diverges from Spark's exact + // `BigDecimal.doubleValue()` -- Decimal(20,-5) holding 1000000 comes back as + // 999999.9999999999. See issue #5013. + private[comet] val negativeScaleDecimalCastReason: String = + "Cast between negative-scale decimal and integer/floating-point/timestamp is not " + + "supported natively" + // When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and // structs with `[]` (instead of `{}`) when casting to string, and omits NULL elements of // structs/maps/arrays (instead of rendering them as the literal "null"). Comet's native @@ -220,8 +231,8 @@ object CometCast canCastToString(fromType, timeZoneId, evalMode) case (DataTypes.TimestampType, _) => canCastFromTimestamp(toType) - case (_: DecimalType, _) => - canCastFromDecimal(toType) + case (d: DecimalType, _) => + canCastFromDecimal(d, toType) case (DataTypes.BooleanType, _) => canCastFromBoolean(toType, evalMode) case (DataTypes.ByteType, _) => @@ -373,6 +384,8 @@ object CometCast Compatible() case DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType => Compatible() + case d: DecimalType if d.scale < 0 => + Unsupported(Some(negativeScaleDecimalCastReason)) case DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType => Compatible() case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) => @@ -389,6 +402,8 @@ object CometCast Compatible() case DataTypes.ByteType | DataTypes.IntegerType | DataTypes.LongType => Compatible() + case d: DecimalType if d.scale < 0 => + Unsupported(Some(negativeScaleDecimalCastReason)) case DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType => Compatible() case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) => @@ -407,6 +422,8 @@ object CometCast Compatible() case DataTypes.FloatType | DataTypes.DoubleType => Compatible() + case d: DecimalType if d.scale < 0 => + Unsupported(Some(negativeScaleDecimalCastReason)) case _: DecimalType => Compatible() case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) => Compatible() @@ -424,6 +441,8 @@ object CometCast Compatible() case DataTypes.FloatType | DataTypes.DoubleType => Compatible() + case d: DecimalType if d.scale < 0 => + Unsupported(Some(negativeScaleDecimalCastReason)) case _: DecimalType => Compatible() case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) => Compatible() @@ -452,13 +471,19 @@ object CometCast case _ => unsupported(DataTypes.DoubleType, toType) } - private def canCastFromDecimal(toType: DataType): SupportLevel = toType match { - case DataTypes.FloatType | DataTypes.DoubleType | DataTypes.ByteType | DataTypes.ShortType | - DataTypes.IntegerType | DataTypes.LongType | DataTypes.BooleanType | - DataTypes.TimestampType => - Compatible() - case _ => Unsupported(Some(s"Cast from DecimalType to $toType is not supported")) - } + private def canCastFromDecimal(fromType: DecimalType, toType: DataType): SupportLevel = + toType match { + // Negative-scale source either overflows or loses precision natively; see #5013. + case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType | + DataTypes.FloatType | DataTypes.DoubleType | DataTypes.TimestampType + if fromType.scale < 0 => + Unsupported(Some(negativeScaleDecimalCastReason)) + case DataTypes.FloatType | DataTypes.DoubleType | DataTypes.ByteType | DataTypes.ShortType | + DataTypes.IntegerType | DataTypes.LongType | DataTypes.BooleanType | + DataTypes.TimestampType => + Compatible() + case _ => Unsupported(Some(s"Cast from DecimalType to $toType is not supported")) + } private def canCastFromDate(toType: DataType, evalMode: CometEvalMode.Value): SupportLevel = toType match { diff --git a/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala b/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala index 23bca9fa15f..da667ae5190 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala @@ -21,7 +21,7 @@ package org.apache.comet.serde import scala.math.min -import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, Cast, Divide, EmptyRow, EqualTo, EvalMode, Expression, If, IntegralDivide, Literal, Multiply, Remainder, Round, Subtract, UnaryMinus} +import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, BinaryArithmetic, Cast, Divide, EmptyRow, EqualTo, EvalMode, Expression, If, IntegralDivide, Literal, Multiply, Remainder, Round, Subtract, UnaryMinus} import org.apache.spark.sql.types.{ByteType, DataType, DecimalType, DoubleType, FloatType, IntegerType, LongType, ShortType} import org.apache.comet.expressions.{CometCast, CometEvalMode} @@ -89,6 +89,25 @@ trait MathBase { Unsupported(Some(s"Unsupported datatype $dt")) } + // Native decimal Add/Subtract/Divide/IntegralDivide/Remainder scale-aligns operands by + // multiplying by 10^|delta|, which overflows (subtract-with-overflow panic in debug, silent + // wrap in release) whenever any operand or the result has negative scale. See issue #5013. + private[comet] val negScaleDecimalArithmeticReason: String = + "Arithmetic on negative-scale decimal is not supported natively" + + private[comet] def negScaleDecimalRejection(expr: BinaryArithmetic): Option[Unsupported] = { + def isNegScale(dt: DataType): Boolean = dt match { + case d: DecimalType => d.scale < 0 + case _ => false + } + if (isNegScale(expr.left.dataType) || isNegScale(expr.right.dataType) || + isNegScale(expr.dataType)) { + Some(Unsupported(Some(negScaleDecimalArithmeticReason))) + } else { + None + } + } + /** * True when an `Add` / `Multiply` chain of `dataType` in `evalMode` can be rebalanced without * changing results. Only integral types in LEGACY (wrapping, modular) eval mode are exactly @@ -151,8 +170,11 @@ trait MathBase { object CometAdd extends CometExpressionSerde[Add] with MathBase { + override def getUnsupportedReasons(): Seq[String] = + Seq(negScaleDecimalArithmeticReason) + override def getSupportLevel(expr: Add): SupportLevel = - mathDataTypeSupportLevel(expr.left.dataType) + negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType)) override def convert( expr: Add, @@ -189,8 +211,11 @@ object CometAdd extends CometExpressionSerde[Add] with MathBase { object CometSubtract extends CometExpressionSerde[Subtract] with MathBase { + override def getUnsupportedReasons(): Seq[String] = + Seq(negScaleDecimalArithmeticReason) + override def getSupportLevel(expr: Subtract): SupportLevel = - mathDataTypeSupportLevel(expr.left.dataType) + negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType)) override def convert( expr: Subtract, @@ -210,6 +235,8 @@ object CometSubtract extends CometExpressionSerde[Subtract] with MathBase { object CometMultiply extends CometExpressionSerde[Multiply] with MathBase { + // No `negScaleDecimalRejection` guard: Multiply doesn't scale-align operands, so negative-scale + // decimals are safe here. Pinned by a regression test in CometExpressionSuite. See issue #5013. override def getSupportLevel(expr: Multiply): SupportLevel = mathDataTypeSupportLevel(expr.left.dataType) @@ -248,16 +275,20 @@ object CometMultiply extends CometExpressionSerde[Multiply] with MathBase { object CometDivide extends CometExpressionSerde[Divide] with MathBase { - override def getSupportLevel(expr: Divide): SupportLevel = { - if (expr.dataType.isInstanceOf[DecimalType] && - (!expr.left.dataType.isInstanceOf[DecimalType] || - !expr.right.dataType.isInstanceOf[DecimalType])) { - // This is only a sanity check; Spark's type coercion should prevent this case. - Unsupported(Some("Decimal division with a decimal result requires decimal operands")) - } else { - mathDataTypeSupportLevel(expr.left.dataType) + override def getUnsupportedReasons(): Seq[String] = + Seq(negScaleDecimalArithmeticReason) + + override def getSupportLevel(expr: Divide): SupportLevel = + negScaleDecimalRejection(expr).getOrElse { + if (expr.dataType.isInstanceOf[DecimalType] && + (!expr.left.dataType.isInstanceOf[DecimalType] || + !expr.right.dataType.isInstanceOf[DecimalType])) { + // This is only a sanity check; Spark's type coercion should prevent this case. + Unsupported(Some("Decimal division with a decimal result requires decimal operands")) + } else { + mathDataTypeSupportLevel(expr.left.dataType) + } } - } override def convert( expr: Divide, @@ -282,8 +313,11 @@ object CometDivide extends CometExpressionSerde[Divide] with MathBase { object CometIntegralDivide extends CometExpressionSerde[IntegralDivide] with MathBase { + override def getUnsupportedReasons(): Seq[String] = + Seq(negScaleDecimalArithmeticReason) + override def getSupportLevel(expr: IntegralDivide): SupportLevel = - mathDataTypeSupportLevel(expr.left.dataType) + negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType)) override def convert( expr: IntegralDivide, @@ -348,8 +382,11 @@ object CometIntegralDivide extends CometExpressionSerde[IntegralDivide] with Mat object CometRemainder extends CometExpressionSerde[Remainder] with MathBase { + override def getUnsupportedReasons(): Seq[String] = + Seq(negScaleDecimalArithmeticReason) + override def getSupportLevel(expr: Remainder): SupportLevel = - mathDataTypeSupportLevel(expr.left.dataType) + negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType)) override def convert( expr: Remainder, diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index 25b43d9fe6e..75e942b9e6c 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -23,7 +23,7 @@ import scala.util.Random import org.apache.hadoop.fs.Path import org.apache.spark.sql.{Column, CometTestBase, DataFrame, Row} -import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, FromUnixTime, Literal, StructsToJson, TruncDate, TruncTimestamp} +import org.apache.spark.sql.catalyst.expressions.{Add, Alias, AttributeReference, Cast, Divide, FromUnixTime, IntegralDivide, Literal, Remainder, StructsToJson, Subtract, TruncDate, TruncTimestamp} import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, OptimizeIn, SimplifyExtractValueOps} import org.apache.spark.sql.comet.CometProjectExec import org.apache.spark.sql.execution.{ProjectExec, SparkPlan} @@ -34,6 +34,7 @@ import org.apache.spark.sql.internal.SQLConf.SESSION_LOCAL_TIMEZONE import org.apache.spark.sql.types._ import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, isSpark41Plus, isSpark42Plus} +import org.apache.comet.serde.{CometAdd, CometDivide, CometIntegralDivide, CometRemainder, CometSubtract, Unsupported} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator} class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -1046,6 +1047,91 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("arithmetic on negative-scale decimal reports unsupported via getSupportLevel") { + // Guards issue #5013: native scale-align multiplies by 10^|delta| which overflows to a + // subtract-with-overflow panic (debug) / silent wrap (release) whenever any operand or the + // result has negative scale. Constructing the expressions requires + // allowNegativeScaleOfDecimal so DecimalType(_, s<0) passes Spark's own check. + withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") { + Seq(DecimalType(10, -1), DecimalType(20, -5)).foreach { negScale => + val posDec = DecimalType(10, 0) + val negAttr = AttributeReference("n", negScale)() + val posAttr = AttributeReference("p", posDec)() + val expected = Unsupported(Some(CometAdd.negScaleDecimalArithmeticReason)) + + assert(CometAdd.getSupportLevel(Add(negAttr, posAttr)) == expected) + assert(CometSubtract.getSupportLevel(Subtract(negAttr, posAttr)) == expected) + assert(CometDivide.getSupportLevel(Divide(negAttr, posAttr)) == expected) + assert(CometIntegralDivide.getSupportLevel(IntegralDivide(negAttr, posAttr)) == expected) + assert(CometRemainder.getSupportLevel(Remainder(negAttr, posAttr)) == expected) + // Guard also fires when negative scale is only on the right operand. + assert(CometAdd.getSupportLevel(Add(posAttr, negAttr)) == expected) + } + } + } + + test("arithmetic on negative-scale decimal falls back and returns correct results") { + // End-to-end proof for issue #5013 that the previously-panicking queries now complete. + // ConvertToLocalRelation is excluded so the arithmetic actually runs on the plan rather + // than being folded at plan time (#4789). `DecimalType(_, s<0)` must be constructed under + // allowNegativeScaleOfDecimal=true because the case class initializer reads the flag, so + // wrap everything (including the type value) in one block. + withSQLConf( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true", + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> + "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation") { + // Cover a narrow and a wide negative-scale type: at precision 20 the combined operand + // width crosses DECIMAL128_MAX_PRECISION, which routes the operation to + // `WideDecimalBinaryExpr` / `decimal_div`'s BigInt path rather than the narrow arrow + // kernels. `unit` is the smallest magnitude representable at that scale. + val negScaleCases = Seq((DecimalType(10, -1), 10), (DecimalType(20, -5), 100000)) + negScaleCases.foreach { case (negScaleType, unit) => + // Build the neg-scale column via string -> Decimal(neg), a known-safe cast, so the + // source construction does not depend on any of the guards under test. + val strs = Seq(10 * unit, 20 * unit, 30 * unit).map(_.toString) + val v = strs.toDF("s").select(col("s").cast(negScaleType).as("v")) + // v % v keeps both operands at the negative-scale type through Spark's coercion so the + // arithmetic guard is the only thing preventing a native panic. + checkSparkAnswer(v.selectExpr("v % v")) + checkSparkAnswer(v.selectExpr("v + cast(2 as decimal(10, 0))")) + checkSparkAnswer(v.selectExpr("v - cast(2 as decimal(10, 0))")) + checkSparkAnswer(v.selectExpr("v / cast(3 as decimal(10, 0))")) + checkSparkAnswer(v.selectExpr("v div cast(3 as decimal(10, 0))")) + } + } + } + + test("safe ops on negative-scale decimal run natively (regression pin)") { + // Multiply and UnaryMinus on negative-scale decimals don't scale-align an integer and + // ran natively at the time #5013 was fixed. The arithmetic guard intentionally does + // not cover them. If a future change ever adds scale-alignment to Multiply or a + // decimal codepath is added to UnaryMinus that does the multiply-by-10^n trick, + // these assertions will start returning a Comet fallback plan and this test will + // catch it before it reaches production as a silent-wrap or debug panic. + withSQLConf( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true", + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> + "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation") { + // The wide case matters most for `v * v`: Decimal(20,-5) * Decimal(20,-5) has + // p1 + p2 >= DECIMAL128_MAX_PRECISION, so it takes `WideDecimalBinaryExpr`'s multiply + // branch (which does compute a scale adjustment against the output precision/scale) + // and its result type is Decimal(38,-10) -- a negative *output* scale, not just + // negative inputs. + val negScaleCases = Seq((DecimalType(10, -1), 10), (DecimalType(20, -5), 100000)) + negScaleCases.foreach { case (negScaleType, unit) => + val strs = Seq(10 * unit, 20 * unit, 30 * unit).map(_.toString) + val v = strs.toDF("s").select(col("s").cast(negScaleType).as("v")) + // Same-scale Multiply - no alignment needed, must stay native. + checkSparkAnswerAndOperator(v.selectExpr("v * v")) + checkSparkAnswerAndOperator(v.selectExpr("v * cast(2 as int)")) + // UnaryMinus on Decimal(neg) - no alignment, must stay native. + checkSparkAnswerAndOperator(v.selectExpr("-v")) + } + } + } + test("rlike with non-scalar pattern runs via codegen dispatcher") { val table = "rlike_non_scalar" withTable(table) { diff --git a/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala index 6f280848164..624de306691 100644 --- a/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala @@ -61,7 +61,10 @@ import org.apache.comet.serde.{Compatible, Incompatible, Unsupported} * through the dispatcher instead. Adding a native cast implementation therefore means moving a * pair out of the `Unsupported` assertions and into the parity matrix below. */ -class CometNativeCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { +class CometNativeCastSuite + extends CometTestBase + with AdaptiveSparkPlanHelper + with CometCodegenAssertions { import testImplicits._ @@ -1014,6 +1017,120 @@ class CometNativeCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("cast between negative-scale decimal and integer/float/timestamp is unsupported") { + // Native casts here have no usable path in either direction. Integer and timestamp + // scale-align by multiplying by 10^|scale|, which overflows the underlying integer (panic + // in debug, silent wrap in release). Float and double divide by 10^scale, which is not + // exactly representable for a negative scale, so they silently diverge from Spark -- a + // Decimal(20,-5) holding 1000000 comes back as 999999.9999999999. See #5013. + // `DecimalType(_, s<0)` must be constructed under allowNegativeScaleOfDecimal=true + // because the case class initializer reads the flag, so wrap everything in one block. + withSQLConf( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true", + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> + "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation") { + // Cover a narrow and a wide negative-scale type. `unit` is the smallest magnitude the + // type can represent exactly (10^|scale|), so the sample values survive the round trip + // instead of all collapsing to zero at the wider scale. + val negScaleCases = Seq((DecimalType(10, -1), 10), (DecimalType(20, -5), 100000)) + negScaleCases.foreach { case (negScaleType, unit) => + val expected = Unsupported(Some(CometCast.negativeScaleDecimalCastReason)) + val intTypes = + Seq(DataTypes.ByteType, DataTypes.ShortType, DataTypes.IntegerType, DataTypes.LongType) + intTypes.foreach { intType => + assert( + CometCast.isSupported(intType, negScaleType, None, CometEvalMode.LEGACY) == expected, + s"expected $intType -> $negScaleType to be Unsupported") + assert( + CometCast.isSupported(negScaleType, intType, None, CometEvalMode.LEGACY) == expected, + s"expected $negScaleType -> $intType to be Unsupported") + } + // Decimal(neg) -> Timestamp: multiply-with-overflow panic observed in debug build. + assert( + CometCast.isSupported( + negScaleType, + DataTypes.TimestampType, + None, + CometEvalMode.LEGACY) == expected) + // Decimal(neg) -> Float/Double: divides by a 10^scale that is not representable. + Seq(DataTypes.FloatType, DataTypes.DoubleType).foreach { fpType => + assert( + CometCast.isSupported(negScaleType, fpType, None, CometEvalMode.LEGACY) == expected, + s"expected $negScaleType -> $fpType to be Unsupported") + } + + // End-to-end: reporting `Unsupported` does not fall the projection back to Spark -- + // `CometCast` mixes in `CodegenDispatchFallback`, so these route through the JVM codegen + // dispatcher (Spark's own `doGenCode`) and stay in the Comet pipeline. That path is safe + // for negative scale: Spark's generated code works on `Decimal` / `BigDecimal` and the + // Arrow vectors carry the unscaled value with the scale as metadata, so none of the + // `10^|scale|` arithmetic that panics in the native kernel is reachable. Assert both that + // the dispatcher actually ran and that results match Spark. + // ConvertToLocalRelation is excluded so the cast actually runs on the plan rather + // than being folded away at plan time (#4789). + val values = Seq(10 * unit, 20 * unit, 30 * unit) + val ints = values.toDF("i") + assertCodegenRan { + checkSparkAnswerAndOperator(ints.select(col("i").cast(negScaleType).as("v"))) + } + // Build the neg-scale column via string -> Decimal(neg), a known-safe cast, so the + // source construction does not depend on any of the guards under test. + val negDec = + values.map(_.toString).toDF("s").select(col("s").cast(negScaleType).as("v")) + assertCodegenRan { + checkSparkAnswerAndOperator(negDec.select(col("v").cast(DataTypes.IntegerType).as("i"))) + } + assertCodegenRan { + checkSparkAnswerAndOperator( + negDec.select(col("v").cast(DataTypes.TimestampType).as("t"))) + } + assertCodegenRan { + checkSparkAnswerAndOperator(negDec.select(col("v").cast(FloatType).as("f"))) + } + assertCodegenRan { + checkSparkAnswerAndOperator(negDec.select(col("v").cast(DoubleType).as("d"))) + } + } + } + } + + test("safe casts around negative-scale decimal run natively (regression pin)") { + // The guard in canCastFromDecimal / canCastFromByte-Short-Int-Long only rejects the + // paths that scale-align an integer (int <-> Decimal(neg), Decimal(neg) -> Timestamp). + // All other cast directions to/from Decimal(neg) run natively today because they + // don't perform that alignment. Pin those paths here so a future change that adds + // scale-alignment to one of them can't silently reintroduce the #5013 panic. + withSQLConf( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true", + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> + "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation") { + val negScaleCases = Seq((DecimalType(10, -1), 10), (DecimalType(20, -5), 100000)) + negScaleCases.foreach { case (negScaleType, unit) => + val strs = Seq(10 * unit, 20 * unit, 30 * unit).map(_.toString) + val negDec = strs.toDF("s").select(col("s").cast(negScaleType).as("v")) + // Casts INTO Decimal(neg) that don't scale-align an integer. + checkSparkAnswerAndOperator(strs.toDF("s").select(col("s").cast(negScaleType))) + checkSparkAnswerAndOperator( + strs.toDF("s").select(col("s").cast(DecimalType(20, 0)).cast(negScaleType))) + withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) -> "true") { + val floats = Seq(1.0f * unit, 2.5f * unit) + val doubles = Seq(1.0d * unit, 2.5d * unit) + checkSparkAnswerAndOperator(floats.toDF("n").select(col("n").cast(negScaleType))) + checkSparkAnswerAndOperator(doubles.toDF("n").select(col("n").cast(negScaleType))) + } + // Casts OUT of Decimal(neg) that neither scale-align an integer nor divide by + // 10^scale. Float and double are deliberately absent: they divide by a 10^scale that + // is not representable for a negative scale, so they are guarded and covered by the + // dispatcher assertions in the test above. + checkSparkAnswerAndOperator(negDec.select(col("v").cast(StringType))) + checkSparkAnswerAndOperator(negDec.select(col("v").cast(DecimalType(38, 10)))) + checkSparkAnswerAndOperator(negDec.select(col("v").cast(BooleanType))) + } + } + } + test("cast DecimalType(10,2) to TimestampType") { castTest(generateDecimalsPrecision10Scale2(), DataTypes.TimestampType) }