Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
0lai0 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e767210
fix: fall back to Spark for negative-scale decimal casts and arithmetic
0lai0 Jul 27, 2026
02c1291
Merge branch 'main' into fix-5013-negative-scale-decimal-cast
0lai0 Jul 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 28 additions & 9 deletions spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,12 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
private[comet] val negativeScaleDecimalToStringReason: String =
"Negative-scale decimal requires spark.sql.legacy.allowNegativeScaleOfDecimal=true"

// Casts between integer/timestamp and negative-scale decimals panic in the native path
// because scale alignment multiplies by 10^|scale|, which overflows in debug builds
// (aborts) and silently wraps in release builds (wrong results). See issue #5013.
private[comet] val negativeScaleDecimalCastReason: String =
"Cast between integer/timestamp and negative-scale decimal 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 only
Expand Down Expand Up @@ -193,8 +199,8 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
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, _) =>
Expand Down Expand Up @@ -342,6 +348,8 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
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) =>
Expand All @@ -358,6 +366,8 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
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) =>
Expand All @@ -376,6 +386,8 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
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()
Expand All @@ -393,6 +405,8 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
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()
Expand Down Expand Up @@ -423,13 +437,18 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
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 overflows natively; see #5013.
case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType |
DataTypes.TimestampType if fromType.scale < 0 =>

@comphead comphead Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we do early return here?

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 {
Expand Down
33 changes: 27 additions & 6 deletions spark/src/main/scala/org/apache/comet/serde/arithmetic.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.CometSparkSessionExtensions.withFallbackReason
Expand Down Expand Up @@ -90,6 +90,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] = {

@comphead comphead Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we really need changes in this file?

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
Expand Down Expand Up @@ -154,7 +173,7 @@ trait MathBase {
object CometAdd extends CometExpressionSerde[Add] with MathBase {

override def getSupportLevel(expr: Add): SupportLevel =
mathDataTypeSupportLevel(expr.left.dataType)
negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType))

override def convert(
expr: Add,
Expand Down Expand Up @@ -192,7 +211,7 @@ object CometAdd extends CometExpressionSerde[Add] with MathBase {
object CometSubtract extends CometExpressionSerde[Subtract] with MathBase {

override def getSupportLevel(expr: Subtract): SupportLevel =
mathDataTypeSupportLevel(expr.left.dataType)
negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType))

override def convert(
expr: Subtract,
Expand All @@ -212,6 +231,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)

Expand Down Expand Up @@ -251,7 +272,7 @@ object CometMultiply extends CometExpressionSerde[Multiply] with MathBase {
object CometDivide extends CometExpressionSerde[Divide] with MathBase {

override def getSupportLevel(expr: Divide): SupportLevel =
mathDataTypeSupportLevel(expr.left.dataType)
negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType))

override def convert(
expr: Divide,
Expand Down Expand Up @@ -292,7 +313,7 @@ object CometDivide extends CometExpressionSerde[Divide] with MathBase {
object CometIntegralDivide extends CometExpressionSerde[IntegralDivide] with MathBase {

override def getSupportLevel(expr: IntegralDivide): SupportLevel =
mathDataTypeSupportLevel(expr.left.dataType)
negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType))

override def convert(
expr: IntegralDivide,
Expand Down Expand Up @@ -355,7 +376,7 @@ object CometIntegralDivide extends CometExpressionSerde[IntegralDivide] with Mat
object CometRemainder extends CometExpressionSerde[Remainder] with MathBase {

override def getSupportLevel(expr: Remainder): SupportLevel =
mathDataTypeSupportLevel(expr.left.dataType)
negScaleDecimalRejection(expr).getOrElse(mathDataTypeSupportLevel(expr.left.dataType))

override def convert(
expr: Remainder,
Expand Down
82 changes: 82 additions & 0 deletions spark/src/test/scala/org/apache/comet/CometCastSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -794,6 +794,88 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}

test("cast between negative-scale decimal and integer/timestamp is unsupported") {
// Native casts here scale-align by multiplying by 10^|scale|, which overflows the
// underlying integer (panic in debug, silent wrap in release). Both directions fall
// back. `Decimal(neg) -> Timestamp` panics for the same reason (probed). 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") {
val negScaleType = DecimalType(10, -1)
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)

// End-to-end: fall back with the same reason and produce Spark-equal results.
// ConvertToLocalRelation is excluded so the cast actually runs on the plan rather
// than being folded away at plan time (#4789).
val ints = Seq(100, 200, 300).toDF("i")
checkSparkAnswerAndFallbackReason(
ints.select(col("i").cast(negScaleType).as("v")),
CometCast.negativeScaleDecimalCastReason)
// 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 =
Seq("100", "200", "300").toDF("s").select(col("s").cast(negScaleType).as("v"))
checkSparkAnswerAndFallbackReason(
negDec.select(col("v").cast(DataTypes.IntegerType).as("i")),
CometCast.negativeScaleDecimalCastReason)
checkSparkAnswerAndFallbackReason(
negDec.select(col("v").cast(DataTypes.TimestampType).as("t")),
CometCast.negativeScaleDecimalCastReason)
}
}

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 negScaleType = DecimalType(10, -1)
val negDec =
Seq("100", "200", "300").toDF("s").select(col("s").cast(negScaleType).as("v"))
// Casts INTO Decimal(neg) that don't scale-align an integer.
checkSparkAnswerAndOperator(Seq("100", "200").toDF("s").select(col("s").cast(negScaleType)))
checkSparkAnswerAndOperator(
Seq("100", "200").toDF("s").select(col("s").cast(DecimalType(10, 0)).cast(negScaleType)))
withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) -> "true") {
checkSparkAnswerAndOperator(Seq(1.0f, 2.5f).toDF("n").select(col("n").cast(negScaleType)))
checkSparkAnswerAndOperator(Seq(1.0, 2.5).toDF("n").select(col("n").cast(negScaleType)))
}
// Casts OUT of Decimal(neg) that don't scale-align an integer.
checkSparkAnswerAndOperator(negDec.select(col("v").cast(FloatType)))
checkSparkAnswerAndOperator(negDec.select(col("v").cast(DoubleType)))
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)
}
Expand Down
72 changes: 71 additions & 1 deletion spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,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.SimplifyExtractValueOps
import org.apache.spark.sql.comet.CometProjectExec
import org.apache.spark.sql.execution.{ProjectExec, SparkPlan}
Expand All @@ -36,6 +36,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 {
Expand Down Expand Up @@ -1050,6 +1051,75 @@ 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(_, -1) passes Spark's own check.
withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") {
val negScale = DecimalType(10, -1)
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") {
val negScaleType = DecimalType(10, -1)
// 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 v = Seq("100", "200", "300").toDF("s").select(col("s").cast(negScaleType).as("v"))
// v % v keeps both operands at DecimalType(10, -1) 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") {
val negScaleType = DecimalType(10, -1)
val v = Seq("100", "200", "300").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) {
Expand Down