fix: fall back to Spark for negative-scale decimal casts and arithmetic

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
 
 <!--BEGIN:CAST_LEGACY_TABLE-->

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 {

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 {

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,