gabotechs commented on code in PR #23184:
URL: https://github.com/apache/datafusion/pull/23184#discussion_r3542595582
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1309,75 +1402,98 @@ pub fn ensure_distribution(
}
};
- let streaming_benefit = if child.data {
- preserving_order_enables_streaming(&plan, &child.plan)?
- } else {
- false
- };
+ Ok(DistributionChildState {
+ context: child,
+ required_input_ordering,
+ maintains_input_order: maintains,
+ requirement,
+ })
+ },
+ )
+ .collect::<Result<Vec<_>>>()?;
- // There is an ordering requirement of the operator:
- if let Some(required_input_ordering) = required_input_ordering {
- // Either:
- // - Ordering requirement cannot be satisfied by preserving
ordering through repartitions, or
- // - using order preserving variant is not desirable.
- let sort_req = required_input_ordering.into_single();
- let ordering_satisfied = child
- .plan
- .equivalence_properties()
- .ordering_satisfy_requirement(sort_req.clone())?;
-
- if (!ordering_satisfied ||
!order_preserving_variants_desirable)
- && !streaming_benefit
- && child.data
- {
- child = replace_order_preserving_variants(child)?;
- // If ordering requirements were satisfied before
repartitioning,
- // make sure ordering requirements are still satisfied
after.
- if ordering_satisfied {
- // Make sure to satisfy ordering requirement:
- child = add_sort_above_with_check(
- child,
- sort_req,
- plan.downcast_ref::<OutputRequirementExec>()
- .map(|output| output.fetch())
- .unwrap_or(None),
- )?;
- }
- }
- // Stop tracking distribution changing operators
- child.data = false;
- } else {
- let streaming_benefit = if child.data {
- preserving_order_enables_streaming(&plan, &child.plan)?
+ enforce_distribution_relationships(
Review Comment:
TBH, it's not super pretty to have this function call here in the middle,
but I'd not now how to make it better without a major refactor of this file...
My impressions is that this file is worth refactoring. In the current state,
it's super difficult to understand what's happening.
This PR is not making it worst though, so one thing we could do is:
1. Make up for it with some good comments at the top of this function call
and probably below it.
2. File an issue for refactoring this file, I might take this one myself.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
