[SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins

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

@@ -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,6 +1332,15 @@ 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 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.
@@ -1398,8 +1421,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 +1443,8 @@ 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>

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, set `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`. (See [SPARK-58904](https://issues.apache.org/jira/browse/SPARK-58904) 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

@@ -23,9 +23,10 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
-import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, Join, LogicalPlan}
 import org.apache.spark.sql.catalyst.plans.logical.EventTimeWatermark._
 import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_DAY
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.CalendarInterval
 
@@ -34,6 +35,34 @@ 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(delayKey)
+    case _ => false
+  }
+
+  private def runtimeWatermarkedAttribute(attributes: Seq[Attribute]): Option[Attribute] = {
+    runtimeWatermarkedAttribute(
+      attributes,
+      allowMultipleEventTimeColumns =
+        !SQLConf.get.getConf(SQLConf.STATEFUL_OPERATOR_ALLOW_MULTIPLE))
+  }
+
+  private def runtimeWatermarkedAttribute(
+      attributes: Seq[Attribute],
+      allowMultipleEventTimeColumns: Boolean): Option[Attribute] = {
+    // Mirror WatermarkSupport.findEventTimeColumn without depending on sql/core from catalyst.
+    val watermarked = attributes.filter(_.metadata.contains(delayKey))
+    if (watermarked.isEmpty) {
+      None
+    } else if (!allowMultipleEventTimeColumns &&
+        watermarked.map(_.exprId).toSet.size > 1) {
+      // Runtime rejects multiple distinct event-time columns in this mode.
+      None
+    } else {
+      Some(watermarked.head)
+    }
+  }
+
   /**
    * Check the provided logical plan to see if its join keys contain a watermark attribute.
    *
@@ -51,6 +80,153 @@ 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). 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(join: Join): Boolean = {
+    join match {
+      case ExtractEquiJoinKeys(_, leftKeys, rightKeys, _, _, _, _, _) =>
+        joinKeyOrdinalForWatermark(leftKeys, rightKeys).exists { ordinal =>
+          val leftRuntimeWatermark = runtimeWatermarkedAttribute(join.left.output)
+          val rightRuntimeWatermark = runtimeWatermarkedAttribute(join.right.output)
+          ordinal < leftKeys.length && ordinal < rightKeys.length &&
+            isWatermarked(leftKeys(ordinal)) && isWatermarked(rightKeys(ordinal)) &&
+            leftRuntimeWatermark.exists(leftKeys(ordinal).references.contains) &&
+            rightRuntimeWatermark.exists(rightKeys(ordinal).references.contains)
+        }
+      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, and is actually a function of the other side's event
+   * watermark. 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.
+   *
+   * Restricting the eligible attributes to watermarked ones is not sufficient on its own:
+   * [[getStateValueWatermark]] requires the evicted side's watermarked attribute to appear in the
+   * condition (it is the attribute the state watermark is solved for), but it treats the
+   * watermark-providing side as satisfied whenever that side merely *has* a watermark, even one
+   * absent from the range bound. A predicate with a constant bound (e.g. `leftTime > <literal>`)
+   * would then still yield a state watermark -- but a constant one, not tied to the other stream's
+   * progress -- so the runtime eviction predicate degenerates to a static constant and the left
+   * state is not really evicted. We therefore additionally require the derived state watermark to
+   * move with the event watermark: a bound that is independent of it does not relate the two sides.
+   */
+  def getStateValueWatermarkOnWatermarkedAttributes(
+      attributesToFindStateWatermarkFor: Seq[Attribute],
+      attributesWithEventWatermark: Seq[Attribute],
+      joinCondition: Option[Expression],
+      eventWatermark: Option[Long],
+      allowMultipleEventTimeColumns: Boolean =
+        !SQLConf.get.getConf(SQLConf.STATEFUL_OPERATOR_ALLOW_MULTIPLE)): Option[Long] = {
+    val evictedSide = runtimeWatermarkedAttribute(
+      attributesToFindStateWatermarkFor, allowMultipleEventTimeColumns)
+      .map(attr => AttributeSet(Seq(attr)))
+    val watermarkProvidingSide = runtimeWatermarkedAttribute(
+      attributesWithEventWatermark, allowMultipleEventTimeColumns)
+      .map(attr => AttributeSet(Seq(attr)))
+    val runtimeEvictedSide = AttributeSet(attributesToFindStateWatermarkFor)
+    val runtimeWatermarkProvidingSide = AttributeSet(attributesWithEventWatermark)
+    eventWatermark match {
+      case Some(wm) if evictedSide.isDefined && watermarkProvidingSide.isDefined =>
+        // Probe with two well-separated event watermark values; a genuine cross-stream range bound
+        // shifts the state watermark forward, whereas a constant or anti-monotonic bound does not.
+        val altWm = wm + 1000000L
+        if (hasNonAdvancingGenericOnlyStateWatermarkPredicate(
+            runtimeEvictedSide, runtimeWatermarkProvidingSide, evictedSide.get,
+            watermarkProvidingSide.get, joinCondition, wm, altWm)) {
+          return None
+        }
+        if (hasNonAdvancingStrictStateWatermarkPredicate(
+            evictedSide.get, watermarkProvidingSide.get, joinCondition, wm, altWm)) {
+          return None
+        }
+        val watermarkAt = getStateValueWatermark(
+          evictedSide.get, watermarkProvidingSide.get, joinCondition, Some(wm),
+          requireWatermarkSideInCondition = true)
+        val watermarkAtAlt = getStateValueWatermark(
+          evictedSide.get, watermarkProvidingSide.get, joinCondition, Some(altWm),
+          requireWatermarkSideInCondition = true)
+        (watermarkAt, watermarkAtAlt) match {
+          case (result @ Some(a), Some(b)) if b > a => result
+          case _ => None
+        }
+      case _ => None
+    }
+  }
+
+  private def hasNonAdvancingGenericOnlyStateWatermarkPredicate(
+      runtimeAttributesToFindStateWatermarkFor: AttributeSet,
+      runtimeAttributesWithEventWatermark: AttributeSet,
+      strictAttributesToFindStateWatermarkFor: AttributeSet,
+      strictAttributesWithEventWatermark: AttributeSet,
+      joinCondition: Option[Expression],
+      eventWatermark: Long,
+      altEventWatermark: Long): Boolean = {
+    joinCondition.exists { condition =>
+      splitConjunctivePredicates(condition).exists { predicate =>
+        val genericWatermark = getStateValueWatermark(
+          runtimeAttributesToFindStateWatermarkFor, runtimeAttributesWithEventWatermark,
+          Some(predicate), Some(eventWatermark))
+        val genericWatermarkAtAlt = getStateValueWatermark(
+          runtimeAttributesToFindStateWatermarkFor, runtimeAttributesWithEventWatermark,
+          Some(predicate), Some(altEventWatermark))
+        val strictWatermark = getStateValueWatermark(
+          strictAttributesToFindStateWatermarkFor, strictAttributesWithEventWatermark,
+          Some(predicate), Some(eventWatermark), requireWatermarkSideInCondition = true)
+        genericWatermark.exists { watermark =>
+          strictWatermark.isEmpty && genericWatermarkAtAlt.forall(_ <= watermark)
+        }
+      }
+    }
+  }
+
+  private def hasNonAdvancingStrictStateWatermarkPredicate(
+      strictAttributesToFindStateWatermarkFor: AttributeSet,
+      strictAttributesWithEventWatermark: AttributeSet,
+      joinCondition: Option[Expression],
+      eventWatermark: Long,
+      altEventWatermark: Long): Boolean = {
+    joinCondition.exists { condition =>
+      splitConjunctivePredicates(condition).exists { predicate =>
+        val watermark = getStateValueWatermark(
+          strictAttributesToFindStateWatermarkFor, strictAttributesWithEventWatermark,
+          Some(predicate), Some(eventWatermark), requireWatermarkSideInCondition = true)
+        val watermarkAtAlt = getStateValueWatermark(
+          strictAttributesToFindStateWatermarkFor, strictAttributesWithEventWatermark,
+          Some(predicate), Some(altEventWatermark), requireWatermarkSideInCondition = true)
+        watermark.exists { value =>
+          watermarkAtAlt.forall(_ <= value)
+        }
+      }
+    }
+  }
+
   /**
    * Get state value watermark (see [[StreamingSymmetricHashJoinExec]] for context about it)
    * given the join condition and the event time watermark. This is how it works.
@@ -73,6 +249,20 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
       attributesWithEventWatermark: AttributeSet,
       joinCondition: Option[Expression],
       eventWatermark: Option[Long]): Option[Long] = {
+    getStateValueWatermark(
+      attributesToFindStateWatermarkFor,
+      attributesWithEventWatermark,
+      joinCondition,
+      eventWatermark,
+      requireWatermarkSideInCondition = false)
+  }
+
+  private def getStateValueWatermark(
+      attributesToFindStateWatermarkFor: AttributeSet,
+      attributesWithEventWatermark: AttributeSet,
+      joinCondition: Option[Expression],
+      eventWatermark: Option[Long],
+      requireWatermarkSideInCondition: Boolean): Option[Long] = {
 
     // If condition or event time watermark is not provided, then cannot calculate state watermark
     if (joinCondition.isEmpty || eventWatermark.isEmpty) return None
@@ -83,7 +273,8 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
     def getStateWatermarkSafely(l: Expression, r: Expression): Option[Long] = {
       try {
         getStateWatermarkFromLessThenPredicate(
-          l, r, attributesToFindStateWatermarkFor, attributesWithEventWatermark, eventWatermark)
+          l, r, attributesToFindStateWatermarkFor, attributesWithEventWatermark, eventWatermark,
+          requireWatermarkSideInCondition)
       } catch {
         case NonFatal(e) =>
           logWarning(log"Error trying to extract state constraint from condition " +
@@ -132,17 +323,34 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
       rightExpr: Expression,
       attributesToFindStateWatermarkFor: AttributeSet,
       attributesWithEventWatermark: AttributeSet,
-      eventWatermark: Option[Long]): Option[Long] = {
+      eventWatermark: Option[Long],
+      requireWatermarkSideInCondition: Boolean): Option[Long] = {
 
-    val attributesInCondition = AttributeSet(
+    val attributesInConditionSeq =
       leftExpr.collect { case a: AttributeReference => a } ++
-      rightExpr.collect { case a: AttributeReference => a }
-    )
-    if (attributesInCondition.count(attributesToFindStateWatermarkFor.contains) > 1 ||
-        attributesInCondition.count(attributesWithEventWatermark.contains) > 1) {
+        rightExpr.collect { case a: AttributeReference => a }
+    val attributesInCondition = AttributeSet(attributesInConditionSeq)
+    val stateSideAttributeCount =
+      if (requireWatermarkSideInCondition) {
+        attributesInConditionSeq.count(attributesToFindStateWatermarkFor.contains)
+      } else {
+        attributesInCondition.count(attributesToFindStateWatermarkFor.contains)
+      }
+    val watermarkSideAttributeCount =
+      if (requireWatermarkSideInCondition) {
+        attributesInConditionSeq.count(attributesWithEventWatermark.contains)
+      } else {
+        attributesInCondition.count(attributesWithEventWatermark.contains)
+      }
+    if (stateSideAttributeCount > 1 || watermarkSideAttributeCount > 1) {
       // If more than attributes present in condition from one side, then it cannot be solved
       return None
     }
+    if (requireWatermarkSideInCondition && watermarkSideAttributeCount == 0) {
+      // A state watermark must move with the other side's event watermark. A bound that does not
+      // reference the watermark-providing side may produce a constant, but not a valid watermark.
+      return None
+    }
 
     def containsAttributeToFindStateConstraintFor(e: Expression): Boolean = {
       e.collectLeaves().collectFirst {