[SPARK-58611][SS] Left anti stream-stream join - #57813
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.
…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.
|
Adding @HeartSaVioR for further review in stream-stream join state format area |
There was a problem hiding this comment.
I was trying to comment each lines but feel like talking about general direction is much better than that.
I see this to be implemented as a variance of LeftOuter, but the spec of LeftAnti is actually a mirror of LeftSemi, meaning so many optimizations we have made for LeftSemi can be applied to LeftAnti. For example, the row in the left side doesn't need to be stored as long as it finds "any" matched row in the right side. This is also applied when the new row in the right side is matched with the left side's row in the state store; we can just remove it from state store rather than updating the matched flag. We don't need to check with matched flag since we wouldn't leave the row which was matched in the state store.
It would give a lot of benefit if we start from LeftSemi, not LeftOuter. Still, LeftAnti should be a hybrid of LeftSemi and LeftOuter (since we need to produce the outer result as the main output), but for the matched rows, the optimization for LeftSemi can be applied to LeftAnti.
… semi Rework the left anti stream-stream join to be a hybrid of left semi and left outer, as suggested in review, rather than a variant of left outer. Left anti is the mirror of left semi: a left row that finds any match can never be anti output, so it can be dropped from state on match instead of being kept with a `matched` flag until eviction. This reuses the left semi optimizations: skip storing a left row that matches on arrival, and remove an already-stored left row via `getJoinedRowsAndRemoveMatched` when a later right row matches it. The right side is processed first, as for left semi. Only the eviction-time emission of the surviving (never-matched) left rows stays left-outer-shaped. As a result the left-side `matched` flag is no longer consulted at eviction (every survivor is unmatched by construction), so its v4 `skipUpdatingMatchedFlag` special-casing is dropped. Row outputs are unchanged; state size for matched left rows now matches left semi rather than left outer. Update the affected `assertNumStateRows` expectations accordingly.
Thanks, agreed. LeftSemi is the right framing. Reworked in
Only the eviction-time emission of surviving never-matched left rows stays left-outer-shaped. Since every survivor is unmatched by construction, the |
…mples Correct the watermark/state-cleanup explanation in the anti join section: for left anti it is the right side watermark that lets the engine decide a left row can no longer match and emit it, so the right watermark drives eviction and output of left side state, while the optional left watermark is what allows the right side state to be cleaned up. The previous text had this reversed. Also add leftAnti / left_anti to the supported join type comments in the stream-stream join example snippets across all language tabs.
The test asserted empty output throughout (the left row gets matched), so its "emits an unmatched left row" name was misleading. Rename it to describe what it verifies and note the positive emission case is covered by "windowed left anti join".
|
Codex found the issue which is actually an existing one but severe with outer-like joins... |
…eft anti stream-stream join A left anti stream-stream join emits its output (unmatched left rows) from left-state eviction, so it needs both the left state evicted and the right side late-filtered on the matching dimension. The generic watermark check guaranteed neither, so the analyzer accepted configs that silently produced wrong output, e.g. a right-watermark-only range join (left state never evicted). Tighten the LeftAnti analyzer check: the range path requires a left-side watermark; the equi-join path requires the watermark on the right join key at the eviction ordinal (new StreamingJoinHelper.isWatermarkOnRightEvictionJoinKey). Left outer/semi are unchanged, pending a separate config-gated fix. Also fix the docs and add analyzer/runtime coverage.
Fix in the latest FOLLOWUP, scoped to left anti (analyzer + docs):
Left outer/semi are untouched here, as you suggested, they need a SQL-config fallback + migration-guide note, so I'd do that as a separate change. Two related gaps are also pre-existing and shared across all stream-stream joins, so I left them for that follow-up: (1) same-batch matches when the eviction watermark runs ahead of the late-event watermark (SPARK-49829-style split), and (2) range-path column-level precision (which attributes the range constraints vs which is watermarked). |
|
Sorry to push you on more work than anticipated, but I think the best direction is to fix the case of left semi & left outer with SQL config as a kill switch, and have left anti on top of the fix. We have common code which has the issue and it's weird we workaround to left anti to not use common code and try to fix it only for left anti. (Actually Codex still found a couple issues and trying to fix the issue in left anti would be thrown out as we will fix the shared code anyway.) Could you please consider expanding your work to fix left semi & left outer as well? I can volunteer if you would like to keep the scope as left anti. Thanks! |
…ent check across left semi/outer/anti Generalize the LeftAnti-only watermark-placement check into a shared check for left semi/outer/anti. The equi-join path now requires both eviction join keys to be watermarked, and the range path requires the state watermark to derive from watermarked attributes on both sides. Left semi/outer are gated by the new config spark.sql.streaming.join.stricterWatermarkRequirements.enabled (default true, set false to restore the old behavior); left anti always enforces it. Also fixes the same-batch (SPARK-49829 split) case for left semi/anti: a non-late right row that is evicting in the batch is now stored so later same-batch left rows can match it, and the matched-row removal probe prunes the V4 state scan by timestamp. Updates the join docs, support matrix, and migration guide.
…new streaming join config spark.sql.streaming.join.stricterWatermarkRequirements.enabled only gates a streaming analyzer validation and never changes a resolved view/UDF/procedure plan, so it uses NOT_APPLICABLE. Required for SparkConfigBindingPolicySuite.
Thanks, expanded in the follow-up commit, no more left-anti workaround. The check now lives in shared code (
PTAL. |
…t-side probe The left-side probe of a left anti join only needs to know whether the right state holds any matching row. Add SymmetricHashJoinStateManager.existsJoinedRow, which short-circuits on the first match and does not update right-side matched flags that left anti never consults, and use it instead of draining getJoinedRows. Also clarify in the docs and analyzer tests that stream-static joins are not stateful and follow the general output-mode rules, so they are allowed in Update mode even when the corresponding stream-stream join is Append-only.
My bad if my proposal wasn't clear. I was thinking of having a new JIRA ticket to change the behavior for Left Semi & Left Outer (it warrants a new ticket due to breaking/behavioral change) and merging the fix (fixing common code in that PR), and getting back to this to scope to Left Anti only. I assume the code change itself won't be complex for the new PR as I expect the code change to be made here already. Could you please help doing it? Again I can volunteer to do that by myself if you mind. Please let me know. |
Thanks for clarifying. I created SPARK-58904 to change the behavior for Left Semi and Left Outer. I'll update this PR to include only the left anti on top of it.. |
|
#58238 (SPARK-58904) is the split-out shared fix from this PR's review discussion. It adds a common analyzer check, Once #58238 lands, I will rebase this PR on top of it and scope it down to left anti only, adding left anti as one more case in that shared check (always enforced, since it is newly supported) rather than the current left-anti-specific handling. |
4 participants
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)