[WIP][SPARK-58724][SS] Incremental state cleanup for streaming dropDuplicates - #57945
Open
jerrypeng wants to merge 6 commits into
Open
[WIP][SPARK-58724][SS] Incremental state cleanup for streaming dropDuplicates#57945jerrypeng wants to merge 6 commits into
jerrypeng wants to merge 6 commits into
Conversation
Streaming deduplication currently evicts all watermark-expired state at once at batch end. For long-running batches this concentrates eviction cost at the commit boundary and lets state grow unbounded within a batch. Aggregation already supports incremental cleanup (SPARK-58635); this brings the same mechanism to StreamingDeduplicateExec and StreamingDeduplicateWithinWatermarkExec. When spark.sql.streaming.statefulOperator.incrementalCleanupFactor is > 0, up to that many eligible state rows are removed per input record processed, spreading eviction across the batch; any remainder is removed at batch end. When 0 (the default), behavior is unchanged -- all eviction happens at batch end. - BaseStreamingDeduplicateExec: add incrementalCleanupFactor/doIncrementalCleanup; replace the evictDupInfoFromState(Unit) hook with iteratorForEviction, an iterator whose next() performs one real removal so the caller can stop after incrementalCleanupFactor removals per input; add numRowsReadDuringEviction and numRowsIncrementallyRemoved metrics. - StreamingDeduplicateExec: implement iteratorForEviction with EvictionIterator (event time read from the state key). - StreamingDeduplicateWithinWatermarkExec: implement iteratorForEviction reading expiresAtMicros from the value row (EvictionIterator cannot be reused there). Under incremental cleanup, eviction uses 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 removal differs (it lags by one batch). Co-authored-by: Isaac
…rk from incremental cleanup Review found that incremental cleanup breaks output-equivalence for StreamingDeduplicateWithinWatermarkExec. 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 sharing a dedup key can have different event times. Evicting an expired entry mid-batch then lets a later non-late record with the same key miss it and be emitted as new, whereas the batch-end (factor-0) path keeps the entry visible to the whole batch and drops the record as a duplicate -- an output that depends on the cleanup factor and store iteration order. Exclude this operator from incremental cleanup: drop its incrementalCleanupFactor override so it inherits the factor-0 default and keeps evicting once at batch end (unchanged behavior). StreamingDeduplicateExec is unaffected -- its key includes the event time, so any record sharing an evictable key is itself late and is dropped by the late-events filter first. Add a regression test asserting a non-zero incrementalCleanupFactor leaves dropDuplicatesWithinWatermark output and per-batch state counts identical to the default, and update the config doc accordingly. Co-authored-by: Isaac
…on in dedup eviction StreamingDeduplicateExec.iteratorForEviction built the EvictionIterator with allowMultipleEventTimeColumns hardcoded to false. The full-eviction path it replaced resolved the event time column with allowMultipleEventTimeColumns = !allowMultipleStatefulOperators (via watermarkExpression -> findEventTimeColumn). Under the non-default spark.sql.streaming.statefulOperator.allowMultiple = false, a dedup key with two or more event-time columns would flip from "use the first column" to throwing MULTIPLE_EVENT_TIME_COLUMNS. Since iteratorForEviction serves both the incremental and the batch-end (factor-0) eviction paths, this affected the default path too. Pass !allowMultipleStatefulOperators to preserve the prior behavior. Co-authored-by: Isaac
…he null-safe conf accessor StreamingDeduplicateExec read the config via session.sessionState.conf, which dereferences `session` (SparkSession.getActiveSession.orNull) directly. On a session-less thread -- e.g. during canonicalization -- that is null and the strict val is evaluated at construction, risking an NPE. Use the SparkPlan.conf accessor instead, which falls back to the active/default conf when session is null, matching the sibling allowMultipleStatefulOperators. Co-authored-by: Isaac
…olumn comment The comment on the EvictionIterator call claimed "behavior is unchanged" and pointed at watermarkExpression -> findEventTimeColumn (which resolves over child.output). The eviction path this replaced actually resolved the event time from keyExpressions (watermarkPredicateForKeysForEviction), which is what the EvictionIterator does and what the runtime and aggregation paths do. Correct the comment to describe the keyExpressions resolution and note the one narrow difference from the old path (which additionally forced the watermark expression over child.output and so could throw MULTIPLE_EVENT_TIME_COLUMNS under the non-default allowMultiple=false with extra event-time columns outside the key). No behavior change. Co-authored-by: Isaac
…he aggregation path Rename the local `evictionWatermark` in StreamingDeduplicateExec.iteratorForEviction to `incrementalAwareEvictionWatermark`, matching the streamline aggregation operator's incremental-cleanup path (and the runtime). Pure rename; no behavior change. Co-authored-by: Isaac
jerrypeng
changed the title
uros-b
reviewed
Aug 12, 2026
uros-b
left a comment
uros-b
left a comment
Member
There was a problem hiding this comment.
Is this PR ready @jerrypeng? cc @HeartSaVioR
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 toStreamingDeduplicateExec: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 thestreamline aggregation operator). When it is
0(the default), behavior is unchanged — all eviction happens at batch end.Details:
BaseStreamingDeduplicateExecgainsincrementalCleanupFactor/doIncrementalCleanup, and its abstract eviction hook changes fromevictDupInfoFromState(store): UnittoiteratorForEviction(store): Iterator[Any]— an iterator whosenext()performs one real state removal, so the record loop can remove up toincrementalCleanupFactorrows per input andthe completion block drains the remainder. This mirrors the mechanism already used by
StatefulStreamlineAggregateExec.StreamingDeduplicateExec.iteratorForEvictionuses the existingEvictionIterator(event time read from the state key). Under incremental cleanup it evicts against the late-eventswatermark 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).
numRowsReadDuringEvictionandnumRowsIncrementallyRemoved(alongside the existingnumRemovedStateRows).StreamingDeduplicateWithinWatermarkExecdeliberately does not participate in incremental cleanup. Its dedup key excludes the event time (the expiry lives in the value row), so tworecords 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.incrementalCleanupFactordefaults to0, which preserves the existing batch-end eviction exactly. Setting it to apositive value opts a
dropDuplicatesquery into incremental cleanup; the deduplicated output is unchanged, and two new state-operator metrics (numRowsReadDuringEviction,numRowsIncrementallyRemoved) appear in the streaming progress.dropDuplicatesWithinWatermarkis unaffected.How was this patch tested?
New unit tests:
StreamingDeduplicationSuite: with a non-zeroincrementalCleanupFactor, deduplication output (and the late-event safety property) is unchanged, andnumRowsIncrementallyRemovedprogresses as expired keys are removed during record processing.
StreamingDeduplicationWithinWatermarkSuite: a non-zeroincrementalCleanupFactorleavesdropDuplicatesWithinWatermarkoutput and per-batch state counts identical to the default(confirming the operator stays opted out).
All existing
StreamingDeduplicationSuiteandStreamingDeduplicationWithinWatermarkSuitetests continue to pass (the factor-0 default path is unchanged), anddev/scalastylepasses.Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Code