[SPARK-58611][SS] Left anti stream-stream join - #57813
Open
ganeshashree wants to merge 2 commits into
Open
Conversation
### What changes were proposed in this pull request? This adds LeftAnti support to stream-stream join, which previously failed at analysis time with "LeftAnti joins with a streaming DataFrame/Dataset on the right are not supported". Unlike left semi, left anti cannot emit while joining: a semi match is positively determined, whereas "no match exists" is only decidable once the watermark guarantees no future right row can match. Left anti is therefore implemented on the eviction path, reusing the existing left outer plumbing in `StreamingSymmetricHashJoinExec`: * nothing is emitted when a left row matches; * at left-side state eviction, rows whose `matched` flag is false are emitted as bare left rows (left outer emits them joined with nulls instead); * a matched left row stays in state carrying `matched = true` so that it is suppressed at eviction time, rather than being dropped from state early the way left semi does; * a left row that fails the pre-join filter can never match, so it is emitted immediately without being added to state. Two details worth calling out for review: * `AddingProcessedRowToStateCompletionIterator` infers the persisted `matched` flag from whether the output iterator is non-empty. Left anti emits nothing on a match, so the match status is now passed explicitly via a new optional `matchedOverride` parameter. Without it every left row would be stored as unmatched and matched rows would be wrongly emitted at eviction. The parameter defaults to the previous behaviour, so the other join types are unaffected. * The joined-row iterator is drained fully rather than short-circuited on the first match, because `getJoinedRows` sets the `matched` flag on the other side's rows lazily as they are produced. Stopping early would leave some matched left rows flagged as unmatched. Requirements, mirroring left outer: a watermark on the right side plus time constraints are mandatory, and Append is the only supported output mode (Update would have to emit rows before the watermark can rule out a future match, and such a row could be invalidated by a later batch). No new state format version is needed -- the `matched` flag already persisted by v2/v3/v4 is exactly the required signal, so existing checkpoints need no migration. RightAnti remains out of scope. ### Why are the changes needed? Stream-stream join supported Inner, LeftOuter, RightOuter, FullOuter and LeftSemi. LeftSemi was added in SPARK-32862 but its complement was never done, leaving the common "find left rows with no match on the right" pattern -- impressions without clicks, orders without shipments, sessions without conversion -- without a native streaming implementation. Users work around it with NOT IN / NOT EXISTS rewrites or hand-written transformWithState logic, both more expensive and easy to get subtly wrong. Note stream-static LEFT ANTI already worked when only the left side was streaming; only a streaming right side was rejected, so the gap was specifically stream-stream. ### Does this PR introduce _any_ user-facing change? Yes. `LEFT ANTI` stream-stream joins are now supported in Append output mode, given a watermark on the right side and time constraints. Queries which previously failed at analysis time now run. No existing behaviour changes. Left anti buffers every left row until eviction, whereas left semi drops matched left rows from state eagerly. This is inherent -- a matched row must be retained so that it can be suppressed at eviction rather than emitted -- so state size for left anti is comparable to left outer, not to left semi. The join support matrix in the Structured Streaming guide gains Left Anti rows for stream-static, static-stream and stream-stream, plus an "Anti Joins with Watermarking" section. ### How was this patch tested? New `StreamingLeftAntiJoinSuite` with virtual-column-family and non-VCF variants, covering windowed anti join across restarts, an unmatched row only being emitted once the watermark passes it, a row matched in a later batch never being emitted, pre-join-filter exclusion on both sides, and Update output mode being rejected. `UnsupportedOperationsSuite` gains LeftAnti coverage for the watermark conditions and for Update/Complete mode rejection. Verified locally: `UnsupportedOperationsSuite` 226/226 pass; the new left anti suite plus the existing left semi suite 32/32 pass on both RocksDB and HDFS-backed state store providers.
uros-b
approved these changes
Aug 6, 2026
…nti join coverage
* Runtime coverage missed state format v4. `skipUpdatingMatchedFlag` in
`StreamingSymmetricHashJoinExec` is gated on `stateFormatVersion == 4`, so left anti
takes a distinct path there, but the anti suites only covered v2/v3. Split
`StreamingLeftAntiJoinSuite` into `StreamingLeftAntiJoinBase` plus a subclass holding
the V1-V3-only tests -- mirroring the existing `StreamingLeftSemiJoinBase` /
`StreamingLeftSemiJoinSuite` split -- and add `StreamingLeftAntiJoinV4Suite` alongside
the other V4 suites so it inherits the runtime tests.
* Range-condition joins use the state value watermark path rather than the state key
watermark path exercised by the windowed anti tests. Add a
`setupJoinWithRangeCondition("left_anti")` test covering it, and extend that helper's
projection to treat `left_anti` like `left_semi` (left columns only).
* The guide said an anti join "must specify watermark on right + time constraints", but
the analyzer routes LeftAnti through the shared `checkForStreamStreamJoinWatermark`, so
a watermarked column in the equality join keys on either side is also accepted. Document
both ways of expressing the event-time constraint.
No change to the implementation -- tests and documentation only.
Member
|
Adding @HeartSaVioR for further review in stream-stream join state format area |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Adds
LeftAntisupport to stream-stream join, which previously failed at analysis time with "LeftAnti joins with a streaming DataFrame/Dataset on the right are not supported".Why are the changes needed?
Left anti is the complement of left semi (SPARK-32862) and the last missing join type in the stream-stream matrix -- "left rows with no match on the right", e.g. impressions without clicks.
LEFT ANTIalready works in batch and stream-static, so this is batch/streaming parity.Does this PR introduce any user-facing change?
Yes.
LEFT ANTIstream-stream joins now run in Append mode given a watermark and event-time constraints; queries that previously failed at analysis time now work. No existing behaviour changes. State size is comparable to left outer, not left semi, since every left row is buffered until eviction. The guide's join matrix and a new "Anti Joins with Watermarking" section are updated.How was this patch tested?
New
StreamingLeftAntiJoinSuitewith three concrete suites covering state formats v2, v3 and v4 (v4 matters becauseskipUpdatingMatchedFlagis gated on it). Covers windowed joins across restarts, watermark-gated emission, matched rows never being emitted, pre-join-filter exclusion, the range-condition (state value watermark) path, and Update mode rejection.UnsupportedOperationsSuitegains LeftAnti watermark and output-mode coverage.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)