dongjoon-hyun commented on code in PR #60117:
URL: https://github.com/apache/spark/pull/60117#discussion_r4483226938
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -3088,12 +3088,23 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
/**
* Whether `e` is expensive. Version 0 of the regex pattern cost estimate
treats every LIKE and
- * RLIKE as expensive. It is checked here rather than in
`StringRegexExpression.expensive`,
- * because `hasExpensiveChild` caches the answer of a parent expression.
+ * RLIKE as expensive, and with `avoidDoubleFilterEvalForPythonUDF` off a
Python UDF is only as
+ * expensive as its arguments. Both are checked here rather than in
`expensive`, because
+ * `hasExpensiveChild` caches the answer of a parent expression.
*/
- private def isExpensive(e: Expression): Boolean = e.expensive ||
- (conf.getConf(SQLConf.REGEX_PATTERN_COST_ESTIMATE_VERSION) == 0 &&
- e.exists(_.isInstanceOf[StringRegexExpression]))
+ private def isExpensive(e: Expression): Boolean = {
+ val classified = if
(conf.getConf(SQLConf.AVOID_DOUBLE_FILTER_EVAL_FOR_PYTHON_UDF)) {
+ e
+ } else {
+ // The stand-in is not foldable either, so a LIKE/RLIKE pattern from a
UDF stays expensive.
+ e.transformUp {
Review Comment:
Thanks. Applied in 90b6e9b459d. `isExpensive` now makes the copy only when
`e.expensive` is true.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -93,6 +93,14 @@ class OffsetSeqLogSuite extends SharedSparkSession {
assert(sessionConf.getConfString(key.key) === "true")
}
+ test("SPARK-60905: Python UDFs are expensive only for checkpoints that
record it") {
+ val entry = SQLConf.AVOID_DOUBLE_FILTER_EVAL_FOR_PYTHON_UDF
+ val (_, offsetSeq) = readFromResource("offset-log-version-2.1.0")
Review Comment:
Thanks. In 90b6e9b459d, the test now reads the offset log of
`checkpoint-version-4.2.0-deduplication-metadata-boundary`, where
`avoidDoubleFilterEval` keeps `true`. If the missing-key default depended on
the checkpoint version, like the one for `avoidDoubleFilterEval`, this
checkpoint would read `true` and the test would fail.
I did not add a restart test. The path from the offset log to the optimizer
is the shared `setSessionConf` code, which this PR does not change, and such a
test would need Python. The restarts from real Spark 4.3.0 checkpoints are in
the PR description.
--
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]
