[SPARK-46064] EliminateEventTimeWatermark does not consider the fact that isStreaming flag can change for current child during resolution

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.3.2, 3.4.1, 3.5.0, 4.0.0
• Fix Version/s: 3.4.2, 4.0.0, 3.5.1
• Component/s: Structured Streaming
• Labels: pull-request-available
• Assignee: Jungtaek Lim
• Reporter: Jungtaek Lim
• Created: 23/Nov/23 05:05
• Updated: 23/Nov/23 11:26
• Resolved: 23/Nov/23 11:26

Description

Looks like this is a long standing bug.

The object `EliminateEventTimeWatermark` is implemented as a rule, but it is not registered in analyzer/optimizer. Instead, it is called directly when withWatermark method is called, which means the rule is applied immediately against the child, regardless whether child is resolved or not.

It is not an issue for the usage of pure DataFrame API because streaming sources have the flag isStreaming set to true even it is yet resolved, but mix-up of SQL and DataFrame API would expose the issue; we may not know the exact value of isStreaming flag on unresolved node and it is subject to change upon resolution.

We should register EliminateEventTimeWatermark as a rule on analysis (or pre-optimization) instead, and do not apply the elimination if the child is not yet resolved.

Issue Links

• links to: GitHub Pull Request #43971