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 dfad9c710c56 [SPARK-58724][SS] Incremental state cleanup for streaming
dropDuplicates
dfad9c710c56 is described below
commit dfad9c710c56aab56533536025b56227927c39a9
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Tue Aug 18 07:14:55 2026 +0900
[SPARK-58724][SS] Incremental state cleanup for streaming dropDuplicates
### What changes were proposed in this pull request?
Streaming deduplication (`dropDuplicates`) evicts all watermark-expired
state at once at the end of each batch. This PR adds **incremental state
cleanup** to `StreamingDeduplicateExec`:
when enabled, up to a configurable number of eviction-eligible state rows
are removed per input record processed, spreading eviction cost across the
batch instead of paying it all at the
batch boundary; any still-eligible rows left over are removed at batch
end as before.
This is gated by the existing config
`spark.sql.streaming.statefulOperator.incrementalCleanupFactor` (added in
SPARK-58635 for streaming aggregation, previously read only by the
streamline aggregation operator). When it is `0` (the default), behavior
is unchanged — all eviction happens at batch end.
Details:
- `BaseStreamingDeduplicateExec` gains `incrementalCleanupFactor` /
`doIncrementalCleanup`, and its abstract eviction hook changes from
`evictDupInfoFromState(store): Unit` to
`iteratorForEviction(store): Iterator[Any]` — an iterator whose `next()`
performs one real state removal, so the record loop can remove up to
`incrementalCleanupFactor` rows per input and
the completion block drains the remainder. This mirrors the mechanism
already used by `StatefulStreamlineAggregateExec`.
- `StreamingDeduplicateExec.iteratorForEviction` uses the existing
`EvictionIterator` (event time read from the state key). Under incremental
cleanup it evicts against the **late-events**
watermark rather than the eviction watermark: within a batch a record can
arrive below the eviction watermark, so state may only be cleaned up to the
timestamp before which no further
input can arrive. The deduplicated output is identical to the default;
only the timing of state reclamation changes (it lags by at most one batch).
- Two new metrics are exposed: `numRowsReadDuringEviction` and
`numRowsIncrementallyRemoved` (alongside the existing `numRemovedStateRows`).
- `StreamingDeduplicateWithinWatermarkExec` deliberately does **not**
participate in incremental cleanup. Its dedup key excludes the event time (the
expiry lives in the value row), so two
records sharing a key can have different event times; evicting an expired
entry mid-batch could make a later, non-late record with the same key be
emitted as new instead of dropped as a
duplicate — an output that would depend on the cleanup factor and store
iteration order. It inherits the factor-0 path and continues to evict once at
batch end.
### Why are the changes needed?
For a long-running streaming query — particularly under Real-Time Mode,
where a batch can run for minutes — evicting all expired deduplication state in
one pass at the batch boundary
concentrates work at the commit point and lets state grow unbounded
within a batch. Incremental cleanup bounds within-batch state growth and
smooths the eviction cost across the batch,
matching the capability streaming aggregation already has.
### Does this PR introduce _any_ user-facing change?
No behavior change by default.
`spark.sql.streaming.statefulOperator.incrementalCleanupFactor` defaults to
`0`, which preserves the existing batch-end eviction exactly. Setting it to a
positive value opts a `dropDuplicates` query into incremental cleanup;
the deduplicated output is unchanged, and two new state-operator metrics
(`numRowsReadDuringEviction`,
`numRowsIncrementallyRemoved`) appear in the streaming progress.
`dropDuplicatesWithinWatermark` is unaffected.
### How was this patch tested?
New unit tests:
- `StreamingDeduplicationSuite`: with a non-zero
`incrementalCleanupFactor`, deduplication output (and the late-event safety
property) is unchanged, and `numRowsIncrementallyRemoved`
progresses as expired keys are removed during record processing.
- `StreamingDeduplicationWithinWatermarkSuite`: a non-zero
`incrementalCleanupFactor` leaves `dropDuplicatesWithinWatermark` output and
per-batch state counts identical to the default
(confirming the operator stays opted out).
All existing `StreamingDeduplicationSuite` and
`StreamingDeduplicationWithinWatermarkSuite` tests continue to pass (the
factor-0 default path is unchanged), and `dev/scalastyle` passes.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Code
Closes #57945 from jerrypeng/oss-dedup-incremental-cleanup.
Authored-by: Boyang Jerry Peng <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
(cherry picked from commit 51b231c20d04002bfe0d8f20c4e2e85dbebf40a1)
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 3 +-
.../operators/stateful/statefulOperators.scala | 171 ++++++++++++++++++---
.../streaming/StreamingDeduplicationSuite.scala | 107 +++++++++++++
...treamingDeduplicationWithinWatermarkSuite.scala | 45 ++++++
4 files changed, 303 insertions(+), 23 deletions(-)
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 7c9fcc679ef3..fd8eb92a6878 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
@@ -3810,7 +3810,8 @@ object SQLConf {
"eviction happens at batch end. When k, up to k eligible records are
removed per input; " +
"any still-eligible records left over are removed at batch end. Only
applies to modes " +
"that evict (e.g. Append/Update); has no effect in Complete mode,
which never evicts. " +
- "Currently read only by the streamline aggregation operator.")
+ "Read by the streamline aggregation and streaming deduplication
(dropDuplicates) " +
+ "operators; dropDuplicatesWithinWatermark always evicts at batch end.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.longConf
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala
index 5fb00fbb2a32..799f6326f7fd 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala
@@ -631,18 +631,6 @@ trait WatermarkSupport extends SparkPlan {
watermarkExpression.map(Predicate.create(_, child.output))
}
- protected def removeKeysOlderThanWatermark(store: StateStore): Unit = {
- if (watermarkPredicateForKeysForEviction.nonEmpty) {
- val numRemovedStateRows = longMetric("numRemovedStateRows")
- store.iterator().foreach { rowPair =>
- if (watermarkPredicateForKeysForEviction.get.eval(rowPair.key)) {
- store.remove(rowPair.key)
- numRemovedStateRows += 1
- }
- }
- }
- }
-
protected def removeKeysOlderThanWatermark(
storeManager: StreamingAggregationStateManager,
store: StateStore): Unit = {
@@ -1385,6 +1373,11 @@ abstract class BaseStreamingDeduplicateExec
protected lazy val stateValueSchema: StructType =
WidenStatefulOpNullability.widenStateSchema(schemaForValueRow)
+ // Initialize it to 0 so that operators explicitly have to opt-in (i.e. set
it to non-zero) to
+ // enable incremental cleanup.
+ protected val incrementalCleanupFactor: Long = 0
+ protected def doIncrementalCleanup: Boolean = incrementalCleanupFactor > 0
+
override protected def doExecute(): RDD[InternalRow] = {
metrics // force lazy init at driver
@@ -1406,6 +1399,7 @@ abstract class BaseStreamingDeduplicateExec
val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
val commitTimeMs = longMetric("commitTimeMs")
val numDroppedDuplicateRows = longMetric("numDroppedDuplicateRows")
+ val numRowsIncrementallyRemoved =
longMetric("numRowsIncrementallyRemoved")
val baseIterator = watermarkPredicateForDataForLateEvents match {
case Some(predicate) => applyRemovingRowsOlderThanWatermark(iter,
predicate)
@@ -1416,10 +1410,37 @@ abstract class BaseStreamingDeduplicateExec
val updatesStartTimeNs = System.nanoTime
+ // Only create the eviction iterator if we are doing incremental
cleanup. It is opened over
+ // the store before the batch processes any rows, so it may not observe
records inserted into
+ // the store later in the batch (the guarantee depends on the store
provider's iterator
+ // semantics). When cleanup is disabled we defer building it to batch
end, after all inserts.
+ val incrementalEvictionIter = if (doIncrementalCleanup) {
+ Some(iteratorForEviction(store))
+ } else {
+ None
+ }
+
val result = baseIterator.filter { r =>
val row = r.asInstanceOf[UnsafeRow]
val key = getKey(row)
val keyExists = store.keyExists(key)
+
+ incrementalEvictionIter.foreach { evictionIter =>
+ allRemovalsTimeMs += timeTakenMs {
+ // Remove up to incrementalCleanupFactor eligible rows for every
input record.
+ var numRemovalsCurrRecord = 0
+ // NOTE: the eviction iterator removes a row inside hasNext, not
next(), so a bare
+ // hasNext already deletes it. To stop at exactly
incrementalCleanupFactor removals per
+ // input we must re-check the count before each hasNext rather
than draining the
+ // iterator.
+ while (numRemovalsCurrRecord < incrementalCleanupFactor &&
evictionIter.hasNext) {
+ evictionIter.next()
+ numRemovalsCurrRecord += 1
+ numRowsIncrementallyRemoved += 1
+ }
+ }
+ }
+
if (!keyExists) {
putDupInfoIntoState(store, row, key, reusedDupInfoRow)
numUpdatedStateRows += 1
@@ -1434,7 +1455,17 @@ abstract class BaseStreamingDeduplicateExec
CompletionIterator[InternalRow, Iterator[InternalRow]](result, {
allUpdatesTimeMs += NANOSECONDS.toMillis(System.nanoTime -
updatesStartTimeNs)
- allRemovalsTimeMs += timeTakenMs { evictDupInfoFromState(store) }
+
+ // Finish eviction: if incremental cleanup ran, drain whatever
eligible rows it did not get
+ // to during record processing; otherwise (incremental cleanup
disabled) build the eviction
+ // iterator now and drain it fully -- the batch-end-only behavior.
+ allRemovalsTimeMs += timeTakenMs {
+ val evictionIter =
incrementalEvictionIter.getOrElse(iteratorForEviction(store))
+ while (evictionIter.hasNext) {
+ evictionIter.next()
+ }
+ }
+
commitTimeMs += timeTakenMs { store.commit() }
setStoreMetrics(store)
setOperatorMetrics()
@@ -1450,15 +1481,46 @@ abstract class BaseStreamingDeduplicateExec
key: UnsafeRow,
reusedDupInfoRow: Option[UnsafeRow]): Unit
- protected def evictDupInfoFromState(store: StateStore): Unit
+ /**
+ * Creates an iterator that removes the state rows eligible for eviction
(older than the eviction
+ * watermark), updating the eviction metrics as it goes. Each `next()`
corresponds to an actual
+ * removal, so a caller can stop early -- after `incrementalCleanupFactor`
removals per input row
+ * -- and the store stays consistent with what was surfaced.
+ *
+ * Note the returned iterator is not necessarily an [[EvictionIterator]]:
that helper reads the
+ * eviction timestamp from the state store key, which is not how every
subclass stores it (e.g.
+ * [[StreamingDeduplicateWithinWatermarkExec]] keeps `expiresAtMicros` in
the value row). The
+ * contents are unused by the caller, hence `Iterator[Any]`.
+ */
+ protected def iteratorForEviction(store: StateStore): Iterator[Any]
override def output: Seq[Attribute] =
WidenStatefulOpNullability.widenOutputForStatefulOp(child.output)
override def outputPartitioning: Partitioning = child.outputPartitioning
+ /**
+ * Three eviction metrics are exposed:
+ *
+ * - numRemovedStateRows: the total number of state rows removed.
+ * - numRowsIncrementallyRemoved: the subset of those removed during
incremental eviction (i.e.
+ * spread across input-record processing rather than at batch end).
+ * - numRowsReadDuringEviction: the number of state rows the eviction
iterator scanned, whether
+ * or not they were removed.
+ *
+ * Since incrementally removed rows are a subset of all removed rows, and
every removed row was
+ * read, the relationship is:
+ *
+ * numRowsReadDuringEviction >= numRemovedStateRows >=
numRowsIncrementallyRemoved
+ */
override def customStatefulOperatorMetrics:
Seq[StatefulOperatorCustomMetric] = {
- Seq(StatefulOperatorCustomSumMetric("numDroppedDuplicateRows", "number of
duplicates dropped"))
+ Seq(
+ StatefulOperatorCustomSumMetric("numDroppedDuplicateRows", "number of
duplicates dropped"),
+ StatefulOperatorCustomSumMetric(
+ "numRowsReadDuringEviction", "number of state rows read during state
eviction"),
+ StatefulOperatorCustomSumMetric(
+ "numRowsIncrementallyRemoved", "number of state rows removed during
incremental eviction")
+ )
}
override def shouldRunAnotherBatch(newInputWatermark: Long): Boolean = {
@@ -1484,6 +1546,12 @@ case class StreamingDeduplicateExec(
protected val extraOptionOnStateStore: Map[String, String] =
Map(StateStoreConf.FORMAT_VALIDATION_CHECK_VALUE_CONFIG -> "false")
+ // Read via the null-safe `conf` accessor rather than
`session.sessionState.conf`: on a
+ // session-less thread (e.g. during canonicalization) `session` is null, and
`conf` falls back to
+ // SparkPlan's active/default conf instead of throwing.
+ override protected val incrementalCleanupFactor: Long =
+ conf.getConf(SQLConf.STREAMING_STATE_INCREMENTAL_CLEANUP_FACTOR)
+
protected def initializeReusedDupInfoRow(): Option[UnsafeRow] = None
protected def putDupInfoIntoState(
@@ -1494,8 +1562,45 @@ case class StreamingDeduplicateExec(
store.put(key, StreamingDeduplicateExec.EMPTY_ROW)
}
- protected def evictDupInfoFromState(store: StateStore): Unit = {
- removeKeysOlderThanWatermark(store)
+ override protected def iteratorForEviction(store: StateStore): Iterator[Any]
= {
+ val numRemovedStateRows = longMetric("numRemovedStateRows")
+ val numRowsReadDuringEviction = longMetric("numRowsReadDuringEviction")
+
+ // When doing incremental cleanup, we have to be careful which watermark
to use. The late-events
+ // watermark is less than or equal to the eviction watermark, so within a
batch it is possible
+ // to receive an event whose timestamp is below the eviction watermark.
Evicting a key at
+ // timestamp t and then, in the same batch, receiving another record at t
would let that record
+ // through as if it were new. Thus, with incremental eviction we may only
clean up records up to
+ // the timestamp before which we will never receive new records -- the
event time watermark for
+ // late events. Without incremental cleanup, all input has been processed
by the time we evict,
+ // so the eviction watermark is safe.
+ val incrementalAwareEvictionWatermark = if (doIncrementalCleanup) {
+ eventTimeWatermarkForLateEvents
+ } else {
+ eventTimeWatermarkForEviction
+ }
+
+ val evictionIterator = EvictionIterator(
+ store,
+ store.iterator(),
+ keyExpressions,
+ // The previous full-eviction predicate
(watermarkPredicateForKeysForEviction) compiled
+ // against `keyExpressions` and only produced a predicate when
`keyExpressions` carried an
+ // event-time column -- so resolving the event-time column from
`keyExpressions` here is
+ // equivalent for the dedup key, and matches the aggregation eviction
path and the runtime.
+ // The `allowMultipleStatefulOperators` knob is carried over unchanged,
hence
+ // !allowMultipleStatefulOperators. (The old path found the event-time
*column* from
+ // child.output, which could throw MULTIPLE_EVENT_TIME_COLUMNS when
child.output -- not the
+ // dedup key -- carried several event-time columns; resolving from the
key here does not, but
+ // that only differs under the non-default allowMultiple=false with 2+
event-time columns
+ // outside the dedup key.)
+ allowMultipleEventTimeColumns = !allowMultipleStatefulOperators,
+ incrementalAwareEvictionWatermark)
+
+ CompletionIterator[UnsafeRowPair,
Iterator[UnsafeRowPair]](evictionIterator, {
+ numRowsReadDuringEviction +=
evictionIterator.numRowsReadDuringEvictionSoFar
+ numRemovedStateRows += evictionIterator.numRowsRemovedSoFar
+ })
}
override def shortName: String =
StatefulOperatorsUtils.DEDUPLICATE_EXEC_OP_NAME
@@ -1538,6 +1643,17 @@ case class StreamingDeduplicateWithinWatermarkExec(
protected val extraOptionOnStateStore: Map[String, String] = Map.empty
+ // Incremental cleanup is deliberately NOT enabled for
dropDuplicatesWithinWatermark. Its dedup
+ // key is only the user's columns; the expiry (`expiresAtMicros`) lives in
the value row and is
+ // decoupled from a record's event time. So two records that share a dedup
key can have different
+ // event times, and one can be non-late while the other's stored entry has
already expired.
+ // Evicting that entry mid-batch (as incremental cleanup would) makes a
later non-late record with
+ // the same key miss the existing entry and be emitted as new -- an output
that depends on the
+ // cleanup factor and store iteration order. Evicting only at batch end (the
inherited factor-0
+ // path) keeps the entry visible to every record in the batch, so the output
is deterministic.
+ // (StreamingDeduplicateExec is immune because its key includes the event
time: any record sharing
+ // an evictable key is itself late and is dropped by the late-events filter
first.)
+
// Below three variables are defined as lazy, as evaluating these variables
does not work with
// canonicalized plan. Specifically, attributes in child won't have an event
time column in
// the canonicalized plan. These variables are NOT referenced in
canonicalized plan, hence
@@ -1571,18 +1687,29 @@ case class StreamingDeduplicateWithinWatermarkExec(
store.put(key, timeoutRow)
}
- protected def evictDupInfoFromState(store: StateStore): Unit = {
+ override protected def iteratorForEviction(store: StateStore): Iterator[Any]
= {
val numRemovedStateRows = longMetric("numRemovedStateRows")
+ val numRowsReadDuringEviction = longMetric("numRowsReadDuringEviction")
- // Convert watermark value to micros.
+ // Incremental cleanup is not enabled for this operator (this class does
not override the base
+ // incrementalCleanupFactor, which defaults to 0 -- see the class-level
comment for why), so
+ // eviction always runs once at batch end against the eviction watermark.
val watermarkForEviction =
DateTimeUtils.millisToMicros(eventTimeWatermarkForEviction.get)
- store.iterator().foreach { rowPair =>
- val valueRow = rowPair.value
- val expiresAt = valueRow.getLong(0)
+ // We cannot reuse `EvictionIterator` here because it reads the eviction
timestamp from the
+ // state store key, whereas this operator stores `expiresAtMicros` in the
value row (the key is
+ // only the dedup key columns). We still match EvictionIterator's
semantics by returning only
+ // the rows that are actually evicted, so each next() corresponds to a
real removal and the
+ // caller's numRemovedStateRows metric counts removals rather than state
rows scanned.
+ store.iterator().filter { rowPair =>
+ numRowsReadDuringEviction += 1
+ val expiresAt = rowPair.value.getLong(0)
if (watermarkForEviction >= expiresAt) {
store.remove(rowPair.key)
numRemovedStateRows += 1
+ true
+ } else {
+ false
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
index 003f71d16437..21261280a280 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
@@ -684,6 +684,113 @@ class StreamingDeduplicationSuite extends
StateStoreMetricsTest
sqlConf = spark.sessionState.conf
)
}
+
+ // Total incremental removals over the query's retained recent progress
(recentProgress is
+ // retention-bounded, so this is not the full batch history), read from the
operator's
+ // numRowsIncrementallyRemoved custom metric. Where a stateful operator ran
(stateOperators is
+ // non-empty) the metric must be present -- assert rather than default it to
0, so a regression
+ // that stops emitting the metric fails the test instead of silently reading
as zero.
+ private def numRowsIncrementallyRemoved(q: StreamingQuery): Long = {
+ q.recentProgress.flatMap(_.stateOperators.headOption).map { op =>
+ assert(op.customMetrics.containsKey("numRowsIncrementallyRemoved"),
+ s"numRowsIncrementallyRemoved custom metric missing; got
${op.customMetrics}")
+ op.customMetrics.get("numRowsIncrementallyRemoved").toLong
+ }.sum
+ }
+
+ // Total state rows removed over the query's retained recent progress
(recentProgress is
+ // retention-bounded, so this is not the full batch history), read from the
operator's first-class
+ // numRowsRemoved metric (the incremental removals are a subset of these).
This asserts that
+ // eviction actually removed state, which numRowsIncrementallyRemoved alone
does not: a batch-end
+ // drain removes rows without incrementing the incremental counter.
+ private def totalStateRowsRemoved(q: StreamingQuery): Long =
+
q.recentProgress.flatMap(_.stateOperators.headOption.map(_.numRowsRemoved)).sum
+
+ test("deduplicate with watermark - incremental cleanup preserves dedup
output") {
+ // With a non-zero incremental cleanup factor, removal of
watermark-expired state is spread
+ // across input-record processing rather than occurring all at once at
batch end. The
+ // deduplicated OUTPUT must be unchanged. (State is evicted against the
late-events watermark
+ // under incremental cleanup, so the timing of state removal lags by a
batch compared to the
+ // factor-0 default -- this test asserts on output and on the late-event
safety property, not
+ // on per-batch state counts.)
+ withSQLConf(SQLConf.STREAMING_STATE_INCREMENTAL_CLEANUP_FACTOR.key ->
"10") {
+ val inputData = MemoryStream[Int]
+ val result = inputData.toDS()
+ .withColumn("eventTime", timestamp_seconds($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .dropDuplicates()
+ .select($"eventTime".cast("long").as[Long])
+
+ testStream(result, Append)(
+ // Duplicates within the batch are dropped; each distinct event time
is emitted once.
+ AddData(inputData, (1 to 5).flatMap(_ => (10 to 15)): _*),
+ CheckAnswer(10 to 15: _*),
+
+ AddData(inputData, 25), // Advances watermark; 25 is new and emitted.
+ CheckNewAnswer(25),
+
+ // A record at 10 is now below the watermark: it must be dropped, and
crucially it must not
+ // be re-emitted even though incremental cleanup may not yet have
removed its key. This is
+ // the safety property behind evicting against the late-events (not
eviction) watermark.
+ AddData(inputData, 10),
+ CheckNewAnswer(),
+
+ AddData(inputData, 45),
+ CheckNewAnswer(45),
+
+ // A duplicate of a surviving recent key (45) is still deduplicated.
+ AddData(inputData, 45),
+ CheckNewAnswer()
+ )
+ }
+ }
+
+ test("deduplicate with watermark - incremental cleanup evicts during record
processing") {
+ // A batch that carries input records AND has state eligible for eviction
under the late-events
+ // watermark should evict incrementally as those records are processed, so
+ // numRowsIncrementallyRemoved is non-zero. (Eviction uses the late-events
watermark under
+ // incremental cleanup, which lags the eviction watermark by one batch.)
+ withSQLConf(SQLConf.STREAMING_STATE_INCREMENTAL_CLEANUP_FACTOR.key ->
"10") {
+ val inputData = MemoryStream[Int]
+ val result = inputData.toDS()
+ .withColumn("eventTime", timestamp_seconds($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .dropDuplicates()
+ .select($"eventTime".cast("long").as[Long])
+
+ testStream(result, Append)(
+ // Batch 0: three distinct keys at 10, 11, 12. Watermark is 0, nothing
evictable yet.
+ AddData(inputData, 10, 11, 12),
+ CheckAnswer(10, 11, 12),
+ assertNumStateRows(total = 3, updated = 3),
+ AssertOnQuery(q => numRowsIncrementallyRemoved(q) == 0,
+ "no incremental removal before any state is evictable"),
+
+ // Batch 1: a new key at 100 advances the eviction watermark to 90,
but the late-events
+ // watermark used by incremental cleanup still lags (it is the
previous batch's eviction
+ // watermark). The keys [10, 11, 12] are not yet below the late-events
watermark, so no
+ // removal happens in this batch; it is deferred to batch 2.
+ AddData(inputData, 100),
+ CheckNewAnswer(100),
+
+ // Batch 2: another new record. By now the late-events watermark has
advanced past the
+ // original keys, so processing this record incrementally evicts them.
+ AddData(inputData, 101),
+ CheckNewAnswer(101),
+ AssertOnQuery(q => numRowsIncrementallyRemoved(q) > 0,
+ "expired keys should be removed incrementally while processing input
records"),
+ // Assert the state was actually removed, not merely that the
incremental counter moved:
+ // the three original keys [10, 11, 12] must be gone, leaving only the
recent keys. Reading
+ // the first-class numRowsRemoved metric (rather than the output) is
what catches a
+ // regression that leaves state behind while still producing correct
output.
+ AssertOnQuery(q => totalStateRowsRemoved(q) >= 3,
+ "expired keys must be removed from state, not just counted as
incremental"),
+ // The surviving state is only the most recent keys; correctness is
unchanged.
+ AddData(inputData, 10), // below watermark, dropped
+ CheckNewAnswer()
+ )
+ }
+ }
}
trait StreamingDeduplicationSuiteBase { self: StreamTest =>
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithinWatermarkSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithinWatermarkSuite.scala
index 9645f82ac241..8a9766c7e875 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithinWatermarkSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithinWatermarkSuite.scala
@@ -22,6 +22,7 @@ import
org.apache.spark.sql.catalyst.streaming.InternalOutputModes.Append
import
org.apache.spark.sql.execution.streaming.operators.stateful.StatefulOperatorsUtils
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.functions.timestamp_seconds
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{LongType, StringType, StructType}
import org.apache.spark.tags.SlowSQLTest
@@ -164,6 +165,50 @@ class StreamingDeduplicationWithinWatermarkSuite extends
StateStoreMetricsTest
)
}
+ test("incremental cleanup factor does not change
dropDuplicatesWithinWatermark output") {
+ // dropDuplicatesWithinWatermark deliberately does not participate in
incremental cleanup
+ // (its dedup key excludes the event time, so mid-batch eviction of a
shared key could change
+ // the output). Setting a non-zero incrementalCleanupFactor must therefore
leave the output and
+ // per-batch state counts identical to the default -- this is the same
sequence as
+ // "deduplicate with subset of columns which event time column is not in
subset".
+ withSQLConf(SQLConf.STREAMING_STATE_INCREMENTAL_CLEANUP_FACTOR.key ->
"10") {
+ val inputData = MemoryStream[(String, Int)]
+ val result = inputData.toDS()
+ .withColumn("eventTime", timestamp_seconds($"_2"))
+ .withWatermark("eventTime", "2 seconds")
+ .dropDuplicatesWithinWatermark("_1")
+ .select($"_1", $"eventTime".cast("long").as[Long])
+
+ testStream(result, Append)(
+ AddData(inputData, "a" -> 17),
+ CheckNewAnswer("a" -> 17),
+ assertNumStateRows(total = 1, updated = 1),
+
+ AddData(inputData, "a" -> 16),
+ CheckNewAnswer(),
+ assertNumStateRows(total = 1, updated = 0),
+
+ AddData(inputData, "a" -> 13),
+ CheckNewAnswer(),
+ assertNumStateRows(total = 1, updated = 0, droppedByWatermark = 1),
+
+ AddData(inputData, "b" -> 22, "c" -> 21),
+ CheckNewAnswer("b" -> 22, "c" -> 21),
+ assertNumStateRows(total = 2, updated = 2),
+
+ // "a" -> 21 is emitted as new because the earlier "a" state expired
and was evicted at the
+ // previous batch end -- unchanged from the factor-0 path.
+ AddData(inputData, "a" -> 21),
+ CheckNewAnswer("a" -> 21),
+ assertNumStateRows(total = 3, updated = 1),
+
+ AddData(inputData, "d" -> 25),
+ CheckNewAnswer("d" -> 25),
+ assertNumStateRows(total = 2, updated = 1)
+ )
+ }
+ }
+
test("SPARK-39650: duplicate with specific keys should allow input to change
schema") {
withTempDir { checkpoint =>
val dedupeInputData = MemoryStream[(String, Int)]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
