[flink][core] Introduce file-size option in Paimon source - #9368
Open
dwangatt wants to merge 2 commits into
Open
[flink][core] Introduce file-size option in Paimon source#9368dwangatt wants to merge 2 commits into
dwangatt wants to merge 2 commits into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
The existing FAIR split assignment uses row count as the split weight. This works well when records have similar sizes, but can create significant workload skew for tables containing records with highly variable payload sizes.
Two splits with the same row count may represent very different amounts of data. As a result, splits containing large-payload records can become concentrated on a small number of source subtasks or TaskManagers, causing:
• Uneven S3 read and network load
• Higher memory and CPU pressure on a subset of TaskManagers
• Long-running straggler tasks
• Longer overall batch and rescale job duration
This change introduces a new split weight mode:
scan.split-enumerator.weight-mode=file-size
When used with the FAIR split assigner, Paimon calculates each DataSplit weight from the total size of its data files rather than its row count. The assigner can therefore distribute byte-heavy splits more evenly across source subtasks.
For split types without file-size information, the implementation falls back to row-count weighting.
The goal is to improve source workload balancing for large-payload tables—particularly during Paimon rescale and other bounded batch reads—by spreading heavy IO tasks across more TaskManagers and reducing execution long tails.
Tests
Add some tests to cover new code. Also tested in our FLINK batch job and new option reduced job running time from 8 hours -> 2 hours.