Revert optimization for Exists subquery without correlated references.

Author: viirya

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

@@ -65,8 +65,7 @@ abstract class Optimizer(sessionCatalog: SessionCatalog, conf: SQLConf)
     Batch("Pullup Correlated Expressions", Once,
       PullupCorrelatedPredicates) ::
     Batch("Subquery", Once,
-      OptimizeSubqueries,
-      RewriteEmptyExists) ::
+      OptimizeSubqueries) ::
     Batch("Replace Operators", fixedPoint,
       ReplaceIntersectWithSemiJoin,
       ReplaceExceptWithAntiJoin,

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

@@ -498,32 +498,3 @@ object RewriteCorrelatedScalarSubquery extends Rule[LogicalPlan] {
       }
   }
 }
-
-/**
- * This rule rewrites a EXISTS predicate sub-queries into an Aggregate with count.
- * So it doesn't be converted to a JOIN later.
- */
-object RewriteEmptyExists extends Rule[LogicalPlan] with PredicateHelper {
-  private def containsAgg(plan: LogicalPlan): Boolean = {
-    plan.collect {
-      case a: Aggregate => a
-    }.nonEmpty
-  }
-
-  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
-    case Filter(condition, child) =>
-      val (withSubquery, withoutSubquery) =
-        splitConjunctivePredicates(condition).partition(SubqueryExpression.hasInOrExistsSubquery)
-      val newWithSubquery = withSubquery.map(_.transform {
-        case e @ Exists(sub, conditions, exprId) if conditions.isEmpty && !containsAgg(sub) =>
-          val countExpr = Alias(Count(Literal(1)).toAggregateExpression(), "count")()
-          val expr = Alias(GreaterThan(countExpr.toAttribute, Literal(0)), e.toString)()
-          ScalarSubquery(
-            Project(Seq(expr),
-              Aggregate(Nil, Seq(countExpr), LocalLimit(Literal(1), sub))),
-            children = Seq.empty,
-            exprId = exprId)
-      })
-      Filter((newWithSubquery ++ withoutSubquery).reduce(And), child)
-  }
-}

sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala

@@ -17,9 +17,6 @@
 
 package org.apache.spark.sql
 
-import org.apache.spark.sql.catalyst.expressions.{Alias, ScalarSubquery}
-import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Count}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Join}
 import org.apache.spark.sql.test.SharedSQLContext
 
 class SubquerySuite extends QueryTest with SharedSQLContext {
@@ -857,23 +854,4 @@ class SubquerySuite extends QueryTest with SharedSQLContext {
       sql("select * from l, r where l.a = r.c + 1 AND (exists (select * from r) OR l.a = r.c)"),
       Row(3, 3.0, 2, 3.0) :: Row(3, 3.0, 2, 3.0) :: Nil)
   }
-
-  test("Convert Exists without correlated references to aggregation with count") {
-    val df =
-      sql("select * from l, r where l.a = r.c + 1 AND (exists (select * from r) OR l.a = r.c)")
-    val joinPlan = df.queryExecution.optimizedPlan.asInstanceOf[Join]
-    val scalarSubquery = joinPlan.condition.get.collect {
-      case s: ScalarSubquery => s
-    }
-    assert(scalarSubquery.length == 1)
-    val aggPlan = scalarSubquery.head.plan.collect {
-      case a: Aggregate => a
-    }
-    assert(aggPlan.length == 1)
-    assert(aggPlan.head.aggregateExpressions.length == 1)
-    val countAggExpr = aggPlan.head.aggregateExpressions.collect {
-      case a @ Alias(AggregateExpression(_: Count, _, _, _), _) => a
-    }
-    assert(countAggExpr.length == 1)
-  }
 }