fix

Author: wangyum

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -23,7 +23,6 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
@@ -1458,8 +1457,8 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
     val (pushDownCandidates, nonDeterministic) = condition.partition(_.deterministic)
     val (leftEvaluateCondition, rest) =
       pushDownCandidates.partition(_.references.subsetOf(left.outputSet))
-    val rightEvaluateCondition = pushDownCandidates.filter(_.references.subsetOf(right.outputSet))
-    val commonCondition = rest.filterNot(_.references.subsetOf(right.outputSet))
+    val (rightEvaluateCondition, commonCondition) =
+        rest.partition(expr => expr.references.subsetOf(right.outputSet))
 
     (leftEvaluateCondition, rightEvaluateCondition, commonCondition ++ nonDeterministic)
   }
@@ -1469,12 +1468,6 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
     case _ => false
   }
 
-  private def pushDownJoinConditions(conditions: Seq[Expression], plan: LogicalPlan) = {
-    conditions
-      .filterNot(_.semanticEquals(TrueLiteral)) // Push down true condition is useless.
-      .reduceLeftOption(And).map(Filter(_, plan)).getOrElse(plan)
-  }
-
   def apply(plan: LogicalPlan): LogicalPlan = plan transform applyLocally
 
   val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
@@ -1533,22 +1526,26 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
       joinType match {
         case _: InnerLike | LeftSemi =>
           // push down the single side only join filter for both sides sub queries
-          val newLeft = pushDownJoinConditions(leftJoinConditions, left)
-          val newRight = pushDownJoinConditions(rightJoinConditions, right)
+          val newLeft = leftJoinConditions.
+            reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
+          val newRight = rightJoinConditions.
+            reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
           val newJoinCond = commonJoinCondition.reduceLeftOption(And)
 
           Join(newLeft, newRight, joinType, newJoinCond, hint)
         case RightOuter =>
           // push down the left side only join filter for left side sub query
-          val newLeft = pushDownJoinConditions(leftJoinConditions, left)
+          val newLeft = leftJoinConditions.
+            reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
           val newRight = right
           val newJoinCond = (rightJoinConditions ++ commonJoinCondition).reduceLeftOption(And)
 
           Join(newLeft, newRight, RightOuter, newJoinCond, hint)
         case LeftOuter | LeftAnti | ExistenceJoin(_) =>
           // push down the right side only join filter for right sub query
           val newLeft = left
-          val newRight = pushDownJoinConditions(rightJoinConditions, right)
+          val newRight = rightJoinConditions.
+            reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
           val newJoinCond = (leftJoinConditions ++ commonJoinCondition).reduceLeftOption(And)
 
           Join(newLeft, newRight, joinType, newJoinCond, hint)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala

@@ -1384,25 +1384,4 @@ class FilterPushdownSuite extends PlanTest {
       condition = Some("x.a".attr === "z.a".attr)).analyze
     comparePlans(optimized, correctAnswer)
   }
-
-  test("SPARK-28220: Push down true join condition for inner join") {
-    val x = testRelation.subquery('x)
-    val y = testRelation.subquery('y)
-    val originalQuery = x.join(y, condition = Some(true))
-
-    val optimized = Optimize.execute(originalQuery.analyze)
-    val correctAnswer = x.join(y, condition = None).analyze
-    comparePlans(optimized, correctAnswer)
-  }
-
-  test("SPARK-28220: Should not push down true join condition for left/right join") {
-    Seq(LeftOuter, RightOuter).foreach { joinType =>
-      val x = testRelation.subquery('x)
-      val y = testRelation.subquery('y)
-      val originalQuery = x.join(y, joinType = joinType, condition = Some(true))
-
-      val optimized = Optimize.execute(originalQuery.analyze)
-      comparePlans(optimized, originalQuery.analyze)
-    }
-  }
 }