Jungtaek Lim created SPARK-60336:
------------------------------------
Summary: 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.0.4, 4.1.3, 4.2.0, 4.3.0
Reporter: Jungtaek Lim
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]
