wangyum commented on code in PR #39908:
URL: https://github.com/apache/spark/pull/39908#discussion_r1106601971
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -202,8 +202,42 @@ object EliminateOuterJoin extends Rule[LogicalPlan] with
PredicateHelper {
})
}
+ /**
+ * Transform LeftOuter Join to Anti Join if:
+ * 1. Only select from the left side
+ * 2. Exists isNull filter on the right side
+ */
+ object LeftOuterJoinsWithNullFilter extends PredicateHelper {
+ def unapply(plan: LogicalPlan): Option[LogicalPlan] = plan match {
+ case Project(projectList, f @ Filter(cond, j @ Join(left, right,
LeftOuter, _, _)))
+ if projectList.forall(canEvaluate(_, left)) =>
+ val filterConditions = splitConjunctivePredicates(cond)
+ val isNullConstraints = f.constraints.collect {
+ case IsNull(e: Attribute) if right.outputSet.contains(e) => e.exprId
+ }
+ val isNotNullConstraints = right.constraints.collect {
+ case IsNotNull(e: Attribute) if right.outputSet.contains(e) =>
e.exprId
+ }
+ if (isNullConstraints.intersect(isNotNullConstraints).nonEmpty) {
+ val newJoin = j.copy(joinType = LeftAnti)
+ val leftConds =
filterConditions.filter(_.references.intersect(right.outputSet).isEmpty)
+ if (leftConds.isEmpty) {
+ Some(Project(projectList, newJoin))
+ } else {
+ Some(Project(projectList, Filter(buildBalancedPredicate(leftConds,
And), newJoin)))
+ }
+ } else {
+ None
+ }
+
+ case _ => None
+ }
+ }
+
def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(OUTER_JOIN), ruleId) {
+ case LeftOuterJoinsWithNullFilter(newProject) =>
+ newProject
Review Comment:
Makes it more clear?
```scala
case p @ Project(projectList, Filter(cond, j @ Join(left, right,
LeftOuter, _, _)))
if projectList.forall(canEvaluate(_, left)) =>
val notNullAttrs =
AttributeSet(right.constraints.collect { case IsNotNull(a:
Attribute) => a })
if (notNullAttrs.nonEmpty) {
val (isNullConditions, rest) =
splitConjunctivePredicates(cond).partition {
case IsNull(a: Attribute) => notNullAttrs.contains(a)
case _ => false
}
if (isNullConditions.nonEmpty) {
val newJoin = j.copy(joinType = LeftAnti)
p.copy(child = rest.reduceLeftOption(And).map(Filter(_,
newJoin)).getOrElse(newJoin))
} else {
p
}
} else {
p
}
```
--
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]
