Make fetch shuffle partition data in parallel #256
Conversation
|
Hi @andygrove, @thinkharderdev, @avantgardnerio, could you help review this PR? |
| let mut partition_locations: Vec<PartitionLocation> = partition_locations | ||
| .into_values() | ||
| .flat_map(|ps| ps.into_iter().enumerate()) | ||
| .sorted_by(|(p1_idx, _), (p2_idx, _)| Ord::cmp(p1_idx, p2_idx)) | ||
| .map(|(_, p)| p) | ||
| .collect(); | ||
| // Shuffle partitions for evenly send fetching partition requests to avoid hot executors within multiple tasks | ||
| partition_locations.shuffle(&mut thread_rng()); |
There was a problem hiding this comment.
Why do we sort by index before shuffling here?
There was a problem hiding this comment.
Sorting first may be helpful for reducing the bias of random chosen. Maybe it's not necessary.
| // TODO make the maximum size configurable, or make it depends on global memory control | ||
| let max_request_num = 50usize; |
There was a problem hiding this comment.
+1 for making this a config option that is documented in the user guide. I can help with this in a follow-on PR.
| .try_flatten() | ||
| }); | ||
| let task_id = context.task_id().unwrap_or_else(|| partition.to_string()); | ||
| info!("ShuffleReaderExec::execute({})", task_id); |
There was a problem hiding this comment.
This could make the logs quite noisy again?
|
@yahoNanJing Could you please explain a little about the error handling case ? |
|
@mingmwang Just added one commit to abort fast when error occurs. |
Which issue does this PR close?
Closes #208.
Rationale for this change
What changes are included in this PR?
Are there any user-facing changes?