[SPARK-38124] Revive HashClusteredDistribution and apply to stream-stream join

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 3.3.0
• Fix Version/s: 3.3.0
• Component/s: SQL, Structured Streaming
• Assignee: Jungtaek Lim
• Reporter: Jungtaek Lim
• Created: 07/Feb/22 07:04
• Updated: 16/Mar/22 07:24
• Resolved: 08/Feb/22 23:32

Description

SPARK-35703 removed HashClusteredDistribution and replaced its usages with ClusteredDistribution.

While this works great for non stateful operators, we still need to have a separate requirement of distribution for stateful operator, because the requirement of ClusteredDistribution is too relaxed while the requirement of physical partitioning on stateful operator is quite strict.

In most cases, stateful operators must require child distribution as HashClusteredDistribution, with below major assumptions:

1. HashClusteredDistribution creates HashPartitioning and we will never ever change it for the future.
2. We will never ever change the implementation of partitionIdExpression in HashPartitioning for the future, so that Partitioner will behave consistently across Spark versions.
3. No partitioning except HashPartitioning can satisfy HashClusteredDistribution.

 

We should revive HashClusteredDistribution (with probably renaming specifically with stateful operator) and apply the distribution to the all stateful operators.

SPARK-35703 only touched stream-stream join, which means stream-stream join hasn't been broken in actual releases. Let's aim the partial revert of SPARK-35703 in this ticket, and have another ticket to deal with other stateful operators, which have been broken for their introduction (2.2+).

Issue Links

• links to: [Github] Pull Request #35419 (HeartSaVioR)
• links to: [Github] Pull Request #35512 (HeartSaVioR)
• links to: [Github] Pull Request #35529 (c21)
• links to: [Github] Pull Request #35529 (c21)