Address comments

Author: c21

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala

@@ -80,7 +80,7 @@ case class AdaptiveSparkPlanExec(
     // TODO add more optimization rules
     override protected def batches: Seq[Batch] = Seq(
       Batch("Demote BroadcastHashJoin", Once, DemoteBroadcastHashJoin(conf)),
-      Batch("Optimize Join to Empty Relation", Once, OptimizeJoinToEmptyRelation)
+      Batch("Eliminate Join to Empty Relation", Once, EliminateJoinToEmptyRelation)
     )
   }
 

…aptive/OptimizeJoinToEmptyRelation.scala …ptive/EliminateJoinToEmptyRelation.scalasql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeJoinToEmptyRelation.scala renamed to sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/EliminateJoinToEmptyRelation.scala sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeJoinToEmptyRelation.scala renamed to sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/EliminateJoinToEmptyRelation.scala

@@ -24,19 +24,18 @@ import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.joins.{EmptyHashedRelation, EmptyHashedRelationWithAllNullKeys, HashedRelation}
 
 /**
- * This optimization rule detects and converts a `Join` to an empty `LocalRelation`:
- * 1. `Join` is single column NULL-aware anti join (NAAJ), and broadcasted `HashedRelation`
- *    is `EmptyHashedRelationWithAllNullKeys`.
+ * This optimization rule detects and converts a Join to an empty [[LocalRelation]]:
+ * 1. Join is single column NULL-aware anti join (NAAJ), and broadcasted [[HashedRelation]]
+ *    is [[EmptyHashedRelationWithAllNullKeys]].
  *
- * 2. `Join` is inner or left semi join, and broadcasted `HashedRelation` is `EmptyHashedRelation`.
+ * 2. Join is inner or left semi join, and broadcasted [[HashedRelation]]
+      is [[EmptyHashedRelation]].
  */
-object OptimizeJoinToEmptyRelation extends Rule[LogicalPlan] {
+object EliminateJoinToEmptyRelation extends Rule[LogicalPlan] {
 
-  private def canEliminate(
-      plan: LogicalPlan,
-      expectedRelation: HashedRelation): Boolean = plan match {
+  private def canEliminate(plan: LogicalPlan, emptyRelation: HashedRelation): Boolean = plan match {
     case LogicalQueryStage(_, stage: BroadcastQueryStageExec) if stage.resultOption.get().isDefined
-      && stage.broadcast.relationFuture.get().value == expectedRelation => true
+      && stage.broadcast.relationFuture.get().value == emptyRelation => true
     case _ => false
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala

@@ -447,9 +447,9 @@ trait HashJoin extends BaseJoinExec with CodegenSupport {
     }
 
     if (isEmptyHashedRelation) {
-      s"""
-         |// If HashedRelation is empty, hash inner join simply returns nothing.
-       """.stripMargin
+      """
+        |// If HashedRelation is empty, hash inner join simply returns nothing.
+      """.stripMargin
     } else if (keyIsUnique) {
       s"""
          |// generate join key for stream side
@@ -573,9 +573,9 @@ trait HashJoin extends BaseJoinExec with CodegenSupport {
     val numOutput = metricTerm(ctx, "numOutputRows")
 
     if (isEmptyHashedRelation) {
-      s"""
-         |// If HashedRelation is empty, hash semi join simply returns nothing.
-       """.stripMargin
+      """
+        |// If HashedRelation is empty, hash semi join simply returns nothing.
+      """.stripMargin
     } else if (keyIsUnique) {
       s"""
          |// generate join key for stream side

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

@@ -1170,20 +1170,24 @@ class AdaptiveQueryExecSuite
     }
   }
 
-  test("SPARK-32573, SPARK-32649: optimize join to empty relation") {
+  test("SPARK-32573: Eliminate NAAJ when BuildSide is EmptyHashedRelationWithAllNullKeys") {
     withSQLConf(
       SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
-      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-      // Test NULL-aware anti join
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) {
       val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
         "SELECT * FROM testData2 t1 WHERE t1.b NOT IN (SELECT b FROM testData3)")
       val bhj = findTopLevelBroadcastHashJoin(plan)
       assert(bhj.size == 1)
       val join = findTopLevelBaseJoin(adaptivePlan)
       assert(join.isEmpty)
       checkNumLocalShuffleReaders(adaptivePlan)
+    }
+  }
 
-      // Test inner and left semi join
+  test("SPARK-32649: Eliminate inner and semi join to empty relation") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
       Seq(
         // inner join (small table at right side)
         "SELECT * FROM testData t1 join testData3 t2 ON t1.key = t2.a WHERE t2.b = 1",