[
https://issues.apache.org/jira/browse/SPARK-60376?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18128978#comment-18128978
]
Jungtaek Lim commented on SPARK-60376:
--------------------------------------
Going to submit a PR soon.
> V4 stream-stream joins can miss matches when range conditions reference
> another timestamp column
> ------------------------------------------------------------------------------------------------
>
> Key: SPARK-60376
> URL: https://issues.apache.org/jira/browse/SPARK-60376
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 4.3.0
> Reporter: Jungtaek Lim
> Priority: Blocker
> Labels: correctness
>
> A stream-stream join using state format V4 can silently omit valid matches
> when its time-range condition references a timestamp column other than the
> selected event-time column.
>
> For example, watermark {{leftTime}} and {{rightTime}} on the respective
> inputs, and consider:
> SELECT l.key, l.leftTime, r.rightTime
> FROM leftStream l JOIN rightStream r
> ON l.key = r.key
> AND r.rightTime BETWEEN l.leftTime AND l.expiry{{}}
>
> Use V4 join state and a 60-second watermark delay on both event-time columns.
> With timestamp values expressed in milliseconds since epoch, add these inputs
> in the same micro-batch: * Left: {{{}(key=1, leftTime=100000,
> expiry=110000){}}}.
> * Right: key {{1}} with {{rightTime}} values {{{}100000{}}}, {{{}100001{}}},
> {{{}100002{}}}, {{{}101000{}}}, {{{}105000{}}}, and {{{}110000{}}}.
> *Expected:* All six right-side rows match.
> *Actual:* The unmodified V4 implementation returns only the rows at
> {{100000}} and {{{}100001{}}}.
>
> The regression also reproduces with either side arriving first and across
> checkpoint restart.
> V4 matching scans use an index on the selected event-time column. The
> constant-offset extractor currently considers all input attributes, allowing
> a constraint on {{expiry}} to be interpreted as a bound on that event-time
> index. This can exclude valid candidate rows before the full join condition
> is evaluated.
>
> The proposed fix restricts matching-scan offset extraction to the selected
> event-time columns. If both safe offsets cannot be derived, it scans the rows
> for the join key and applies the full join condition. Existing fixed-interval
> extraction, state eviction, output-watermark calculation, and
> checkpoint/state layout remain unchanged.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
