viirya commented on code in PR #6035:
URL: https://github.com/apache/datafusion-comet/pull/6035#discussion_r4052348832
##########
spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala:
##########
@@ -216,6 +219,41 @@ class CometScalarFunctionSuite extends CometTestBase {
assertRejectReason(withContext, "CometScalarFunction", "evalContext")
}
+ test("literal cast failure in an unvisited conditional branch does not fail
planning") {
Review Comment:
This suite is entirely `CometScalarFunction` ANSI-sensitivity unit tests —
all 15 call `.convert(...)` directly, with no SparkSession, no Parquet and no
query execution. This test is end-to-end and exercises `CometCast`, so
`CometNativeCastSuite` looks like the natural home; it is already the canonical
place for cast behavior and already imports `CometCast` constants for exactly
this kind of assertion.
##########
spark/src/main/scala/org/apache/comet/expressions/CometCast.scala:
##########
@@ -109,7 +111,15 @@ object CometCast
val cometEvalMode = evalMode(cast)
cast.child match {
case _: Literal =>
- exprToProtoInternal(Literal.create(cast.eval(), cast.dataType),
inputs, binding)
+ val value =
+ try {
+ cast.eval()
+ } catch {
+ case NonFatal(_) =>
+ withFallbackReason(cast, "Literal cast requires Spark's
conditional evaluation")
+ return None
+ }
Review Comment:
This is the block that decides the expression falls all the way back to
Spark. Because `getSupportLevel` returned `Compatible`, `exprToProtoInternal`
took the branch that calls `convert` directly, so `dispatchIfFallback` is never
reached and the whole projection leaves the Comet pipeline.
Consider moving the decision into `getSupportLevel`: try the evaluation
there, and on `NonFatal` return
`Unsupported(Some(literalCastConditionalEvalReason))`. The framework then tries
the JVM codegen dispatcher first — which can run `Cast.doGenCode` in-pipeline
and raise only if the branch is actually visited — and falls back to Spark only
if the dispatcher declines. `convert` can then keep its current unconditional
`cast.eval()`, since the failing case no longer reaches it.
Two smaller things on this same block:
**The reason string should be a shared constant.** This file already
establishes the convention, with the rationale spelled out at the top:
```scala
// Shared with CometNativeCastSuite so the asserted reason cannot drift from
production.
private[comet] val negativeScaleDecimalToStringReason: String = ...
```
Here the string is written literally in both production and the test. Since
`checkSparkAnswerAndFallbackReasons` matches with `contains`, a future edit
would not break compilation and would not necessarily fail the test either —
exactly the drift that comment guards against. On the wording: "Literal cast
requires Spark's conditional evaluation" states the remedy but not the cause,
and a user seeing this in EXPLAIN is asking why their query left Comet.
Something self-explanatory would read better, e.g. "Cast of a literal threw
during planning; Spark leaves it for conditional evaluation so it may never be
reached at runtime".
**Please add a comment explaining why the catch exists.** The reason is
genuinely non-obvious: Spark's `ConstantFolding.tryFold` catches `NonFatal` for
expressions inside a conditional branch, tags them `FAILED_TO_EVALUATE` and
leaves them unfolded, so a cast that throws can legitimately survive into the
plan without ever being evaluated at runtime. Without that context, `catch {
case NonFatal(_) => return None }` reads like swallowing an error. Worth also
noting that `FAILED_TO_EVALUATE` would be the precise signal but is
`private[sql]` and unusable from this package — otherwise someone will
reasonably wonder why it is not used.
##########
spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala:
##########
@@ -216,6 +219,41 @@ class CometScalarFunctionSuite extends CometTestBase {
assertRejectReason(withContext, "CometScalarFunction", "evalContext")
}
+ test("literal cast failure in an unvisited conditional branch does not fail
planning") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(2)
+ .selectExpr("id", "CAST(id AS STRING) AS value")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ withParquetTable(path.getCanonicalPath, "cast_branch_rows") {
+ val cast = "CAST(IF(id = 1, 'bad', value) AS INT)"
+ val masked = s"SELECT $cast AS parsed FROM cast_branch_rows LIMIT 1"
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assert(sql(masked).collect().toSeq == Seq(Row(0)))
+ }
+ val (_, plan) = checkSparkAnswerAndFallbackReason(
+ masked,
+ "Literal cast requires Spark's conditional evaluation")
+ assert(collect(plan) { case scan: CometNativeScanExec => scan
}.nonEmpty)
+ val (sparkError, cometError) =
+ checkSparkAnswerMaybeThrows(sql(s"SELECT $cast FROM
cast_branch_rows"))
+ assert(sparkError.nonEmpty && cometError.nonEmpty)
+ val errors = Seq(sparkError.get, cometError.get).map { error =>
+ causeChain(error).collect { case e: SparkThrowable => e }.last
+ }
+ assert(errors.forall(_.getErrorClass == "CAST_INVALID_INPUT"))
+ assert(errors(0).getClass == errors(1).getClass)
+ assert(errors(0).getSqlState == errors(1).getSqlState)
+ }
+ }
+ }
+ }
Review Comment:
Could you add a case asserting the fix is not over-corrected — that a
*successful* literal cast (say `CAST('1' AS INT)` in the same branch position)
still folds and stays native, with no fallback reason? #5623 paired its guard
with exactly this kind of counterpart test, and without one there is nothing
stopping the catch from being widened later.
A non-ANSI case would also be worth having, or a note explaining why ANSI
coverage is sufficient.
##########
spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala:
##########
@@ -216,6 +219,41 @@ class CometScalarFunctionSuite extends CometTestBase {
assertRejectReason(withContext, "CometScalarFunction", "evalContext")
}
+ test("literal cast failure in an unvisited conditional branch does not fail
planning") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(2)
+ .selectExpr("id", "CAST(id AS STRING) AS value")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ withParquetTable(path.getCanonicalPath, "cast_branch_rows") {
+ val cast = "CAST(IF(id = 1, 'bad', value) AS INT)"
+ val masked = s"SELECT $cast AS parsed FROM cast_branch_rows LIMIT 1"
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assert(sql(masked).collect().toSeq == Seq(Row(0)))
Review Comment:
The test's validity rests on `ConstantFolding` refusing to fold `CAST('bad'
AS INT)` because it sits in a conditional branch — but nothing here says so. If
Spark's folding behavior ever changes, this test would keep passing while
silently no longer covering the bug.
A brief comment, or an assertion that the `Cast` is still present in the
optimized plan, would pin the premise down.
##########
spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala:
##########
@@ -216,6 +219,41 @@ class CometScalarFunctionSuite extends CometTestBase {
assertRejectReason(withContext, "CometScalarFunction", "evalContext")
}
+ test("literal cast failure in an unvisited conditional branch does not fail
planning") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(2)
+ .selectExpr("id", "CAST(id AS STRING) AS value")
+ .coalesce(1)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ withParquetTable(path.getCanonicalPath, "cast_branch_rows") {
+ val cast = "CAST(IF(id = 1, 'bad', value) AS INT)"
+ val masked = s"SELECT $cast AS parsed FROM cast_branch_rows LIMIT 1"
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assert(sql(masked).collect().toSeq == Seq(Row(0)))
+ }
+ val (_, plan) = checkSparkAnswerAndFallbackReason(
+ masked,
+ "Literal cast requires Spark's conditional evaluation")
+ assert(collect(plan) { case scan: CometNativeScanExec => scan
}.nonEmpty)
+ val (sparkError, cometError) =
+ checkSparkAnswerMaybeThrows(sql(s"SELECT $cast FROM
cast_branch_rows"))
+ assert(sparkError.nonEmpty && cometError.nonEmpty)
+ val errors = Seq(sparkError.get, cometError.get).map { error =>
+ causeChain(error).collect { case e: SparkThrowable => e }.last
+ }
+ assert(errors.forall(_.getErrorClass == "CAST_INVALID_INPUT"))
+ assert(errors(0).getClass == errors(1).getClass)
+ assert(errors(0).getSqlState == errors(1).getSqlState)
Review Comment:
`CometTestBase.checkSparkError(df, errorClass)` already does all of this,
and is stricter in two ways worth keeping:
- it asserts no `CometNativeException` appears in the cause chain, so the
test cannot pass on an error that surfaced from native code with a
coincidentally matching class;
- it uses `lastOption.getOrElse(fail(...))`, whereas `.last` here throws
`UnsupportedOperationException` if the chain contains no `SparkThrowable`,
turning a meaningful assertion failure into a confusing one.
The whole block collapses to:
```scala
checkSparkError(sql(s"SELECT $cast FROM cast_branch_rows"),
"CAST_INVALID_INPUT")
```
which also reads better than the `errors(0)` / `errors(1)` indexing.
(Unrelated to this PR: `getErrorClass` is deprecated in favor of
`getCondition()`, but the shared helper still uses it, so that is better
changed there on its own.)
--
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]
