[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,50 @@ case class InSet(child: Expression, hset: Set[Any]) extends UnaryExpression with
   }
 }
 
+case class ExistsSubquery(child: Expression,
+                          subQuery: String,
+                          hset: Set[Any]) extends UnaryExpression with Predicate {
+
+  require(hset != null, "hset could not be null")
+
+  override def toString: String = s"Exists ${subQuery}"
+
+  override def nullable: Boolean = child.nullable
+
+  protected override def nullSafeEval(value: Any): Any = {
+    if (set.contains(value)) {
+      true
+    } else {
+      false
+    }
+  }
+
+  @transient lazy val set: Set[Any] = child.dataType match {
+    case t: AtomicType if !t.isInstanceOf[BinaryType] => hset
+    case _: NullType => hset
+    case _ =>
+      // for structs use interpreted ordering to be able to compare UnsafeRows with non-UnsafeRows
+      TreeSet.empty(TypeUtils.getInterpretedOrdering(child.dataType)) ++ (hset - null)
+  }
+
+  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("set", set)
+      s"""
+         |${ev.value} = $setTerm.size() > 0;
+       """.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/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,69 @@ 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[Array[Any]] = null)
+  extends ExecSubqueryExpression {
+
+  @transient private var result: Array[Any] = _
+
+  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 = {
+    val rows = plan.executeCollect()
+    result = child.dataType match {
+      case _: StructType => rows.toArray
+      case _ => rows.map(_.get(0, child.dataType))
+    }
+    resultBroadcast = plan.sqlContext.sparkContext.broadcast(result)
+  }
+
+  def values(): Option[Array[Any]] = Option(resultBroadcast).map(_.value)
+
+  private def prepareResult(): Unit = {
+    require(resultBroadcast != null, s"$this has not finished")
+    if (result == null) {
+      result = resultBroadcast.value
+    }
+  }
+
+  override def eval(input: InternalRow): Any = {
+    prepareResult()
+    !result.isEmpty
+  }
+
+  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.toSet).doGenCode(ctx, ev)
+  }
+}
+
 /**
  * Plans subqueries that are present in the given [[SparkPlan]].
  */
@@ -194,6 +257,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)
     }
   }
 }