Skip to content

[SPARK-58611][SS] Left anti stream-stream join #57813

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ganeshashree wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[SPARK-58611][SS] Left anti stream-stream join #57813

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c6d1876
[SPARK-58611][SS] Left anti stream-stream join
ganeshas-db Aug 6, 2026
d5e7e6d
[SPARK-58611][SS][FOLLOWUP] Add state format v4 and range-condition a…
ganeshas-db Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 43 additions & 6 deletions docs/streaming/apis-on-dataframes-and-datasets.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

<table>
Expand All @@ -1343,8 +1365,8 @@ regarding watermark delays and whether data will be dropped or not.
</td>
</tr>
<tr>
<td rowspan="5" style="vertical-align: middle;">Stream</td>
<td rowspan="5" style="vertical-align: middle;">Static</td>
<td rowspan="6" style="vertical-align: middle;">Stream</td>
<td rowspan="6" style="vertical-align: middle;">Static</td>
<td style="vertical-align: middle;">Inner</td>
<td style="vertical-align: middle;">Supported, not stateful</td>
</tr>
Expand All @@ -1365,8 +1387,12 @@ regarding watermark delays and whether data will be dropped or not.
<td style="vertical-align: middle;">Supported, not stateful</td>
</tr>
<tr>
<td rowspan="5" style="vertical-align: middle;">Static</td>
<td rowspan="5" style="vertical-align: middle;">Stream</td>
<td style="vertical-align: middle;">Left Anti</td>
<td style="vertical-align: middle;">Supported, not stateful</td>
</tr>
<tr>
<td rowspan="6" style="vertical-align: middle;">Static</td>
<td rowspan="6" style="vertical-align: middle;">Stream</td>
<td style="vertical-align: middle;">Inner</td>
<td style="vertical-align: middle;">Supported, not stateful</td>
</tr>
Expand All @@ -1387,8 +1413,12 @@ regarding watermark delays and whether data will be dropped or not.
<td style="vertical-align: middle;">Not supported</td>
</tr>
<tr>
<td rowspan="5" style="vertical-align: middle;">Stream</td>
<td rowspan="5" style="vertical-align: middle;">Stream</td>
<td style="vertical-align: middle;">Left Anti</td>
<td style="vertical-align: middle;">Not supported</td>
</tr>
<tr>
<td rowspan="6" style="vertical-align: middle;">Stream</td>
<td rowspan="6" style="vertical-align: middle;">Stream</td>
<td style="vertical-align: middle;">Inner</td>
<td style="vertical-align: middle;">
Supported, optionally specify watermark on both sides +
Expand Down Expand Up @@ -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
</td>
</tr>
<tr>
<td style="vertical-align: middle;">Left Anti</td>
<td style="vertical-align: middle;">
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
</td>
</tr>
<tr>
<td></td>
<td></td>
Expand Down
19 changes: 14 additions & 5 deletions ...t/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
16 changes: 10 additions & 6 deletions ...st/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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,
Expand All @@ -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",
Expand All @@ -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(
Expand Down
Loading