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

[0lai0]

Which issue does this PR close?

Closes #5013

Rationale for this change

With spark.sql.legacy.allowNegativeScaleOfDecimal=true, Spark allows DecimalType(p, s) with s < 0. Comet's native cast and decimal arithmetic kernels scale-align values by computing 10^|scale| (e.g. 10_i128.pow(scale as u32)). When scale is negative, casting it to u32 produces a huge value, so the power computation overflows: it panics (attempt to multiply/subtract with overflow) in debug builds and silently wraps to an incorrect value in release builds (the native workspace sets overflow-checks = false for release).

This is reachable from two places:

• Casting an integer type (Byte/Short/Integer/Long) to/from a negative-scale DecimalType, or casting a negative-scale DecimalType to TimestampType, both hit the same 10^|scale| computation in numeric.rs / cast_decimal_to_timestamp.
• Arithmetic on negative-scale decimals (Add/Subtract/Divide/IntegralDivide/Remainder), which scale-aligns operands the same way.

CometCast already special-cased negative-scale decimals for the Decimal -> String direction, but that guard was unreachable in practice because producing the negative-scale value in the first place would panic before a -> String cast could ever run.

Since allowNegativeScaleOfDecimal is a rarely-used legacy flag (default false), and the native kernels don't implement negative-scale-safe rescaling, the fix makes Comet detect these cases at planning time and fall back to Spark instead of attempting them natively.

What changes are included in this PR?

• CometCast.scala: added negativeScaleDecimalCastReason and marked as Unsupported:

  • int -> Decimal(neg) via canCastFromByte/Short/Int/Long
  • Decimal(neg) -> int and Decimal(neg) -> Timestamp via canCastFromDecimal (now takes the source DecimalType so it can check fromType.scale)

  All other cast directions to/from negative-scale decimals (e.g. String <-> Decimal(neg), Decimal(neg) -> Float/Double/String/Boolean/wider-Decimal) don't perform integer scale-alignment and are unaffected, they keep running natively.

• arithmetic.scala: added negScaleDecimalRejection(expr: BinaryArithmetic) in MathBase, returning Unsupported when the left operand, right operand, or result type is a negative-scale DecimalType. Wired into CometAdd, CometSubtract, CometDivide, CometIntegralDivide, CometRemainder.

  CometMultiply and UnaryMinus are intentionally left unguarded, neither scale-aligns, so they stay native

How are these changes tested?

Added tests to CometCastSuite and CometExpressionSuite
make core
./mvnw -pl spark test -Dsuites='org.apache.comet.CometCastSuite'
./mvnw -pl spark test -Dsuites='org.apache.comet.CometExpressionSuite'

Note: the TimestampType case is not called out in the linked issue's reproducer, it was found while auditing the same 10^|scale| code path during this fix and is included here since it shares the exact same root cause and fix.

[andygrove]

This may overlap with #4799

[0lai0]

Thanks @andygrove and @comphead, sorry for missing #4799! That would be great if it covers it.
I'd be glad to help review #4799.

[andygrove]

This is a small targeted PR that seems ready for review and #4799 is still in flux. Let's reopen this one.

[andygrove]

Thanks for this @0lai0. The analysis in the description is careful and mostly holds up. I traced every pow(scale as u32) site in native/ against your guards, and the ones you deliberately left native really are safe:

• float/double → Decimal(neg): cast_floating_point_to_decimal128 uses 10_f64.powi(scale as i32) (numeric.rs:901)
• string ↔ Decimal(neg): sign-checked i32 branches (string.rs:603-620)
• Decimal(neg) → Boolean: spark_cast_decimal_to_boolean has no scale math (numeric.rs:860)
• Multiply: the wide path's scale_diff branches are sign-checked (wide_decimal_binary_expr.rs:256-275), and the narrow path short-circuits past the panicking subtraction in the planner guard, so leaving it unguarded is correct
• Avg: target_scale - sum_scale is (s+4) - s = 4 regardless of sign, so avg_decimal.rs:382 is fine
• Ceil/Floor/Round were already guarded

Also agree the Scala suites are the right home here rather than CometSqlFileTestSuite, since negative scale cannot be written in SQL.

A few things to address.

Rebase needed

This is showing as conflicting with main and no CI has run on the branch. CometCast has changed since 27 July, it now mixes in CometTypeShim and CodegenDispatchFallback, and getSupportLevel gained a literal-child short-circuit. Could you rebase so we get a green CI signal? I would like to see the new tests pass on a debug build before merging, since debug is where the panic surfaces.

Boolean → Decimal(negative scale) looks like the same bug, unguarded

CometCast.scala:347 (canCastFromBoolean) returns Compatible() for _: DecimalType with no scale check, and this PR does not touch it. cast.rs:441 sends (Boolean, Decimal128(p, s)) to cast_boolean_to_decimal, which does 10_i128.pow(scale as u32) at boolean.rs:37. That is the identical pattern you are guarding everywhere else.

Does col("b").cast(DecimalType(10, -1)) still panic with attempt to multiply with overflow? If so it needs the same case d: DecimalType if d.scale < 0 guard, plus a test alongside the integer ones.

Consolidating the negative-scale check

After this PR there are five places testing d.scale < 0 with four different message strings: the two new constants here, negativeScaleDecimalToStringReason, and CometCeil/CometFloor/CometRound each carrying their own "Decimal type has negative scale".

Would it be worth pulling out a single shared predicate, something like CometCast.isNegativeScaleDecimal(dt) with one reason string? The Boolean gap above is exactly the kind of thing that is easy to miss when the check is scattered, and this family of guards is likely to keep growing.

The native underflow stays armed

The attempt to subtract with overflow in #5013 comes from the match guard at native/core/src/execution/planner.rs:918:

max(s1, s2) as u8 + max(p1 - s1 as u8, p2 - s2 as u8) >= DECIMAL128_MAX_PRECISION

With s1 = -1, s1 as u8 is 255 and p1 - 255 underflows the u8. In release builds overflow-checks = false means it wraps to p1 + 1 and we quietly take a different branch instead of panicking.

Your Scala guards mean we should not reach this anymore, but could we also do that arithmetic in i16? It is a small change and it turns a landmine into an ordinary error for any future path that reaches native with a negative scale. If you would rather keep this PR tight, could you file a follow-up issue and link it here?

Test coverage is only DecimalType(10, -1)

CometCastSuite.scala:797 and CometExpressionSuite.scala:1054 both use DecimalType(10, -1), but #5013 also reproduced with DecimalType(20, -5), and the scale magnitude selects a different native branch. create_binary_expr_with_options routes wide operands to WideDecimalBinaryExpr rather than arrow's BinaryExpr, and decimal_div picks its BigInt path over the i128 one.

Could you parameterise over both DecimalType(10, -1) and DecimalType(20, -5)? It matters most for the v * v pin in "safe ops on negative-scale decimal run natively", since right now that only proves the narrow arrow path is safe.

Documentation

Two gaps worth closing.

The generated cast matrix only samples createDecimalType(10, 2) from CometCast.supportedTypes, so none of the new Unsupported pairs appear in the table and users have no way to learn about them. _category_template/cast.md:153 already has a hand-maintained "Decimal with Negative Scale to String" section, so a companion section covering integer ↔ Decimal(neg) and Decimal(neg) → Timestamp would fit naturally there.

On the arithmetic side, Add, Subtract, Divide, IntegralDivide and Remainder are all in QueryPlanSerde.mathExpressions:111, which means they get a math.md entry built from getUnsupportedReasons(). Could the five serdes override that to return negScaleDecimalArithmeticReason? Otherwise the compat page shows them as fully supported with no caveat.

[comphead]

should we do early return here?

[comphead]

do we really need changes in this file?

[comphead]

@andygrove @0lai0 WDYT instead of listing ops if we scan an entire input plan and fallback if there is negative decimals?
Something like

object NegativeScaleDecimalFinder {

    /** True if `dt` is (or nests) a `DecimalType` with negative scale. */
    def hasNegativeScaleDecimal(dt: DataType): Boolean = dt match {
      case d: DecimalType         => d.scale < 0
      case ArrayType(elem, _)     => hasNegativeScaleDecimal(elem)
      case MapType(k, v, _)       => hasNegativeScaleDecimal(k) || hasNegativeScaleDecimal(v)
      case s: StructType          => s.fields.exists(f => hasNegativeScaleDecimal(f.dataType))
      case _                      => false
    }

    /** True if any expression in the plan (or its subqueries) produces a negative-scale decimal. */
    def planHasNegativeScaleDecimal(plan: QueryPlan[_]): Boolean = {
      def exprHas(e: Expression): Boolean =
        hasNegativeScaleDecimal(e.dataType) || e.children.exists(exprHas)

      plan.exists { node =>
        node.output.exists(a => hasNegativeScaleDecimal(a.dataType)) ||
        node.expressions.exists(exprHas) ||
        node.subqueries.exists(planHasNegativeScaleDecimal)
      }
    }
  
    /** Returns the offending nodes for easier debugging. */
    def findNegativeScaleDecimalNodes(plan: QueryPlan[_]): Seq[QueryPlan[_]] = {
      plan.collect {
        case node if node.output.exists(a => hasNegativeScaleDecimal(a.dataType)) ||
                     node.expressions.exists(e =>
                       hasNegativeScaleDecimal(e.dataType) ||
                       e.find(x => hasNegativeScaleDecimal(x.dataType)).isDefined) =>
          node
      }
    }
  }

[0lai0]

Thanks @comphead. Good point on nested types, my guards match case d: DecimalType if d.scale < 0, so they don't match ArrayType(DecimalType(10, -1)) or a struct field with negative scale.

On plan-level scanning or per-expression guards, I don't have a strong view either way.
@andygrove, do you have a preference?