[SPARK-58611][SS] Left anti stream-stream join

docs/streaming/apis-on-dataframes-and-datasets.md

@@ -1230,7 +1230,7 @@ impressionsWithWatermark.join(
     clickTime >= impressionTime AND
     clickTime <= impressionTime + interval 1 hour
     """),
-  "leftOuter"                 # can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
+  "leftOuter"     # can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi", "leftAnti"
 )
 
 {% endhighlight %}
@@ -1248,7 +1248,7 @@ impressionsWithWatermark.join(
     clickTime >= impressionTime AND
     clickTime <= impressionTime + interval 1 hour
     """),
-  joinType = "leftOuter"      // can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
+  joinType = "leftOuter"  // "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi", "leftAnti"
  )
 
 {% endhighlight %}
@@ -1264,7 +1264,7 @@ impressionsWithWatermark.join(
     "clickAdId = impressionAdId AND " +
     "clickTime >= impressionTime AND " +
     "clickTime <= impressionTime + interval 1 hour "),
-  "leftOuter"                 // can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
+  "leftOuter"     // can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi", "leftAnti"
 );
 
 {% endhighlight %}
@@ -1283,7 +1283,7 @@ joined <- join(
       "clickAdId = impressionAdId AND",
       "clickTime >= impressionTime AND",
       "clickTime <= impressionTime + interval 1 hour"),
-  "left_outer"                 # can be "inner", "left_outer", "right_outer", "full_outer", "left_semi"
+  "left_outer"  # "inner", "left_outer", "right_outer", "full_outer", "left_semi", "left_anti"
 ))
 
 {% endhighlight %}
@@ -1293,6 +1293,20 @@ joined <- join(
 </div>
 
 
+For a **left outer** join, the surviving unmatched left rows are emitted when the left-side state is
+evicted, and an already-emitted `NULL`-extended row can be invalidated by a right row that arrives
+later. Correct results therefore require the watermark to be placed so that (1) the left state is
+actually evicted and (2) both sides are late-filtered on the dimension that bounds matching.
+Concretely: for an equality join on a watermarked event-time key, that key must be watermarked on
+**both** sides; for a time range condition, the range bound must relate watermarked event-time
+columns from both sides. The recommended, always-correct configuration -- watermarking both sides on
+the event-time column used by the join, as in the example above -- satisfies both. Configurations
+that leave the left state un-evicted (a watermark on only the right side of a range condition) or
+leave either side unfiltered on the eviction key (a watermark on only one equality join key) are
+rejected at analysis time. To restore the previous, looser behavior, set
+`spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`; note that the looser
+behavior can silently produce incorrect (missing) outer results.
+
 ###### Semantic Guarantees of Stream-stream Outer Joins with Watermarking
 Outer joins have the same guarantees as [inner joins](#semantic-guarantees-of-stream-stream-inner-joins-with-watermarking)
 regarding watermark delays and whether data will be dropped or not.
@@ -1318,10 +1332,51 @@ constraints must be specified for semi join. This is to evict unmatched input ro
 the engine must know when an input row on left side is not going to match with anything on right
 side in future.
 
+As with a left outer join, the left state must be evicted so that never-matched left rows do not
+accumulate: for an equality join the watermarked join key evicts the left key state; for a time
+range condition the range bound must relate watermarked event-time columns from both sides. Unlike
+outer and anti joins, a semi join has no additional late-filtering requirement, because it emits a
+row on match rather than at eviction, so there is no already-emitted row for a late right row to
+invalidate. A range condition watermarked only on one side is rejected at analysis time; to restore
+the previous, looser behavior (which leaves the left state unbounded), set
+`spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`.
+
 ###### Semantic Guarantees of Stream-stream Semi Joins with Watermarking
 Semi joins have the same guarantees as [inner joins](#semantic-guarantees-of-stream-stream-inner-joins-with-watermarking)
 regarding watermark delays and whether data will be dropped or not.
 
+##### Anti Joins with Watermarking
+An anti join returns values from the left side of the relation that has no match with the right.
+It is also referred to as a left anti join. As with semi joins, watermarking and event-time
+constraints must be specified for an anti join: since a row is emitted precisely because it has
+*no* match, the engine generally has to wait until the watermark guarantees that no matching row
+can arrive on the right side in future before it can emit the row. (The exception is a left row
+that fails a deterministic left-side-only predicate in the join condition: it can never match any
+right row, so it is emitted immediately without waiting for the watermark.)
+
+Like a left outer join, an anti join has stricter watermark requirements, because two independent
+things must both hold. First, both sides must be late-filtered on the matching dimension, so rows
+that arrive too late to matter are dropped rather than processed; without this a late row could
+match a left row that has *already* been emitted as an anti row, silently corrupting the result.
+Second, the left state must be evicted, since eviction is what emits the surviving unmatched left
+rows. Concretely: for an equality join on a watermarked event-time key, that key must be
+watermarked on **both** sides; for a time range condition (for example `leftTime BETWEEN
+rightTime - INTERVAL 1 HOUR AND rightTime`), the range bound must relate watermarked event-time
+columns from both sides. The recommended, always-correct configuration is to watermark both sides on
+the event-time column used by the join. Configurations that leave either side unfiltered on the
+eviction key (for example a watermark on only one equality join key) or leave the left state never
+evicted (a watermark on only one side of a range condition) are rejected at analysis time.
+
+Note that anti join is only supported in Append output mode. Update mode would have to emit rows
+early, before the watermark can rule out a future match, and such a row could be invalidated by a
+later batch.
+
+###### Semantic Guarantees of Stream-stream Anti Joins with Watermarking
+Anti joins have the same guarantees regarding watermark delays and whether data will be dropped as
+[outer joins](#outer-joins-with-watermarking), because surviving unmatched rows are likewise only
+emitted once the watermark has passed them (the sole exception being left rows that fail a
+deterministic left-side-only predicate, which are emitted immediately as noted above).
+
 ##### Support matrix for joins in streaming queries
 
 <table>
@@ -1343,8 +1398,8 @@ regarding watermark delays and whether data will be dropped or not.
       </td>
   </tr>
   <tr>
-    <td rowspan="5" style="vertical-align: middle;">Stream</td>
-    <td rowspan="5" style="vertical-align: middle;">Static</td>
+    <td rowspan="6" style="vertical-align: middle;">Stream</td>
+    <td rowspan="6" style="vertical-align: middle;">Static</td>
     <td style="vertical-align: middle;">Inner</td>
     <td style="vertical-align: middle;">Supported, not stateful</td>
   </tr>
@@ -1365,8 +1420,12 @@ regarding watermark delays and whether data will be dropped or not.
     <td style="vertical-align: middle;">Supported, not stateful</td>
   </tr>
   <tr>
-    <td rowspan="5" style="vertical-align: middle;">Static</td>
-    <td rowspan="5" style="vertical-align: middle;">Stream</td>
+    <td style="vertical-align: middle;">Left Anti</td>
+    <td style="vertical-align: middle;">Supported, not stateful</td>
+  </tr>
+  <tr>
+    <td rowspan="6" style="vertical-align: middle;">Static</td>
+    <td rowspan="6" style="vertical-align: middle;">Stream</td>
     <td style="vertical-align: middle;">Inner</td>
     <td style="vertical-align: middle;">Supported, not stateful</td>
   </tr>
@@ -1387,8 +1446,12 @@ regarding watermark delays and whether data will be dropped or not.
     <td style="vertical-align: middle;">Not supported</td>
   </tr>
   <tr>
-    <td rowspan="5" style="vertical-align: middle;">Stream</td>
-    <td rowspan="5" style="vertical-align: middle;">Stream</td>
+    <td style="vertical-align: middle;">Left Anti</td>
+    <td style="vertical-align: middle;">Not supported</td>
+  </tr>
+  <tr>
+    <td rowspan="6" style="vertical-align: middle;">Stream</td>
+    <td rowspan="6" style="vertical-align: middle;">Stream</td>
     <td style="vertical-align: middle;">Inner</td>
     <td style="vertical-align: middle;">
       Supported, optionally specify watermark on both sides +
@@ -1398,8 +1461,9 @@ regarding watermark delays and whether data will be dropped or not.
   <tr>
     <td style="vertical-align: middle;">Left Outer</td>
     <td style="vertical-align: middle;">
-      Conditionally supported, must specify watermark on right + time constraints for correct
-      results, optionally specify watermark on left for all state cleanup
+      Conditionally supported. For equality-key joins, watermark both sides of the join key used for
+      eviction. For range-condition joins, the range bound must use watermarked columns from both
+      sides.
     </td>
   </tr>
   <tr>
@@ -1419,8 +1483,16 @@ regarding watermark delays and whether data will be dropped or not.
   <tr>
     <td style="vertical-align: middle;">Left Semi</td>
     <td style="vertical-align: middle;">
-      Conditionally supported, must specify watermark on right + time constraints for correct
-      results, optionally specify watermark on left for all state cleanup
+      Conditionally supported. Equality-key joins require a watermark on a join key for state
+      cleanup. Range-condition joins require watermarked range-bound columns from both sides.
+    </td>
+  </tr>
+  <tr>
+    <td style="vertical-align: middle;">Left Anti</td>
+    <td style="vertical-align: middle;">
+      Conditionally supported. For equality-key joins, watermark both sides of the join key used for
+      eviction. For range-condition joins, the range bound must use watermarked columns from both
+      sides. Append output mode only.
     </td>
   </tr>
   <tr>
@@ -1435,7 +1507,10 @@ Additional details on supported joins:
 
 - Joins can be cascaded, that is, you can do `df1.join(df2, ...).join(df3, ...).join(df4, ....)`.
 
-- As of Spark 2.4, you can use joins only when the query is in Append output mode. Other output modes are not yet supported.
+- For stream-stream joins, inner and left semi joins support Append and Update output modes.
+  Stream-stream left outer, right outer, full outer, and left anti joins support Append output mode
+  only. Complete output mode is not supported for stream-stream joins. Supported stream-static joins
+  are not stateful and follow the general output-mode rules for non-aggregation queries.
 
 - You cannot use mapGroupsWithState and flatMapGroupsWithState before and after joins.
 
@@ -1512,7 +1587,8 @@ joined = impressionsWithWatermark.join(
     clickTime >= impressionTime AND
     clickTime <= impressionTime + interval 1 hour
     """),
-  "leftOuter"                 # can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
+  "leftOuter"     # "inner", "leftOuter", "rightOuter", "fullOuter" (not "leftSemi"/"leftAnti":
+                  # they output left columns only, which this aggregation on click* does not have)
 )
 
 joined.groupBy(
@@ -1534,7 +1610,9 @@ val joined = impressionsWithWatermark.join(
     clickTime >= impressionTime AND
     clickTime <= impressionTime + interval 1 hour
   """),
-  joinType = "leftOuter"      // can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
+  // "inner", "leftOuter", "rightOuter", "fullOuter" (not "leftSemi"/"leftAnti": they output left
+  // columns only, which this aggregation on click* does not have)
+  joinType = "leftOuter"
 )
 
 joined
@@ -1553,7 +1631,9 @@ Dataset<Row> joined = impressionsWithWatermark.join(
     "clickAdId = impressionAdId AND " +
     "clickTime >= impressionTime AND " +
     "clickTime <= impressionTime + interval 1 hour "),
-  "leftOuter"                 // can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
+  // "inner", "leftOuter", "rightOuter", "fullOuter" (not "leftSemi"/"leftAnti": they output left
+  // columns only, which this aggregation on click* does not have)
+  "leftOuter"
 );
 
 joined
@@ -2066,9 +2146,12 @@ Here is the compatibility matrix.
   </tr>
   <tr>
       <td colspan="2" style="vertical-align: middle;">Queries with <code>joins</code></td>
-      <td style="vertical-align: middle;">Append</td>
+      <td style="vertical-align: middle;">Append, Update</td>
       <td style="vertical-align: middle;">
-        Update and Complete mode not supported yet. See the
+        For stream-stream joins, Update mode is supported only for inner and left semi joins.
+        Supported stream-static joins are not stateful and follow the general output-mode rules for
+        non-aggregation queries. Complete mode is not supported.
+        See the
         <a href="#support-matrix-for-joins-in-streaming-queries">support matrix in the Join Operations section</a>
          for more details on what types of joins are supported.
       </td>

docs/streaming/ss-migration-guide.md

@@ -23,6 +23,10 @@ Note that this migration guide describes the items specific to Structured Stream
 Many items of SQL migration can be applied when migrating Structured Streaming to higher versions.
 Please refer [Migration Guide: SQL, Datasets and DataFrame](../sql-migration-guide.html).
 
+## Upgrading from Structured Streaming 4.3 to 4.4
+
+- Since Spark 4.4, stream-stream left semi and left outer joins enforce stricter watermark-placement requirements at analysis time, so that the left-side state their output (or bounded state size) depends on is actually evicted, and, for left outer, so that late rows cannot invalidate an already-emitted unmatched row. Configurations that previously ran but could silently produce incorrect results or unbounded state -- for example a range-condition join whose range bound is not between watermarked attributes on both sides, or a left outer equality join whose eviction key is not watermarked on both sides -- now fail with an `AnalysisException`. To restore the previous behavior for left semi and left outer joins, set `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`. Newly supported stream-stream left anti joins always use the stricter requirements. (See [SPARK-58611](https://issues.apache.org/jira/browse/SPARK-58611) for more details.)
+
 ## Upgrading from Structured Streaming 4.1 to 4.2
 
 - Since Spark 4.2, restarting a streaming query from a checkpoint whose metadata file is missing while the offset or commit logs contain data fails with `STREAMING_CHECKPOINT_MISSING_METADATA_FILE`, instead of silently generating a new query ID (which can duplicate data in exactly-once sinks). Restore the metadata file or use a new checkpoint location. To restore the previous behavior, set `spark.sql.streaming.checkpoint.verifyMetadataExists.enabled` to `false`. (See [SPARK-55058](https://issues.apache.org/jira/browse/SPARK-55058) for more details.)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala

@@ -34,6 +34,15 @@ import org.apache.spark.unsafe.types.CalendarInterval
  */
 object StreamingJoinHelper extends PredicateHelper with Logging {
 
+  private def isWatermarked(expression: Expression): Boolean = expression match {
+    case ne: NamedExpression => ne.metadata.contains(EventTimeWatermark.delayKey)
+    case _ => false
+  }
+
+  private def watermarkedAttributes(attributes: AttributeSet): AttributeSet = {
+    AttributeSet(attributes.filter(_.metadata.contains(delayKey)).toSeq)
+  }
+
   /**
    * Check the provided logical plan to see if its join keys contain a watermark attribute.
    *
@@ -51,6 +60,60 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
     }
   }
 
+  /**
+   * Whether both equality join keys at the state-key eviction ordinal are watermarked.
+   *
+   * This is required by outer-like equality joins (left outer and left anti). Eviction of left
+   * state must be aligned with late-event filtering on both sides: a right watermark drops late
+   * right rows after unmatched rows have been emitted, and a left watermark drops late left rows
+   * after the right state that could have matched them has been evicted.
+   *
+   * The eviction ordinal is chosen in the same way as
+   * StreamingSymmetricHashJoinHelper.findJoinKeyOrdinalForWatermark.
+   */
+  def isWatermarkOnBothEvictionJoinKeys(plan: LogicalPlan): Boolean = {
+    plan match {
+      case ExtractEquiJoinKeys(_, leftKeys, rightKeys, _, _, _, _, _) =>
+        joinKeyOrdinalForWatermark(leftKeys, rightKeys).exists { ordinal =>
+          ordinal < leftKeys.length && ordinal < rightKeys.length &&
+            isWatermarked(leftKeys(ordinal)) && isWatermarked(rightKeys(ordinal))
+        }
+      case _ => false
+    }
+  }
+
+  private def joinKeyOrdinalForWatermark(
+      leftKeys: Seq[Expression],
+      rightKeys: Seq[Expression]): Option[Int] = {
+    leftKeys.indexWhere(isWatermarked) match {
+      case i if i >= 0 => Some(i)
+      case _ =>
+        rightKeys.indexWhere(isWatermarked) match {
+          case i if i >= 0 => Some(i)
+          case _ => None
+        }
+    }
+  }
+
+  /**
+   * Like [[getStateValueWatermark]], but only succeeds when the state watermark is derived from
+   * watermarked attributes on both sides. This is useful for analysis-time validation of range
+   * conditions: the runtime value-watermark predicate is applied to the watermarked attribute on
+   * the side being evicted, so accepting a range bound over some other attribute would make the
+   * predicate either ineffective or incorrect.
+   */
+  def getStateValueWatermarkOnWatermarkedAttributes(
+      attributesToFindStateWatermarkFor: AttributeSet,
+      attributesWithEventWatermark: AttributeSet,
+      joinCondition: Option[Expression],
+      eventWatermark: Option[Long]): Option[Long] = {
+    getStateValueWatermark(
+      watermarkedAttributes(attributesToFindStateWatermarkFor),
+      watermarkedAttributes(attributesWithEventWatermark),
+      joinCondition,
+      eventWatermark)
+  }
+
   /**
    * Get state value watermark (see [[StreamingSymmetricHashJoinExec]] for context about it)
    * given the join condition and the event time watermark. This is how it works.