sunchao commented on code in PR #5533:
URL: https://github.com/apache/datafusion-comet/pull/5533#discussion_r4041852152
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -714,6 +722,132 @@ case class CometExecRule(session: SparkSession)
}
}
+ /** Keep operators that can skip malformed unbase64 inputs in Spark's row
pipeline. */
+ private def preserveUnbase64EvaluationMasks(plan: SparkPlan): SparkPlan = {
+ val limitReason = "unbase64 requires Spark evaluation below LIMIT"
+ val joinReason = "unbase64 requires Spark evaluation in first-match join
conditions"
+
+ def containsUnbase64(expr: Expression): Boolean =
+ expr.exists(_.isInstanceOf[UnBase64])
Review Comment:
Addressed in `7df125174`.
The trigger now comes from the serde policy, `RequiresSparkEvaluationMask`,
and only `UnBase64` is enrolled, including subclasses inside dispatched
expression trees. ANSI Add/Cast returning `None` records the current enrollment
scope; I added an explicit link beside those assertions to the broader audit in
#6006.
Filed #6006 for the rest of the throwing-expression family, per-row AND/CASE
masks, and aggregate FILTER. It calls for skipped-invalid and consumed-invalid
controls plus execution-path assertions, rather than assuming every conditional
shape currently diverges. The compatibility guide now records this scope as
well.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -714,6 +722,132 @@ case class CometExecRule(session: SparkSession)
}
}
+ /** Keep operators that can skip malformed unbase64 inputs in Spark's row
pipeline. */
+ private def preserveUnbase64EvaluationMasks(plan: SparkPlan): SparkPlan = {
Review Comment:
Addressed in `7df125174`.
Added `spark.comet.exec.preserveEvaluationMasks.enabled` (default `true`).
Setting it to `false` disables this operator-level fallback, including the
conservative AQE aggregate-buffer protection, for users who know their Base64
input is valid. Existing expression support/dispatch restrictions still apply.
The config, expression docs, audit, and compatibility guide explain that opting
out can raise errors on malformed rows Spark would skip.
The new regression checks native decoding under LIMIT, JVM dispatch for
compound inputs, native semi/anti joins, and native aggregate planning under
AQE with the gate disabled; enabled controls retain the fallback.
--
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]
