You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2026/10/21 20:33:20 UTC
(spark) branch branch-4.x updated: [SPARK-60375][SS] Base left semi join output watermarks on emitted event time
This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 6999ea40839f [SPARK-60375][SS] Base left semi join output watermarks on emitted event time
6999ea40839f is described below
commit 6999ea40839f7aa95a88f8efa31825450ba2114a
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Thu Oct 22 05:32:56 2026 +0900
[SPARK-60375][SS] Base left semi join output watermarks on emitted event time
### What changes were proposed in this pull request?
Derive a left semi stream-stream join's output watermark from the event-time column that survives in its output. A left semi join returns only left columns:
- If the output contains a watermark-bearing event-time column, use the minimum of the left state watermark and input watermark.
- If no output column carries watermark metadata, return no output watermark (`None`). A watermark defined only on the right is lost with the right columns.
### Why are the changes needed?
An output watermark describes the event times a downstream operator may receive. A left semi join emits left rows, including buffered left rows that match a later right input. Its output watermark should therefore follow the bound on that left event-time column.
The existing calculation also includes the right-state watermark. Retaining older right rows is necessary for future matches, but their timestamps are never emitted by a left semi join. Including that bound unnecessarily delays downstream window completion and state cleanup.
For `L > R - INTERVAL 5 SECONDS AND L < R + INTERVAL 30 SECONDS`, an input watermark of 120 seconds gives left and right state watermarks of 115 and 90 seconds. Using the bound on emitted left event time advances the output watermark to 115 seconds and closes the downstream window `[100, 110)`. Including the right-state bound holds the output watermark at 90 seconds and leaves the window buffered.
When the left output has no watermark-bearing column, reporting no output watermark explicitly represents that loss.
### Does this PR introduce _any_ user-facing change?
Yes. Append-mode aggregations after bounded left semi joins can emit completed windows and clean up downstream state earlier. A left semi join without watermark metadata on its output no longer propagates a numeric output watermark.
This is an improvement to watermark propagation and downstream progress. Join matching, state eviction, and checkpoint/state layout are unchanged.
### How was this patch tested?
Added equality and range streaming-query cases with and without a watermark-bearing left output column in `StreamingLeftSemiJoinBase`. The four cases run across six state-format/provider configurations, for 24 cases. They check both the output-watermark value and its absence when watermark metadata is lost.
Added an end-to-end left semi join followed by windowed aggregation in `MultiStatefulOperatorsSuite`, including a checkpoint restart. The three-batch scenario verifies that window `[100, 110)` is emitted when the output watermark reaches 115 seconds. It runs in six configurations: V2 with HDFS/RocksDB, and V3/V4 with RocksDB changelog checkpointing off/on. This regression fails with the original both-side calculation and passes with the change.
The 24 metadata cases and six end-to-end configurations passed. SQL Test scalastyle passed with zero errors and warnings.
```
build/sbt 'sql/testOnly *StreamingLeftSemiJoin*Suite -- -z "left semi output watermark"'
build/sbt 'sql/testOnly *MultiStatefulOperatorsSuite -- -z "left semi output watermark closes"'
build/sbt 'sql/Test/scalastyle'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59626 from HeartSaVioR/jungtaek-lim_data/fix-left-semi-output-watermark.
Authored-by: Jungtaek Lim <ka...@gmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit 080731eb2a69ee30eb24023b6868b7db36a28abf)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../join/StreamingSymmetricHashJoinExec.scala | 19 ++++---
.../streaming/MultiStatefulOperatorsSuite.scala | 61 +++++++++++++++++++++-
.../spark/sql/streaming/StreamingJoinSuite.scala | 41 +++++++++++++++
3 files changed, 114 insertions(+), 7 deletions(-)
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 f63c82371bad..23c58c5556e4 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
@@ -1117,17 +1117,24 @@ case class StreamingSymmetricHashJoinExec(
Nil
}
- // This operator will evict based on the state watermark on both side of inputs; we would like
- // to let users leverage both sides of event time column for output of join, so the watermark
- // must be lower bound of both sides of event time column. The lower bound of event time column
- // for each side is determined by state watermark, hence we take a minimum of (left state
- // watermark, right state watermark, input watermark) to decide the output watermark.
+ // The output watermark bounds the event-time columns emitted by the join. State watermarks
+ // bound buffered rows that can still be emitted, and the input watermark bounds new rows.
+ // Joins emitting both sides take the minimum of both state watermarks and the input watermark.
override def produceOutputWatermark(inputWatermarkMs: Long): Option[Long] = {
val (leftStateWatermark, rightStateWatermark) =
StreamingSymmetricHashJoinHelper.getStateWatermark(
left.output, right.output, leftKeys, rightKeys, condition.full, Some(inputWatermarkMs),
!allowMultipleStatefulOperators)
- Some((leftStateWatermark ++ rightStateWatermark ++ Some(inputWatermarkMs)).min)
+ if (joinType == LeftSemi) {
+ // Only left columns are output. Without a watermarked left column, the watermark is lost;
+ // the right state watermark cannot replace it or constrain the emitted left event time.
+ WatermarkSupport.findEventTimeColumn(output,
+ allowMultipleEventTimeColumns = !allowMultipleStatefulOperators).map { _ =>
+ (leftStateWatermark ++ Some(inputWatermarkMs)).min
+ }
+ } else {
+ Some((leftStateWatermark ++ rightStateWatermark ++ Some(inputWatermarkMs)).min)
+ }
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/MultiStatefulOperatorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/MultiStatefulOperatorsSuite.scala
index cd901deae8e1..92d326234b96 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/MultiStatefulOperatorsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/MultiStatefulOperatorsSuite.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.execution.streaming.operators.stateful.StateStoreSaveExec
import org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinExec
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
-import org.apache.spark.sql.execution.streaming.state.StateStore
+import org.apache.spark.sql.execution.streaming.state.{HDFSBackedStateStoreProvider, RocksDBStateStoreProvider, StateStore}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.tags.SlowSQLTest
@@ -887,6 +887,65 @@ class MultiStatefulOperatorsSuite
testOutputWatermarkInJoin(join3, input1, -40L * 1000 - 1)
}
+ Seq(
+ (2, classOf[HDFSBackedStateStoreProvider], false),
+ (2, classOf[RocksDBStateStoreProvider], false),
+ (3, classOf[RocksDBStateStoreProvider], false),
+ (3, classOf[RocksDBStateStoreProvider], true),
+ (4, classOf[RocksDBStateStoreProvider], false),
+ (4, classOf[RocksDBStateStoreProvider], true)
+ ).foreach { case (stateFormatVersion, provider, changelogEnabled) =>
+ test("left semi output watermark closes downstream windows across restart: " +
+ s"state format=$stateFormatVersion, ${provider.getSimpleName}, " +
+ s"changelog=$changelogEnabled") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val left = leftInput.toDF()
+ .select($"_1".as("leftKey"), timestamp_seconds($"_2").as("leftTime"))
+ .withWatermark("leftTime", "0 seconds")
+ val right = rightInput.toDF()
+ .select($"_1".as("rightKey"), timestamp_seconds($"_2").as("rightTime"))
+ .withWatermark("rightTime", "0 seconds")
+ val result = left.join(right, expr(
+ "leftKey = rightKey AND leftTime > rightTime - INTERVAL 5 SECONDS AND " +
+ "leftTime < rightTime + INTERVAL 30 SECONDS"), "left_semi")
+ .groupBy(window($"leftTime", "10 seconds"))
+ .count()
+ .select($"window.start".cast("long"), $"count")
+
+ withSQLConf(
+ SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key -> stateFormatVersion.toString,
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key -> provider.getName,
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" ->
+ changelogEnabled.toString,
+ SQLConf.STATEFUL_OPERATOR_ALLOW_MULTIPLE.key -> "true",
+ SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key -> "false") {
+ withTempDir { checkpointDir =>
+ // Watermarks below are in seconds and describe the bounds used in each batch.
+ // inputW is the join's eviction watermark; leftW/rightW are its state bounds;
+ // outputW is the downstream aggregation's eviction watermark.
+ // Join rows are (key, event time); aggregation state is [window start, end): count.
+ testStream(result)(
+ StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
+ // Batch 0: inputW=0, leftW=-5, rightW=-30; propagation starts at outputW=0.
+ // Join emits (1, 105); aggregation buffers [100, 110): 1.
+ MultiAddData(leftInput, (1, 105))(rightInput, (1, 105)),
+ CheckNewAnswer(),
+ // Batch 1: inputW=105, leftW=100, rightW=75, outputW=100.
+ // Join emits (2, 120); aggregation buffers [100, 110): 1 and [120, 130): 1.
+ MultiAddData(leftInput, (2, 120))(rightInput, (2, 120)),
+ CheckNewAnswer(),
+ StopStream,
+ StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
+ // Batch 2: inputW=120, leftW=115, rightW=90, outputW=115 (after restart).
+ // Join emits nothing; aggregation emits [100, 110): 1 and keeps [120, 130): 1.
+ AddData(rightInput, (9, 120)),
+ CheckNewAnswer((100L, 1L)))
+ }
+ }
+ }
+ }
+
test("SPARK-49829 time window agg per each source followed by stream-stream join") {
val inputStream1 = MemoryStream[Long]
val inputStream2 = MemoryStream[Long]
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 2adb5b25bc00..37ed60e9370b 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
@@ -2328,6 +2328,47 @@ abstract class StreamingLeftSemiJoinBase extends StreamingJoinSuite {
import testImplicits._
+ for (leftWatermark <- Seq(false, true); joinOnTime <- Seq(false, true)) {
+ test(s"left semi output watermark follows output event time: " +
+ s"left watermark=$leftWatermark, equality=$joinOnTime") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val leftWithoutWatermark = leftInput.toDF()
+ .select($"_1".as("leftKey"), timestamp_seconds($"_2").as("leftTime"))
+ val left = if (leftWatermark) {
+ leftWithoutWatermark.withWatermark("leftTime", "0 seconds")
+ } else {
+ leftWithoutWatermark
+ }
+ val right = rightInput.toDF()
+ .select($"_1".as("rightKey"), timestamp_seconds($"_2").as("rightTime"))
+ .withWatermark("rightTime", "0 seconds")
+ val timeCondition = if (joinOnTime) {
+ "leftTime = rightTime"
+ } else {
+ "leftTime > rightTime - INTERVAL 5 SECONDS AND " +
+ "leftTime < rightTime + INTERVAL 30 SECONDS"
+ }
+ val joined = left.join(right, expr(s"leftKey = rightKey AND $timeCondition"), "left_semi")
+ .select($"leftKey", $"leftTime".cast("int"))
+ val expectedWatermark = if (leftWatermark) {
+ Some(if (joinOnTime) 120000L else 115000L)
+ } else {
+ None
+ }
+
+ testStream(joined)(
+ MultiAddData(leftInput, (1, 5))(rightInput, (1, 5)),
+ CheckNewAnswer((1, 5)),
+ Execute { query =>
+ val joinOperator = query.lastExecution.executedPlan.collect {
+ case j: StreamingSymmetricHashJoinExec => j
+ }.head
+ assert(joinOperator.produceOutputWatermark(120000L) === expectedWatermark)
+ })
+ }
+ }
+
testWithAppendAndUpdate("windowed left semi join") { outputMode =>
withTempDir { checkpointDir =>
val (leftInput, rightInput, joined) = setupWindowedJoin("left_semi")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org