HyukjinKwon commented on code in PR #59344:
URL: https://github.com/apache/spark/pull/59344#discussion_r4267146118
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/SplitStreamedSideJoinConditionSuite.scala:
##########
@@ -151,6 +163,15 @@ class SplitStreamedSideJoinConditionSuite extends
QueryTest with SharedSparkSess
|ON s.id = b.bid AND s.a < 5
""".stripMargin))
+ testWithCodegenOnAndOff(
+ "left outer join with lossless widening cast in streamed-only condition")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, s.pad, b.bid
+ |FROM (SELECT id, CAST(id AS int) AS a, pad FROM streamed_t WHERE pad %
2 = 0) s
+ |LEFT OUTER JOIN (SELECT DISTINCT bid FROM build_t) b
+ |ON s.id = b.bid AND CAST(s.a AS bigint) < 5L
Review Comment:
**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.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:
##########
@@ -262,6 +264,9 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
private def canEvaluateUnconditionallyInternal(e: Expression): Boolean = e
match {
case _: AttributeReference | _: Literal => true
+ case Cast(child, toType, _, _) =>
+ child.dataType.isInstanceOf[NumericType] &&
toType.isInstanceOf[NumericType] &&
+ Cast.canUpCast(child.dataType, toType) &&
canEvaluateUnconditionallyInternal(child)
Review Comment:
**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
```sql
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.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:
##########
@@ -243,14 +243,16 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
*
* Only a whitelist of total, deterministic expressions qualifies:
* - leaves: attribute references and literals;
+ * - lossless widening numeric casts: Cast where both types are numeric
and Cast.canUpCast
Review Comment:
**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.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExprUtilsSuite.scala:
##########
@@ -44,6 +46,10 @@ class ExprUtilsSuite extends SparkFunSuite {
assert(!arithmetic.throwable)
assert(!ExprUtils.canEvaluateUnconditionally(arithmetic))
assert(!ExprUtils.canEvaluateUnconditionally(EqualTo(Remainder(a,
Literal(3L)), Literal(0L))))
+ // Narrowing cast (Long -> Int), parsing cast (String -> Int), and
non-numeric cast.
Review Comment:
**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.
--
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]
