diff --git a/docs/streaming/apis-on-dataframes-and-datasets.md b/docs/streaming/apis-on-dataframes-and-datasets.md index 86585caead51f..5c3236a165b99 100644 --- a/docs/streaming/apis-on-dataframes-and-datasets.md +++ b/docs/streaming/apis-on-dataframes-and-datasets.md @@ -1322,6 +1322,28 @@ side in future. Semi joins have the same guarantees as [inner joins](#semantic-guarantees-of-stream-stream-inner-joins-with-watermarking) regarding watermark delays and whether data will be dropped or not. +##### Anti Joins with Watermarking +An anti join returns values from the left side of the relation that has no match with the right. +It is also referred to as a left anti join. As with semi joins, watermarking and event-time +constraints must be specified for an anti join: since a row is emitted precisely because it has +*no* match, the engine has to wait until the watermark guarantees that no matching row can arrive +on the right side in future before it can emit the row. + +As for the other stateful join types, the event-time constraint can be expressed in either of two +ways: a watermarked event-time column can appear in the equality join keys, or a watermark can be +defined on the right side together with a time range condition (for example +`leftTime BETWEEN rightTime - INTERVAL 1 HOUR AND rightTime`). Defining a watermark on the left +side as well is optional, and is what allows the left side state to be cleaned up. + +Note that anti join is only supported in Append output mode. Update mode would have to emit rows +early, before the watermark can rule out a future match, and such a row could be invalidated by a +later batch. + +###### Semantic Guarantees of Stream-stream Anti Joins with Watermarking +Anti joins have the same guarantees regarding watermark delays and whether data will be dropped as +[outer joins](#outer-joins-with-watermarking), because unmatched rows are likewise only emitted once +the watermark has passed them. + ##### Support matrix for joins in streaming queries @@ -1343,8 +1365,8 @@ regarding watermark delays and whether data will be dropped or not. - - + + @@ -1365,8 +1387,12 @@ regarding watermark delays and whether data will be dropped or not. - - + + + + + + @@ -1387,8 +1413,12 @@ regarding watermark delays and whether data will be dropped or not. - - + + + + + + + + + + diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala index eddcee169e377..363cb10b75f10 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala @@ -491,7 +491,10 @@ object UnsupportedOperationChecker extends Logging { joinType match { // The behavior for unmatched rows in outer joins with update mode // hasn't been defined yet. - case LeftOuter | RightOuter | FullOuter => + // LeftAnti is included here because its unmatched rows are only emitted once the + // watermark guarantees no future match, so early-firing in Update mode would + // produce rows which a later batch could invalidate. + case LeftOuter | RightOuter | FullOuter | LeftAnti => if (outputMode != InternalOutputModes.Append) { throwError(s"$joinType join between two streaming DataFrames/Datasets" + s" is not supported in ${outputMode} output mode, only in Append output mode") @@ -522,10 +525,16 @@ object UnsupportedOperationChecker extends Logging { checkForStreamStreamJoinWatermark(j) } + // We support streaming left anti joins with stream on both sides under the + // appropriate conditions. A streaming right with a static left is not supported: + // unmatched left rows are determined at watermark-based eviction of the left state, + // which a static left side does not have. case LeftAnti => - if (right.isStreaming) { - throwError(s"$LeftAnti joins with a streaming DataFrame/Dataset " + - "on the right are not supported") + if (!left.isStreaming && right.isStreaming) { + throwError(s"$LeftAnti join with a streaming DataFrame/Dataset " + + "on the right and a static DataFrame/Dataset on the left is not supported") + } else if (left.isStreaming && right.isStreaming) { + checkForStreamStreamJoinWatermark(j) } // We support streaming left outer and left semi joins with static on the right always, @@ -687,7 +696,7 @@ object UnsupportedOperationChecker extends Logging { // Check if the nullable side has a watermark, and there's a range condition which // implies a state value watermark on the first side. val hasValidWatermarkRange = join.joinType match { - case LeftOuter | LeftSemi => StreamingJoinHelper.getStateValueWatermark( + case LeftOuter | LeftSemi | LeftAnti => StreamingJoinHelper.getStateValueWatermark( join.left.outputSet, join.right.outputSet, join.condition, Some(1000000)).isDefined case RightOuter => StreamingJoinHelper.getStateValueWatermark( join.right.outputSet, join.left.outputSet, join.condition, Some(1000000)).isDefined diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala index 293523b86f998..ae114e76f0aa9 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala @@ -424,7 +424,8 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { streamBatchSupported = false, expectedMsg = "FullOuter join") - // Left outer, left semi, left anti join: *-stream not allowed + // Left outer, left semi, left anti join: batch-stream not allowed, and stream-stream join is + // allowed 'conditionally' - see the watermark checks below Seq((LeftOuter, "LeftOuter join"), (LeftSemi, "LeftSemi join"), (LeftAnti, "LeftAnti join")) .foreach { case (joinType, name) => testBinaryOperationInStreamingPlan( @@ -443,8 +444,10 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { streamStreamSupported = false, expectedMsg = "RightOuter join") - // Left outer, right outer, full outer joins: Update mode not allowed - Seq(LeftOuter, RightOuter, FullOuter).foreach { joinType => + // Left outer, right outer, full outer, left anti joins: Update mode not allowed. Left anti is + // included because its unmatched rows are only emitted at watermark-based eviction, so + // early-firing could emit a row which a later batch would invalidate. + Seq(LeftOuter, RightOuter, FullOuter, LeftAnti).foreach { joinType => assertNotSupportedInStreamingPlan( s"$joinType join with stream-stream relations and update mode", streamRelation.join(streamRelation, joinType = joinType, @@ -467,7 +470,8 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { (LeftSemi, "only in Append and Update output modes"), (LeftOuter, "only in Append output mode"), (RightOuter, "only in Append output mode"), - (FullOuter, "only in Append output mode") + (FullOuter, "only in Append output mode"), + (LeftAnti, "only in Append output mode") ).foreach { case (joinType, allowedModesMsg) => assertNotSupportedInStreamingPlan( s"$joinType join with stream-stream relations and complete mode", @@ -477,8 +481,8 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { Seq("is not supported in Complete output mode", allowedModesMsg)) } - // Left outer, right outer, full outer, left semi joins - Seq(LeftOuter, RightOuter, FullOuter, LeftSemi).foreach { joinType => + // Left outer, right outer, full outer, left semi, left anti joins + Seq(LeftOuter, RightOuter, FullOuter, LeftSemi, LeftAnti).foreach { joinType => // Stream-stream allowed with join on watermark attribute // Note that the attribute need not be watermarked on both sides. assertSupportedInStreamingPlan( diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala index 8f90a603c7efb..de0e33edee84c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala @@ -187,7 +187,7 @@ case class StreamingSymmetricHashJoinExec( require( joinType == Inner || joinType == LeftOuter || joinType == RightOuter || joinType == FullOuter || - joinType == LeftSemi, + joinType == LeftSemi || joinType == LeftAnti, errorMessageForJoinType) outputMode.foreach { mode => @@ -238,7 +238,7 @@ case class StreamingSymmetricHashJoinExec( case LeftOuter => left.output ++ right.output.map(_.withNullability(true)) case RightOuter => left.output.map(_.withNullability(true)) ++ right.output case FullOuter => (left.output ++ right.output).map(_.withNullability(true)) - case LeftSemi => left.output + case LeftSemi | LeftAnti => left.output case _ => throwBadJoinTypeException() } WidenStatefulOpNullability.widenOutputForStatefulOp(base) @@ -251,7 +251,7 @@ case class StreamingSymmetricHashJoinExec( case LeftOuter => left.outputPartitioning case RightOuter => right.outputPartitioning case FullOuter => UnknownPartitioning(left.outputPartitioning.numPartitions) - case LeftSemi => left.outputPartitioning + case LeftSemi | LeftAnti => left.outputPartitioning case _ => throwBadJoinTypeException() } @@ -422,6 +422,12 @@ case class StreamingSymmetricHashJoinExec( // For Left Semi Join, we process the right side first so that new right input is stored // before left input probes against it. This way, new left rows that match new right rows // in the same microbatch are emitted immediately without being buffered in state. + // + // Left Anti Join generates nothing while joining, on either side: whether a left row has no + // match is only decided once the watermark guarantees no future right row can match it, so + // unmatched left rows are emitted when the left side state is evicted (see `outputIter`). Both + // sides still store their input, and matching a new right input marks the corresponding stored + // left rows as matched so that they are suppressed at eviction time. val leftOutputIter = joinerManager.leftSideJoiner.storeAndJoinWithOtherSide(joinerManager.rightSideJoiner) { (input: InternalRow, matched: InternalRow) => joinedRow.withLeft(input).withRight(matched) @@ -542,10 +548,40 @@ case class StreamingSymmetricHashJoinExec( val rightSideOutputIter = new LazilyInitializingJoinedRowIterator(rightSideInitIterFn) hashJoinOutputIter ++ leftSideOutputIter ++ rightSideOutputIter + case LeftAnti => + // Left anti join is structurally an eviction-time join, like left outer: whether a left + // row has "no match" is only known once the watermark guarantees no future right row can + // match it. So we reuse the left outer eviction path, with two differences: + // * nothing is emitted when a left row matches, and + // * an unmatched left row is emitted as the bare left row rather than joined with nulls. + // + // State format version 1 is already rejected for non-inner joins (see the constructor), + // so only the 'matched' flag path needs to be handled. + val initIterFn = { () => + val removedRowIter = joinerManager.leftSideJoiner.removeAndReturnOldState() + removedRowIter.filterNot { kv => + stateFormatVersion match { + case 2 | 3 | 4 => kv.matched + case _ => throwBadStateFormatVersionException() + } + }.map(_.value) + } + + // NOTE: we need to make sure `antiOutputIter` is evaluated "after" exhausting all of + // elements in `hashJoinOutputIter`, otherwise it may lead to out of sync according to + // the interface contract on StateStore.iterator and end up with correctness issue. + // Please refer SPARK-38684 for more details. + val antiOutputIter = new LazilyInitializingRowIterator(initIterFn) + + // `hashJoinOutputIter` still has to be consumed: draining it is what appends input rows to + // the state stores and persists the 'matched' flag for left rows which found a match. For + // left anti it only carries the rows which failed the pre-join filter (see + // `generateFilteredJoinedRow`), which are genuine anti output, so it is concatenated as-is. + hashJoinOutputIter ++ antiOutputIter case _ => throwBadJoinTypeException() } - val outputProjection = if (joinType == LeftSemi) { + val outputProjection = if (joinType == LeftSemi || joinType == LeftAnti) { UnsafeProjection.create(output, output) } else { UnsafeProjection.create(left.output ++ right.output, output) @@ -587,10 +623,14 @@ case class StreamingSymmetricHashJoinExec( // // For full outer joins, we have already removed unnecessary states from both sides, so // nothing needs to be outputted here. + // + // For left anti joins, the left side state was already consumed and removed while + // generating the unmatched ("anti") output, same as the left side of a left outer join, + // so only the right side remains to be removed greedily. numRemovedStateRows += ( joinType match { case Inner | LeftSemi => joinerManager.removeOldState() - case LeftOuter => joinerManager.rightSideJoiner.removeOldState() + case LeftOuter | LeftAnti => joinerManager.rightSideJoiner.removeOldState() case RightOuter => joinerManager.leftSideJoiner.removeOldState() case FullOuter => 0L case _ => throwBadJoinTypeException() @@ -779,6 +819,9 @@ case class StreamingSymmetricHashJoinExec( (row: InternalRow) => Iterator(generateJoinedRow(row, nullRight)) case RightSide if joinType == RightOuter || joinType == FullOuter => (row: InternalRow) => Iterator(generateJoinedRow(row, nullLeft)) + // A left row which fails the pre-join filter can never satisfy the join condition, so it + // is unmatched by definition and is emitted right away without being added to the state. + case LeftSide if joinType == LeftAnti => (row: InternalRow) => Iterator(row) case _ => (_: InternalRow) => Iterator.empty } @@ -787,12 +830,15 @@ case class StreamingSymmetricHashJoinExec( // unmatched rows) and on the left side of left semi (for matched-rows removal). // For older versions, we do not apply the optimization as it is a behavioral change, // although the optimization is valid for all versions. + // + // Left anti join needs the flag on the left side rows only, so that unmatched left rows can + // be identified at eviction time. That flag is written while processing the right side. val needToUpdateMatchedOnOtherSide = joinType match { case Inner => false case LeftOuter => joinSide == RightSide case RightOuter => joinSide == LeftSide case FullOuter => true - case LeftSemi => joinSide == RightSide + case LeftSemi | LeftAnti => joinSide == RightSide case _ => true } val skipUpdatingMatchedFlag = stateFormatVersion == 4 && !needToUpdateMatchedOnOtherSide @@ -845,21 +891,48 @@ case class StreamingSymmetricHashJoinExec( timestampRange = computeTimestampRange(thisRow), skipUpdatingMatchedFlag) } - val outputIter = generateOutputIter(thisRow, joinedRowIter) - new AddingProcessedRowToStateCompletionIterator(key, thisRow, outputIter) + if (joinType == LeftAnti) { + // Left anti join emits nothing when rows match, on either side; unmatched left rows + // are emitted later during watermark-based eviction of the left side state. The match + // status is passed explicitly, since it can no longer be inferred from the (empty) + // output iterator. + // + // The iterator must be drained fully rather than short-circuited on the first match: + // `getJoinedRows` sets the 'matched' flag on the other side's rows lazily, as they are + // produced, so stopping early would leave some matched left rows flagged as unmatched + // and they would then be wrongly emitted as anti output at eviction time. Draining is + // also what the state manager API requires of callers. + var matched = false + while (joinedRowIter.hasNext) { + joinedRowIter.next() + matched = true + } + new AddingProcessedRowToStateCompletionIterator( + key, thisRow, Iterator.empty, Some(matched)) + } else { + val outputIter = generateOutputIter(thisRow, joinedRowIter) + new AddingProcessedRowToStateCompletionIterator(key, thisRow, outputIter) + } } else { generateFilteredJoinedRow(thisRow) } } } + /** + * @param matchedOverride the match status of `thisRow`, for join types whose output iterator + * does not reflect whether a match was found (left anti emits nothing + * on a match). When empty, the match status is inferred from whether + * the output iterator is non-empty. + */ private class AddingProcessedRowToStateCompletionIterator( key: UnsafeRow, thisRow: UnsafeRow, - subIter: Iterator[InternalRow]) + subIter: Iterator[InternalRow], + matchedOverride: Option[Boolean] = None) extends CompletionIterator[InternalRow, Iterator[InternalRow]](subIter) { - private val iteratorNotEmpty: Boolean = super.hasNext + private val iteratorNotEmpty: Boolean = matchedOverride.getOrElse(super.hasNext) override def completion(): Unit = { // The criteria of whether the input has to be added into state store or not: @@ -874,6 +947,10 @@ case class StreamingSymmetricHashJoinExec( // if input is going to be evicted in this batch. Though, input should be added to the // state store if it's right outer join or full outer join, as unmatched output is // handled during state eviction. + // + // Note left anti deliberately does not get the left semi skip-on-match treatment: a matched + // left row has to stay in state (carrying matched = true) so that it is suppressed, rather + // than emitted, when the left side state is evicted. val isLeftSemiWithMatch = joinType == LeftSemi && joinSide == LeftSide && iteratorNotEmpty val shouldAddToState = if (isLeftSemiWithMatch) { false @@ -1084,6 +1161,18 @@ case class StreamingSymmetricHashJoinExec( override def next(): JoinedRow = iter.next() } + /** + * Same as [[LazilyInitializingJoinedRowIterator]], but for join types which emit rows from a + * single side (left anti) rather than joined rows. + */ + private class LazilyInitializingRowIterator( + initFn: () => Iterator[UnsafeRow]) extends Iterator[UnsafeRow] { + private lazy val iter: Iterator[UnsafeRow] = initFn() + + override def hasNext: Boolean = iter.hasNext + override def next(): UnsafeRow = iter.next() + } + // If `STATE_STORE_SKIP_NULLS_FOR_STREAM_STREAM_JOINS` is enabled, counting the number // of skipped null values as custom metric of stream join operator. override def customStatefulOperatorMetrics: Seq[StatefulOperatorCustomMetric] = diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala index c0983f338abe5..ead8e406e349f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala @@ -157,7 +157,7 @@ abstract class StreamingJoinSuite val windowed2 = df2 .select($"key", window($"rightTime", "10 second"), $"rightValue") val joined = windowed1.join(windowed2, Seq("key", "window"), joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select($"key", $"window.end".cast("long"), $"leftValue") } else { joined.select($"key", $"window.end".cast("long"), $"leftValue", @@ -185,7 +185,7 @@ abstract class StreamingJoinSuite && $"leftValue" > 4, joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue") } else if (joinType == "left_outer") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue", @@ -219,7 +219,7 @@ abstract class StreamingJoinSuite && $"rightValue".cast("int") > 7, joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue") } else if (joinType == "left_outer") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue", @@ -262,7 +262,7 @@ abstract class StreamingJoinSuite s"leftTime BETWEEN rightTime - $lowerBound AND rightTime + $upperBound"), joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select($"leftKey", $"leftTime".cast("int")) } else { joined.select($"leftKey", $"rightKey", $"leftTime".cast("int"), @@ -2632,6 +2632,212 @@ class StreamingFullOuterJoinWithoutVCFSuite extends StreamingFullOuterJoinSuite override protected def testMode = Mode.WithoutVCF } +abstract class StreamingLeftAntiJoinBase extends StreamingJoinSuite { + + test("windowed left anti join") { + withTempDir { checkpointDir => + val (leftInput, rightInput, joined) = setupWindowedJoin("left_anti") + + testStream(joined, OutputMode.Append())( + StartStream(checkpointLocation = checkpointDir.getCanonicalPath), + MultiAddData(leftInput, 1, 2, 3, 4, 5)(rightInput, 3, 4, 5, 6, 7), + // Nothing is emitted yet: left 1 and 2 are unmatched, but the watermark cannot yet rule + // out a future match for them. Left 3, 4, 5 matched and are suppressed forever. + CheckNewAnswer(), + // states + // left: 1, 2, 3, 4, 5 (all buffered; 3, 4, 5 carry matched = true) + // right: 3, 4, 5, 6, 7 + assertNumStateRows( + total = Seq(10), updated = Seq(10), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + MultiAddData(leftInput, 21)(rightInput, 22), + // Watermark = 11, so window=[0,10] is evicted from the left side: the unmatched left rows + // 1 and 2 are emitted now, while the matched 3, 4, 5 are dropped without output. + CheckNewAnswer(Row(1, 10, 2), Row(2, 10, 4)), + // states + // left: 21 + // right: 22 + // + // states evicted + // left: 1, 2, 3, 4, 5 (below watermark) + // right: 3, 4, 5, 6, 7 (below watermark) + // + // Only the 5 right side rows are reported as removed: the left side rows are evicted + // through removeAndReturnOldState while generating the anti output, which does not feed + // numRemovedStateRows. Left outer join reports removals the same way. + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(5))), + StopStream, + // Restart the join query from the same checkpoint + StartStream(checkpointLocation = checkpointDir.getCanonicalPath), + AddData(leftInput, 22), + // Left 22 matches right 22, so it is buffered with matched = true and never emitted. + CheckNewAnswer(), + // states + // left: 21, 22 + // right: 22 + assertNumStateRows( + total = Seq(3), updated = Seq(1), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + StopStream, + // Restart the query from the same checkpoint + StartStream(checkpointLocation = checkpointDir.getCanonicalPath), + AddData(leftInput, 1), + // Row not added as 1 < state key watermark = 12. + CheckNewAnswer(), + // states + // left: 21, 22 + // right: 22 + assertNumStateRows( + total = Seq(3), updated = Seq(0), + droppedByWatermark = Seq(1), removed = Some(Seq(0))) + ) + } + } + + test("left anti join emits an unmatched left row only once the watermark passes it") { + val (leftInput, rightInput, joined) = setupWindowedJoin("left_anti") + + testStream(joined, OutputMode.Append())( + AddData(leftInput, 3), + // Unmatched so far, but not yet provably unmatched. + CheckNewAnswer(), + // A matching right row arrives in a later batch, so left 3 must never be emitted. + AddData(rightInput, 3), + CheckNewAnswer(), + // Advance the watermark past window=[0,10]. + MultiAddData(leftInput, 21)(rightInput, 21), + CheckNewAnswer(), + // Left 21 matched right 21, and left 3 was matched before eviction, so still nothing. + MultiAddData(leftInput, 31)(rightInput, 31), + CheckNewAnswer() + ) + } + + test("left anti early state exclusion on left") { + val (leftInput, rightInput, joined) = setupWindowedJoinWithLeftCondition("left_anti") + + testStream(joined, OutputMode.Append())( + MultiAddData(leftInput, 1, 2, 3)(rightInput, 3, 4, 5), + // The left rows with leftValue <= 4 (i.e. left 1 and 2) fail the pre-join filter, so they can + // never match and are emitted immediately without being added to the state. + CheckNewAnswer(Row(1, 10, 2), Row(2, 10, 4)), + // states + // left: 3 (matched right 3 in the same batch, buffered with matched = true) + // right: 3, 4, 5 + assertNumStateRows( + total = Seq(4), updated = Seq(4), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + // Left 3 matched, so advancing the watermark must not produce an anti row for it. + MultiAddData(leftInput, 20)(rightInput, 21), + CheckNewAnswer(), + // states + // left: 20 + // right: 21 + // + // states evicted + // left: 3 (below watermark, matched so no output) + // right: 3, 4, 5 (below watermark) + // + // Only the 3 right side rows are counted as removed - see the note in + // "windowed left anti join" about left side eviction not feeding numRemovedStateRows. + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(3))) + ) + } + + test("left anti early state exclusion on right") { + val (leftInput, rightInput, joined) = setupWindowedJoinWithRightCondition("left_anti") + + testStream(joined, OutputMode.Append())( + MultiAddData(leftInput, 3, 4, 5)(rightInput, 1, 2, 3), + // The right rows with rightValue <= 7 (i.e. right 1 and 2) fail the pre-join filter, so they + // are not added to the state and cannot suppress any left row. + CheckNewAnswer(), + // states + // left: 3, 4, 5 (3 matched right 3, so carries matched = true) + // right: 3 + assertNumStateRows( + total = Seq(4), updated = Seq(4), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + // Advance the watermark: left 4 and 5 were never matched, so they are emitted now. + MultiAddData(leftInput, 20)(rightInput, 21), + CheckNewAnswer(Row(4, 10, 8), Row(5, 10, 10)), + // states + // left: 20 + // right: 21 + // + // states evicted + // left: 3, 4, 5 (below watermark) + // right: 3 (below watermark) + // + // Only the single right side row is counted as removed - see the note in + // "windowed left anti join" about left side eviction not feeding numRemovedStateRows. + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(1))) + ) + } + + test("left anti join with watermark range condition") { + val (leftInput, rightInput, joined) = setupJoinWithRangeCondition("left_anti") + + testStream(joined, OutputMode.Append())( + AddData(leftInput, (1, 5), (3, 5)), + // Neither left row is provably unmatched yet. + CheckNewAnswer(), + // states + // left: (1, 5), (3, 5) + // right: nothing + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + AddData(rightInput, (1, 10), (2, 5)), + // Right (1, 10) satisfies the range condition against left (1, 5), so that left row is + // marked as matched in state and must never be emitted. Unlike left semi, it is kept in + // state so that it can be suppressed at eviction time. + CheckNewAnswer(), + // states + // left: (1, 5) (now matched), (3, 5) + // right: (1, 10), (2, 5) + assertNumStateRows( + total = Seq(4), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + // Advance the watermark to 20 by adding rows with event time 30 on both sides. A left row is + // only safe to evict once no future right row can match it, which the range condition puts + // at leftTime + 5 < watermark, so the left rows with leftTime < 15 are evicted here: + // (3, 5) was never matched so it is emitted, while (1, 5) was matched so it is suppressed. + AddData(leftInput, (1, 30)), + CheckNewAnswer(), + AddData(rightInput, (0, 30)), + CheckNewAnswer(Row(3, 5)), + // Advance the watermark to 50, which evicts the left rows with leftTime < 45. Left (1, 30) + // never matched -- the only right row with key 1 was evicted long before -- so it is + // emitted now. + AddData(leftInput, (2, 60)), + CheckNewAnswer(), + AddData(rightInput, (0, 60)), + CheckNewAnswer(Row(1, 30)) + ) + } +} + +abstract class StreamingLeftAntiJoinSuite extends StreamingLeftAntiJoinBase { + + test("left anti join is not supported in Update output mode") { + val (_, _, joined) = setupWindowedJoin("left_anti") + + val e = intercept[AnalysisException] { + joined.writeStream.format("memory").queryName("leftAntiUpdate") + .outputMode(OutputMode.Update()).start() + } + assert(e.getMessage.contains("LeftAnti join between two streaming DataFrames/Datasets " + + "is not supported in Update output mode, only in Append output mode")) + } +} + @SlowSQLTest class StreamingLeftSemiJoinWithVCFSuite extends StreamingLeftSemiJoinSuite { override protected def testMode = Mode.WithVCF @@ -2641,3 +2847,13 @@ class StreamingLeftSemiJoinWithVCFSuite extends StreamingLeftSemiJoinSuite { class StreamingLeftSemiJoinWithoutVCFSuite extends StreamingLeftSemiJoinSuite { override protected def testMode = Mode.WithoutVCF } + +@SlowSQLTest +class StreamingLeftAntiJoinWithVCFSuite extends StreamingLeftAntiJoinSuite { + override protected def testMode = Mode.WithVCF +} + +@SlowSQLTest +class StreamingLeftAntiJoinWithoutVCFSuite extends StreamingLeftAntiJoinSuite { + override protected def testMode = Mode.WithoutVCF +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala index 6d4a97861efef..fa064e64fec6e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala @@ -476,3 +476,8 @@ class StreamingFullOuterJoinV4Suite class StreamingLeftSemiJoinV4Suite extends StreamingLeftSemiJoinBase with TestWithV4StateFormat + +@SlowSQLTest +class StreamingLeftAntiJoinV4Suite + extends StreamingLeftAntiJoinBase + with TestWithV4StateFormat
StreamStaticStreamStatic Inner Supported, not stateful
Supported, not stateful
StaticStreamLeft AntiSupported, not stateful
StaticStream Inner Supported, not stateful
Not supported
StreamStreamLeft AntiNot supported
StreamStream Inner Supported, optionally specify watermark on both sides + @@ -1423,6 +1453,13 @@ regarding watermark delays and whether data will be dropped or not. results, optionally specify watermark on left for all state cleanup
Left Anti + Conditionally supported, must specify watermark on right + time constraints for correct + results, optionally specify watermark on left for all state cleanup. Append output mode only +