[
https://issues.apache.org/jira/browse/SPARK-60378?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim updated SPARK-60378:
---------------------------------
Labels: correctness (was: )
> 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
> Priority: Blocker
> Labels: correctness
>
> 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]
