Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/19327#discussion_r141987664
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingSymmetricHashJoinExec.scala
---
@@ -207,31 +221,108 @@ case class StreamingSymmetricHashJoinExec(
// matching new left input with new right input, since the new left
input has become stored
// by that point. This tiny asymmetry is necessary to avoid
duplication.
val leftOutputIter =
leftSideJoiner.storeAndJoinWithOtherSide(rightSideJoiner) {
- (inputRow: UnsafeRow, matchedRow: UnsafeRow) =>
+ (inputRow: UnsafeRow, matchedRow: UnsafeRow) => {
joinedRow.withLeft(inputRow).withRight(matchedRow)
+ }
}
val rightOutputIter =
rightSideJoiner.storeAndJoinWithOtherSide(leftSideJoiner) {
- (inputRow: UnsafeRow, matchedRow: UnsafeRow) =>
+ (inputRow: UnsafeRow, matchedRow: UnsafeRow) => {
joinedRow.withLeft(matchedRow).withRight(inputRow)
+ }
}
// Filter the joined rows based on the given condition.
- val outputFilterFunction =
- newPredicate(condition.getOrElse(Literal(true)), left.output ++
right.output).eval _
- val filteredOutputIter =
- (leftOutputIter ++ rightOutputIter).filter(outputFilterFunction).map
{ row =>
- numOutputRows += 1
- row
- }
+ val outputFilterFunction =
newPredicate(condition.getOrElse(Literal(true)), output).eval _
+
+ // We need to save the time that the inner join output iterator
completes, since outer join
+ // output counts as both update and removal time.
+ var innerOutputCompletionTimeNs: Long = 0
+ def onInnerOutputCompletion = {
+ innerOutputCompletionTimeNs = System.nanoTime
+ }
+ val filteredInnerOutputIter = CompletionIterator[InternalRow,
Iterator[InternalRow]](
+ (leftOutputIter ++ rightOutputIter).filter(outputFilterFunction),
onInnerOutputCompletion)
+
+ val outputIter: Iterator[InternalRow] = joinType match {
+ case Inner =>
+ filteredInnerOutputIter
+ case LeftOuter =>
+ // We generate the outer join input by:
+ // * Getting an iterator over the rows that have aged out on the
left side. These rows are
+ // candidates for being null joined. Note that to avoid doing
two passes, this iterator
+ // removes the rows from the state manager as they're processed.
+ // * Checking whether the current row matches a key in the right
side state, and that key
+ // has any value which satisfies the filter function when
joined. If it doesn't,
+ // we know we can join with null, since there was never
(including this batch) a match
+ // within the watermark period. If it does, there must have been
a match at some point, so
+ // we know we can't join with null.
+ val nullRight = new
GenericInternalRow(right.output.map(_.withNullability(true)).length)
+ val removedRowIter = leftSideJoiner.removeOldState()
+ val outerOutputIter = removedRowIter
+ .filterNot(pair => {
+ rightSideJoiner.get(pair.key).exists(
+ rightValue => {
+ outputFilterFunction(
+ joinedRow.withLeft(pair.value).withRight(rightValue))
+ })
--- End diff --
this is very nested and hard to read. can you make in an inline function.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
