nahtonaj opened a new pull request, #58261:
URL: https://github.com/apache/spark/pull/58261
### 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.
--
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]
