[jira] [Resolved] (SPARK-60376) V4 stream-stream joins can miss matches when range conditions reference another timestamp column

["Jungtaek Lim (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-60376?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Jungtaek Lim resolved SPARK-60376.
----------------------------------
    Fix Version/s: 4.3.0
       Resolution: Fixed

Issue resolved by pull request 59628
[https://github.com/apache/spark/pull/59628]

> 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
>            Assignee: Jungtaek Lim
>            Priority: Blocker
>              Labels: correctness, pull-request-available
>             Fix For: 4.3.0
>
>
> 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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org