[SPARK-30657] Streaming limit after streaming dropDuplicates can throw error

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 2.3.0, 2.3.1, 2.3.2, 2.3.3, 2.3.4, 2.4.0, 2.4.1, 2.4.2, 2.4.3, 2.4.4
• Fix Version/s: 3.0.0
• Component/s: Structured Streaming
• Target Version/s: 3.0.0
• Assignee: Tathagata Das
• Reporter: Tathagata Das
• Created: 28/Jan/20 09:08
• Updated: 01/Feb/20 22:32
• Resolved: 01/Feb/20 00:30

Description

LocalLimitExec does not consume the iterator of the child plan. So if there is a limit after a stateful operator like streaming dedup in append mode (e.g. streamingdf.dropDuplicates().limit(5)), the state changes of streaming duplicate may not be committed (most stateful ops commit state changes only after the generated iterator is fully consumed). This leads to the next batch failing with java.lang.IllegalStateException: Error reading delta file .../N.delta does not exist as the state store delta file was never generated.

Issue Links

• links to: GitHub Pull Request #27373