HeartSaVioR opened a new pull request, #59594: URL: https://github.com/apache/spark/pull/59594
### What changes were proposed in this pull request? For [SPARK-60336](https://issues.apache.org/jira/browse/SPARK-60336), preserve repeated arithmetic terms when extracting a streaming join state watermark. Count attribute occurrences before deduplication and remove only the selected state term when evaluating the remaining expression. ### Why are the changes needed? `ExpressionSet` treats repeated arithmetic terms as duplicates. This can advance the state eviction cutoff too far and silently lose valid join output. Repeated timestamp attributes can also be mistaken for a unit-coefficient constraint. The end-to-end regression uses an inner join with 10-second watermarks on both timestamp columns and this condition: ```sql leftKey = rightKey AND CAST(leftTime AS BIGINT) > CAST(rightTime AS BIGINT) - 2 - 2 ``` Input tuples are `(key, event-time seconds)`; output tuples are `(key, left-time seconds, right-time seconds)`. | Step | Left input | Right input | Expected new output | | --- | --- | --- | --- | | Establish a matching row in state | `(1, 8)` | `(1, 9)` | `(1, 8, 9)` | | Advance both watermarks to 10 seconds and run cleanup | `(2, 20)` | `(3, 20)` | None | | Send another on-time match | None | `(1, 11)` | `(1, 8, 11)` | The correct left-state cutoff is `10 - 2 - 2 = 6` seconds. The old helper deduplicates the two `-2` terms and computes 8 seconds, evicting the stored left row. The final right row is above the watermark and satisfies `8 > 11 - 2 - 2`, but the old code returns no match. The fix retains the left row and emits `(1, 8, 11)`. ### Does this PR introduce _any_ user-facing change? Yes. Repeated constant terms contribute their full value to state eviction cutoffs, preventing premature eviction and missing join output. Expressions with repeated timestamp attributes are conservatively rejected by the watermark extractor. Depending on join type and other constraints, this can disable cleanup or cause an outer/semi join to be rejected during analysis. Previously lost state/output is not repaired. ### How was this patch tested? Added helper coverage for repeated positive/negative constants and repeated attributes on both inputs, plus an end-to-end regression in `StreamingInnerJoinBase` that verifies the initial match, watermark advancement, and the later match after cleanup. ```text build/sbt 'catalyst/testOnly *StreamingJoinHelperSuite *UnsupportedOperationsSuite' build/sbt 'sql/testOnly *StreamingInnerJoin*Suite -- -z "repeated subtraction terms"' build/sbt 'sql/testOnly *StreamingInnerJoin*Suite -- -z "time range"' ``` - Helper/analyzer suites: 236 tests passed. - End-to-end regression: all 6 configurations fail with the original helper, missing `(1, 8, 11)`, and pass with this fix. The matrix covers state format V2 with HDFS/RocksDB and V3/V4 with RocksDB changelog checkpointing disabled/enabled. - Existing time-range cases: 24 tests passed. - Catalyst Compile/Test and SQL Test scalastyle: zero errors and warnings. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
