[SPARK-29800][SQL] Rewrite non-correlated EXISTS subquery use ScalaSubquery to optimize perf

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

@@ -64,9 +64,10 @@ object SubqueryExpression {
   /**
    * Returns true when an expression contains an IN or EXISTS subquery and false otherwise.
    */
-  def hasInOrExistsSubquery(e: Expression): Boolean = {
+  def hasInOrCorrelatedExistsSubquery(e: Expression): Boolean = {
     e.find {
-      case _: ListQuery | _: Exists => true
+      case _: ListQuery => true
+      case _: Exists if e.children.nonEmpty => true
       case _ => false
     }.isDefined
   }

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

@@ -96,7 +96,8 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case Filter(condition, child) =>
       val (withSubquery, withoutSubquery) =
-        splitConjunctivePredicates(condition).partition(SubqueryExpression.hasInOrExistsSubquery)
+        splitConjunctivePredicates(condition)
+          .partition(SubqueryExpression.hasInOrCorrelatedExistsSubquery)
 
       // Construct the pruned filter condition.
       val newFilter: LogicalPlan = withoutSubquery match {

sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala

@@ -23,8 +23,9 @@ import scala.collection.mutable.ArrayBuffer
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.{expressions, InternalRow}
-import org.apache.spark.sql.catalyst.expressions.{AttributeSeq, CreateNamedStruct, Expression, ExprId, InSet, ListQuery, Literal, PlanExpression}
+import org.apache.spark.sql.catalyst.expressions.{CreateNamedStruct, Expression, ExprId, InSet, ListQuery, Literal, PlanExpression}
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
+import org.apache.spark.sql.catalyst.plans.logical.Project
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{BooleanType, DataType, StructType}
@@ -171,6 +172,44 @@ case class InSubqueryExec(
   }
 }
 
+/**
+ * The physical node of non-correlated EXISTS subquery.
+ */
+case class ExistsExec(
+    plan: BaseSubqueryExec,
+    exprId: ExprId)
+  extends ExecSubqueryExpression {
+
+  @volatile private var result: Option[Boolean] = None
+
+  override def dataType: DataType = BooleanType
+  override def children: Seq[Expression] = Nil
+  override def nullable: Boolean = false
+  override def toString: String = s"EXISTS (${plan.simpleString(SQLConf.get.maxToStringFields)})"
+  override def withNewPlan(plan: BaseSubqueryExec): ExistsExec = copy(plan = plan)
+
+  override def semanticEquals(other: Expression): Boolean = other match {
+    case in: ExistsExec => plan.sameResult(in.plan)
+    case _ => false
+  }
+
+  def updateResult(): Unit = {
+    result = Some(plan.executeTake(1).length == 1)
+  }
+
+  def values(): Option[Boolean] = result
+
+  override def eval(input: InternalRow): Any = {
+    require(result.isDefined, s"$this has not finished")
+    result.get
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    require(result.isDefined, s"$this has not finished")
+    Literal.create(result.get, dataType).doGenCode(ctx, ev)
+  }
+}
+
 /**
  * Plans subqueries that are present in the given [[SparkPlan]].
  */
@@ -194,6 +233,9 @@ case class PlanSubqueries(sparkSession: SparkSession) extends Rule[SparkPlan] {
         }
         val executedPlan = new QueryExecution(sparkSession, query).executedPlan
         InSubqueryExec(expr, SubqueryExec(s"subquery#${exprId.id}", executedPlan), exprId)
+      case expressions.Exists(sub, children, exprId) =>
+        val executedPlan = new QueryExecution(sparkSession, Project(Nil, sub)).executedPlan
+        ExistsExec(SubqueryExec(s"subquery#${exprId.id}", executedPlan), exprId)
     }
   }
 }