Jungtaek Lim created SPARK-60378:
------------------------------------
Summary: V4 stream-stream joins can skip rows during state eviction
Key: SPARK-60378
URL: https://issues.apache.org/jira/browse/SPARK-60378
Project: Spark
Issue Type: Bug
Components: Structured Streaming
Affects Versions: 4.3.0
Reporter: Jungtaek Lim
Stream-stream joins using state format V4 can skip newly admitted rows during
state eviction. For outer joins, this can lose the null-extended output for
unmatched rows and retain those rows in state.
V4 uses the previous derived state cutoff as a lower bound for its
timestamp-index eviction scan. This cutoff can be ahead of the watermark used
to reject late input. A newly accepted row can therefore fall below the scan’s
lower bound even though it is already eligible for eviction.
For example, use a left outer stream-stream join with state format V4,
zero-second watermarks on both event-time columns, and this condition:
leftKey = rightKey AND leftTime > rightTime + INTERVAL 5 SECONDS
Disable no-data micro-batches and supply these input batches. Tuples are
{{{}(key, event time in seconds){}}}: # Left {{{}(9, 20){}}}, right {{{}(9,
10){}}}: emits their match.
# Right {{{}(2, 20){}}}: advances the watermark without producing output.
# Left {{{}(1, 12){}}}, right {{{}(2, 30){}}}: should emit {{{}(1, null, 12,
null){}}}, projecting {{{}(leftKey, rightKey, leftTime, rightTime){}}}.
For this condition, the left-state cutoff is the input watermark used for
eviction plus five seconds. Batch 2 uses a 10-second input watermark, giving a
15-second state cutoff. In batch 3, late-input filtering still uses 10 seconds,
while the current left-state cutoff is 25 seconds. The 12-second row is
accepted and eligible for finalization, but the eviction scan starts above the
previous 15-second cutoff and misses it.
*Expected:* Batch 3 emits the unmatched left row with null right-side values
and removes it from state.
*Actual:* Batch 3 emits no outer-null row and leaves the row in state.
The fix should retain the previous cutoff as the scan lower bound only when it
does not exceed the late-input watermark. Otherwise, eviction should scan from
the beginning of the timestamp index through the current cutoff. This issue is
specific to V4’s indexed eviction scan; V1–V3 do not use this lower-bound
optimization.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
