cloud-fan commented on code in PR #56535:
URL: https://github.com/apache/spark/pull/56535#discussion_r3440761168
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -210,12 +215,28 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
* InMemoryRelation is intentionally excluded because cache() and persist()
are lazy: its
* presence does not guarantee the cached data has been materialized, and
missing or evicted
* blocks may require evaluating the upstream computation again.
+ *
+ * The supported operators are intentionally narrow. DPP is optional, and
logical-plan
+ * determinism does not cover user functions stored outside Catalyst
expressions.
*/
- private def hasSelectivePredicateOrLocalOrCheckpointedInput(plan:
LogicalPlan): Boolean = {
- plan.exists {
- case f: Filter => isLikelySelective(f.condition)
+ private def isRepeatableMaterializedPlan(plan: LogicalPlan): Boolean = {
+ def isRepeatableExpression(expression: Expression): Boolean = {
+ expression.deterministic && !SubqueryExpression.hasSubquery(expression)
&&
+ !expression.exists {
+ case _: NonSQLExpression | _: UserDefinedExpression | _:
UserDefinedGenerator => true
+ case _ => false
+ }
+ }
+
+ plan match {
case _: LocalRelation => true
case r: LogicalRDD => r.isCheckpointedInput
+ case Project(projectList, child) if
projectList.forall(isRepeatableExpression) =>
+ isRepeatableMaterializedPlan(child)
+ case Filter(condition, child) if isRepeatableExpression(condition) =>
+ isRepeatableMaterializedPlan(child)
+ case u: Union => u.children.forall(isRepeatableMaterializedPlan)
+ case SubqueryAlias(_, child) => isRepeatableMaterializedPlan(child)
Review Comment:
This `SubqueryAlias` case looks unreachable in the optimizer:
`EliminateSubqueryAliases` runs in the catalyst `FinishAnalysis` batch (part of
`super.defaultBatches`), which completes before the `PartitionPruning` batch in
`SparkOptimizer`, so no `SubqueryAlias` node survives to reach this rule.
Non-blocking, and a question rather than a request: is it intended as
defensive coding (in case the batch ordering changes), or can it be dropped to
keep the allowlist as tight as the Scaladoc advertises? If kept, a one-word
comment noting it's defensive would help future readers.
--
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]
