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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
