diff --git a/docs/streaming/apis-on-dataframes-and-datasets.md b/docs/streaming/apis-on-dataframes-and-datasets.md index 86585caead51f..71ec5b0df1fcd 100644 --- a/docs/streaming/apis-on-dataframes-and-datasets.md +++ b/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( +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 @@ -1343,8 +1398,8 @@ regarding watermark delays and whether data will be dropped or not. - - + + @@ -1365,8 +1420,12 @@ regarding watermark delays and whether data will be dropped or not. - - + + + + + + @@ -1387,8 +1446,12 @@ regarding watermark delays and whether data will be dropped or not. - - + + + + + + @@ -1419,8 +1483,16 @@ regarding watermark delays and whether data will be dropped or not. + + + + @@ -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 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. - + diff --git a/docs/streaming/ss-migration-guide.md b/docs/streaming/ss-migration-guide.md index 210f752d9d17d..582eff56134b4 100644 --- a/docs/streaming/ss-migration-guide.md +++ b/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.) 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 c4549a189e8e1..597f4bac6a32a 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 @@ -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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala index eddcee169e377..bd53b67b17d12 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala @@ -491,7 +491,10 @@ object UnsupportedOperationChecker extends Logging { joinType match { // The behavior for unmatched rows in outer joins with update mode // hasn't been defined yet. - case LeftOuter | RightOuter | FullOuter => + // LeftAnti is included here because its unmatched rows are only emitted once the + // watermark guarantees no future match, so early-firing in Update mode would + // produce rows which a later batch could invalidate. + case LeftOuter | RightOuter | FullOuter | LeftAnti => if (outputMode != InternalOutputModes.Append) { throwError(s"$joinType join between two streaming DataFrames/Datasets" + s" is not supported in ${outputMode} output mode, only in Append output mode") @@ -522,10 +525,16 @@ object UnsupportedOperationChecker extends Logging { checkForStreamStreamJoinWatermark(j) } + // We support streaming left anti joins with stream on both sides under the + // appropriate conditions. A streaming right with a static left is not supported: + // unmatched left rows are determined at watermark-based eviction of the left state, + // which a static left side does not have. case LeftAnti => - if (right.isStreaming) { - throwError(s"$LeftAnti joins with a streaming DataFrame/Dataset " + - "on the right are not supported") + if (!left.isStreaming && right.isStreaming) { + throwError(s"$LeftAnti join with a streaming DataFrame/Dataset " + + "on the right and a static DataFrame/Dataset on the left is not supported") + } else if (left.isStreaming && right.isStreaming) { + checkForStreamStreamJoinWatermark(j) } // We support streaming left outer and left semi joins with static on the right always, @@ -687,7 +696,7 @@ object UnsupportedOperationChecker extends Logging { // Check if the nullable side has a watermark, and there's a range condition which // implies a state value watermark on the first side. val hasValidWatermarkRange = join.joinType match { - case LeftOuter | LeftSemi => StreamingJoinHelper.getStateValueWatermark( + case LeftOuter | LeftSemi | LeftAnti => StreamingJoinHelper.getStateValueWatermark( join.left.outputSet, join.right.outputSet, join.condition, Some(1000000)).isDefined case RightOuter => StreamingJoinHelper.getStateValueWatermark( join.right.outputSet, join.left.outputSet, join.condition, Some(1000000)).isDefined @@ -709,5 +718,116 @@ object UnsupportedOperationChecker extends Logging { "is not supported without a watermark in the join keys, or a watermark on " + "the nullable side and an appropriate range condition")(join) } + + // The generic check above is side- and ordinal-insensitive: it accepts any watermark in the + // join keys, or any watermark that yields a state value watermark on the nullable side. That is + // not enough for the join types whose output (or bounded state size) depends on the left state + // being evicted: left semi, left outer, and left anti. + // + // Left semi/outer are existing join types, so users can temporarily restore the previous loose + // behavior with SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED. Left anti is + // new, so it always uses the stricter rules. + if (join.joinType == LeftAnti || SQLConf.get.streamingJoinStricterWatermarkRequirements) { + checkStreamStreamJoinWatermarkPlacement( + join, + watermarkInJoinKeys, + canRestorePreviousBehavior = join.joinType != LeftAnti) + } + } + + /** + * Enforce stricter watermark-placement requirements for the stream-stream join types whose + * correctness or bounded state depends on the left-side state being evicted: left semi, left + * outer, and left anti. These requirements are stricter than the generic check in + * [[checkForStreamStreamJoinWatermark]]. For left semi and left outer, these checks are gated + * by [[SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED]]. Left anti is new and + * always enforces these requirements. + * + * Two independent requirements apply: + * + * 1. Left-state eviction. Left semi/outer/anti all rely on the left state being evicted: left + * outer/anti emit their surviving unmatched left rows from that eviction (so without it the + * unmatched/outer output is never produced), and left semi cleans up its never-matched left + * rows there (so without it the left state grows without bound). In the equi-join path the + * left key state is evicted through the shared watermarked join key. In the range-condition + * path the left *value* state is evicted only when the range condition derives the left state + * watermark from watermarked attributes on both sides. + * + * 2. No invalidation by late rows. This applies only to the outer-like types (left outer and + * left anti), whose already-emitted unmatched row can be invalidated by a later match. Both + * sides must be late-filtered on the same dimension the left state is evicted by; otherwise + * a row that is old on the eviction key can still match an already-emitted unmatched row. In + * the equi-join path this requires both join keys at the eviction ordinal to be watermarked. + * Left semi does not need this: it emits on match while joining, never at eviction, so there + * is nothing to invalidate. + */ + private def checkStreamStreamJoinWatermarkPlacement( + join: Join, + watermarkInJoinKeys: Boolean, + canRestorePreviousBehavior: Boolean): Unit = { + // (requiresLeftStateEviction, isOuterLike) per join type. Only the left-side-eviction join + // types are constrained here; inner, right outer, and full outer keep their existing behavior. + val (requiresLeftStateEviction, isOuterLike) = join.joinType match { + case LeftOuter => (true, true) + case LeftAnti => (true, true) + case LeftSemi => (true, false) + case _ => (false, false) + } + + if (requiresLeftStateEviction) { + if (watermarkInJoinKeys) { + // Equi-join path: the left key state is evicted via the join-key watermark, so the + // eviction requirement is met. For the outer-like types we additionally need both sides + // late-filtered on the same key ordinal used for eviction. A watermark on only one side, on + // an unrelated column, or on a different key position (composite keys) would let a late row + // match an already-emitted unmatched row. + if (isOuterLike && !StreamingJoinHelper.isWatermarkOnBothEvictionJoinKeys(join)) { + val restore = if (canRestorePreviousBehavior) { + " To restore the previous behavior, set " + + s"${SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED.key}=false." + } else { + "" + } + throwError( + s"Stream-stream ${join.joinType} join between two streaming DataFrame/Datasets with " + + "a " + + "watermarked join key requires watermarks on both join keys used for state " + + "eviction (for a composite key, the left and right keys at the same position), so " + + "that late rows on either side are dropped and cannot invalidate already-emitted " + + "unmatched " + + s"rows. A watermark on only one side is not supported.$restore")(join) + } + } else { + // Range-condition path (hasValidWatermarkRange holds here): the left state is evicted only + // when the range condition derives its state watermark from watermarked attributes on both + // sides. A right-only watermark, or a watermark on an unrelated left/right attribute, + // leaves the runtime eviction predicate ineffective or incorrect. + val hasWatermarkRangeOnBothSides = + StreamingJoinHelper.getStateValueWatermarkOnWatermarkedAttributes( + join.left.outputSet, join.right.outputSet, join.condition, Some(1000000)).isDefined + if (!hasWatermarkRangeOnBothSides) { + val reason = + if (isOuterLike) { + "so that the left state whose surviving unmatched rows form the " + + s"${join.joinType} output is evicted" + } else { + "so that the left state is evicted and does not grow without bound" + } + val restore = if (canRestorePreviousBehavior) { + " To restore the previous behavior, set " + + s"${SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED.key}=false." + } else { + "" + } + throwError( + s"Stream-stream ${join.joinType} join between two streaming DataFrame/Datasets with " + + "a " + + "range condition requires a range bound between watermarked attributes on both " + + s"sides, $reason. A watermark on only one side, or on an unrelated attribute, is " + + "not " + + s"supported.$restore")(join) + } + } + } } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 8a0d88346b63c..7fb3485c6df84 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -3784,6 +3784,22 @@ object SQLConf { .checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2, 3, and 4") .createWithDefault(2) + val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED = + buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled") + .doc("When true, the analyzer enforces stricter watermark-placement requirements for " + + "stream-stream left semi and left outer joins, so that the left-side state which 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 " + + "would otherwise silently produce incorrect results or unbounded state (e.g. 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) are rejected. Set this to false to restore the previous, looser behavior for " + + "left semi and left outer joins. Left anti joins always use the stricter requirements.") + .version("4.4.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .booleanConf + .createWithDefault(true) + val STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION = buildConf("spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition") .doc("When true, streaming session window sorts and merge sessions in local partition " + @@ -8500,6 +8516,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf { def stateStoreSkipNullsForStreamStreamJoins: Boolean = getConf(STATE_STORE_SKIP_NULLS_FOR_STREAM_STREAM_JOINS) + def streamingJoinStricterWatermarkRequirements: Boolean = + getConf(STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED) + def stateStoreCoordinatorMultiplierForMinVersionDiffToLog: Long = getConf(STATE_STORE_COORDINATOR_MULTIPLIER_FOR_MIN_VERSION_DIFF_TO_LOG) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala index 293523b86f998..fcd267b06c688 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala @@ -424,7 +424,8 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { streamBatchSupported = false, expectedMsg = "FullOuter join") - // Left outer, left semi, left anti join: *-stream not allowed + // Left outer, left semi, left anti join: batch-stream not allowed, and stream-stream join is + // allowed 'conditionally' - see the watermark checks below Seq((LeftOuter, "LeftOuter join"), (LeftSemi, "LeftSemi join"), (LeftAnti, "LeftAnti join")) .foreach { case (joinType, name) => testBinaryOperationInStreamingPlan( @@ -443,8 +444,10 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { streamStreamSupported = false, expectedMsg = "RightOuter join") - // Left outer, right outer, full outer joins: Update mode not allowed - Seq(LeftOuter, RightOuter, FullOuter).foreach { joinType => + // Left outer, right outer, full outer, left anti joins: Update mode not allowed. Left anti is + // included because its unmatched rows are only emitted at watermark-based eviction, so + // early-firing could emit a row which a later batch would invalidate. + Seq(LeftOuter, RightOuter, FullOuter, LeftAnti).foreach { joinType => assertNotSupportedInStreamingPlan( s"$joinType join with stream-stream relations and update mode", streamRelation.join(streamRelation, joinType = joinType, @@ -453,6 +456,20 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { Seq("is not supported in Update output mode")) } + // Supported stream-static joins are not stateful and follow the general output-mode rules for + // non-aggregation queries, even when the corresponding stream-stream join is Append-only. + Seq((LeftOuter, "LeftOuter join"), (LeftAnti, "LeftAnti join")).foreach { + case (joinType, name) => + assertSupportedInStreamingPlan( + s"$name with stream-batch relations and update mode", + streamRelation.join(batchRelation, joinType = joinType), + OutputMode.Update()) + } + assertSupportedInStreamingPlan( + "RightOuter join with batch-stream relations and update mode", + batchRelation.join(streamRelation, joinType = RightOuter), + OutputMode.Update()) + // LeftSemi join: Update mode allowed (equivalent to Append mode for non-outer joins) assertSupportedInStreamingPlan( s"LeftSemi join with stream-stream relations and update mode", @@ -467,7 +484,8 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { (LeftSemi, "only in Append and Update output modes"), (LeftOuter, "only in Append output mode"), (RightOuter, "only in Append output mode"), - (FullOuter, "only in Append output mode") + (FullOuter, "only in Append output mode"), + (LeftAnti, "only in Append output mode") ).foreach { case (joinType, allowedModesMsg) => assertNotSupportedInStreamingPlan( s"$joinType join with stream-stream relations and complete mode", @@ -477,8 +495,11 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { Seq("is not supported in Complete output mode", allowedModesMsg)) } - // Left outer, right outer, full outer, left semi joins - Seq(LeftOuter, RightOuter, FullOuter, LeftSemi).foreach { joinType => + // Right outer and full outer joins keep the side-insensitive watermark behavior. Left outer, + // left semi, and left anti are handled in dedicated blocks below: they have stricter + // watermark-placement requirements (see checkStreamStreamJoinWatermarkPlacement), so the + // side-insensitive cases here do not all apply to them. + Seq(RightOuter, FullOuter).foreach { joinType => // Stream-stream allowed with join on watermark attribute // Note that the attribute need not be watermarked on both sides. assertSupportedInStreamingPlan( @@ -525,6 +546,299 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper { "the nullable side and an appropriate range condition")) } + // Left outer and left semi joins: stricter watermark-placement requirements, shared with left + // anti (see checkStreamStreamJoinWatermarkPlacement) and gated by + // STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED. Their left-side state must be + // evicted -- left outer emits its unmatched rows from that eviction, left semi cleans up its + // never-matched rows there -- and left outer, being outer-like, additionally needs both sides + // late-filtered on the eviction key so late rows cannot invalidate already-emitted unmatched + // rows. Distinct attributes per side avoid exprId collisions in these synthetic plans. + { + val leftWm = AttributeReference("olw", IntegerType)().withMetadata(watermarkMetadata) + val rightWm = AttributeReference("orw", IntegerType)().withMetadata(watermarkMetadata) + val leftPlain = AttributeReference("olp", IntegerType)() + val rightPlain = AttributeReference("orp", IntegerType)() + val leftOtherWm = AttributeReference("olo", IntegerType)().withMetadata(watermarkMetadata) + val rightOtherWm = AttributeReference("oro", IntegerType)().withMetadata(watermarkMetadata) + val strictKey = SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED.key + + Seq(LeftOuter, LeftSemi).foreach { joinType => + // Supported: range condition with a watermark on both sides (the left value state is evicted + // on its own watermark, derived from the right watermark). + assertSupportedInStreamingPlan( + s"$joinType join with stream-stream relations and range condition, watermark on both sides", + new TestStreamingRelation(leftWm).join(new TestStreamingRelation(rightWm), + joinType = joinType, condition = Some(leftWm > rightWm + 10)), + OutputMode.Append()) + + // Not supported: range condition with a watermark on the right only -- the left state is + // never evicted (so left outer never emits its unmatched rows, and left semi's left state + // grows without bound). + assertNotSupportedInStreamingPlan( + s"$joinType join with stream-stream relations and range condition, right watermark only", + new TestStreamingRelation(leftPlain).join(new TestStreamingRelation(rightWm), + joinType = joinType, condition = Some(leftPlain > rightWm + 10)), + OutputMode.Append(), + Seq("requires a range bound between watermarked attributes on both sides")) + + // Not supported: the left watermark is unrelated to the range-bound left attribute, so the + // runtime eviction predicate would be applied to the wrong column. + assertNotSupportedInStreamingPlan( + s"$joinType join with stream-stream relations and range condition, unrelated left " + + "watermark", + new TestStreamingRelation(Seq(leftPlain, leftOtherWm)).join( + new TestStreamingRelation(rightWm), + joinType = joinType, condition = Some(leftPlain > rightWm + 10)), + OutputMode.Append(), + Seq("requires a range bound between watermarked attributes on both sides")) + + // Not supported: the right watermark is unrelated to the range-bound right attribute, so the + // state watermark cannot be derived from the runtime late-filtering column. + assertNotSupportedInStreamingPlan( + s"$joinType join with stream-stream relations and range condition, unrelated right " + + "watermark", + new TestStreamingRelation(leftWm).join( + new TestStreamingRelation(Seq(rightPlain, rightOtherWm)), + joinType = joinType, condition = Some(leftWm > rightPlain + 10)), + OutputMode.Append(), + Seq("without a watermark in the join keys")) + + // Kill switch: with the stricter requirements disabled, the right-only range config falls + // back to the previous (looser) behavior and is accepted again. + assertSupportedInStreamingPlan( + s"$joinType join, right watermark only, accepted when stricter requirements disabled", + new TestStreamingRelation(leftPlain).join(new TestStreamingRelation(rightWm), + joinType = joinType, condition = Some(leftPlain > rightWm + 10)), + OutputMode.Append(), + strictKey -> "false") + } + + // Left semi is not outer-like: it emits on match, never at eviction, so the equi-join path has + // no no-invalidation requirement. A one-sided key watermark is enough to evict the left key + // state through the shared equality key. + assertSupportedInStreamingPlan( + "left semi join with stream-stream relations and join key watermarked on the right side only", + new TestStreamingRelation(leftPlain).join(new TestStreamingRelation(rightWm), + joinType = LeftSemi, condition = Some(leftPlain === rightWm)), + OutputMode.Append()) + assertSupportedInStreamingPlan( + "left semi join with stream-stream relations and join key watermarked on the left side only", + new TestStreamingRelation(leftWm).join(new TestStreamingRelation(rightPlain), + joinType = LeftSemi, condition = Some(leftWm === rightPlain)), + OutputMode.Append()) + + // Left outer is outer-like, so equality joins require both keys at the eviction ordinal to be + // watermarked. Otherwise one side is not late-filtered on the dimension used to evict state. + assertSupportedInStreamingPlan( + "left outer join with stream-stream relations and both join keys watermarked", + new TestStreamingRelation(leftWm).join(new TestStreamingRelation(rightWm), + joinType = LeftOuter, condition = Some(leftWm === rightWm)), + OutputMode.Append()) + Seq( + ("right side only", leftPlain, rightWm), + ("left side only", leftWm, rightPlain) + ).foreach { case (side, leftKey, rightKey) => + assertNotSupportedInStreamingPlan( + s"left outer join with stream-stream relations and join key watermarked on $side", + new TestStreamingRelation(leftKey).join(new TestStreamingRelation(rightKey), + joinType = LeftOuter, condition = Some(leftKey === rightKey)), + OutputMode.Append(), + Seq("requires watermarks on both join keys used for state eviction")) + } + + assertNotSupportedInStreamingPlan( + "left outer join with stream-stream relations and watermark on an unrelated right column", + new TestStreamingRelation(leftWm).join( + new TestStreamingRelation(Seq(rightPlain, rightWm)), + joinType = LeftOuter, condition = Some(leftWm === rightPlain)), + OutputMode.Append(), + Seq("requires watermarks on both join keys used for state eviction")) + + // Composite equality keys. State-key eviction uses a single ordinal (the first watermarked + // left key), so both keys at that ordinal must be watermarked. + val outerLeftWm1 = AttributeReference("olw1", IntegerType)().withMetadata(watermarkMetadata) + val outerLeftPlain2 = AttributeReference("olp2", IntegerType)() + val outerRightWm1 = AttributeReference("orw1", IntegerType)().withMetadata(watermarkMetadata) + val outerRightPlain1 = AttributeReference("orp1", IntegerType)() + val outerRightPlain2 = AttributeReference("orp2", IntegerType)() + val outerRightWm2 = AttributeReference("orw2", IntegerType)().withMetadata(watermarkMetadata) + + assertSupportedInStreamingPlan( + "left outer join with stream-stream relations and composite key watermarked at same ordinal", + new TestStreamingRelation(Seq(outerLeftWm1, outerLeftPlain2)).join( + new TestStreamingRelation(Seq(outerRightWm1, outerRightPlain2)), + joinType = LeftOuter, + condition = Some(outerLeftWm1 === outerRightWm1 && + outerLeftPlain2 === outerRightPlain2)), + OutputMode.Append()) + + assertNotSupportedInStreamingPlan( + "left outer join with stream-stream relations and composite key watermarked on mismatched " + + "ordinals", + new TestStreamingRelation(Seq(outerLeftWm1, outerLeftPlain2)).join( + new TestStreamingRelation(Seq(outerRightPlain1, outerRightWm2)), + joinType = LeftOuter, + condition = Some(outerLeftWm1 === outerRightPlain1 && + outerLeftPlain2 === outerRightWm2)), + OutputMode.Append(), + Seq("requires watermarks on both join keys used for state eviction")) + + // Kill switch: accepted again when the stricter requirements are disabled. + assertSupportedInStreamingPlan( + "left outer join, right-only key watermark, accepted when stricter requirements disabled", + new TestStreamingRelation(leftPlain).join(new TestStreamingRelation(rightWm), + joinType = LeftOuter, condition = Some(leftPlain === rightWm)), + OutputMode.Append(), + strictKey -> "false") + assertSupportedInStreamingPlan( + "left outer join, left-only key watermark, accepted when stricter requirements disabled", + new TestStreamingRelation(leftWm).join(new TestStreamingRelation(rightPlain), + joinType = LeftOuter, condition = Some(leftWm === rightPlain)), + OutputMode.Append(), + strictKey -> "false") + } + + // Left anti join uses the same strict requirements as left outer, but it is new in this PR, so + // the compatibility flag does not disable them. + { + val leftWm = AttributeReference("lw", IntegerType)().withMetadata(watermarkMetadata) + val rightWm = AttributeReference("rw", IntegerType)().withMetadata(watermarkMetadata) + val leftPlain = AttributeReference("lp", IntegerType)() + val rightPlain = AttributeReference("rp", IntegerType)() + val leftOtherWm = AttributeReference("lo", IntegerType)().withMetadata(watermarkMetadata) + val rightOtherWm = AttributeReference("ro", IntegerType)().withMetadata(watermarkMetadata) + + // Supported: equality join keys watermarked on both sides. + assertSupportedInStreamingPlan( + "left anti join with stream-stream relations and both join keys watermarked", + new TestStreamingRelation(leftWm).join(new TestStreamingRelation(rightWm), + joinType = LeftAnti, condition = Some(leftWm === rightWm)), + OutputMode.Append()) + + // Supported: range condition with a watermark on both sides. + assertSupportedInStreamingPlan( + "left anti join with stream-stream relations and range condition, watermark on both sides", + new TestStreamingRelation(leftWm).join(new TestStreamingRelation(rightWm), + joinType = LeftAnti, condition = Some(leftWm > rightWm + 10)), + OutputMode.Append()) + + Seq( + ("right side only", leftPlain, rightWm), + ("left side only", leftWm, rightPlain) + ).foreach { case (side, leftKey, rightKey) => + assertNotSupportedInStreamingPlan( + s"left anti join with stream-stream relations and join key watermarked on $side", + new TestStreamingRelation(leftKey).join(new TestStreamingRelation(rightKey), + joinType = LeftAnti, condition = Some(leftKey === rightKey)), + OutputMode.Append(), + Seq("requires watermarks on both join keys used for state eviction")) + } + + // Not supported: the watermark is on an unrelated right column, not the right join key, so the + // right side is not late-filtered on the key that bounds matching. + assertNotSupportedInStreamingPlan( + "left anti join with stream-stream relations and watermark on an unrelated right column", + new TestStreamingRelation(leftWm).join( + new TestStreamingRelation(Seq(rightPlain, rightOtherWm)), + joinType = LeftAnti, condition = Some(leftWm === rightPlain)), + OutputMode.Append(), + Seq("requires watermarks on both join keys used for state eviction")) + + // Composite equality keys. State-key eviction uses a single ordinal (the first watermarked + // left key), so both keys at that ordinal must be watermarked. + val leftWm1 = AttributeReference("lw1", IntegerType)().withMetadata(watermarkMetadata) + val leftPlain2 = AttributeReference("lp2", IntegerType)() + val rightWm1 = AttributeReference("rw1", IntegerType)().withMetadata(watermarkMetadata) + val rightPlain1 = AttributeReference("rp1", IntegerType)() + val rightPlain2 = AttributeReference("rp2", IntegerType)() + val rightWm2 = AttributeReference("rw2", IntegerType)().withMetadata(watermarkMetadata) + + // Supported: composite key watermarked at the same ordinal on both sides (ordinal 0). + assertSupportedInStreamingPlan( + "left anti join with stream-stream relations and composite key watermarked at same ordinal", + new TestStreamingRelation(Seq(leftWm1, leftPlain2)).join( + new TestStreamingRelation(Seq(rightWm1, rightPlain2)), + joinType = LeftAnti, + condition = Some(leftWm1 === rightWm1 && leftPlain2 === rightPlain2)), + OutputMode.Append()) + + // Not supported: composite key watermarked on mismatched ordinals -- left key 0 and right + // key 1 are watermarked, but eviction uses ordinal 0, whose right key is not watermarked, so + // the right side is not late-filtered on the eviction key. + assertNotSupportedInStreamingPlan( + "left anti join with stream-stream relations and composite key watermarked on mismatched " + + "ordinals", + new TestStreamingRelation(Seq(leftWm1, leftPlain2)).join( + new TestStreamingRelation(Seq(rightPlain1, rightWm2)), + joinType = LeftAnti, + condition = Some(leftWm1 === rightPlain1 && leftPlain2 === rightWm2)), + OutputMode.Append(), + Seq("requires watermarks on both join keys used for state eviction")) + + // Not supported: range condition with a watermark on the right only -- the left state is never + // evicted, so no anti row is ever produced. + assertNotSupportedInStreamingPlan( + "left anti join with stream-stream relations and range condition, right watermark only", + new TestStreamingRelation(leftPlain).join(new TestStreamingRelation(rightWm), + joinType = LeftAnti, condition = Some(leftPlain > rightWm + 10)), + OutputMode.Append(), + Seq("requires a range bound between watermarked attributes on both sides")) + + // Not supported: the left watermark is unrelated to the range-bound left attribute. + assertNotSupportedInStreamingPlan( + "left anti join with stream-stream relations and range condition, unrelated left watermark", + new TestStreamingRelation(Seq(leftPlain, leftOtherWm)).join( + new TestStreamingRelation(rightWm), + joinType = LeftAnti, condition = Some(leftPlain > rightWm + 10)), + OutputMode.Append(), + Seq("requires a range bound between watermarked attributes on both sides")) + + // Not supported: the right watermark is unrelated to the range-bound right attribute. + assertNotSupportedInStreamingPlan( + "left anti join with stream-stream relations and range condition, unrelated right watermark", + new TestStreamingRelation(leftWm).join( + new TestStreamingRelation(Seq(rightPlain, rightOtherWm)), + joinType = LeftAnti, condition = Some(leftWm > rightPlain + 10)), + OutputMode.Append(), + Seq("without a watermark in the join keys")) + + test("streaming plan - left anti strict watermark rules ignore compatibility flag") { + withSQLConf(SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED.key -> "false") { + val rangeError = intercept[AnalysisException] { + UnsupportedOperationChecker.checkForStreaming( + wrapInStreaming(new TestStreamingRelation(leftPlain).join( + new TestStreamingRelation(rightWm), + joinType = LeftAnti, + condition = Some(leftPlain > rightWm + 10))), + OutputMode.Append()) + } + assert(rangeError.getMessage.contains( + "requires a range bound between watermarked attributes on both sides")) + + Seq((leftPlain, rightWm), (leftWm, rightPlain)).foreach { case (leftKey, rightKey) => + val equalityError = intercept[AnalysisException] { + UnsupportedOperationChecker.checkForStreaming( + wrapInStreaming(new TestStreamingRelation(leftKey).join( + new TestStreamingRelation(rightKey), + joinType = LeftAnti, + condition = Some(leftKey === rightKey))), + OutputMode.Append()) + } + assert(equalityError.getMessage.contains( + "requires watermarks on both join keys used for state eviction")) + } + } + } + + // Not supported: no watermark at all (rejected by the generic stream-stream join check). + assertNotSupportedInStreamingPlan( + "left anti join with stream-stream relations and no watermark", + new TestStreamingRelation(leftPlain).join(new TestStreamingRelation(rightPlain), + joinType = LeftAnti, condition = Some(leftPlain === rightPlain)), + OutputMode.Append(), + Seq("without a watermark in the join keys")) + } + // multi-aggregations only supported in Append mode assertPassOnGlobalWatermarkLimit( "aggregate - multiple streaming aggregations - append", diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala index 8f90a603c7efb..9785c65897cbe 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala @@ -187,7 +187,7 @@ case class StreamingSymmetricHashJoinExec( require( joinType == Inner || joinType == LeftOuter || joinType == RightOuter || joinType == FullOuter || - joinType == LeftSemi, + joinType == LeftSemi || joinType == LeftAnti, errorMessageForJoinType) outputMode.foreach { mode => @@ -238,7 +238,7 @@ case class StreamingSymmetricHashJoinExec( case LeftOuter => left.output ++ right.output.map(_.withNullability(true)) case RightOuter => left.output.map(_.withNullability(true)) ++ right.output case FullOuter => (left.output ++ right.output).map(_.withNullability(true)) - case LeftSemi => left.output + case LeftSemi | LeftAnti => left.output case _ => throwBadJoinTypeException() } WidenStatefulOpNullability.widenOutputForStatefulOp(base) @@ -251,7 +251,7 @@ case class StreamingSymmetricHashJoinExec( case LeftOuter => left.outputPartitioning case RightOuter => right.outputPartitioning case FullOuter => UnknownPartitioning(left.outputPartitioning.numPartitions) - case LeftSemi => left.outputPartitioning + case LeftSemi | LeftAnti => left.outputPartitioning case _ => throwBadJoinTypeException() } @@ -408,6 +408,10 @@ case class StreamingSymmetricHashJoinExec( // new left input with stored right input, and also stores all the left input. // - Left Semi Join: generates all new left input rows from matching new left input with // stored right input, and also stores all the non-matched left input. + // - Left Anti Join: like left semi, but generates nothing while joining; it stores only the + // non-matched new left input (a matched left row can never be anti output, so it is not + // buffered). A left row that fails the pre-join filter is unmatched by definition and is + // emitted immediately without being stored (see `generateFilteredJoinedRow`). // // - `rightSideJoiner.storeAndJoinWithOtherSide(leftSideJoiner)` // - Inner, Left Outer, Right Outer, Full Outer Join: generates all rows from matching @@ -418,10 +422,19 @@ case class StreamingSymmetricHashJoinExec( // - Left Semi Join: generates all stored left input rows, from matching new right input // with stored left input, and also stores all the right input. Note only first-time // matched left input rows will be generated, this is to guarantee left semi semantics. + // - Left Anti Join: generates nothing, but removes the matched left rows from state (they + // can no longer become anti output), and also stores all the right input. // - // For Left Semi Join, we process the right side first so that new right input is stored - // before left input probes against it. This way, new left rows that match new right rows - // in the same microbatch are emitted immediately without being buffered in state. + // For Left Semi and Left Anti Join, we process the right side first so that new right input is + // stored before left input probes against it. This way, new left rows that match new right + // rows in the same microbatch are handled immediately without being buffered in state (emitted + // for left semi, dropped for left anti). + // + // Left Anti differs from left semi only in what it emits: for a row that passes the pre-join + // filter, nothing during the join (a left row that fails the pre-join filter is emitted + // immediately, since it can never match). A surviving (never-matched) left row is not provably + // unmatched until the watermark rules out any future right match, so -- like left outer -- + // those rows are emitted when the left state is evicted (see `outputIter`). val leftOutputIter = joinerManager.leftSideJoiner.storeAndJoinWithOtherSide(joinerManager.rightSideJoiner) { (input: InternalRow, matched: InternalRow) => joinedRow.withLeft(input).withRight(matched) @@ -441,10 +454,10 @@ case class StreamingSymmetricHashJoinExec( // this will be prepended to a second iterator producing other rows; for inner and left semi // joins, this is the full output. // - // For left semi join, right side is processed first so new right rows are available in - // state when left rows probe, avoiding unnecessary left-side state buffering. + // For left semi and left anti join, right side is processed first so new right rows are + // available in state when left rows probe, avoiding unnecessary left-side state buffering. val hashJoinOutputIter = CompletionIterator[InternalRow, Iterator[InternalRow]]( - if (joinType == LeftSemi) rightOutputIter ++ leftOutputIter + if (joinType == LeftSemi || joinType == LeftAnti) rightOutputIter ++ leftOutputIter else leftOutputIter ++ rightOutputIter, onHashJoinOutputCompletion()) @@ -542,10 +555,36 @@ case class StreamingSymmetricHashJoinExec( val rightSideOutputIter = new LazilyInitializingJoinedRowIterator(rightSideInitIterFn) hashJoinOutputIter ++ leftSideOutputIter ++ rightSideOutputIter + case LeftAnti => + // Left anti join is a hybrid of left semi and left outer. The matched-row handling is the + // left semi optimization: a left row that finds any match can never be anti output, so it + // is dropped from state on match rather than kept (see `storeAndJoinWithOtherSide`), + // exactly mirroring left semi. As a result every left row that survives in state is, by + // construction, still unmatched. + // + // Whether such a surviving row is truly unmatched is only decided once the watermark + // guarantees no future right row can match it, so -- like the left side of a left outer + // join -- the survivors are emitted at eviction time, as bare left rows. Unlike left + // outer, no matched-flag filter is needed here, since matched rows are already gone. + val initIterFn = { () => + joinerManager.leftSideJoiner.removeAndReturnOldState().map(_.value) + } + + // NOTE: we need to make sure `antiOutputIter` is evaluated "after" exhausting all of + // elements in `hashJoinOutputIter`, otherwise it may lead to out of sync according to + // the interface contract on StateStore.iterator and end up with correctness issue. + // Please refer SPARK-38684 for more details. + val antiOutputIter = new LazilyInitializingRowIterator(initIterFn) + + // `hashJoinOutputIter` still has to be consumed: draining it is what appends unmatched left + // input to state and removes newly-matched left rows. For left anti it only carries the + // rows which failed the pre-join filter (see `generateFilteredJoinedRow`), which are + // genuine anti output, so it is concatenated as-is. + hashJoinOutputIter ++ antiOutputIter case _ => throwBadJoinTypeException() } - val outputProjection = if (joinType == LeftSemi) { + val outputProjection = if (joinType == LeftSemi || joinType == LeftAnti) { UnsafeProjection.create(output, output) } else { UnsafeProjection.create(left.output ++ right.output, output) @@ -569,8 +608,8 @@ case class StreamingSymmetricHashJoinExec( } // Count rows removed from the other side's state during the join phase - // (e.g., left semi join optimization removes matched left-side rows while processing - // right-side input, instead of a separate eviction pass). + // (e.g., the left semi / left anti join optimization removes matched left-side rows while + // processing right-side input, instead of a separate eviction pass). numRemovedStateRows += joinerManager.numRemovedFromOtherSideDuringJoin allRemovalsTimeMs += timeTakenMs { @@ -587,10 +626,14 @@ case class StreamingSymmetricHashJoinExec( // // For full outer joins, we have already removed unnecessary states from both sides, so // nothing needs to be outputted here. + // + // For left anti joins, the left side state was already consumed and removed while + // generating the unmatched ("anti") output, same as the left side of a left outer join, + // so only the right side remains to be removed greedily. numRemovedStateRows += ( joinType match { case Inner | LeftSemi => joinerManager.removeOldState() - case LeftOuter => joinerManager.rightSideJoiner.removeOldState() + case LeftOuter | LeftAnti => joinerManager.rightSideJoiner.removeOldState() case RightOuter => joinerManager.leftSideJoiner.removeOldState() case FullOuter => 0L case _ => throwBadJoinTypeException() @@ -779,6 +822,9 @@ case class StreamingSymmetricHashJoinExec( (row: InternalRow) => Iterator(generateJoinedRow(row, nullRight)) case RightSide if joinType == RightOuter || joinType == FullOuter => (row: InternalRow) => Iterator(generateJoinedRow(row, nullLeft)) + // A left row which fails the pre-join filter can never satisfy the join condition, so it + // is unmatched by definition and is emitted right away without being added to the state. + case LeftSide if joinType == LeftAnti => (row: InternalRow) => Iterator(row) case _ => (_: InternalRow) => Iterator.empty } @@ -787,12 +833,15 @@ case class StreamingSymmetricHashJoinExec( // unmatched rows) and on the left side of left semi (for matched-rows removal). // For older versions, we do not apply the optimization as it is a behavioral change, // although the optimization is valid for all versions. + // + // Left anti mirrors left semi here: matched left rows are removed from state on match, so no + // matched flag is consulted at eviction time (every survivor is unmatched by construction). val needToUpdateMatchedOnOtherSide = joinType match { case Inner => false case LeftOuter => joinSide == RightSide case RightOuter => joinSide == LeftSide case FullOuter => true - case LeftSemi => joinSide == RightSide + case LeftSemi | LeftAnti => joinSide == RightSide case _ => true } val skipUpdatingMatchedFlag = stateFormatVersion == 4 && !needToUpdateMatchedOnOtherSide @@ -815,8 +864,11 @@ case class StreamingSymmetricHashJoinExec( case _ => (_: InternalRow, joinedRowIter: Iterator[JoinedRow]) => joinedRowIter } + // Both left semi and left anti remove a matched left row from state while processing the + // right side: once matched, a left row can no longer become semi output (already emitted) or + // anti output (permanently disqualified), so there is no reason to keep it. val removeMatchedFromOtherSideState = - joinType == LeftSemi && joinSide == RightSide + (joinType == LeftSemi || joinType == LeftAnti) && joinSide == RightSide nonLateRows.flatMap { row => val thisRow = row.asInstanceOf[UnsafeRow] @@ -826,40 +878,81 @@ case class StreamingSymmetricHashJoinExec( // the case of inner join). if (preJoinFilter(thisRow)) { val key = keyGenerator(thisRow) - // If the join type is Left Semi and this is the right side, we can remove the matched - // row from the other (left) side's state, since the row won't be produced anymore for - // the following input rows. - val joinedRowIter: Iterator[JoinedRow] = if (removeMatchedFromOtherSideState) { - otherSideJoiner.joinStateManager.getJoinedRowsAndRemoveMatched( - key, - thatRow => generateJoinedRow(thisRow, thatRow), - postJoinFilter).map { row => - numRemovedFromOtherSideDuringJoinCount += 1 - row + // If the join type is Left Semi or Left Anti and this is the right side, we can remove + // the matched row from the other (left) side's state, since the row won't be produced + // anymore for the following input rows. + val joinedRowIter: Iterator[JoinedRow] = + if (joinType == LeftAnti && joinSide == LeftSide) { + Iterator.empty + } else if (removeMatchedFromOtherSideState) { + otherSideJoiner.joinStateManager.getJoinedRowsAndRemoveMatched( + key, + thatRow => generateJoinedRow(thisRow, thatRow), + postJoinFilter, + timestampRange = computeTimestampRange(thisRow)).map { row => + numRemovedFromOtherSideDuringJoinCount += 1 + row + } + } else { + otherSideJoiner.joinStateManager.getJoinedRows( + key, + thatRow => generateJoinedRow(thisRow, thatRow), + postJoinFilter, + timestampRange = computeTimestampRange(thisRow), + skipUpdatingMatchedFlag) } + if (joinType == LeftAnti) { + // Left anti join emits nothing while joining, on either side; unmatched left rows are + // emitted later during watermark-based eviction of the left side state. The match + // status is passed explicitly, since it can no longer be inferred from the (empty) + // output iterator: on the left side it drives skip-on-match (a matched left row is not + // stored), mirroring left semi. + // + val matched = if (joinSide == LeftSide) { + // On the left-side probe path, anti semantics only need to know whether any right + // state row matches. Do not use getJoinedRows here: it may scan all rows for the key + // and update right-side matched flags that left anti never consults. + otherSideJoiner.joinStateManager.existsJoinedRow( + key, + thatRow => generateJoinedRow(thisRow, thatRow), + postJoinFilter, + timestampRange = computeTimestampRange(thisRow)) + } else { + // On the right-side input path, draining is required because the iterator removes all + // matched left rows from state; those rows can no longer become anti output. + var foundMatch = false + while (joinedRowIter.hasNext) { + joinedRowIter.next() + foundMatch = true + } + foundMatch + } + new AddingProcessedRowToStateCompletionIterator( + key, thisRow, Iterator.empty, Some(matched)) } else { - otherSideJoiner.joinStateManager.getJoinedRows( - key, - thatRow => generateJoinedRow(thisRow, thatRow), - postJoinFilter, - timestampRange = computeTimestampRange(thisRow), - skipUpdatingMatchedFlag) + val outputIter = generateOutputIter(thisRow, joinedRowIter) + new AddingProcessedRowToStateCompletionIterator(key, thisRow, outputIter) } - val outputIter = generateOutputIter(thisRow, joinedRowIter) - new AddingProcessedRowToStateCompletionIterator(key, thisRow, outputIter) } else { generateFilteredJoinedRow(thisRow) } } } + /** + * @param matchedOverride the match status of `thisRow`, for join types whose output iterator + * does not reflect whether a match was found (left anti emits nothing + * on a match). When empty, the match status is inferred from whether + * the output iterator is non-empty. + */ private class AddingProcessedRowToStateCompletionIterator( key: UnsafeRow, thisRow: UnsafeRow, - subIter: Iterator[InternalRow]) + subIter: Iterator[InternalRow], + matchedOverride: Option[Boolean] = None) extends CompletionIterator[InternalRow, Iterator[InternalRow]](subIter) { - private val iteratorNotEmpty: Boolean = super.hasNext + private val iteratorNotEmpty: Boolean = matchedOverride.getOrElse(super.hasNext) override def completion(): Unit = { // The criteria of whether the input has to be added into state store or not: @@ -867,15 +960,21 @@ case class StreamingSymmetricHashJoinExec( // and the join type is left semi. // For other cases, the input should be added, including the case it's going to be evicted // in this batch. It hasn't yet evaluated with inputs from right side for this batch. - // Refer to the classdoc of SteramingSymmetricHashJoinExec about how stream-stream join + // Refer to the classdoc of StreamingSymmetricHashJoinExec about how stream-stream join // works. - // - Right side: for this side, the evaluation with inputs from left side for this batch - // is done at this point. That said, input can be skipped to be added to the state store - // if input is going to be evicted in this batch. Though, input should be added to the - // state store if it's right outer join or full outer join, as unmatched output is - // handled during state eviction. - val isLeftSemiWithMatch = joinType == LeftSemi && joinSide == LeftSide && iteratorNotEmpty - val shouldAddToState = if (isLeftSemiWithMatch) { + // - Right side: for this side, the evaluation with earlier inputs from left side for this + // batch is done at this point. That said, input can be skipped to be added to the state + // store if input is going to be evicted in this batch. Two join types still need to store + // such rows temporarily: + // - right/full outer, because unmatched output is handled during state eviction; + // - left semi/anti, because the right side is processed first and later left input from + // the same batch must still be able to match these non-late right rows. + // + // Left anti gets the same left-side skip-on-match treatment as left semi: a matched left + // row can never be anti output, so there is no reason to keep it in state. + val isLeftSemiOrAntiWithMatch = + (joinType == LeftSemi || joinType == LeftAnti) && joinSide == LeftSide && iteratorNotEmpty + val shouldAddToState = if (isLeftSemiOrAntiWithMatch) { false } else if (joinSide == LeftSide) { true @@ -885,8 +984,11 @@ case class StreamingSymmetricHashJoinExec( // if the input is not evicted in this batch (hence need to be persisted) val isNotEvictingInThisBatch = !stateKeyWatermarkPredicateFunc(key) && !stateValueWatermarkPredicateFunc(thisRow) + val isRightSideOfLeftSemiOrAnti = + joinType == LeftSemi || joinType == LeftAnti isNotEvictingInThisBatch || + isRightSideOfLeftSemiOrAnti || // if the input is producing "unmatched row" in this batch ( (joinType == RightOuter && !iteratorNotEmpty) || @@ -1084,6 +1186,18 @@ case class StreamingSymmetricHashJoinExec( override def next(): JoinedRow = iter.next() } + /** + * Same as [[LazilyInitializingJoinedRowIterator]], but for join types which emit rows from a + * single side (left anti) rather than joined rows. + */ + private class LazilyInitializingRowIterator( + initFn: () => Iterator[UnsafeRow]) extends Iterator[UnsafeRow] { + private lazy val iter: Iterator[UnsafeRow] = initFn() + + override def hasNext: Boolean = iter.hasNext + override def next(): UnsafeRow = iter.next() + } + // If `STATE_STORE_SKIP_NULLS_FOR_STREAM_STREAM_JOINS` is enabled, counting the number // of skipped null values as custom metric of stream join operator. override def customStatefulOperatorMetrics: Seq[StatefulOperatorCustomMetric] = diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala index 97878b33c778f..46a8df46eeb21 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala @@ -83,6 +83,23 @@ trait SymmetricHashJoinStateManager { timestampRange: Option[(Long, Long)] = None, skipUpdatingMatchedFlag: Boolean = false): Iterator[JoinedRow] + /** + * Return whether any joined row for the given key satisfies the provided predicate. + * + * This method does not update matched flags and may stop at the first matching row. It is + * suitable for join paths that only need match existence, such as probing the right side for a + * left anti join input row. + * + * @param timestampRange Optional optimization hint as (minTimestamp, maxTimestamp), both + * inclusive. Derived classes may use it to reduce scan scope but are free to ignore it. + * The predicate must produce correct output regardless of whether this hint is leveraged. + */ + def existsJoinedRow( + key: UnsafeRow, + generateJoinedRow: InternalRow => JoinedRow, + predicate: JoinedRow => Boolean, + timestampRange: Option[(Long, Long)] = None): Boolean + /** * Retrieve all joined rows for the given key and remove the matched rows from state. The joined * rows are generated with the provided generateJoinedRow function and filtered with the provided @@ -94,11 +111,14 @@ trait SymmetricHashJoinStateManager { * * It is caller's responsibility to consume the whole iterator. * - * NOTE: For the rows which already have been marked as matched in the state, this method removes - * them from the state without returning them. Under normal operation, this should not happen and - * this should not be an issue, but it can occur if the join type was changed during query - * restart. We do not define an expected behavior for such changes, so we just optimize it rather - * than trying to provide some best-effort results. + * @param timestampRange optional inclusive timestamp range for implementations that can prune the + * state scan by event time. + * + * NOTE: For the rows which already have been marked as matched in the scanned state, this method + * removes them from the state without returning them. Under normal operation, this should not + * happen and this should not be an issue, but it can occur if the join type was changed during + * query restart. We do not define an expected behavior for such changes, so we just optimize it + * rather than trying to provide some best-effort results. * * NOTE2: There is a further optimization opportunity -- if a row does not pass the predicate * (postJoinFilter), it may never match in future batches as long as expressions are @@ -110,7 +130,8 @@ trait SymmetricHashJoinStateManager { def getJoinedRowsAndRemoveMatched( key: UnsafeRow, generateJoinedRow: InternalRow => JoinedRow, - predicate: JoinedRow => Boolean): Iterator[JoinedRow] + predicate: JoinedRow => Boolean, + timestampRange: Option[(Long, Long)] = None): Iterator[JoinedRow] /** * Provide all key-value pairs in the state manager. @@ -515,10 +536,42 @@ class SymmetricHashJoinStateManagerV4( ret.filter(_ != null) } + override def existsJoinedRow( + key: UnsafeRow, + generateJoinedRow: InternalRow => JoinedRow, + predicate: JoinedRow => Boolean, + timestampRange: Option[(Long, Long)]): Boolean = { + def existsInValues(valuesAndMatched: Iterator[ValueAndMatchPair]): Boolean = { + valuesAndMatched.exists { vmp => + predicate(generateJoinedRow(vmp.value)) + } + } + + extractEventTimeFnFromKey(key) match { + case Some(ts) => + existsInValues(keyWithTsToValues.get(key, ts)) + + case _ => + val (minTs, maxTs) = timestampRange.getOrElse((Long.MinValue, Long.MaxValue)) + val valuesByTimestamp = keyWithTsToValues.getValuesInRange(key, minTs, maxTs) + try { + valuesByTimestamp.exists { result => + existsInValues(result.values.iterator) + } + } finally { + valuesByTimestamp match { + case nextIterator: NextIterator[_] => nextIterator.closeIfNeeded() + case _ => + } + } + } + } + override def getJoinedRowsAndRemoveMatched( key: UnsafeRow, generateJoinedRow: InternalRow => JoinedRow, - predicate: JoinedRow => Boolean): Iterator[JoinedRow] = { + predicate: JoinedRow => Boolean, + timestampRange: Option[(Long, Long)]): Iterator[JoinedRow] = { def getJoinedRowsFromTsAndValues( ts: Long, valuesAndMatched: Array[ValueAndMatchPair]): Iterator[JoinedRow] = { @@ -578,7 +631,8 @@ class SymmetricHashJoinStateManagerV4( getJoinedRowsFromTsAndValues(ts, valuesAndMatchedIter.toArray) case _ => - keyWithTsToValues.getValues(key).flatMap { result => + val (minTs, maxTs) = timestampRange.getOrElse((Long.MinValue, Long.MaxValue)) + keyWithTsToValues.getValuesInRange(key, minTs, maxTs).flatMap { result => val ts = result.timestamp val valuesAndMatched = result.values.toArray getJoinedRowsFromTsAndValues(ts, valuesAndMatched) @@ -1191,7 +1245,8 @@ abstract class SymmetricHashJoinStateManagerBase( override def getJoinedRowsAndRemoveMatched( key: UnsafeRow, generateJoinedRow: InternalRow => JoinedRow, - predicate: JoinedRow => Boolean): Iterator[JoinedRow] = { + predicate: JoinedRow => Boolean, + timestampRange: Option[(Long, Long)]): Iterator[JoinedRow] = { new NextIterator[JoinedRow] { private var numValues: Long = keyToNumValues.get(key) private var index: Long = 0L @@ -1263,6 +1318,17 @@ abstract class SymmetricHashJoinStateManagerBase( }.filter(_ != null) } + override def existsJoinedRow( + key: UnsafeRow, + generateJoinedRow: InternalRow => JoinedRow, + predicate: JoinedRow => Boolean, + timestampRange: Option[(Long, Long)]): Boolean = { + val numValues = keyToNumValues.get(key) + keyWithIndexToValue.getAll(key, numValues).exists { keyIdxToValue => + predicate(generateJoinedRow(keyIdxToValue.value)) + } + } + /** Remove using a predicate on keys. */ override def evictByKeyCondition(removalCondition: UnsafeRow => Boolean): Long = { var numRemoved = 0L diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManagerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManagerSuite.scala index 0460a41f4cc5b..036f93201093f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManagerSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManagerSuite.scala @@ -544,6 +544,28 @@ class SymmetricHashJoinStateManagerEventTimeInKeySuite } } + versionsInTest.foreach { ver => + test(s"StreamingJoinStateManager V$ver - existsJoinedRow returns match existence") { + withJoinStateManager( + inputValueAttributes, joinKeyExpressions, stateFormatVersion = ver) { manager => + implicit val mgr = manager + + append(20, 2) + append(20, 3) + + val dummyRow = new GenericInternalRow(0) + assert(manager.existsJoinedRow( + toJoinKeyRow(20), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 2)) + assert(!manager.existsJoinedRow( + toJoinKeyRow(20), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 99)) + } + } + } + // V1 excluded: V1 converter does not persist matched flags (SPARK-26154) versionsInTest.filter(_ >= 2).foreach { ver => test(s"StreamingJoinStateManager V$ver - matched flag update + eviction roundtrip") { @@ -586,6 +608,47 @@ class SymmetricHashJoinStateManagerEventTimeInKeySuite } } + // V1 excluded: V1 converter does not persist matched flags (SPARK-26154) + versionsInTest.filter(_ >= 2).foreach { ver => + test(s"StreamingJoinStateManager V$ver - existsJoinedRow does not update matched flag") { + withTempDir { checkpointDir => + withJoinStateManagerWithCheckpointDir( + inputValueAttributes, joinKeyExpressions, ver, + checkpointDir, storeVersion = 0, changelogCheckpoint = false) { manager => + implicit val mgr = manager + + append(20, 2) + append(20, 3) + + val dummyRow = new GenericInternalRow(0) + assert(manager.existsJoinedRow( + toJoinKeyRow(20), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 2)) + assert(!manager.existsJoinedRow( + toJoinKeyRow(20), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 99)) + + mgr.commit() + } + + withJoinStateManagerWithCheckpointDir( + inputValueAttributes, joinKeyExpressions, ver, + checkpointDir, storeVersion = 1, changelogCheckpoint = false) { manager => + implicit val mgr = manager + + val evicted = removeAndReturnByKey(25) + val matchedByValue = evicted.map(p => (toValueInt(p.value), p.matched)).toMap + assert(matchedByValue(2) === false) + assert(matchedByValue(3) === false) + + mgr.commit() + } + } + } + } + // V1 excluded: V1 converter does not persist matched flags (SPARK-26154) versionsInTest.filter(_ >= 2).foreach { ver => test(s"StreamingJoinStateManager V$ver - getJoinedRowsAndRemoveMatched partial") { @@ -914,6 +977,27 @@ class SymmetricHashJoinStateManagerEventTimeInValueSuite } } + versionsInTest.foreach { ver => + test(s"StreamingJoinStateManager V$ver - existsJoinedRow returns match existence") { + withJoinStateManager( + inputValueAttributes, joinKeyExpressions, stateFormatVersion = ver) { manager => + implicit val mgr = manager + + appendAndTest(40, 100, 200, 300) + + val dummyRow = new GenericInternalRow(0) + assert(manager.existsJoinedRow( + toJoinKeyRow(40), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 100)) + assert(!manager.existsJoinedRow( + toJoinKeyRow(40), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 999)) + } + } + } + // V1 excluded: V1 converter does not persist matched flags (SPARK-26154) versionsInTest.filter(_ >= 2).foreach { ver => test(s"StreamingJoinStateManager V$ver - matched flag update + eviction roundtrip") { @@ -952,6 +1036,47 @@ class SymmetricHashJoinStateManagerEventTimeInValueSuite } } + // V1 excluded: V1 converter does not persist matched flags (SPARK-26154) + versionsInTest.filter(_ >= 2).foreach { ver => + test(s"StreamingJoinStateManager V$ver - existsJoinedRow does not update matched flag") { + withTempDir { checkpointDir => + withJoinStateManagerWithCheckpointDir( + inputValueAttributes, joinKeyExpressions, ver, + checkpointDir, storeVersion = 0, changelogCheckpoint = false) { manager => + implicit val mgr = manager + + appendAndTest(40, 100, 200, 300) + + val dummyRow = new GenericInternalRow(0) + assert(manager.existsJoinedRow( + toJoinKeyRow(40), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 100, + timestampRange = Some((100L, 100L)))) + assert(!manager.existsJoinedRow( + toJoinKeyRow(40), + row => new JoinedRow(row, dummyRow), + jr => jr.getInt(1) == 999, + timestampRange = Some((100L, 100L)))) + + mgr.commit() + } + + withJoinStateManagerWithCheckpointDir( + inputValueAttributes, joinKeyExpressions, ver, + checkpointDir, storeVersion = 1, changelogCheckpoint = false) { manager => + implicit val mgr = manager + + val evicted = removeAndReturnByValue(125) + val matchedByValue = evicted.map(p => (toValueInt(p.value), p.matched)).toMap + assert(matchedByValue(100) === false) + + mgr.commit() + } + } + } + } + // V1 excluded: V1 converter does not persist matched flags (SPARK-26154) versionsInTest.filter(_ >= 2).foreach { ver => test(s"StreamingJoinStateManager V$ver - getJoinedRowsAndRemoveMatched partial") { @@ -1063,6 +1188,20 @@ class SymmetricHashJoinStateManagerEventTimeInValueSuite ).map(_.getInt(1)).toSeq.sorted } + private def removeMatchedRowTimestamps( + key: Int, + range: Option[(Long, Long)], + predicate: JoinedRow => Boolean = (_: JoinedRow) => true)( + implicit manager: SymmetricHashJoinStateManager): Seq[Int] = { + val dummyRow = new GenericInternalRow(0) + manager.getJoinedRowsAndRemoveMatched( + toJoinKeyRow(key), + row => new JoinedRow(row, dummyRow), + predicate, + timestampRange = range + ).map(_.getInt(1)).toSeq.sorted + } + test("StreamingJoinStateManager V4 - getJoinedRows with timestampRange") { withJoinStateManager( inputValueAttributes, joinKeyExpressions, stateFormatVersion = 4) { manager => @@ -1082,6 +1221,52 @@ class SymmetricHashJoinStateManagerEventTimeInValueSuite } } + test("StreamingJoinStateManager V4 - existsJoinedRow with timestampRange") { + withJoinStateManager( + inputValueAttributes, joinKeyExpressions, stateFormatVersion = 4) { manager => + implicit val mgr = manager + val dummyRow = new GenericInternalRow(0) + + Seq(10, 20, 30, 40, 50).foreach(append(40, _)) + + def existsInRange(range: Option[(Long, Long)]): Boolean = { + manager.existsJoinedRow( + toJoinKeyRow(40), + row => new JoinedRow(row, dummyRow), + _ => true, + timestampRange = range) + } + + assert(existsInRange(Some((20L, 40L)))) + assert(existsInRange(Some((20L, 20L)))) + assert(existsInRange(Some((25L, 35L)))) + assert(!existsInRange(Some((60L, 100L)))) + assert(!existsInRange(Some((0L, 5L)))) + assert(existsInRange(None)) + } + } + + test("StreamingJoinStateManager V4 - getJoinedRowsAndRemoveMatched with timestampRange") { + withJoinStateManager( + inputValueAttributes, joinKeyExpressions, stateFormatVersion = 4) { manager => + implicit val mgr = manager + + manager.append(toJoinKeyRow(40), toInputValue(40, 5), matched = true) + Seq(10, 20, 30, 40, 50).foreach(append(40, _)) + + assert(removeMatchedRowTimestamps(40, Some((20L, 40L))) === Seq(20, 30, 40)) + assert(get(40) === Seq(5, 10, 50)) + + Seq(20, 30, 40).foreach(append(40, _)) + val matched = removeMatchedRowTimestamps( + 40, + Some((20L, 40L)), + joinedRow => joinedRow.getInt(1) != 30) + assert(matched === Seq(20, 40)) + assert(get(40) === Seq(5, 10, 30, 50)) + } + } + test("StreamingJoinStateManager V4 - timestampRange with multiple values per timestamp") { withJoinStateManager( inputValueAttributes, joinKeyExpressions, stateFormatVersion = 4) { manager => 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 c0983f338abe5..708a7b85e26ed 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 @@ -157,7 +157,7 @@ abstract class StreamingJoinSuite val windowed2 = df2 .select($"key", window($"rightTime", "10 second"), $"rightValue") val joined = windowed1.join(windowed2, Seq("key", "window"), joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select($"key", $"window.end".cast("long"), $"leftValue") } else { joined.select($"key", $"window.end".cast("long"), $"leftValue", @@ -185,7 +185,7 @@ abstract class StreamingJoinSuite && $"leftValue" > 4, joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue") } else if (joinType == "left_outer") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue", @@ -219,7 +219,7 @@ abstract class StreamingJoinSuite && $"rightValue".cast("int") > 7, joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue") } else if (joinType == "left_outer") { joined.select(left("key"), left("window.end").cast("long"), $"leftValue", @@ -239,16 +239,19 @@ abstract class StreamingJoinSuite joinType: String, watermark: String = "10 seconds", lowerBound: String = "interval 5 seconds", - upperBound: String = "interval 5 seconds") + upperBound: String = "interval 5 seconds", + leftWatermark: Boolean = true) : (MemoryStream[(Int, Int)], MemoryStream[(Int, Int)], DataFrame) = { val leftInput = MemoryStream[(Int, Int)] val rightInput = MemoryStream[(Int, Int)] - val df1 = leftInput.toDF().toDF("leftKey", "time") + val df1Base = leftInput.toDF().toDF("leftKey", "time") .select($"leftKey", timestamp_seconds($"time") as "leftTime", ($"leftKey" * 2) as "leftValue") - .withWatermark("leftTime", watermark) + // Left semi/outer/anti require range bounds over watermarked attributes on both sides. Allow + // tests to omit the left watermark to exercise that analyzer path. + val df1 = if (leftWatermark) df1Base.withWatermark("leftTime", watermark) else df1Base val df2 = rightInput.toDF().toDF("rightKey", "time") .select($"rightKey", timestamp_seconds($"time") as "rightTime", @@ -262,7 +265,7 @@ abstract class StreamingJoinSuite s"leftTime BETWEEN rightTime - $lowerBound AND rightTime + $upperBound"), joinType) - val select = if (joinType == "left_semi") { + val select = if (joinType == "left_semi" || joinType == "left_anti") { joined.select($"leftKey", $"leftTime".cast("int")) } else { joined.select($"leftKey", $"rightKey", $"leftTime".cast("int"), @@ -272,6 +275,23 @@ abstract class StreamingJoinSuite (leftInput, rightInput, select) } + protected def setupEqualityJoinWithZeroWatermark(joinType: String) + : (MemoryStream[(String, Int)], MemoryStream[(String, Int)], DataFrame) = { + val leftInput = MemoryStream[(String, Int)] + val rightInput = MemoryStream[(String, Int)] + + val left = leftInput.toDF() + .selectExpr("_1 AS key", "timestamp_seconds(_2) AS eventTime") + .withWatermark("eventTime", "0 seconds") + val right = rightInput.toDF() + .selectExpr("_1 AS key", "timestamp_seconds(_2) AS eventTime") + .withWatermark("eventTime", "0 seconds") + val joined = left.join(right, Seq("key", "eventTime"), joinType) + .selectExpr("key", "CAST(eventTime AS long) AS eventTime") + + (leftInput, rightInput, joined) + } + protected def setupSelfJoin(joinType: String) : (MemoryStream[(Int, Long)], DataFrame) = { @@ -1983,6 +2003,57 @@ abstract class StreamingOuterJoinSuite extends StreamingOuterJoinBase { } } } + + test("left outer join with a range condition requires watermarked bounds on both sides") { + // Only the right side is watermarked. With a range condition (no watermark in the join keys), + // the operator never builds a left-side eviction predicate, so the unmatched (null-extended) + // left rows would never be emitted. This must be rejected at analysis time rather than silently + // dropping the outer output. + val (_, _, joined) = setupJoinWithRangeCondition("left_outer", leftWatermark = false) + + val e = intercept[AnalysisException] { + joined.writeStream.format("memory").queryName("leftOuterRightWatermarkOnly") + .outputMode(OutputMode.Append()).start() + } + assert(e.getMessage.contains( + "requires a range bound between watermarked attributes on both sides")) + } + + test("left outer join with an equi join key watermarked on the left only is rejected") { + // The join key is watermarked on the left only, so the right side is not late-filtered on the + // eviction key: a late right row could match a left row already emitted as a null-extended + // outer row. Rejected at analysis time. + val leftInput = MemoryStream[(Int, Int)] + val rightInput = MemoryStream[(Int, Int)] + val df1 = leftInput.toDF().toDF("key", "time") + .select($"key", timestamp_seconds($"time") as "leftTime") + .withWatermark("leftTime", "10 seconds") + val df2 = rightInput.toDF().toDF("key", "time") + .select($"key", timestamp_seconds($"time") as "rightTime") + val joined = df1.join(df2, expr("leftTime = rightTime"), "left_outer") + + val e = intercept[AnalysisException] { + joined.writeStream.format("memory").queryName("leftOuterLeftKeyWatermarkOnly") + .outputMode(OutputMode.Append()).start() + } + assert(e.getMessage.contains("requires watermarks on both join keys used for state eviction")) + } + + test("stricter watermark requirements can be disabled for left outer join") { + // With the kill switch off, the right-only range config that is otherwise rejected is accepted + // again (falling back to the previous, looser behavior). + withSQLConf( + SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED.key -> "false") { + val (_, _, joined) = setupJoinWithRangeCondition("left_outer", leftWatermark = false) + val query = joined.writeStream.format("memory") + .queryName("leftOuterFallback").outputMode(OutputMode.Append()).start() + try { + query.processAllAvailable() + } finally { + query.stop() + } + } + } } @SlowSQLTest @@ -2426,6 +2497,36 @@ abstract class StreamingLeftSemiJoinBase extends StreamingJoinSuite { ) } + testWithAppendAndUpdate("left semi join matches same-batch right row between late and " + + "eviction watermarks") { outputMode => + withTempDir { checkpoint => + withSQLConf(SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key -> "false") { + val (leftInput, rightInput, joined) = + setupEqualityJoinWithZeroWatermark("left_semi") + + testStream(joined, outputMode)( + StartStream(checkpointLocation = checkpoint.getCanonicalPath), + // batch 0 + // WM: late record = 0, eviction = 0 + MultiAddData( + (leftInput, Seq(("a", 1), ("b", 2))), + (rightInput, Seq(("b", 2), ("c", 1))) + ), + CheckNewAnswer(("b", 2)), + // batch 1 + // WM: late record = 0, eviction = 2. Right ("d", 1) is not late, but it is + // evicting in this batch. It still has to be visible to left ("d", 1) from the same + // batch before watermark cleanup removes it. + MultiAddData( + (rightInput, Seq(("d", 1))), + (leftInput, Seq(("d", 1))) + ), + CheckNewAnswer(("d", 1)) + ) + } + } + } + testWithAppendAndUpdate("self left semi join") { outputMode => val (inputStream, query) = setupSelfJoin("left_semi") @@ -2598,6 +2699,36 @@ abstract class StreamingLeftSemiJoinSuite extends StreamingLeftSemiJoinBase { */ } } + + test("left semi join with a range condition requires watermarked bounds on both sides") { + // With only the right side watermarked and a range condition (no watermark in the join keys), + // the left state is never evicted, so never-matched left rows accumulate without bound. This is + // rejected at analysis time. + val (_, _, joined) = setupJoinWithRangeCondition("left_semi", leftWatermark = false) + + val e = intercept[AnalysisException] { + joined.writeStream.format("memory").queryName("leftSemiRightWatermarkOnly") + .outputMode(OutputMode.Append()).start() + } + assert(e.getMessage.contains( + "requires a range bound between watermarked attributes on both sides")) + } + + test("stricter watermark requirements can be disabled for left semi join") { + // With the kill switch off, the right-only range config falls back to the previous behavior and + // is accepted again. + withSQLConf( + SQLConf.STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED.key -> "false") { + val (_, _, joined) = setupJoinWithRangeCondition("left_semi", leftWatermark = false) + val query = joined.writeStream.format("memory") + .queryName("leftSemiFallback").outputMode(OutputMode.Append()).start() + try { + query.processAllAvailable() + } finally { + query.stop() + } + } + } } // Concrete single-mode suites for parallel CI execution and failure isolation. @@ -2632,6 +2763,257 @@ class StreamingFullOuterJoinWithoutVCFSuite extends StreamingFullOuterJoinSuite override protected def testMode = Mode.WithoutVCF } +abstract class StreamingLeftAntiJoinBase extends StreamingJoinSuite { + + import testImplicits._ + + test("windowed left anti join") { + withTempDir { checkpointDir => + val (leftInput, rightInput, joined) = setupWindowedJoin("left_anti") + + testStream(joined, OutputMode.Append())( + StartStream(checkpointLocation = checkpointDir.getCanonicalPath), + MultiAddData(leftInput, 1, 2, 3, 4, 5)(rightInput, 3, 4, 5, 6, 7), + // Nothing is emitted yet: left 1 and 2 are unmatched, but the watermark cannot yet rule + // out a future match for them. Left 3, 4, 5 matched and are suppressed forever. + CheckNewAnswer(), + // states + // left: 1, 2 (3, 4, 5 matched right rows, so they are not buffered -- see left semi) + // right: 3, 4, 5, 6, 7 + assertNumStateRows( + total = Seq(7), updated = Seq(7), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + MultiAddData(leftInput, 21)(rightInput, 22), + // Watermark = 11, so window=[0,10] is evicted from the left side: the surviving unmatched + // left rows 1 and 2 are emitted now (3, 4, 5 were never buffered, having matched). + CheckNewAnswer(Row(1, 10, 2), Row(2, 10, 4)), + // states + // left: 21 + // right: 22 + // + // states evicted + // left: 1, 2 (below watermark) + // right: 3, 4, 5, 6, 7 (below watermark) + // + // Only the 5 right side rows are reported as removed: the left side rows are evicted + // through removeAndReturnOldState while generating the anti output, which does not feed + // numRemovedStateRows. Left outer join reports removals the same way. + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(5))), + StopStream, + // Restart the join query from the same checkpoint + StartStream(checkpointLocation = checkpointDir.getCanonicalPath), + AddData(leftInput, 22), + // Left 22 matches right 22, so it is not buffered and never emitted (as in left semi). + CheckNewAnswer(), + // states + // left: 21 + // right: 22 + assertNumStateRows( + total = Seq(2), updated = Seq(0), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + StopStream, + // Restart the query from the same checkpoint + StartStream(checkpointLocation = checkpointDir.getCanonicalPath), + AddData(leftInput, 1), + // Row not added as 1 < state key watermark = 12. + CheckNewAnswer(), + // states + // left: 21 + // right: 22 + assertNumStateRows( + total = Seq(2), updated = Seq(0), + droppedByWatermark = Seq(1), removed = Some(Seq(0))) + ) + } + } + + test("left anti join does not emit a left row matched by a right row in a later batch") { + val (leftInput, rightInput, joined) = setupWindowedJoin("left_anti") + + testStream(joined, OutputMode.Append())( + AddData(leftInput, 3), + // Unmatched so far, but not yet provably unmatched, so nothing is emitted. This is the + // window in which a match can still arrive and cancel the eventual anti output -- the + // positive "emitted once the watermark passes" case is covered by "windowed left anti join". + CheckNewAnswer(), + // A matching right row arrives in a later batch, so left 3 must never be emitted. + AddData(rightInput, 3), + CheckNewAnswer(), + // Advance the watermark past window=[0,10]. + MultiAddData(leftInput, 21)(rightInput, 21), + CheckNewAnswer(), + // Left 21 matched right 21, and left 3 was matched before eviction, so still nothing. + MultiAddData(leftInput, 31)(rightInput, 31), + CheckNewAnswer() + ) + } + + test("left anti early state exclusion on left") { + val (leftInput, rightInput, joined) = setupWindowedJoinWithLeftCondition("left_anti") + + testStream(joined, OutputMode.Append())( + MultiAddData(leftInput, 1, 2, 3)(rightInput, 3, 4, 5), + // The left rows with leftValue <= 4 (i.e. left 1 and 2) fail the pre-join filter, so they can + // never match and are emitted immediately without being added to the state. + CheckNewAnswer(Row(1, 10, 2), Row(2, 10, 4)), + // states + // left: empty (3 matched right 3 in the same batch, so it is not buffered) + // right: 3, 4, 5 + assertNumStateRows( + total = Seq(3), updated = Seq(3), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + // Left 3 matched, so advancing the watermark must not produce an anti row for it. + MultiAddData(leftInput, 20)(rightInput, 21), + CheckNewAnswer(), + // states + // left: 20 + // right: 21 + // + // states evicted + // left: nothing (3 was never buffered, having matched) + // right: 3, 4, 5 (below watermark) + // + // Only the 3 right side rows are counted as removed. + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(3))) + ) + } + + test("left anti early state exclusion on right") { + val (leftInput, rightInput, joined) = setupWindowedJoinWithRightCondition("left_anti") + + testStream(joined, OutputMode.Append())( + MultiAddData(leftInput, 3, 4, 5)(rightInput, 1, 2, 3), + // The right rows with rightValue <= 7 (i.e. right 1 and 2) fail the pre-join filter, so they + // are not added to the state and cannot suppress any left row. + CheckNewAnswer(), + // states + // left: 4, 5 (3 matched right 3, so it is not buffered) + // right: 3 + assertNumStateRows( + total = Seq(3), updated = Seq(3), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + // Advance the watermark: left 4 and 5 were never matched, so they are emitted now. + MultiAddData(leftInput, 20)(rightInput, 21), + CheckNewAnswer(Row(4, 10, 8), Row(5, 10, 10)), + // states + // left: 20 + // right: 21 + // + // states evicted + // left: 4, 5 (below watermark) + // right: 3 (below watermark) + // + // Only the single right side row is counted as removed - the left side rows are evicted + // through removeAndReturnOldState while generating the anti output, which does not feed + // numRemovedStateRows. + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(1))) + ) + } + + test("left anti join with watermark range condition") { + val (leftInput, rightInput, joined) = setupJoinWithRangeCondition("left_anti") + + testStream(joined, OutputMode.Append())( + AddData(leftInput, (1, 5), (3, 5)), + // Neither left row is provably unmatched yet. + CheckNewAnswer(), + // states + // left: (1, 5), (3, 5) + // right: nothing + assertNumStateRows( + total = Seq(2), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(0))), + AddData(rightInput, (1, 10), (2, 5)), + // Right (1, 10) satisfies the range condition against left (1, 5), so that left row is + // removed from state and must never be emitted, exactly as in left semi. + CheckNewAnswer(), + // states + // left: (3, 5) ((1, 5) removed on match) + // right: (1, 10), (2, 5) + assertNumStateRows( + total = Seq(3), updated = Seq(2), + droppedByWatermark = Seq(0), removed = Some(Seq(1))), + // Advance the watermark to 20 by adding rows with event time 30 on both sides. A left row is + // only safe to evict once no future right row can match it, which the range condition puts + // at leftTime + 5 < watermark, so the left rows with leftTime < 15 are evicted here: + // (3, 5) was never matched so it is emitted. (1, 5) is not among them -- it was already + // removed from state when it matched in the previous batch. + AddData(leftInput, (1, 30)), + CheckNewAnswer(), + AddData(rightInput, (0, 30)), + CheckNewAnswer(Row(3, 5)), + // Advance the watermark to 50, which evicts the left rows with leftTime < 45. Left (1, 30) + // never matched -- the only right row with key 1 was evicted long before -- so it is + // emitted now. + AddData(leftInput, (2, 60)), + CheckNewAnswer(), + AddData(rightInput, (0, 60)), + CheckNewAnswer(Row(1, 30)) + ) + } + + test("left anti join suppresses same-batch right row between late and eviction watermarks") { + withTempDir { checkpoint => + withSQLConf(SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key -> "false") { + val (leftInput, rightInput, joined) = setupEqualityJoinWithZeroWatermark("left_anti") + + testStream(joined, OutputMode.Append())( + StartStream(checkpointLocation = checkpoint.getCanonicalPath), + // batch 0 + // WM: late record = 0, eviction = 0 + MultiAddData( + (leftInput, Seq(("a", 1), ("b", 2))), + (rightInput, Seq(("b", 2), ("c", 1))) + ), + CheckNewAnswer(), + // batch 1 + // WM: late record = 0, eviction = 2. Right ("d", 1) is not late, but it is evicting + // in this batch. It must suppress left ("d", 1) from the same batch before cleanup. + MultiAddData( + (rightInput, Seq(("d", 1))), + (leftInput, Seq(("d", 1))) + ), + CheckNewAnswer(("a", 1)) + ) + } + } + } +} + +abstract class StreamingLeftAntiJoinSuite extends StreamingLeftAntiJoinBase { + + test("left anti join is not supported in Update output mode") { + val (_, _, joined) = setupWindowedJoin("left_anti") + + val e = intercept[AnalysisException] { + joined.writeStream.format("memory").queryName("leftAntiUpdate") + .outputMode(OutputMode.Update()).start() + } + assert(e.getMessage.contains("LeftAnti join between two streaming DataFrames/Datasets " + + "is not supported in Update output mode, only in Append output mode")) + } + + test("left anti join with a range condition requires watermarked bounds on both sides") { + // Only the right side is watermarked. With a range condition (no watermark in the join keys), + // the operator would never build a left-side eviction predicate, so no anti row would ever be + // emitted. This must be rejected at analysis rather than silently producing no output. + val (_, _, joined) = setupJoinWithRangeCondition("left_anti", leftWatermark = false) + + val e = intercept[AnalysisException] { + joined.writeStream.format("memory").queryName("leftAntiRightWatermarkOnly") + .outputMode(OutputMode.Append()).start() + } + assert(e.getMessage.contains( + "requires a range bound between watermarked attributes on both sides")) + } +} + @SlowSQLTest class StreamingLeftSemiJoinWithVCFSuite extends StreamingLeftSemiJoinSuite { override protected def testMode = Mode.WithVCF @@ -2641,3 +3023,13 @@ class StreamingLeftSemiJoinWithVCFSuite extends StreamingLeftSemiJoinSuite { class StreamingLeftSemiJoinWithoutVCFSuite extends StreamingLeftSemiJoinSuite { override protected def testMode = Mode.WithoutVCF } + +@SlowSQLTest +class StreamingLeftAntiJoinWithVCFSuite extends StreamingLeftAntiJoinSuite { + override protected def testMode = Mode.WithVCF +} + +@SlowSQLTest +class StreamingLeftAntiJoinWithoutVCFSuite extends StreamingLeftAntiJoinSuite { + override protected def testMode = Mode.WithoutVCF +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala index 6d4a97861efef..fa064e64fec6e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala @@ -476,3 +476,8 @@ class StreamingFullOuterJoinV4Suite class StreamingLeftSemiJoinV4Suite extends StreamingLeftSemiJoinBase with TestWithV4StateFormat + +@SlowSQLTest +class StreamingLeftAntiJoinV4Suite + extends StreamingLeftAntiJoinBase + with TestWithV4StateFormat
StreamStaticStreamStatic Inner Supported, not stateful
Supported, not stateful
StaticStreamLeft AntiSupported, not stateful
StaticStream Inner Supported, not stateful
Not supported
StreamStreamLeft AntiNot supported
StreamStream Inner Supported, optionally specify watermark on both sides + @@ -1398,8 +1461,9 @@ regarding watermark delays and whether data will be dropped or not.
Left Outer - 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.
Left Semi - 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. +
Left Anti + 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.
Queries with joinsAppendAppend, Update - 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 support matrix in the Join Operations section for more details on what types of joins are supported.