Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/19452#discussion_r144937226
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingSymmetricHashJoinExec.scala
---
@@ -221,43 +228,29 @@ 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) {
- (input: UnsafeRow, matched: UnsafeRow) =>
joinedRow.withLeft(input).withRight(matched)
+ (input: InternalRow, matched: InternalRow) =>
joinedRow.withLeft(input).withRight(matched)
}
val rightOutputIter =
rightSideJoiner.storeAndJoinWithOtherSide(leftSideJoiner) {
- (input: UnsafeRow, matched: UnsafeRow) =>
joinedRow.withLeft(matched).withRight(input)
+ (input: InternalRow, matched: InternalRow) =>
joinedRow.withLeft(matched).withRight(input)
}
- // Filter the joined rows based on the given condition.
- 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)
-
- def matchesWithRightSideState(leftKeyValue: UnsafeRowPair) = {
- rightSideJoiner.get(leftKeyValue.key).exists(
- rightValue => {
- outputFilterFunction(
- joinedRow.withLeft(leftKeyValue.value).withRight(rightValue))
- })
- }
+ // This is the iterator which produces the inner join rows. For outer
joins, this will be
+ // prepended to a second iterator producing outer join rows; for inner
joins, this is the full
+ // output.
+ val innerOutputIter = CompletionIterator[InternalRow,
Iterator[InternalRow]](
+ (leftOutputIter ++ rightOutputIter), onInnerOutputCompletion)
- def matchesWithLeftSideState(rightKeyValue: UnsafeRowPair) = {
- leftSideJoiner.get(rightKeyValue.key).exists(
- leftValue => {
- outputFilterFunction(
- joinedRow.withLeft(leftValue).withRight(rightKeyValue.value))
- })
- }
+ val postJoinFilter =
newPredicate(condition.bothSides.getOrElse(Literal(true)), output).eval _
--- End diff --
This is also incorrect.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
