Jungtaek Lim created SPARK-60376:
------------------------------------
Summary: 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
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]
