[
https://issues.apache.org/jira/browse/SPARK-60336?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18128658#comment-18128658
]
Jungtaek Lim commented on SPARK-60336:
--------------------------------------
Going to submit a PR for this soon.
> Repeated arithmetic terms in stream-stream join conditions can cause
> premature state eviction
> ---------------------------------------------------------------------------------------------
>
> Key: SPARK-60336
> URL: https://issues.apache.org/jira/browse/SPARK-60336
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 4.2.0, 4.3.0, 4.1.3, 4.0.4
> Reporter: Jungtaek Lim
> Priority: Critical
> Labels: correctness
>
> This seems to be an ancient bug from the introduction of stream-stream join.
> It's correctness but only happens in a way where users are writing their
> query in (slightly) uncommon way.
> Streaming join state watermark extraction can discard repeated arithmetic
> terms, compute an overly aggressive eviction cutoff, and silently lose valid
> join output.
> Reproduction:
> Configure a 10-second watermark on both timestamp columns and use this inner
> join condition:
>
> leftKey = rightKey ANDCAST(leftTime AS BIGINT) > CAST(rightTime AS BIGINT) -
> 2 - 2
>
> Input rows below are {{{}(key, event-time seconds){}}}; output rows are
> {{{}(key, left-time seconds, right-time seconds){}}}. # Add left {{(1, 8)}}
> and right {{{}(1, 9){}}}. The query emits {{{}(1, 8, 9){}}}.
> # Add left {{(2, 20)}} and right {{{}(3, 20){}}}. These keys do not match.
> Both watermarks advance to 10 seconds; allow state cleanup to run.
> # Add right {{{}(1, 11){}}}. This row is newer than the watermark and
> satisfies the join condition with the stored left row: {{{}8 > 11 - 2 - 2{}}}.
> Expected: the final step emits {{{}(1, 8, 11){}}}.
> Actual: the final step emits nothing because the left row was prematurely
> evicted.
> The correct left-state cutoff is {{10 - 2 - 2 = 6}} seconds.
> {{StreamingJoinHelper}} uses {{ExpressionSet}} for the arithmetic terms,
> which collapses the repeated {{-2}} terms and produces an 8-second cutoff.
> The join predicate retains both terms, but cleanup removes the left row at 8
> seconds.
> The extraction should preserve arithmetic term multiplicity. Attribute
> occurrences must also be counted before deduplication so expressions with
> repeated timestamp attributes are not incorrectly treated as constraints with
> a coefficient of one.
> An end-to-end regression reproduces the missing output with the original
> helper in all six tested configurations: state format V2 with HDFS/RocksDB,
> and V3/V4 with RocksDB changelog checkpointing off/on. The identical test
> passes in all six configurations with the proposed fix.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
