[FLINK-25796] Avoid record copy for result partition of sort-shuffle if there are enough buffers for better performance

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.15.0
• Component/s: Runtime / Network
• Labels: pull-request-available
• Assignee: Yingjie Cao
• Reporter: Yingjie Cao
• Created: 25/Jan/22 03:10
• Updated: 12/Feb/22 03:52
• Resolved: 12/Feb/22 03:52

Description

Currently, for result partition of sort-shuffle, there is extra record copy overhead Introduced by clustering records by subpartition index. For small records, this overhead can cause even 20% performance regression. This ticket aims to solve the problem.

In fact, the hash-based implementation is a nature way to achieve the goal of sorting records by partition index. However, it incurs some serious weaknesses. For example, when there is no enough buffers or there is data skew, it can waste buffers and influence compression efficiency which can cause performance regression.

This ticket tries to solve the issue by dynamically switching between the two implementations, that is, if there are enough buffers, the hash-based implementation will be used and if there is no enough buffers, the sort-based implementation will be used.

Issue Links

• links to: GitHub Pull Request #18505