[FLINK-28512] Select HashBasedDataBuffer and SortBasedDataBuffer dynamically based on the number of network buffers can be allocated for SortMergeResultPartition

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.16.0
• Labels: pull-request-available
• Assignee: Yuxin Tan
• Reporter: Yingjie Cao
• Created: 12/Jul/22 10:29
• Updated: 21/Jul/22 10:16
• Resolved: 21/Jul/22 10:16

Description

Currently, the SortMergeResultPartition select to use HashBasedDataBuffer and SortBasedDataBuffer based on the number of required buffers per result partition decided by 'taskmanager.network.sort-shuffle.min-buffers'. If the configured value is large enough, HashBasedDataBuffer will be used, otherwise, SortBasedDataBuffer will be used. Usually, the HashBasedDataBuffer has better performance. However, it is not easy to tune this value, because if a user tries to increase it for better performance, he/she is easy to encounter the 'Insufficient number of network buffers' error. This patch improves this case by selecting HashBasedDataBuffer and SortBasedDataBuffer dynamically based on the number of network buffers can be allocated. More specifically, if there is enough buffers at runtime, HashBasedDataBuffer will be used, otherwise, SortBasedDataBuffer will be used. To achieve better performance, the user only need to increase total amount of network memory per task manager.

Issue Links

• links to: GitHub Pull Request #20315