[SPARK-50492] Fix java.util.NoSuchElementException when event time column is dropped after dropDuplicatesWithinWatermark

• Type: Task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.5.0, 4.0.0, 3.5.1, 3.5.2, 3.5.3
• Fix Version/s: 4.0.0, 3.5.4
• Component/s: Structured Streaming
• Labels: pull-request-available
• Assignee: Livia Zhu
• Reporter: Livia Zhu
• Created: 04/Dec/24 22:56
• Updated: 06/Dec/24 08:37
• Resolved: 06/Dec/24 02:29

Description

Consider the following query:

```
val result = inputData.toDF()
.select("_1", "_2")
.withColumn("timestamp", to_timestamp($"_2", "yyyy-MM-dd HH:mm:ss"))
.withWatermark("timestamp", "24 hours")
.dropDuplicatesWithinWatermark("timestamp")
.select("_1")[]
```
 
Currently, the ColumnPruning optimization will prune the `timestamp` column since it is not selected in the final Project, leading to a `java.util.NoSuchElementException` when we try to get the event time column in DeduplicateWithinWatermarkExec.
 
We need to update the references for the DeduplicateWithinWatermark logical plan node so that the event time column is included in the references.

Issue Links

• links to: GitHub Pull Request #49065
• links to: GitHub Pull Request #49087