jerrypeng opened a new pull request, #58673:
URL: https://github.com/apache/spark/pull/58673
### What changes were proposed in this pull request?
This PR enables `dropDuplicatesWithinWatermark` in Structured Streaming
Real-Time Mode (RTM).
It:
- Adds `StreamingDeduplicateWithinWatermarkExec` to the RTM operator
allowlist.
- Applies the existing RTM restriction on stateful operators below a
`Union` to `DeduplicateWithinWatermark`.
- Adds positive and negative analysis tests for the directional union rule.
- Adds an RTM test covering duplicate suppression, watermark-based
eviction, late-data filtering, key re-admission, and pipelined shuffle
execution.
- Updates the RTM documentation and corrects the PySpark `dropDuplicates`
subset syntax.
The existing batch-end eviction behavior is preserved. Incremental
mid-batch eviction is unsafe for this operator because the expiry timestamp is
stored in the value rather than the
deduplication key. Evicting a key while processing a batch could allow a
later non-late row with the same key to be emitted, making results depend on
input and state-store iteration
order.
### Why are the changes needed?
RTM supports `dropDuplicates`, but currently rejects the bounded-state
`dropDuplicatesWithinWatermark` physical operator because it is missing from
the allowlist.
Supporting this operator lets RTM applications use event-time watermarks
to bound deduplication state instead of retaining every observed key
indefinitely. The existing planning,
watermark propagation, state-store, and checkpoint mechanisms already
support the operator; this PR enables it for RTM and adds the corresponding
validation and coverage.
### Does this PR introduce _any_ user-facing change?
Yes.
Previously, an RTM query using `dropDuplicatesWithinWatermark` failed to
start with `STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST`.
After this change, queries such as the following are supported:
```python
query = (
df.withWatermark("eventTime", "10 minutes")
.dropDuplicatesWithinWatermark(["id"])
.writeStream
.outputMode("update")
.trigger(realTime="5 minutes")
.start()
)
Watermark advancement and state eviction take effect at RTM batch
boundaries.
This changes behavior only on the unreleased master branch and does not
add a new API, configuration, or state format.
### How was this patch tested?
Added tests covering:
- Direct RTM support for DeduplicateWithinWatermark.
- A stateful operator after Union, which is supported.
- A stateful operator before Union, which remains unsupported.
- Live duplicate suppression in RTM.
- State retention across batch boundaries.
- Watermark-driven eviction and subsequent key re-admission.
- Late-event filtering.
- Physical planning as StreamingDeduplicateWithinWatermarkExec.
- Pipelined shuffle execution.
The following suites passed:
- UnsupportedOperationsSuite: 225 tests
- StreamRealTimeModeWithManualClockSuite: 15 tests
- StreamingDeduplicationWithinWatermarkSuite: 9 tests
Catalyst and SQL main/test Scalastyle checks also passed.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with OpenAI Codex (GPT-5)
--
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]
