Redistribute file groups across size preserving order
(
&self,
file_groups: &[FileGroup],
)
| 268 | |
| 269 | /// Redistribute file groups across size preserving order |
| 270 | fn repartition_preserving_order( |
| 271 | &self, |
| 272 | file_groups: &[FileGroup], |
| 273 | ) -> Option<Vec<FileGroup>> { |
| 274 | // Can't repartition and preserve order if there are more groups |
| 275 | // than partitions |
| 276 | if file_groups.len() >= self.target_partitions { |
| 277 | return None; |
| 278 | } |
| 279 | let num_new_groups = self.target_partitions - file_groups.len(); |
| 280 | |
| 281 | // If there is only a single file |
| 282 | if file_groups.len() == 1 && file_groups[0].len() == 1 { |
| 283 | return self.repartition_evenly_by_size(file_groups); |
| 284 | } |
| 285 | |
| 286 | // Find which files could be split (single file groups) |
| 287 | let mut heap: BinaryHeap<_> = file_groups |
| 288 | .iter() |
| 289 | .enumerate() |
| 290 | .filter_map(|(group_index, group)| { |
| 291 | // ignore groups that do not have exactly 1 file |
| 292 | if group.len() == 1 { |
| 293 | Some(ToRepartition { |
| 294 | source_index: group_index, |
| 295 | file_size: group[0].effective_size(), |
| 296 | new_groups: vec![group_index], |
| 297 | }) |
| 298 | } else { |
| 299 | None |
| 300 | } |
| 301 | }) |
| 302 | .map(CompareByRangeSize) |
| 303 | .collect(); |
| 304 | |
| 305 | // No files can be redistributed |
| 306 | if heap.is_empty() { |
| 307 | return None; |
| 308 | } |
| 309 | |
| 310 | // Add new empty groups to which we will redistribute ranges of existing files |
| 311 | // Add new empty groups to which we will redistribute ranges of existing files |
| 312 | let mut file_groups: Vec<_> = file_groups |
| 313 | .iter() |
| 314 | .cloned() |
| 315 | .chain(repeat_with(|| FileGroup::new(Vec::new())).take(num_new_groups)) |
| 316 | .collect(); |
| 317 | |
| 318 | // Divide up empty groups |
| 319 | for (group_index, group) in file_groups.iter().enumerate() { |
| 320 | if !group.is_empty() { |
| 321 | continue; |
| 322 | } |
| 323 | // Pick the file that has the largest ranges to read so far |
| 324 | let mut largest_group = heap.pop().unwrap(); |
| 325 | largest_group.new_groups.push(group_index); |
| 326 | heap.push(largest_group); |
| 327 | } |
no test coverage detected