review

Author: dilipbiswal

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/subquery.scala

@@ -267,13 +267,6 @@ object ScalarSubquery {
       case _ => false
     }.isDefined
   }
-
-  def hasScalarSubquery(e: Expression): Boolean = {
-    e.find {
-      case _: ScalarSubquery => true
-      case _ => false
-    }.isDefined
-  }
 }
 
 /**

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

@@ -130,24 +130,16 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] with PredicateHelper {
         val newGrandChildren = union.children.map { Join(_, rightOp, joinType, joinCond, hint) }
         union.withNewChildren(newGrandChildren)
       } else {
-        val pushDown = splitConjunctivePredicates(joinCond.get)
-
-        if (pushDown.nonEmpty) {
-          val pushDownCond = pushDown.reduceLeft(And)
-          val output = union.output
-          val newGrandChildren = union.children.map { grandchild =>
-            val newCond = pushDownCond transform {
-              case e if output.exists(_.semanticEquals(e)) =>
-                grandchild.output(output.indexWhere(_.semanticEquals(e)))
-            }
-            assert(newCond.references.subsetOf(grandchild.outputSet ++ rightOp.outputSet))
-            Join(grandchild, rightOp, joinType, Option(newCond), hint)
+        val output = union.output
+        val newGrandChildren = union.children.map { grandchild =>
+          val newCond = joinCond.get transform {
+            case e if output.exists(_.semanticEquals(e)) =>
+              grandchild.output(output.indexWhere(_.semanticEquals(e)))
           }
-          union.withNewChildren(newGrandChildren)
-        } else {
-          // Nothing to push down
-          join
+          assert(newCond.references.subsetOf(grandchild.outputSet ++ rightOp.outputSet))
+          Join(grandchild, rightOp, joinType, Option(newCond), hint)
         }
+        union.withNewChildren(newGrandChildren)
       }
 
     // LeftSemi/LeftAnti over UnaryNode
@@ -182,22 +174,22 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] with PredicateHelper {
   private def pushDownJoin(
       join: Join,
       grandchild: LogicalPlan)(insertFilter: Expression => LogicalPlan): LogicalPlan = {
-    val (pushDown, stayUp) = if (join.condition.isDefined) {
-      splitConjunctivePredicates(join.condition.get)
-        .partition {_.references.subsetOf(grandchild.outputSet ++ join.right.outputSet)}
+    if (join.condition.isEmpty) {
+      insertFilter(null)
     } else {
-      (Nil, Nil)
-    }
+      val (pushDown, stayUp) = splitConjunctivePredicates(join.condition.get)
+        .partition {_.references.subsetOf(grandchild.outputSet ++ join.right.outputSet)}
 
-    if (pushDown.nonEmpty) {
-      val newChild = insertFilter(pushDown.reduceLeft(And))
-      if (stayUp.nonEmpty) {
-        Filter(stayUp.reduceLeft(And), newChild)
+      if (pushDown.nonEmpty) {
+        val newChild = insertFilter(pushDown.reduceLeft(And))
+        if (stayUp.nonEmpty) {
+          Filter(stayUp.reduceLeft(And), newChild)
+        } else {
+          newChild
+        }
       } else {
-        newChild
+        join
       }
-    } else {
-      join
     }
   }
 }

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

@@ -257,4 +257,19 @@ class LeftSemiPushdownSuite extends PlanTest {
       .analyze
     comparePlans(optimized, correctAnswer)
   }
+
+  test("Unary: LeftSemiAnti join pushdown - empty join condition") {
+    val originalQuery = testRelation
+      .select(star())
+      .repartition(1)
+      .join(testRelation1, joinType = LeftSemi, condition = None)
+
+    val optimized = Optimize.execute(originalQuery.analyze)
+    val correctAnswer = testRelation
+      .join(testRelation1, joinType = LeftSemi, condition = None)
+      .select('a, 'b, 'c)
+      .repartition(1)
+      .analyze
+    comparePlans(optimized, correctAnswer)
+  }
 }