[SPARK-60137][SQL] Allow lossless widening cast in ExprUtils.canEvaluateUnconditionally

[wangyum]

What changes were proposed in this pull request?

ExprUtils.canEvaluateUnconditionally (introduced in SPARK-58066) determines whether an expression is safe to evaluate on rows where the original execution plan would not have evaluated it (e.g. hoisting a streamed-side join conjunct in outer/anti joins, or relocating expressions out of short-circuited positions).

Previously, canEvaluateUnconditionallyInternal conservatively excluded all Cast expressions. This pull request adds support for safe, lossless widening numeric casts by checking:

case Cast(child, toType, _, _) =>
  child.dataType.isInstanceOf[NumericType] && toType.isInstanceOf[NumericType] &&
    Cast.canUpCast(child.dataType, toType) && canEvaluateUnconditionallyInternal(child)

Where both types are numeric and Cast.canUpCast guarantees that the cast involves no truncation, no precision loss, and no possible runtime failure (even in ANSI mode). Non-numeric upcasts (such as LongType <-> TimestampType or Any -> StringType) remain conservatively excluded to avoid issues related to timezone handling, collation, or overflow in date/time math.

Why are the changes needed?

Lossless numeric widening casts (e.g. Byte -> Short -> Integer -> Long -> Float -> Double and wider DecimalType) are total functions on their inputs and cannot fail at runtime.

Conservatively rejecting all Cast expressions prevents optimizing queries where predicates or join conditions contain safe widening casts (for example, t1.id.cast(LongType) < 100L in StreamedSideJoinCondition or range constraint propagation). Restricting to numeric upcasts expands the set of safe expressions that can be hoisted and relocated without introducing any risk of runtime errors.

Does this PR introduce any user-facing change?

No. This is an internal optimizer and expression utility improvement.

How was this patch tested?

• Added unit test cases in ExprUtilsSuite:
  • Positive cases: Verified that lossless widening numeric casts (IntegerType -> LongType) can be evaluated unconditionally.
  • Negative cases: Verified that narrowing casts (LongType -> IntegerType), parsing casts (StringType -> IntegerType), and non-numeric casts (LongType -> StringType) remain excluded.
• Added integration test cases in SplitStreamedSideJoinConditionSuite:
  • Unit test verifying StreamedSideJoinCondition.split hoists conditions with lossless widening cast into streamedOnly.
  • End-to-end query test with whole-stage codegen on and off for left outer join with widening cast in streamed-only condition.
• Passed:
  • build/sbt "catalyst/testOnly *ExprUtilsSuite"
  • build/sbt "sql/testOnly *SplitStreamedSideJoinConditionSuite"
  • build/sbt "catalyst/scalastyle" and build/sbt "sql/scalastyle"

Was this patch authored or co-authored using generative AI tooling?

Yes, Generated-by: github-copilot

[HyukjinKwon]

Review summary

The approach is sound. Admitting numeric up-casts is safe for almost every pair Cast.canUpCast accepts: integral/float widening, integral -> wider decimal, decimal -> wider decimal, and decimal -> int/short/byte at precisions that fit. The exception is decimal -> BIGINT. DecimalType.isTighterThan lets DECIMAL(19,0) through even though 19-digit values overflow a long, so once the conjunct is hoisted under ANSI the cast can fail on streamed rows the original plan never evaluated. Keeping decimal sources limited to decimal targets in the new arm, plus a negative unit case, closes it.

On the tests: the new end-to-end join case loses its cast to UnwrapCastInBinaryComparison before planning, so it never exercises a hoisted Cast, and nothing pins the new arm's child recursion. Separately, the doc bullet's "lossless" wording overstates what canUpCast guarantees for Int/Long -> Float/Double; those casts are still safe here because they never throw.

Cross-cutting observations

• Both issues come from the PR's premise, stated in the description and the new doc bullet, that Cast.canUpCast guarantees 'no truncation, no precision loss, and no possible runtime failure (even in ANSI mode)'. canUpCast admits Int/Long -> Float/Double, which lose precision, and Decimal(19,0) -> Long, which can overflow. The whitelist should state and enforce the property it needs (non-throwing) instead of inheriting that claim, and the PR description should be corrected to match.
  Separation: The two counterexamples break different properties and need different, self-contained repairs (an admission exclusion versus a wording change). The shared premise does not create shared ownership or require the repairs to coordinate.

Findings

4 total: 0 P0, 0 P1, 1 P2, 3 P3.

Non-blocking (P2)

• Cast.canUpCast admits DECIMAL(19,0) -> BIGINT, which can throw after hoisting — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:269 — see inline.

Nit (P3)

• The end-to-end widening-cast test never reaches a hoisted Cast — sql/core/src/test/scala/org/apache/spark/sql/execution/joins/SplitStreamedSideJoinConditionSuite.scala:172 — see inline.
• The doc bullet calls the admitted casts 'lossless', but some lose precision — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:246 — see inline.
• No test covers the new Cast arm's child recursion — sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExprUtilsSuite.scala:49 — see inline.

PR description suggestions

• The description says Cast.canUpCast guarantees "no truncation, no precision loss, and no possible runtime failure (even in ANSI mode)". That does not hold for Int/Long -> Float/Double (precision loss) or DECIMAL(19,0) -> BIGINT (ANSI overflow). Please restate the property the whitelist actually relies on (non-throwing), and mention any decimal exclusion added in response.
• The testing section says the end-to-end query covers a widening cast in the streamed-only condition, but the optimizer removes that cast before planning. Please update it once the test is changed to keep a cast through optimization.

Generated by Omnigent on Databricks.

[HyukjinKwon]

Non-blocking (P2): Cast.canUpCast isn't a no-throw guarantee for decimal -> integral. UpCastRule admits (DecimalType, NumericType) via DecimalType.isTighterThan, which for an integral target only checks precision < DecimalType.forType(target).precision && scale == 0. LongDecimal is DecimalType(20, 0), so DECIMAL(19,0) -> BIGINT is admitted, but DECIMAL(19,0) can hold 9999999999999999999, which is larger than Long.MaxValue. Under ANSI (the default), both the interpreted path (exactNumeric.toLong -> Decimal.roundToLong) and codegen (castDecimalToIntegralTypeCode) call longValueExact and throw CAST_OVERFLOW.

With spark.sql.join.splitStreamedSideJoinCondition=true, something like

SELECT * FROM s LEFT JOIN b ON s.k = b.k AND CAST(s.d AS BIGINT) > s.id  -- s.d: DECIMAL(19,0)

now gets the cast conjunct hoisted by StreamedSideJoinCondition.split. A non-literal comparand, so UnwrapCastInBinaryComparison leaves the cast alone. A streamed row with d = 9999999999999999999 and no matching b.k used to come back null-extended, and now fails the whole query. That is exactly the outer/anti-result-to-exception case this whitelist exists to prevent.

The narrowest fix is to not rely on canUpCast for decimal sources with non-decimal targets in this arm. For example, also require !child.dataType.isInstanceOf[DecimalType] || toType.isInstanceOf[DecimalType], and add a negative ExprUtilsSuite case for Cast(<DECIMAL(19,0) attr>, LongType). Changing isTighterThan itself would also affect analyzer up-cast validation, so I'd keep the fix local here.

Verification:

• Regression: A unit assertion that canEvaluateUnconditionally rejects a Cast of a DECIMAL(19,0) attribute to BIGINT. It fails against the current PR head.
• Behavior: Unit assertions that integral -> wider-decimal and decimal -> wider-decimal casts over attributes remain admitted, and that the existing Int -> Long admission still holds.

[HyukjinKwon]

Nit (P3): This conjunct doesn't survive the optimizer. UnwrapCastInBinaryComparison (operator optimization batch) matches Cast(s.a: INT, LongType) < Literal(5L): canUnwrapCast(IntegerType, LongType) falls through to Cast.canUpCast, and 5 is in INT range. simplifyNumericComparison therefore rewrites it to s.a < 5 before planning. By the time StreamedSideJoinCondition.split runs there is no Cast, so this is effectively the existing "left outer join, unique build key" case, and it passes with the ExprUtils change reverted. checkDataFrame only compares hoisting on vs. off, so it can't tell either way.

To actually cover a hoisted cast, use a pair that canUpCast admits but canUnwrapCast rejects, e.g. CAST(s.id AS DOUBLE) < 5.0D (Long -> Double is excluded by SPARK-39476), or compare against a non-literal streamed column. Then assert from the executed BroadcastHashJoinExec that split puts the Cast conjunct in the streamed-only part.

Verification:

• Regression: With the Cast admission arm reverted, the end-to-end case fails its hoisting assertion. With the arm present, it passes.
• Behavior: A left outer broadcast hash join whose streamed-only condition contains a surviving numeric up-cast returns the same rows with hoisting enabled and disabled, with whole-stage codegen on and off.

[HyukjinKwon]

Nit (P3): "Lossless" isn't accurate for everything Cast.canUpCast admits: via legalNumericPrecedence it also accepts Int -> Float and Long -> Float/Double. For example, 16777217 -> 1.6777216E7, or 9007199254740993L -> 9.007199254740992E15. UnwrapCastInBinaryComparison.canUnwrapCast explicitly excludes those pairs for that reason (SPARK-39476). They never throw, so the whitelist is still safe for them. What it actually relies on is that the cast is total and non-throwing, so wording like "non-throwing widening numeric up-casts (Cast.canUpCast)" would describe the contract accurately and avoid someone later treating these casts as injective.

Verification:

• Inspection: The whitelist doc's description of admitted casts matches the implemented admission rule and makes no losslessness claim that Int/Long -> Float/Double casts would violate.

[HyukjinKwon]

Nit (P3): None of these negative cases depends on the canEvaluateUnconditionallyInternal(child) recursion in the new Cast arm. All three are rejected by the type and canUpCast checks first, and the existing "whitelisted predicate over a non-whitelisted child" case uses IsNotNull, not Cast. If the recursion were dropped, everything here would still pass while Cast(Add(i, Literal(1)), LongType) (with Add able to overflow under ANSI) became hoistable. A one-line negative assertion like assert(!ExprUtils.canEvaluateUnconditionally(Cast(Add(i, Literal(1)), LongType))) would pin it.

Verification:

• Regression: With the child-recursion conjunct removed from the Cast arm, the new assertion fails. With it present, the assertion passes.