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

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

@@ -487,6 +487,36 @@ case class InSet(child: Expression, hset: Set[Any]) extends UnaryExpression with
   }
 }
 
+case class ExistsSubquery(child: Expression,
+                          subQuery: String,
+                          result: Boolean) extends UnaryExpression with Predicate {
+
+  override def toString: String = s"Exists ${subQuery}"
+
+  override def nullable: Boolean = child.nullable
+
+  protected override def nullSafeEval(value: Any): Any = {
+    true
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    genCodeWithSet(ctx, ev)
+  }
+
+  private def genCodeWithSet(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    nullSafeCodeGen(ctx, ev, c => {
+      val setTerm = ctx.addReferenceObj("result", result)
+      s"""
+         |${ev.value} = $setTerm;
+       """.stripMargin
+    })
+  }
+
+  override def sql: String = {
+    s"(EXISTS (${subQuery}))"
+  }
+}
+
 @ExpressionDescription(
   usage = "expr1 _FUNC_ expr2 - Logical AND.")
 case class And(left: Expression, right: Expression) extends BinaryOperator with Predicate {

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

@@ -106,12 +106,20 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
 
       // Filter the plan by applying left semi and left anti joins.
       withSubquery.foldLeft(newFilter) {
-        case (p, Exists(sub, conditions, _)) =>
-          val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
-          buildJoin(outerPlan, sub, LeftSemi, joinCond)
-        case (p, Not(Exists(sub, conditions, _))) =>
-          val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
-          buildJoin(outerPlan, sub, LeftAnti, joinCond)
+        case (p, exists @ Exists(sub, conditions, _)) =>
+          if (SubqueryExpression.hasCorrelatedSubquery(exists)) {
+            val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
+            buildJoin(outerPlan, sub, LeftSemi, joinCond)
+          } else {
+            Filter(exists, newFilter)
+          }
+        case (p, Not(exists @ Exists(sub, conditions, _))) =>
+          if (SubqueryExpression.hasCorrelatedSubquery(exists)) {
+            val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
+            buildJoin(outerPlan, sub, LeftAnti, joinCond)
+          } else {
+            Filter(Not(exists), newFilter)
+          }
         case (p, InSubquery(values, ListQuery(sub, conditions, _, _))) =>
           // Deduplicate conflicting attributes if any.
           val newSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))

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

@@ -23,7 +23,7 @@ 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, ExistsSubquery, Expression, ExprId, InSet, ListQuery, Literal, PlanExpression}
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.internal.SQLConf
@@ -171,6 +171,63 @@ case class InSubqueryExec(
   }
 }
 
+/**
+ * The physical node of exists-subquery. This is for support use exists in join's on condition,
+ * since some join type we can't pushdown exists condition, we plan it here
+ */
+case class ExistsExec(child: Expression,
+                      subQuery: String,
+                      plan: BaseSubqueryExec,
+                      exprId: ExprId,
+                      private var resultBroadcast: Broadcast[Boolean] = null)
+  extends ExecSubqueryExpression {
+
+  @transient private var result: Boolean = _
+
+  override def dataType: DataType = BooleanType
+  override def children: Seq[Expression] = child :: Nil
+  override def nullable: Boolean = child.nullable
+  override def toString: String = s"EXISTS ${plan.name}"
+  override def withNewPlan(plan: BaseSubqueryExec): ExistsExec = copy(plan = plan)
+
+  override def semanticEquals(other: Expression): Boolean = other match {
+    case in: ExistsExec => child.semanticEquals(in.child) && plan.sameResult(in.plan)
+    case _ => false
+  }
+
+
+  def updateResult(): Unit = {
+    result = !plan.execute().isEmpty()
+    resultBroadcast = plan.sqlContext.sparkContext.broadcast[Boolean](result)
+  }
+
+  def values(): Option[Boolean] = Option(resultBroadcast).map(_.value)
+
+  private def prepareResult(): Unit = {
+    require(resultBroadcast != null, s"$this has not finished")
+    result = resultBroadcast.value
+  }
+
+  override def eval(input: InternalRow): Any = {
+    prepareResult()
+    result
+  }
+
+  override lazy val canonicalized: ExistsExec = {
+    copy(
+      child = child.canonicalized,
+      subQuery = subQuery,
+      plan = plan.canonicalized.asInstanceOf[BaseSubqueryExec],
+      exprId = ExprId(0),
+      resultBroadcast = null)
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    prepareResult()
+    ExistsSubquery(child, subQuery, result).doGenCode(ctx, ev)
+  }
+}
+
 /**
  * Plans subqueries that are present in the given [[SparkPlan]].
  */
@@ -194,6 +251,19 @@ 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 expr = if (children.length == 1) {
+          children.head
+        } else {
+          CreateNamedStruct(
+            children.zipWithIndex.flatMap { case (v, index) =>
+              Seq(Literal(s"col_$index"), v)
+            }
+          )
+        }
+        val executedPlan = new QueryExecution(sparkSession, sub).executedPlan
+        ExistsExec(expr, sub.treeString,
+          SubqueryExec(s"subquery#${exprId.id}", executedPlan), exprId)
     }
   }
 }