MCPcopy Create free account
hub / github.com/apache/datafusion / repartition_preserving_order

Method repartition_preserving_order

datafusion/datasource/src/file_groups.rs:270–362  ·  view source on GitHub ↗

Redistribute file groups across size preserving order

(
        &self,
        file_groups: &[FileGroup],
    )

Source from the content-addressed store, hash-verified

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 }

Callers 1

Calls 15

newFunction · 0.85
collectMethod · 0.80
effective_sizeMethod · 0.80
range_sizeMethod · 0.80
lenMethod · 0.45
mapMethod · 0.45
iterMethod · 0.45
is_emptyMethod · 0.45
clonedMethod · 0.45
takeMethod · 0.45
popMethod · 0.45

Tested by

no test coverage detected