shenavaa opened a new pull request, #59953:
URL: https://github.com/apache/spark/pull/59953
### What changes were proposed in this pull request?
`StreamingDeduplicateWithinWatermarkExec` now honours
`spark.sql.streaming.statefulOperator.incrementalCleanupFactor`, as
`StreamingDeduplicateExec` already does.
- `BaseStreamingDeduplicateExec` gets a `protected def isDuplicate(store,
key)` hook. The default is the existing `store.keyExists(key)`, so
`StreamingDeduplicateExec` is
unchanged.
- `StreamingDeduplicateWithinWatermarkExec`:
- reads `incrementalCleanupFactor` from the conf (previously fixed at
0).
- overrides `isDuplicate`: with incremental cleanup on, a key is a
duplicate only if its stored `expiresAtMicros` is **above the late-events
watermark**. Without
incremental cleanup it still uses `keyExists`.
- `iteratorForEviction` uses the late-events watermark when incremental
cleanup is on (eviction watermark otherwise, as before), removes in `next()` so
each `next()` is
exactly one removal (the contract the base class relies on for the
per-record budget and the `numRowsIncrementallyRemoved` metric), and **re-reads
the current value before
removing** a candidate.
- `SQLConf` doc for the factor updated to say it now applies to
`dropDuplicatesWithinWatermark`.
Why this is safe: this operator's dedup key does not contain the event
time; the expiry lives in the value row. Under the factor-0 path, at any point
in a batch every entry
in the store satisfies `expiresAt > lateEventsWatermark` (batch-end
eviction removed everything `<= previous eviction watermark == this batch's
late-events watermark`, and
new inserts have `expiresAt = eventTime + delay` with `eventTime >
lateEventsWatermark` because of the late-record filter). So `keyExists` is
equivalent to `present &&
expiresAt > lateEventsWatermark`. The new `isDuplicate` computes exactly
that predicate, and the incremental iterator only removes entries with
`expiresAt <=
lateEventsWatermark`, which can never change the predicate's result.
Output is therefore identical to the factor-0 path regardless of when, or in
which order, expired entries
are physically removed. This is the same late-events-watermark reasoning
already used by `StreamingDeduplicateExec.iteratorForEviction`.
The re-read is required because the store iterator may be a snapshot
taken when the batch started (RocksDB): a record processed earlier in the batch
can refresh a
logically-expired key with a new expiry, and the snapshot would still
show the stale value. Without the re-read, that live entry would be evicted and
a later duplicate
emitted.
### Why are the changes needed?
`dropDuplicatesWithinWatermark` was explicitly excluded from incremental
cleanup, so all state eviction for it happened at batch end. The reason for the
exclusion (mid-batch
eviction could change the output because expiry is in the value, not the
key) is addressed by checking logical expiry instead of physical presence. With
this change the
eviction cost can be spread across record processing for this operator
too, as it already can for `dropDuplicates`.
### Does this PR introduce _any_ user-facing change?
No new config and no change with the default settings
(`incrementalCleanupFactor = 0`).
When the factor is already set to a non-zero value,
`dropDuplicatesWithinWatermark` now evicts incrementally instead of ignoring
the setting. Emitted rows are unchanged. The
only observable difference is in state metrics: physical reclamation can
trail by one batch (entries between the late-events and eviction watermarks are
removed in the next
batch rather than at this batch's end), which matches `dropDuplicates`
under incremental cleanup. No state schema change; existing checkpoints are
compatible.
### How was this patch tested?
`StreamingDeduplicationWithinWatermarkSuite`:
- Updated "incremental cleanup preserves dropDuplicatesWithinWatermark
output" (factor 10): same input sequence as the factor-0 test, asserts the
emitted rows are identical
and documents the one-batch lag in state-row counts.
- New "incremental cleanup must not evict a key refreshed earlier in the
same batch (RocksDB)": factor 1, one shuffle partition, byte-ordered keys, so
the eviction iterator
deterministically reaches a key *after* a record in the same batch
refreshed it. Verified this test fails without the re-read (the refreshed key
is evicted and the duplicate
is emitted: 4 rows instead of 3) and passes with it.
- All existing tests in the suite pass (10/10).
`StreamingDeduplicationSuite` (shares the modified base class): 32/32
pass.
`scalastyle:check` on `sql/core` and `sql/catalyst`: clean.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude (Anthropic). Some of the code and tests were
produced with the tool's assistance and were reviewed, run and verified by the
author.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
