sunchao commented on code in PR #6035:
URL: https://github.com/apache/datafusion-comet/pull/6035#discussion_r4321302685
##########
spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala:
##########
@@ -65,6 +66,70 @@ class CometNativeCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
import testImplicits._
+ 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)))
+ }
+ // ConstantFolding must leave the throwing literal cast inside the
conditional.
+ val badCasts = sql(masked).queryExecution.optimizedPlan
+ .flatMap(_.expressions)
+ .flatMap(_.collect {
+ case c: Cast
+ if c.child.isInstanceOf[Literal] &&
+
Option(c.child.asInstanceOf[Literal].value).exists(_.toString == "bad") =>
+ c
+ })
+ assert(badCasts.nonEmpty)
+ val (_, plan) =
+ checkSparkAnswerAndFallbackReason(masked,
CometCast.literalCastConditionalEvalReason)
+ assert(collect(plan) { case scan: CometNativeScanExec => scan
}.nonEmpty)
+ checkSparkError(
+ sql(s"SELECT $cast FROM cast_branch_rows"),
+ "CAST_INVALID_INPUT",
+ checkNative = false)
+ }
+ }
+ }
+ }
+
+ test("successful literal casts and non-ANSI invalid casts retain native
projection") {
+ withParquetTable(Seq(Tuple1(0), Tuple1(1)), "cast_branch_controls") {
+ for (ansi <- Seq("true", "false")) {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi) {
+ val query = "SELECT CAST(IF(_1 = 1, '1', CAST(_1 AS STRING)) AS INT)
" +
+ "FROM cast_branch_controls"
+ val (_, plan) = checkSparkAnswerAndOperator(sql(query))
+ assert(collect(plan) { case project: CometProjectExec => project
}.nonEmpty)
+ assert(
+ !plan.exists(_.getTagValue(CometExplainInfo.FALLBACK_REASONS)
+ .exists(_.contains(CometCast.literalCastConditionalEvalReason))))
Review Comment:
Removed the fallback-reason checks and redundant projection checks from
these controls. They now use `checkSparkAnswerAndOperator` for Spark parity and
native execution. The revised PR also removes the whole-projection fallback, so
there is no new literal-cast fallback reason to assert against. Fixed in the
current head, `1dc896732`.
##########
spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala:
##########
@@ -65,6 +66,70 @@ class CometNativeCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
import testImplicits._
+ 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)))
+ }
+ // ConstantFolding must leave the throwing literal cast inside the
conditional.
+ val badCasts = sql(masked).queryExecution.optimizedPlan
+ .flatMap(_.expressions)
+ .flatMap(_.collect {
+ case c: Cast
+ if c.child.isInstanceOf[Literal] &&
+
Option(c.child.asInstanceOf[Literal].value).exists(_.toString == "bad") =>
+ c
+ })
+ assert(badCasts.nonEmpty)
+ val (_, plan) =
+ checkSparkAnswerAndFallbackReason(masked,
CometCast.literalCastConditionalEvalReason)
+ assert(collect(plan) { case scan: CometNativeScanExec => scan
}.nonEmpty)
+ checkSparkError(
+ sql(s"SELECT $cast FROM cast_branch_rows"),
+ "CAST_INVALID_INPUT",
+ checkNative = false)
+ }
+ }
+ }
+ }
+
+ test("successful literal casts and non-ANSI invalid casts retain native
projection") {
Review Comment:
Added direct serialization assertions in `successful and non-ANSI literal
casts fold and retain Comet projection`: `QueryPlanSerde.exprToProto` must
produce literal `1` for the successful cast in both ANSI modes, and literal
`null` for the invalid non-ANSI cast. Calling the serializer directly prevents
Spark's ConstantFolding from hiding Comet's folding behavior. The Parquet
queries still check Spark parity and native execution. These tests passed on
Spark 4.0 at `1dc896732`.
##########
spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala:
##########
@@ -452,11 +452,14 @@ abstract class CometTestBase
}
}
- /** Checks native execution and Spark exception type, error class and
SQLSTATE parity. */
+ /** Checks Spark exception parity and, by default, native execution. */
Review Comment:
The revised approach keeps the projection native, so the `checkNative`
parameter is no longer needed and I removed it. `CometTestBase.scala` now
matches main again. The new demanded-error cases use the original strict
`checkSparkError` helper, including its native-plan and exception/cause-chain
checks. This supersedes my earlier reply about opting out of the native
assertion.
--
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]
