This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 990671c9c2ea [SPARK-58979][SS] Preserve event time through streaming
dedup pruning
990671c9c2ea is described below
commit 990671c9c2eab3da13ce66866b1decdbc6122070
Author: Jonathan Gao <[email protected]>
AuthorDate: Thu Oct 1 12:20:26 2026 +0900
[SPARK-58979][SS] Preserve event time through streaming dedup pruning
### What changes were proposed in this pull request?
Include event-time watermark attributes in streaming
`Deduplicate.references`, so `ColumnPruning` cannot remove the event-time
column that `StreamingDeduplicateExec` needs for late-record filtering.
Batch deduplication continues to reference only its key attributes.
This follows the existing `DeduplicateWithinWatermark.references` pattern
added by SPARK-50492.
### Why are the changes needed?
Two otherwise equivalent streaming queries currently behave differently
depending on their final projection.
When the event-time column is retained:
```scala
input.toDF()
.selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
.withWatermark("ts", "10 seconds")
.dropDuplicates("id")
.selectExpr("id", "CAST(ts AS LONG) AS tsl")
```
the watermark attribute reaches `StreamingDeduplicateExec`, and late input
is filtered.
When the event-time column is projected away:
```scala
input.toDF()
.selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
.withWatermark("ts", "10 seconds")
.dropDuplicates("id")
.select("id")
```
`ColumnPruning` removes `ts` below `Deduplicate`. The physical
deduplication operator can no longer construct its late-event predicate, so the
same late row is emitted if it has a new deduplication key.
A downstream projection should not change the internal watermark behavior
of streaming deduplication.
JIRA: https://issues.apache.org/jira/browse/SPARK-58979
Related precedent: https://issues.apache.org/jira/browse/SPARK-50492
### Does this PR introduce _any_ user-facing change?
Yes.
Streaming `dropDuplicates` now preserves late-record filtering when the
event-time column is not selected downstream. Previously, sufficiently late
rows with new deduplication keys could be emitted only because column pruning
removed the watermark attribute.
### How was this patch tested?
Added paired tests to `StreamingDeduplicationSuite` using identical input:
- Event-time projected away after `dropDuplicates("id")`.
- Event-time retained above `dropDuplicates("id")` as the control.
Both tests verify that the same late row is filtered.
Local verification:
- `dev/lint-scala`: Scalastyle passed.
- `dev/lint-scala`: Scalafmt passed.
- `git diff --check`: passed.
The focused SBT suite could not run locally because this host could not
resolve Maven/SBT dependencies. GitHub Actions will run the complete test
matrix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: KiroCrew using Databricks GPT-5.6 Sol.
Closes #58261 from nahtonaj/fix/spark-dedup-watermark-pruning.
Lead-authored-by: Jonathan Gao <[email protected]>
Co-authored-by: Jon Gao <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../plans/logical/basicLogicalOperators.scala | 8 ++++++
.../streaming/StreamingDeduplicationSuite.scala | 32 ++++++++++++++++++++++
2 files changed, 40 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index dd44a8b25912..0de24ec1d166 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -2441,6 +2441,14 @@ case class Deduplicate(
keys: Seq[Attribute],
child: LogicalPlan,
dedupSpec: Option[DeduplicateSpec] = None) extends UnaryNode {
+ // Streaming deduplication filters late rows even when event time is not
part of the key.
+ // A watermarked key already preserves event time; let unused alternatives
be pruned.
+ override def references: AttributeSet = AttributeSet(keys) ++ AttributeSet(
+ if (child.isStreaming &&
!keys.exists(_.metadata.contains(EventTimeWatermark.delayKey))) {
+ child.output.filter(_.metadata.contains(EventTimeWatermark.delayKey))
+ } else {
+ Seq.empty
+ })
override def maxRows: Option[Long] = child.maxRows
override def output: Seq[Attribute] = {
val base = child.output
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 21261280a280..f473e26a8a35 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
@@ -376,6 +376,38 @@ class StreamingDeduplicationSuite extends
StateStoreMetricsTest
)
}
+ test("SPARK-58979: dedup event-time projected away") {
+ val input = MemoryStream[(String, Long)]
+ val result = input.toDF()
+ .selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
+ .withWatermark("ts", "10 seconds")
+ .dropDuplicates("id")
+ .select("id")
+
+ testStream(result, Append)(
+ AddData(input, ("a", 1000000L)),
+ CheckNewAnswer("a"),
+ AddData(input, ("b", 1000L)),
+ CheckNewAnswer()
+ )
+ }
+
+ test("SPARK-58979: dedup event-time retained") {
+ val input = MemoryStream[(String, Long)]
+ val result = input.toDF()
+ .selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
+ .withWatermark("ts", "10 seconds")
+ .dropDuplicates("id")
+ .selectExpr("id", "CAST(ts AS LONG) AS tsl")
+
+ testStream(result, Append)(
+ AddData(input, ("a", 1000000L)),
+ CheckNewAnswer(("a", 1000000L)),
+ AddData(input, ("b", 1000L)),
+ CheckNewAnswer()
+ )
+ }
+
test("test no-data flag") {
val flagKey = SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
