andygrove commented on code in PR #5050:
URL: https://github.com/apache/datafusion-comet/pull/5050#discussion_r4303075483
##########
native/core/src/execution/planner.rs:
##########
@@ -1511,14 +1511,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.
Review Comment:
I need to walk back something from my first review. The `u8` version of this
test never picked the wrong branch in release builds. `max(s1, s2)` is taken on
the `i8` values before the cast, and with overflow checks off the `u8`
subtraction and addition wrap modulo 256 right back to the true value. I
checked the wrapped expression against your `i16` one for every precision from
1 to 38 and every scale from -128 up to the precision, and they agree on all
31,843,449 combinations. So the bug was the debug-build panic. The `i16`
rewrite is still the right fix. Could this comment, the one in
`create_modulo_expr`, the note on `negScaleDecimalRejection` and the PR
description say that debug builds panicked, rather than that release took the
wrong branch?
##########
spark/src/main/scala/org/apache/comet/expressions/CometCast.scala:
##########
@@ -511,12 +527,26 @@ object CometCast
private def canCastFromDecimal(fromType: DecimalType, toType: DataType):
SupportLevel =
toType match {
- case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType |
DataTypes.LongType
- if fromType.scale < 0 =>
- Unsupported(Some(negativeScaleDecimalToIntegralReason))
- case DataTypes.FloatType | DataTypes.DoubleType | DataTypes.ByteType |
DataTypes.ShortType |
- DataTypes.IntegerType | DataTypes.LongType | DataTypes.BooleanType |
- DataTypes.TimestampType =>
+ // The integral and timestamp kernels scale-align through a power of ten
that overflows for
+ // a negative scale. Matching each target group once keeps the scale
check next to the list
+ // it applies to. See issue #5013.
+ case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType |
+ DataTypes.LongType =>
+ if (fromType.scale < 0) {
+ Unsupported(Some(negativeScaleDecimalToIntegralReason))
+ } else {
+ Compatible()
+ }
+ case DataTypes.TimestampType =>
+ if (fromType.scale < 0) {
+ Unsupported(Some(negativeScaleDecimalToTimestampReason))
+ } else {
+ Compatible()
+ }
+ // Float and double are not gated on scale: `decimal128_to_f64` /
`decimal128_to_f32`
+ // round the exact decimal value once and handle a negative scale
deliberately (#5684).
+ case DataTypes.FloatType | DataTypes.DoubleType | DataTypes.BooleanType
=>
+ Compatible()
Compatible()
Review Comment:
Looks like the merge in `7c940c626` left a second `Compatible()` here. It's
harmless since the block returns the last expression, but could you drop one?
##########
spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala:
##########
@@ -245,6 +245,18 @@ private[codegen] object CometBatchKernelCodegenInput
extends CometTypeShim {
}
}
+ /**
+ * True when a decimal column can be read through the unscaled-long fast
path.
+ *
+ * `Decimal.createUnsafe` produces a long-backed `Decimal` whose
`decimalVal` is null, and
+ * Spark's conversions then index `Decimal.POW_10(scale)`. That throws for a
negative scale,
+ * which `spark.sql.legacy.allowNegativeScaleOfDecimal=true` allows, so a
`try_cast` would
+ * silently return NULL and a plain cast would fail. The `BigDecimal`-backed
reader has no such
+ * assumption, so negative scales take it regardless of precision.
+ */
+ private def canReadDecimalAsUnscaledLong(dt: DecimalType): Boolean =
+ dt.precision <= Decimal.MAX_LONG_DIGITS && dt.scale >= 0
Review Comment:
`CometVector.getDecimal` has the same fast path, at `CometVector.java:91`,
and Spark reads every Comet batch through it after `CometColumnarToRow`. The
new division fallback sends exactly that shape to Spark. With your test confs
(legacy flag on, local table scan on, `ConvertToLocalRelation` excluded) and
`CollapseProject` left alone, this keeps two projections because `v` is used
twice:
```scala
Seq("100", "-100", "0").toDF("s")
.select(col("s").cast(DecimalType(10, -1)).as("v"))
.selectExpr("v / cast(3 as decimal(10, 0)) as q", "cast(v as int) as i")
```
Spark returns `[33.3333333333,100]`, `[-33.3333333333,-100]` and
`[0E-10,0]`. Comet runs the inner projection natively, falls back for the outer
one, and fails with `ArrayIndexOutOfBoundsException: Index -1 out of bounds for
length 19` when Spark's `cast(v as int)` indexes `POW_10(-1)`. `Decimal(20,
-5)` is fine since it takes the `BigDecimal` branch. On `main` the division
runs natively and reaches the `10.pow(~4.29e9)` in `decimal_div`, so this isn't
a regression, but it is the same root cause you fixed here. Adding `&& scale >=
0` to that condition fixed it for me, and the negative-scale tests in both
suites still passed. Could you fold that in, with this query as a test?
##########
spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala:
##########
@@ -1486,6 +1487,111 @@ class CometExpressionSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
}
+ test("division on negative-scale decimal reports unsupported via
getSupportLevel") {
+ // Issue #5013: `spark_decimal_div` computes its scale-alignment exponents
in `u32`, so a
+ // negative scale wraps to roughly 4.29e9 and the BigInt branch attempts
an unbounded
+ // allocation. Only division is affected. The other arithmetic operators
scale-align by
+ // `10^(max_scale - s)`, which is never negative, so they stay native and
are pinned by
+ // "safe ops on negative-scale decimal run natively" below. 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(CometDivide.negScaleDecimalArithmeticReason))
+
+ assert(CometDivide.getSupportLevel(Divide(negAttr, posAttr)) ==
expected)
+ assert(CometIntegralDivide.getSupportLevel(IntegralDivide(negAttr,
posAttr)) == expected)
+ // Guard also fires when negative scale is only on the right operand.
+ assert(CometDivide.getSupportLevel(Divide(posAttr, negAttr)) ==
expected)
+
+ // The unguarded operators must not report unsupported for the same
operands.
+ assert(CometAdd.getSupportLevel(Add(negAttr, posAttr)) == Compatible())
+ assert(CometSubtract.getSupportLevel(Subtract(negAttr, posAttr)) ==
Compatible())
+ assert(CometRemainder.getSupportLevel(Remainder(negAttr, posAttr)) ==
Compatible())
+ assert(CometMultiply.getSupportLevel(Multiply(negAttr, posAttr)) ==
Compatible())
+ }
+ }
+ }
+
+ test("division on negative-scale decimal falls back and returns correct
results") {
+ // End-to-end proof for issue #5013 that the previously-panicking queries
now complete.
+ // `checkSparkAnswerAndFallbackReason` rather than `checkSparkAnswer`:
these five serdes do
+ // not mix in `CodegenDispatchFallback`, so `Unsupported` falls the whole
projection back to
+ // Spark, and asserting the reason is what proves the guard fired. A plain
answer comparison
+ // would pass whether or not it did.
+ // 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"))
+ Seq("v / cast(3 as decimal(10, 0))", "v div cast(3 as decimal(10,
0))").foreach { e =>
+ checkSparkAnswerAndFallbackReason(
+ v.selectExpr(e),
+ CometDivide.negScaleDecimalArithmeticReason)
+ }
+ }
+ }
+ }
+
+ 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"))
+ // Add, Subtract and Remainder scale-align by `10^(max_scale - s)`,
which is never
+ // negative. They were guarded in an earlier revision of this PR
because the `u8`
+ // underflow in the two width tests made them fail. Those are
rewritten in `i16` here,
+ // so they run natively again. If either rewrite is reverted these
assertions fail.
Review Comment:
CI builds the native library with `--profile ci`, which inherits
`overflow-checks = false` from release. So if either width rewrite were
reverted, these assertions would still pass in CI. The `cargo nextest` job runs
with overflow checks on, though. Would you be up for adding a negative-scale
case next to `test_modulo_basic_decimal` in `modulo_expr.rs` and next to
`test_decimal_divide_by_zero_returns_spark_error` in `div.rs`? That would pin
the modulo rewrite and also cover the new `decimal_div` rejection, which has no
test yet. The planner copy is harder to reach from a unit test. Moving the
width calculation into one small function that both sites call would make it
testable and leave one copy of the expression instead of two.
##########
spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala:
##########
@@ -1486,6 +1487,111 @@ class CometExpressionSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
}
+ test("division on negative-scale decimal reports unsupported via
getSupportLevel") {
+ // Issue #5013: `spark_decimal_div` computes its scale-alignment exponents
in `u32`, so a
+ // negative scale wraps to roughly 4.29e9 and the BigInt branch attempts
an unbounded
+ // allocation. Only division is affected. The other arithmetic operators
scale-align by
+ // `10^(max_scale - s)`, which is never negative, so they stay native and
are pinned by
+ // "safe ops on negative-scale decimal run natively" below. 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(CometDivide.negScaleDecimalArithmeticReason))
+
+ assert(CometDivide.getSupportLevel(Divide(negAttr, posAttr)) ==
expected)
+ assert(CometIntegralDivide.getSupportLevel(IntegralDivide(negAttr,
posAttr)) == expected)
+ // Guard also fires when negative scale is only on the right operand.
+ assert(CometDivide.getSupportLevel(Divide(posAttr, negAttr)) ==
expected)
+
+ // The unguarded operators must not report unsupported for the same
operands.
+ assert(CometAdd.getSupportLevel(Add(negAttr, posAttr)) == Compatible())
+ assert(CometSubtract.getSupportLevel(Subtract(negAttr, posAttr)) ==
Compatible())
+ assert(CometRemainder.getSupportLevel(Remainder(negAttr, posAttr)) ==
Compatible())
+ assert(CometMultiply.getSupportLevel(Multiply(negAttr, posAttr)) ==
Compatible())
+ }
+ }
+ }
+
+ test("division on negative-scale decimal falls back and returns correct
results") {
+ // End-to-end proof for issue #5013 that the previously-panicking queries
now complete.
+ // `checkSparkAnswerAndFallbackReason` rather than `checkSparkAnswer`:
these five serdes do
+ // not mix in `CodegenDispatchFallback`, so `Unsupported` falls the whole
projection back to
Review Comment:
A few comments in these tests still describe earlier revisions. "These five
serdes" is two now. The note at line 1533 about precision 20 routing to
`WideDecimalBinaryExpr` or the BigInt path describes a native path this test
never reaches, since it asserts the Spark fallback, and division doesn't use
`WideDecimalBinaryExpr`. Line 1579 and the comment in `arithmetic.scala` at
line 100 refer to "an earlier revision of this PR" and "this PR rewrites",
which won't mean much after merge. Could you bring them in line with the code
as it is?
##########
docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md:
##########
@@ -165,6 +165,22 @@ 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 and Timestamp
+
+Casting a `DecimalType` with a negative scale to `ByteType`, `ShortType`,
`IntegerType`,
+`LongType` 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 align scale by computing `10^|scale|` in fixed-point
arithmetic, which
+overflows when the scale is negative. See
+[#5013](https://github.com/apache/datafusion-comet/issues/5013).
+
+All other cast directions to and from negative-scale decimals run natively:
`FloatType` and
+`DoubleType`, `StringType` in both directions (subject to the config gate
described above),
+`BooleanType`, and widening to another `DecimalType`.
Review Comment:
`BooleanType` only runs natively in one direction. A negative-scale decimal
to `BooleanType` is native, but `canCastFromBoolean` sends every `BooleanType`
to `DecimalType` cast through the dispatcher, whatever the scale. Could this
say casts from a negative-scale decimal to `BooleanType`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
