mustafasrepo commented on code in PR #5661:
URL: https://github.com/apache/arrow-datafusion/pull/5661#discussion_r1142947802
##########
datafusion/core/src/physical_optimizer/sort_enforcement.rs:
##########
@@ -343,77 +347,60 @@ impl PhysicalOptimizerRule for EnforceSorting {
}
/// This function turns plans of the form
-/// "SortExec: expr=[a@0 ASC]",
+/// "SortExec: expr=\[a@0 ASC\]",
/// " CoalescePartitionsExec",
/// " RepartitionExec: partitioning=RoundRobinBatch(8),
input_partitions=1",
/// to
-/// "SortPreservingMergeExec: [a@0 ASC]",
-/// " SortExec: expr=[a@0 ASC]",
+/// "SortPreservingMergeExec: \[a@0 ASC\]",
+/// " SortExec: expr=\[a@0 ASC\]",
/// " RepartitionExec: partitioning=RoundRobinBatch(8),
input_partitions=1",
-/// by following connections from `CoalescePartitionsExec`s to `SortExec`s.
+/// by following connections from [`CoalescePartitionsExec`]s to [`SortExec`]s.
/// By performing sorting in parallel, we can increase performance in some
scenarios.
fn parallelize_sorts(
requirements: PlanWithCorrespondingCoalescePartitions,
) -> Result<Option<PlanWithCorrespondingCoalescePartitions>> {
let plan = requirements.plan;
- if plan.children().is_empty() {
- return Ok(None);
- }
let mut coalesce_onwards = requirements.coalesce_onwards;
- // We know that `plan` has children, so `coalesce_onwards` is non-empty.
- if coalesce_onwards[0].is_some() {
- if (is_sort(&plan) || is_sort_preserving_merge(&plan))
- // Make sure that Sort is actually global sort
- && plan.output_partitioning().partition_count() == 1
- {
- // If there is a connection between a `CoalescePartitionsExec` and
a
- // Global Sort that satisfy the requirements (i.e. intermediate
- // executors don't require single partition), then we can
- // replace the `CoalescePartitionsExec`+ GlobalSort cascade with
- // the `SortExec` + `SortPreservingMergeExec`
- // cascade to parallelize sorting.
- let mut prev_layer = plan.clone();
- update_child_to_remove_coalesce(&mut prev_layer, &mut
coalesce_onwards[0])?;
- let sort_exprs = get_sort_exprs(&plan)?;
- add_sort_above(&mut prev_layer, sort_exprs.to_vec())?;
- let spm = SortPreservingMergeExec::new(sort_exprs.to_vec(),
prev_layer);
- return Ok(Some(PlanWithCorrespondingCoalescePartitions {
- plan: Arc::new(spm),
- coalesce_onwards: vec![None],
- }));
- } else if plan.as_any().is::<CoalescePartitionsExec>() {
- // There is an unnecessary `CoalescePartitionExec` in the plan.
- let mut prev_layer = plan.clone();
- update_child_to_remove_coalesce(&mut prev_layer, &mut
coalesce_onwards[0])?;
- let new_plan = plan.with_new_children(vec![prev_layer])?;
- return Ok(Some(PlanWithCorrespondingCoalescePartitions {
- plan: new_plan,
- coalesce_onwards: vec![None],
- }));
- }
+ if plan.children().is_empty() || coalesce_onwards[0].is_none() {
Review Comment:
I have just removed nested `if` by early returning in `is_none` case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
