c21 commented on a change in pull request #32476:
URL: https://github.com/apache/spark/pull/32476#discussion_r630566578
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
##########
@@ -431,6 +433,66 @@ case class SortMergeJoinExec(
// Copy the streamed keys as class members so they could be used in next
function call.
val matchedKeyVars = copyKeys(ctx, streamedKeyVars)
+ // Handle the case when streamed rows has any NULL keys.
+ val handleStreamedAnyNull = joinType match {
+ case _: InnerLike =>
+ // Skip streamed row.
+ s"""
+ |$streamedRow = null;
+ |continue;
+ """.stripMargin
+ case LeftOuter | RightOuter =>
+ // Eagerly return streamed row. Only call `matches.clear()` when
`matches.isEmpty()` is
+ // false, to reduce unnecessary computation.
+ s"""
+ |if (!$matches.isEmpty()) {
+ | $matches.clear();
+ |}
+ |return false;
+ """.stripMargin
+ case x =>
+ throw new IllegalArgumentException(
+ s"SortMergeJoin.genScanner should not take $x as the JoinType")
+ }
+
+ // Handle the case when streamed keys has no match with buffered side.
+ val handleStreamedWithoutMatch = joinType match {
+ case _: InnerLike =>
+ // Skip streamed row.
+ s"$streamedRow = null;"
+ case LeftOuter | RightOuter =>
+ // Eagerly return with streamed row.
+ "return false;"
+ case x =>
+ throw new IllegalArgumentException(
+ s"SortMergeJoin.genScanner should not take $x as the JoinType")
+ }
+
+ // Generate a function to scan both streamed and buffered sides to find a
match.
+ // Return whether a match is found.
+ //
+ // `streamedIter`: the iterator for streamed side.
+ // `bufferedIter`: the iterator for buffered side.
+ // `streamedRow`: the current row from streamed side.
+ // When `streamedIter` is empty, `streamedRow` is null.
+ // `matches`: the rows from buffered side already matched with
`streamedRow`.
+ // `matches` is buffered and reused for all `streamedRow`s
having same join keys.
+ // If there is no match with `streamedRow`, `matches` is empty.
+ // `bufferedRow`: the current matched row from buffered side.
+ //
+ // The function has the following step:
+ // - Step 1: Find the next `streamedRow` with non-null join keys.
+ // For `streamedRow` with null join keys
(`handleStreamedAnyNull`):
+ // 1. Inner join: skip the row.
+ // 2. Left/Right Outer join: clear the previous `matches` if
needed, keep the row,
Review comment:
For streamed row with null keys, inner join will skip it, but left/right
outer join (and left semi/anti join in followup) needs to output it. It seems
to me it's not very good to have a streamed row with a non-cleared `matches`
(which matches previous rows but not itself). We can change coed to live with
it but I feel it's not very clean.
--
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
