(spark) branch branch-4.x updated: [SPARK-60336][SS] Preserve repeated arithmetic terms in streaming join watermarks

[ka...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new e92b239b6088 [SPARK-60336][SS] Preserve repeated arithmetic terms in streaming join watermarks
e92b239b6088 is described below

commit e92b239b608890efefe5a2d2a9b1ecd1c120c3c3
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Tue Oct 20 08:43:34 2026 +0900

    [SPARK-60336][SS] Preserve repeated arithmetic terms in streaming join watermarks
    
    ### 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)
    
    Closes #59594 from HeartSaVioR/jungtaek-lim_data/fix-streaming-join-repeated-terms.
    
    Authored-by: Jungtaek Lim <ka...@gmail.com>
    Signed-off-by: Jungtaek Lim <ka...@gmail.com>
    (cherry picked from commit 0d31a7d06cdcbe28a3093527ea0db0f0b82ba764)
    Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
 .../catalyst/analysis/StreamingJoinHelper.scala    | 10 ++++----
 .../analysis/StreamingJoinHelperSuite.scala        | 25 ++++++++++++++++++-
 .../spark/sql/streaming/StreamingJoinSuite.scala   | 29 ++++++++++++++++++++++
 3 files changed, 58 insertions(+), 6 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
index c4549a189e8e..7062ced703e8 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
@@ -134,13 +134,12 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
       attributesWithEventWatermark: AttributeSet,
       eventWatermark: Option[Long]): Option[Long] = {
 
-    val attributesInCondition = AttributeSet(
+    val attributesInCondition =
       leftExpr.collect { case a: AttributeReference => a } ++
       rightExpr.collect { case a: AttributeReference => a }
-    )
     if (attributesInCondition.count(attributesToFindStateWatermarkFor.contains) > 1 ||
         attributesInCondition.count(attributesWithEventWatermark.contains) > 1) {
-      // If more than attributes present in condition from one side, then it cannot be solved
+      // Multiple attribute occurrences on either side require coefficient handling.
       return None
     }
 
@@ -157,7 +156,8 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
 
     // Canonicalization step 2: extract commutative terms
     //    rightTime-with-watermark, c1, -leftTime, -c2
-    val terms = ExpressionSet(collectTerms(allOnLeftExpr))
+    // Equal terms still contribute separately to the arithmetic sum.
+    val terms = collectTerms(allOnLeftExpr)
     logDebug("Terms extracted from join condition:\n\t" + terms.mkString("\n\t"))
 
     // Find the term that has leftTime (i.e. the one present in attributesToFindConstraintFor
@@ -190,7 +190,7 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
     // from the other terms. That is,
     // rightTime-with-watermark, c1, -c2  =>  watermark, c1, -c2  =>  watermark + c1 + (-c2)
     logDebug(s"Constraint term from join condition:\t$constraintTerm")
-    val exprWithWatermarkSubstituted = (terms - constraintTerm).map { term =>
+    val exprWithWatermarkSubstituted = terms.filterNot(_ eq constraintTerm).map { term =>
       term.transform {
         case a @ AttributeReference(_, _, _, metadata)
           if attributesWithEventWatermark.contains(a) && metadata.contains(delayKey) =>
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala
index aaaeb9bb5118..d25e231e0fcf 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala
@@ -17,14 +17,37 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet}
 import org.apache.spark.sql.catalyst.optimizer.SimpleTestOptimizer
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, Filter, LeafNode, Statistics}
-import org.apache.spark.sql.types.{IntegerType, MetadataBuilder, TimestampType}
+import org.apache.spark.sql.types.{IntegerType, LongType, MetadataBuilder, TimestampType}
 
 class StreamingJoinHelperSuite extends AnalysisTest {
 
+  test("state watermark extraction preserves arithmetic term multiplicity") {
+    val leftTime = AttributeReference("leftTime", TimestampType)()
+    val metadata = new MetadataBuilder().putLong(EventTimeWatermark.delayKey, 1000).build()
+    val rightTime = AttributeReference("rightTime", TimestampType, metadata = metadata)()
+    val left = leftTime.cast(LongType)
+    val right = rightTime.cast(LongType)
+
+    val cases = Seq(
+      (left > right + 2 + 2, Some(14000L)),
+      (left > right - 2 - 2, Some(6000L)),
+      (left + left > right, None),
+      (left > right + right, None),
+      (left > right - right, None))
+    cases.foreach { case (condition, expected) =>
+      withClue(condition.sql) {
+        assert(StreamingJoinHelper.getStateValueWatermark(
+          AttributeSet(Seq(leftTime)), AttributeSet(Seq(rightTime)),
+          Some(condition), Some(10000L)) === expected)
+      }
+    }
+  }
+
   test("extract watermark from time condition") {
     val attributesToFindConstraintFor = Seq(
       AttributeReference("leftTime", TimestampType)(),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala
index 9b91dea19606..fed678c77585 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala
@@ -461,6 +461,35 @@ abstract class StreamingInnerJoinBase extends StreamingJoinSuite {
     )
   }
 
+  test("stream stream inner join preserves repeated subtraction terms") {
+    val leftInput = MemoryStream[(Int, Int)]
+    val rightInput = MemoryStream[(Int, Int)]
+    val left = leftInput.toDF().toDF("leftKey", "time")
+      .select($"leftKey", timestamp_seconds($"time").as("leftTime"))
+      .withWatermark("leftTime", "10 seconds")
+    val right = rightInput.toDF().toDF("rightKey", "time")
+      .select($"rightKey", timestamp_seconds($"time").as("rightTime"))
+      .withWatermark("rightTime", "10 seconds")
+    val joined = left.join(right, expr("leftKey = rightKey AND " +
+      "CAST(leftTime AS BIGINT) > CAST(rightTime AS BIGINT) - 2 - 2"))
+      .select($"leftKey", $"leftTime".cast("int"), $"rightTime".cast("int"))
+
+    testStream(joined)(
+      MultiAddData(leftInput, (1, 8))(rightInput, (1, 9)),
+      CheckNewAnswer((1, 8, 9)),
+      // Advance both input watermarks to 10s using keys that cannot match.
+      MultiAddData(leftInput, (2, 20))(rightInput, (3, 20)),
+      CheckNewAnswer(),
+      Execute { query =>
+        assert(query.lastProgress.eventTime.get("watermark") == "1970-01-01T00:00:10.000Z")
+      },
+      // The state cutoff is 10 - 2 - 2 = 6s. Deduplicating the constants gives 8s,
+      // evicting the left row even though it can still match an on-time right row.
+      AddData(rightInput, (1, 11)),
+      CheckNewAnswer((1, 8, 11))
+    )
+  }
+
   testWithAppendAndUpdate("stream stream inner join with time range - with watermark" +
       " - two side conditions") { outputMode =>
     import org.apache.spark.sql.functions._


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org