eason-yuchen-liu opened a new pull request, #59024:
URL: https://github.com/apache/spark/pull/59024
### What changes were proposed in this pull request?
The join state watermark, which is tracked in milliseconds, is converted
back to
microseconds at two sites:
1. The state eviction predicate built in
`WatermarkSupport.watermarkExpression`
(`ms * 1000`).
2. The time-interval range scan in `StreamingSymmetricHashJoinExec`, which
computes each scan bound as `eventTime + offset`.
For a join range condition with an astronomical interval, the resulting state
watermark can exceed the representable timestamp range. In that case `ms *
1000`
and `eventTime + offset` overflow `Long`:
- In the eviction predicate the overflow is silent and flips the comparison,
so
the operator can evict all state.
- In the range scan the overflow produces a wrong, too-narrow scan window, so
the join can miss matches.
This PR routes both conversions through saturating helpers added to
`WatermarkSupport`:
- `millisToMicrosSaturating(millis: Long): Long` clamps `millis * 1000` to
`Long.MaxValue` / `Long.MinValue` instead of wrapping.
- `addMicrosSaturating(a: Long, b: Long): Long` performs saturating addition
for
the `eventTime + offset` bound.
An out-of-range bound now saturates to the timestamp boundary (evict all /
evict
none / full scan) instead of wrapping. Values within the representable range
are
unchanged.
### Why are the changes needed?
Silent `Long` overflow when re-expanding a millisecond state watermark to
microseconds is a correctness bug for stream-stream joins with large range
conditions: it can either evict all state or narrow the range-scan window and
drop otherwise-matching rows. Saturating to the timestamp boundary preserves
the
intended semantics at the extremes.
### Does this PR introduce _any_ user-facing change?
No. Behavior only changes for state watermarks that fall outside the
representable timestamp range, where the previous behavior was an
overflow/wrap-around bug. Values within range are converted exactly as
before.
### How was this patch tested?
Added unit tests in `StreamingSymmetricHashJoinHelperSuite` covering both
helpers, including exactly-scalable boundary values and one-past-the-boundary
inputs that previously overflowed:
- `millisToMicrosSaturating saturates instead of overflowing`
- `addMicrosSaturating saturates instead of overflowing`
```
build/sbt 'sql/testOnly *StreamingSymmetricHashJoinHelperSuite'
...
Tests: succeeded 9, failed 0, canceled 0, ignored 0, pending 0
All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Opus 4.8)
--
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]
