leanken commented on a change in pull request #29104:
URL: https://github.com/apache/spark/pull/29104#discussion_r457791849
##########
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:
i will try do the refine.
----------------------------------------------------------------
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:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
