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/22 05:38:59 UTC
(spark) branch branch-4.3 updated: [SPARK-60378][SS] Fix skipped rows in V4 streaming join eviction
This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 0360cdc81d5d [SPARK-60378][SS] Fix skipped rows in V4 streaming join eviction
0360cdc81d5d is described below
commit 0360cdc81d5d37e6753fb80d8741358513ac679b
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Thu Oct 22 14:38:18 2026 +0900
[SPARK-60378][SS] Fix skipped rows in V4 streaming join eviction
### What changes were proposed in this pull request?
Fix V4 timestamp-index eviction so it includes newly admitted rows below the previous state cutoff. Use that cutoff as the scan lower bound only when late-input filtering prevents newly admitted rows from falling below it. For value predicates, the derived cutoff must not exceed the late-input watermark. For key predicates, late filtering must apply to that same equality key. Otherwise scan from the beginning of the index through the current eviction cutoff.
### Why are the changes needed?
For a left outer stream-stream join with zero-second watermarks on both event-time columns:
```sql
leftKey = rightKey AND leftTime > rightTime + INTERVAL 5 SECONDS
```
The previous derived state cutoff can be ahead of the watermark used to reject late input. A new row can therefore be accepted yet already be eligible for eviction. Starting the V4 scan above that cutoff skips the row and loses its outer-null output.
The regression disables no-data micro-batches and uses these input batches (times in seconds):
1. Left `(9, 20)`, right `(9, 10)`: emits their match.
2. Right `(2, 20)`: advances the watermark.
3. Left `(1, 12)`, right `(2, 30)`: must emit `(1, null, 12, null)`.
In batch 3, the late-input watermark is 10 seconds and the previous state cutoff is 15 seconds. The accepted 12-second row falls below the old scan lower bound.
The equality-key path has the same omission when only the opposite input has a watermark:
```sql
leftKey = rightKey AND leftTime = rightTime
```
With a watermark only on `rightTime`, the left input has no late filter. After the previous eviction cutoff reaches 10 seconds, a new left row `(2, 5)` is accepted and must produce its outer-null result. Reusing the old scan boundary skips it. The added regressions cover left and right outer joins, continuous execution and restart.
### Does this PR introduce _any_ user-facing change?
Yes. V4 joins finalize these eligible rows instead of skipping them. V1-V3 do not use the timestamp-index scan lower bound. State encoding and the eviction predicate are unchanged.
For affected join conditions, scanning from the beginning can read more state; the current eviction cutoff still bounds the scan. No performance benchmark was run.
### How was this patch tested?
- The new key-path E2E regression failed in all eight V4 cases before the fix, while all sixteen V2/V3 controls passed.
- After the fix, all 38 tests passed: 10 helper tests, 24 key-path E2E cases across V2/V3/V4, and the four existing V4 value-path regressions. E2E coverage includes left/right outer joins, continuous execution/restart, both changelog modes, and no duplicate outer-null result in the next batch.
- SQL Compile/Test scalastyle passed with zero errors and warnings; ASCII/100-column checks and `git diff --check` passed.
```text
build/sbt 'sql/testOnly *StreamingSymmetricHashJoinHelperSuite' \
'sql/testOnly *StreamingOuterJoinV4Suite *StreamingOuterJoinWithVCFSuite *StreamingOuterJoinWithoutVCFSuite -- -z "old unwatermarked keys"' \
'sql/testOnly *StreamingOuterJoinV4Suite -- -z "newly admitted rows"' \
'sql/scalastyle' 'sql/Test/scalastyle'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59629 from HeartSaVioR/jungtaek-lim_data/fix-streaming-join-v4-eviction-scan.
Authored-by: Jungtaek Lim <ka...@gmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit c7a468dbee5e582ae50ba710f8405335b36112d8)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../join/StreamingSymmetricHashJoinHelper.scala | 47 ++++++++++++++-------
.../join/SymmetricHashJoinStateManager.scala | 20 +++++----
.../spark/sql/streaming/StreamingJoinSuite.scala | 49 ++++++++++++++++++++++
.../spark/sql/streaming/StreamingJoinV4Suite.scala | 48 ++++++++++++++++++++-
.../StreamingSymmetricHashJoinHelperSuite.scala | 29 ++++++++++++-
5 files changed, 169 insertions(+), 24 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala
index e96c9dbfd7e1..83a49d917a2a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala
@@ -50,16 +50,19 @@ object StreamingSymmetricHashJoinHelper extends Logging {
*
* @param stateWatermark Current batch's eviction watermark. Entries with timestamp
* at or below this value are eligible for eviction in this batch.
- * @param prevStateWatermark Previous batch's eviction watermark, i.e. the watermark
- * used for filtering late events in the current batch.
- * Entries with timestamp at or below this value were already
- * evicted in prior batches, so the effective range of entries
- * to evict in this batch is `(prevStateWatermark, stateWatermark]`.
- * State manager implementations can leverage this lower bound
- * to optimize eviction (e.g. narrowing the scan range to skip
- * already-evicted entries). `None` means we do not have a known
- * lower bound (e.g. the first batch after restart), in which
- * case eviction must consider all entries up to `stateWatermark`.
+ * @param prevStateWatermark Safe exclusive lower bound in the same timestamp domain as
+ * `stateWatermark`. It is derived from a prior eviction cutoff
+ * only when prior eviction and late-input filtering prove that
+ * the skipped range is empty: entries at or below this value
+ * were already evicted, and newly admitted rows cannot land
+ * there. The effective range of entries to evict in this batch
+ * is `(prevStateWatermark, stateWatermark]`. State manager
+ * implementations can leverage this lower bound to optimize
+ * eviction (e.g. narrowing the scan range to skip already-evicted
+ * entries). `None` means we do not have a known lower bound, or
+ * cannot prove that late-input filtering excludes newly admitted
+ * rows at or below it. In that case eviction must consider all
+ * entries up to `stateWatermark`.
*/
case class JoinStateKeyWatermarkPredicate(
expr: Expression,
@@ -238,12 +241,25 @@ object StreamingSymmetricHashJoinHelper extends Logging {
oneSideJoinKeys(joinKeyOrdinalForWatermark.get).nullable)
val expr = watermarkExpression(Some(keyExprWithWatermark), eventTimeWatermarkForEviction)
expr.map { e =>
+ // V4 can skip keys at or below the previous watermark only if late-input filtering
+ // rejects new rows in that range on this same key. A watermark on the other input's
+ // equality key can drive eviction even when this input has no watermark, or has one
+ // on a different column. New rows can then pass late-input filtering with keys
+ // already eligible for eviction. V4 must scan from the beginning to find them;
+ // otherwise, outer joins can fail to emit null rows for these unmatched inputs.
+ val prevStateKeyWatermark =
+ if (oneSideInputAttributes.find(_.metadata.contains(delayKey))
+ .exists(oneSideJoinKeys(joinKeyOrdinalForWatermark.get).semanticEquals)) {
+ eventTimeWatermarkForLateEvents
+ } else {
+ None
+ }
// watermarkExpression only provides the expression when eventTimeWatermarkForEviction
- // is defined
+ // is defined.
JoinStateKeyWatermarkPredicate(
e,
eventTimeWatermarkForEviction.get,
- eventTimeWatermarkForLateEvents)
+ prevStateKeyWatermark)
}
} else if (isWatermarkDefinedOnInput) { // case 2 in the StreamingSymmetricHashJoinExec docs
val stateValueWatermark = StreamingJoinHelper.getStateValueWatermark(
@@ -251,12 +267,15 @@ object StreamingSymmetricHashJoinHelper extends Logging {
attributesWithEventWatermark = AttributeSet(otherSideInputAttributes),
condition,
eventTimeWatermarkForEviction)
- val prevStateValueWatermark = eventTimeWatermarkForLateEvents.flatMap { _ =>
+ // A row can pass late-input filtering but already be eligible for eviction in this batch.
+ // When the previous join-derived state cutoff is ahead of the late-input watermark, a new
+ // row can fall at or below that cutoff. V4 must scan from the beginning to find such rows.
+ val prevStateValueWatermark = eventTimeWatermarkForLateEvents.flatMap { lateWatermark =>
StreamingJoinHelper.getStateValueWatermark(
attributesToFindStateWatermarkFor = AttributeSet(oneSideInputAttributes),
attributesWithEventWatermark = AttributeSet(otherSideInputAttributes),
condition,
- eventTimeWatermarkForLateEvents)
+ eventTimeWatermarkForLateEvents).filter(_ <= lateWatermark)
}
val inputAttributeWithWatermark = oneSideInputAttributes.find(_.metadata.contains(delayKey))
val expr = watermarkExpression(inputAttributeWithWatermark, stateValueWatermark)
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 97878b33c778..817775c5271c 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
@@ -188,9 +188,10 @@ trait SupportsEvictByTimestamp { self: SymmetricHashJoinStateManager =>
* Evict the state by timestamp. Returns the number of values evicted.
*
* @param endTimestamp Inclusive upper bound: evicts entries with timestamp <= endTimestamp.
- * @param startTimestamp Exclusive lower bound: entries with timestamp <= startTimestamp are
- * assumed to have been evicted already (e.g. from the previous batch). When provided,
- * the scan starts from startTimestamp + 1.
+ * @param startTimestamp Exclusive lower bound. When provided, the caller guarantees that
+ * entries with timestamp <= startTimestamp were evicted already and no newly admitted rows
+ * have timestamps <= startTimestamp, so the scan starts from startTimestamp + 1. `None`
+ * scans all timestamps up to endTimestamp.
*/
def evictByTimestamp(endTimestamp: Long, startTimestamp: Option[Long] = None): Long
@@ -200,9 +201,10 @@ trait SupportsEvictByTimestamp { self: SymmetricHashJoinStateManager =>
* It is caller's responsibility to consume the whole iterator.
*
* @param endTimestamp Inclusive upper bound: evicts entries with timestamp <= endTimestamp.
- * @param startTimestamp Exclusive lower bound: entries with timestamp <= startTimestamp are
- * assumed to have been evicted already (e.g. from the previous batch). When provided,
- * the scan starts from startTimestamp + 1.
+ * @param startTimestamp Exclusive lower bound. When provided, the caller guarantees that
+ * entries with timestamp <= startTimestamp were evicted already and no newly admitted rows
+ * have timestamps <= startTimestamp, so the scan starts from startTimestamp + 1. `None`
+ * scans all timestamps up to endTimestamp.
*/
def evictAndReturnByTimestamp(
endTimestamp: Long, startTimestamp: Option[Long] = None): Iterator[KeyToValuePair]
@@ -921,8 +923,10 @@ class SymmetricHashJoinStateManagerV4(
*
* @param endTimestamp Inclusive upper bound: entries with timestamp <= endTimestamp are
* eligible for eviction.
- * @param startTimestamp Exclusive lower bound: entries with timestamp <= startTimestamp
- * are assumed to have been evicted already. The scan starts from startTimestamp + 1.
+ * @param startTimestamp Exclusive lower bound. When provided, the caller guarantees that
+ * entries with timestamp <= startTimestamp were evicted already and no newly admitted rows
+ * have timestamps <= startTimestamp, so the scan starts from startTimestamp + 1. `None`
+ * scans all timestamps up to endTimestamp.
*/
def scanEvictedKeys(
endTimestamp: 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 37ed60e9370b..7027625a8d34 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
@@ -32,6 +32,7 @@ import org.apache.spark.SparkUnsupportedOperationException
import org.apache.spark.scheduler.ExecutorCacheTaskLocation
import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SparkSession}
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
+import org.apache.spark.sql.catalyst.plans.{LeftOuter, RightOuter}
import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
import org.apache.spark.sql.execution.datasources.v2.state.StateSourceOptions
import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
@@ -1356,6 +1357,54 @@ abstract class StreamingOuterJoinBase extends StreamingJoinSuite {
import testImplicits._
import org.apache.spark.sql.functions._
+ private object RestartMode extends Enumeration {
+ val WithoutRestart, WithRestart = Value
+ }
+
+ for {
+ joinType <- Seq(LeftOuter, RightOuter)
+ restartMode <- RestartMode.values
+ } {
+ test(s"SPARK-60378: $joinType eviction includes old unwatermarked keys, $restartMode") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val leftOuter = joinType == LeftOuter
+ val left = {
+ val df = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time").as("leftTime"))
+ if (leftOuter) df else df.withWatermark("leftTime", "10 seconds")
+ }
+ val right = {
+ val df = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time").as("rightTime"))
+ if (leftOuter) df.withWatermark("rightTime", "10 seconds") else df
+ }
+ val joined = left.join(right,
+ expr("leftKey = rightKey AND leftTime = rightTime"), joinType.toString)
+ .select($"leftKey", $"rightKey", $"leftTime".cast("int"), $"rightTime".cast("int"))
+ val unwatermarkedInput = if (leftOuter) leftInput else rightInput
+ val watermarkedInput = if (leftOuter) rightInput else leftInput
+ val expected = if (leftOuter) Row(2, null, 5, null) else Row(null, 2, null, 5)
+
+ withSQLConf(SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key -> "false") {
+ withTempDir { checkpointDir =>
+ val start = StartStream(checkpointLocation = checkpointDir.getCanonicalPath)
+ val setup = Seq(start,
+ AddData(watermarkedInput, (9, 20)), CheckNewAnswer(),
+ AddData(watermarkedInput, (9, 30)), CheckNewAnswer()) ++
+ (if (restartMode == RestartMode.WithRestart) Seq(StopStream, start) else Nil)
+ testStream(joined)((setup ++ Seq(
+ // The setup batches advance the watermark to 10s and then 20s. This batch uses
+ // 10s for late-input filtering and 20s for eviction. Reusing the previous 10s cutoff
+ // as a scan bound would miss time=5s, which the unwatermarked input accepts.
+ // Eviction must find this row and emit its unmatched outer-join result.
+ AddData(unwatermarkedInput, (2, 5)), CheckNewAnswer(expected),
+ AddData(watermarkedInput, (9, 40)), CheckNewAnswer(), StopStream)): _*)
+ }
+ }
+ }
+ }
+
Seq("left_outer", "right_outer").foreach { joinType =>
test(s"stream-stream $joinType join does not support Update mode") {
val input1 = MemoryStream[Int]
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 6436c0c8d6ba..ca39a96d4e52 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
@@ -21,6 +21,7 @@ import org.apache.hadoop.fs.Path
import org.scalatest.Tag
import org.apache.spark.SparkIllegalArgumentException
+import org.apache.spark.sql.Row
import org.apache.spark.sql.execution.datasources.v2.state.StateSourceOptions
import org.apache.spark.sql.execution.streaming.checkpointing.CheckpointFileManager
import org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinExec
@@ -542,7 +543,52 @@ class StreamingInnerJoinV4Suite
@SlowSQLTest
class StreamingOuterJoinV4Suite
extends StreamingOuterJoinBase
- with TestWithV4StateFormat
+ with TestWithV4StateFormat {
+
+ import testImplicits._
+
+ for (history <- Seq("continuous", "restart")) {
+ test(s"V4 eviction includes newly admitted rows below the previous bound, $history") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val left = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time").as("leftTime"))
+ .withWatermark("leftTime", "0 seconds")
+ val right = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time").as("rightTime"))
+ .withWatermark("rightTime", "0 seconds")
+ val joined = left.join(right,
+ expr("leftKey = rightKey AND leftTime > rightTime + INTERVAL 5 SECONDS"),
+ "left_outer")
+ .select($"leftKey", $"rightKey", $"leftTime".cast("long"), $"rightTime".cast("long"))
+
+ withSQLConf(SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key -> "false") {
+ withTempDir { checkpointDir =>
+ val start = StartStream(checkpointLocation = checkpointDir.getCanonicalPath)
+ // All watermarks below are in seconds. inputW is the minimum of the two sides'
+ // maximum event times before the batch and drives eviction. Late input uses the
+ // previous batch's inputW. The condition gives leftStateW = inputW + 5, but no
+ // rightStateW: old right rows can still match future left rows.
+ val setup = Seq(start,
+ // Batch 1: inputW = 0, leftStateW = 5, rightStateW = None.
+ MultiAddData(leftInput, (9, 20))(rightInput, (9, 10)),
+ CheckNewAnswer(Row(9, 9, 20, 10)),
+ // Batch 2: inputW = 10, leftStateW = 15, rightStateW = None.
+ AddData(rightInput, (2, 20)), CheckNewAnswer()) ++
+ (if (history == "restart") Seq(StopStream, start) else Nil)
+ // Batch 3: inputW = 20, leftStateW = 25, rightStateW = None.
+ // Late input uses inputW = 10 from batch 2, so the 12-second row is accepted.
+ // The old scan started above batch 2's leftStateW = 15, skipping this row.
+ testStream(joined)((setup ++ Seq(
+ MultiAddData(leftInput, (1, 12))(rightInput, (2, 30)),
+ CheckNewAnswer(Row(1, null, 12, null)),
+ // Batch 4: inputW = 20, leftStateW = 25, rightStateW = None. No duplicate output.
+ AddData(rightInput, (2, 50)), CheckNewAnswer(), StopStream)): _*)
+ }
+ }
+ }
+ }
+}
@SlowSQLTest
class StreamingFullOuterJoinV4Suite
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSymmetricHashJoinHelperSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSymmetricHashJoinHelperSuite.scala
index 3336cc680ac0..39dc9f7400da 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSymmetricHashJoinHelperSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSymmetricHashJoinHelperSuite.scala
@@ -19,11 +19,18 @@ package org.apache.spark.sql.streaming
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.EventTimeWatermark
import org.apache.spark.sql.execution.LocalTableScanExec
-import org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinHelper.JoinConditionSplitPredicates
+import org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinHelper
+import org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinHelper.{JoinConditionSplitPredicates, JoinStateKeyWatermarkPredicate}
import org.apache.spark.sql.types._
class StreamingSymmetricHashJoinHelperSuite extends StreamTest {
+
+ private object WatermarkColumn extends Enumeration {
+ val NoWatermark, JoinKey, Other = Value
+ }
+
val leftAttributeA = AttributeReference("a", IntegerType)()
val leftAttributeB = AttributeReference("b", IntegerType)()
val rightAttributeC = AttributeReference("c", IntegerType)()
@@ -127,4 +134,24 @@ class StreamingSymmetricHashJoinHelperSuite extends StreamTest {
leftAttributeA === rightAttributeC && randAttribute > Literal(0)))
assert(split.full.contains(predicate))
}
+
+ for (watermarkColumn <- WatermarkColumn.values) {
+ test(s"SPARK-60378: previous key scan bound with watermark on $watermarkColumn") {
+ val watermark = new MetadataBuilder().putLong(EventTimeWatermark.delayKey, 1000L).build()
+ val key = AttributeReference("key", TimestampType,
+ metadata = if (watermarkColumn == WatermarkColumn.JoinKey) watermark else Metadata.empty)()
+ val other = AttributeReference("other", TimestampType,
+ metadata = if (watermarkColumn == WatermarkColumn.Other) watermark else Metadata.empty)()
+ val rightKey = AttributeReference("rightKey", TimestampType, metadata = watermark)()
+ val predicates = StreamingSymmetricHashJoinHelper.getStateWatermarkPredicates(
+ Seq(key, other), Seq(rightKey), Seq(key), Seq(rightKey), None,
+ Some(20000L), Some(10000L), false)
+ val expectedPrevious =
+ if (watermarkColumn == WatermarkColumn.JoinKey) Some(10000L) else None
+ assert(predicates.left.exists {
+ case JoinStateKeyWatermarkPredicate(_, 20000L, previous) => previous == expectedPrevious
+ case _ => false
+ })
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org