sunchao commented on code in PR #6035:
URL: https://github.com/apache/datafusion-comet/pull/6035#discussion_r4178682477
##########
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:
Added a shared explanatory reason and the ConstantFolding/FAILED_TO_EVALUATE
comment. I kept full projection fallback deliberately: batch codegen dispatch
can evaluate the bad cast on a row Spark never reaches after LIMIT. The
regression covers both dispatcher settings and passes; successful/non-ANSI
controls stay native.
##########
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:
Moved the end-to-end regression into CometNativeCastSuite; both focused cast
tests pass on Spark 4.1.
##########
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 regression now asserts the optimized plan still contains the cast of the
invalid literal before executing the query, pinning the ConstantFolding premise.
##########
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:
Switched to checkSparkError. Its new optional checkNative argument defaults
to true; this intentional projection-fallback case disables only that plan
assertion and retains the shared exception, error-class, SQLSTATE and
cause-chain checks.
##########
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:
Added Parquet-backed successful literal-cast controls in ANSI and non-ANSI
modes plus an invalid non-ANSI control. All assert native CometProject
execution and absence of the fallback reason; the focused tests pass.
--
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]
