You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ganeshashree (via GitHub)" <gi...@apache.org> on 2026/08/06 05:02:21 UTC
[PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
ganeshashree opened a new pull request, #57813:
URL: https://github.com/apache/spark/pull/57813
### What changes were proposed in this pull request?
This adds `LeftAnti` support to stream-stream join, which previously failed at analysis time:
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:
the moment a left row matches, it can be emitted. "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 — the opposite of SPARK-56384, which allowed Update for Inner/LeftSemi precisely because
those have no early-firing concern).
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, as it is not among the join types planned onto this operator.
This is a natural follow-up to SPARK-32862 (left semi) and SPARK-32863 (full outer) under the
SPARK-32883 umbrella.
### 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, alerts without acknowledgement — 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.
`LEFT ANTI` is supported in stream-stream joins by Flink. Note stream-static `LEFT ANTI` already
worked in Spark 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. This is a new capability compared to released Spark versions; no existing behaviour changes.
Before:
```scala
left.join(right, expr("leftKey = rightKey AND ..."), "left_anti").writeStream...
// AnalysisException: LeftAnti joins with a streaming DataFrame/Dataset
// on the right are not supported
After: the query runs, and each left row with no matching right row is emitted once the
watermark guarantees no future match can arrive.
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 query restarts, including state row counts;
- an unmatched left row only being emitted once the watermark passes it;
- a left row matched in a later batch never being emitted (the main correctness risk);
- pre-join-filter exclusion on the left side and on the right side;
- Update output mode being rejected.
UnsupportedOperationsSuite gains LeftAnti coverage for the stream-stream watermark
conditions and for Update / Complete output mode rejection.
Verified locally:
- UnsupportedOperationsSuite — 226/226 pass
- new left anti suite + existing left semi suite — 32/32 pass
- left anti (VCF) + left outer + full outer suites — 64/64 pass
all on both RocksDBStateStoreProvider and HDFSBackedStateStoreProvider. The left semi, left
outer and full outer suites are included deliberately as regression coverage for the two shared
code paths this PR touches.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree closed pull request #57813: [SPARK-58611][SS] Left anti stream-stream join
URL: https://github.com/apache/spark/pull/57813
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5477640450
Closing this PR for now because there are existing issues with watermark placement in the common code. If we simply support Left anti without addressing them, those issues would carry over to Left anti. Addressing the existing issues requires an SS expert who has worked on or reviewed the existing work. I would be happy to contribute to Left anti support once the existing issues are resolved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5243652215
> 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.
Thanks, agreed. LeftSemi is the right framing. Reworked in `772c64e` as a hybrid of left semi and left outer:
- left row that matches on arrival isn't stored;
- already-stored left row matched by a later right row is removed from state (`getJoinedRowsAndRemoveMatched`), not flag-flipped;
- right side processed first, as in LeftSemi.
Only the eviction-time emission of surviving never-matched left rows stays left-outer-shaped. Since every survivor is unmatched by construction, the `matched` flag is no longer read at eviction, so I dropped the v4 `skipUpdatingMatchedFlag` special-casing too. Outputs unchanged; state size for matched left rows now matches LeftSemi. Updated `assertNumStateRows` accordingly. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5202467963
Adding @HeartSaVioR for further review in stream-stream join state format area
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5391302615
#58238 (SPARK-58904) is the split-out shared fix from this PR's review discussion. It adds a common analyzer check, `checkStreamStreamJoinWatermarkPlacement`, that enforces correct watermark placement for stream-stream left semi and left outer joins: the left-side state must be evictable (the equi eviction key must be watermarked, or the range bound must sit between watermarked attributes on both sides), and for left outer both eviction-ordinal join keys must be watermarked so a late row cannot invalidate an already-emitted unmatched row. It is gated by a kill switch, `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` (default true).
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5328164000
> Codex found the issue which is actually an existing one but severe with outer-like joins... If we have only right side watermark (as we documented left side watermark is optional), it actually does not build the predicate to evict the state on the left side. For outer-like joins e.g. left outer and left anti, this won't produce unmatched outer results. Can you check with the above as constructing tests and confirming the behavior, and if Codex is right about it, block analyzer to allow only right watermark to be present, and also update the doc? We should make a fix for left semi and left outer as well, but for them we may need a SQL config to fall back to old behavior, and update the migration guide doc.
Fix in the latest FOLLOWUP, scoped to left anti (analyzer + docs):
- Range-condition path: now requires a watermark on the left side too (right-only is rejected), which lets the left state be evicted.
- Equi-join path: requires the watermark on the right join key at the ordinal used for eviction (isWatermarkOnRightEvictionJoinKey), so the right side is actually late-filtered on the matching dimension. This also rejects a left-only key watermark, a watermark on an unrelated right column, and a mismatched-ordinal composite key, all of which would otherwise let a late right row invalidate an already-emitted anti-row.
- Docs/matrix updated, plus the immediate-emission caveat for left rows failing a deterministic left-only pre-join filter.
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5323410873
Codex found the issue which is actually an existing one but severe with outer-like joins...
If we have only right side watermark (as we documented left side watermark is optional), it actually does not build the predicate to evict the state on the left side. For outer-like joins e.g. left outer and left anti, this won't produce unmatched outer results.
Can you check with the above as constructing tests and confirming the behavior, and if Codex is right about it, block analyzer to allow only right watermark to be present, and also update the doc? We should make a fix for left semi and left outer as well, but for them we may need a SQL config to fall back to old behavior, and update the migration guide doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5335211252
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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5369017922
> > 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.
>
> 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..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5351085288
> 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.
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58611][SS] Left anti stream-stream join [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #57813:
URL: https://github.com/apache/spark/pull/57813#issuecomment-5338341767
> 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!
Thanks, expanded in the follow-up commit, no more left-anti workaround. The check now lives in shared code (`checkStreamStreamJoinWatermarkPlacement`) across left semi/outer/anti:
- Left semi/outer are gated by a kill switch, `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` (default true); left anti sits on top and always enforces it.
- Equi path now requires both eviction keys watermarked; range path requires the bound between watermarked attributes on both sides.
PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org