[GitHub] [spark] cloud-fan commented on a change in pull request #29104: [SPARK-32290][SQL] SingleColumn Null Aware Anti Join Optimize

[GitBox <gi...@apache.org>]

cloud-fan commented on a change in pull request #29104:
URL: https://github.com/apache/spark/pull/29104#discussion_r457411741



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
##########
@@ -385,55 +408,101 @@ case class BroadcastHashJoinExec(
     val (matched, checkCondition, _) = getJoinCondition(ctx, input)
     val numOutput = metricTerm(ctx, "numOutputRows")
 
-    if (uniqueKeyCodePath) {
-      val found = ctx.freshName("found")
-      s"""
-         |boolean $found = false;
-         |// generate join key for stream side
-         |${keyEv.code}
-         |// Check if the key has nulls.
-         |if (!($anyNull)) {
-         |  // Check if the HashedRelation exists.
-         |  UnsafeRow $matched = (UnsafeRow)$relationTerm.getValue(${keyEv.value});
-         |  if ($matched != null) {
-         |    // Evaluate the condition.
-         |    $checkCondition {
-         |      $found = true;
-         |    }
-         |  }
-         |}
-         |if (!$found) {
-         |  $numOutput.add(1);
-         |  ${consume(ctx, input)}
-         |}
-       """.stripMargin
+    if (isNullAwareAntiJoin) {
+      require(leftKeys.length == 1, "leftKeys length should be 1")
+      require(rightKeys.length == 1, "rightKeys length should be 1")
+      require(right.output.length == 1, "not in subquery hash join optimize only single column.")
+      require(joinType == LeftAnti, "joinType must be LeftAnti.")
+      require(buildSide == BuildRight, "buildSide must be BuildRight.")
+      require(SQLConf.get.nullAwareAntiJoinOptimizeEnabled,
+        "nullAwareAntiJoinOptimizeEnabled must turn on for BroadcastNullAwareHashJoinExec.")
+      require(checkCondition == "", "not in subquery hash join optimize empty condition.")
+
+      if (broadcastRelation.value.inputEmpty) {
+        s"""
+           |// singleColumn NAAJ inputEmpty(true) accept all
+           |$numOutput.add(1);
+           |${consume(ctx, input)}
+         """.stripMargin
+      } else if (broadcastRelation.value.anyNullKeyExists) {
+        s"""
+           |// singleColumn NAAJ inputEmpty(false) anyNullKeyExists(true) reject all
+         """.stripMargin
+      } else {
+        val found = ctx.freshName("found")
+        s"""
+           |// singleColumn NAAJ inputEmpty(false) anyNullKeyExists(false)
+           |boolean $found = false;
+           |// generate join key for stream side
+           |${keyEv.code}
+           |// Check if the key has nulls.
+           |if (!($anyNull)) {
+           |  // Check if the HashedRelation exists.
+           |  UnsafeRow $matched = (UnsafeRow)$relationTerm.getValue(${keyEv.value});
+           |  if ($matched != null) {
+           |    $found = true;
+           |  }
+           |} else {
+           |  $found = true;
+           |}
+           |
+           |if (!$found) {
+           |  $numOutput.add(1);
+           |  ${consume(ctx, input)}
+           |}
+        """.stripMargin
+      }
     } else {

Review comment:
       nit: we can use `else if` to reduce code diff




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org