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 with 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.
##########
datafusion/physical-plan/src/distribution_requirements.rs:
##########
@@ -0,0 +1,484 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Input distribution requirements for physical execution plans.
+
+use std::sync::Arc;
+
+use datafusion_common::{Result, internal_err};
+use datafusion_physical_expr::{
+ Distribution, EquivalenceProperties, Partitioning,
PartitioningSatisfaction,
+ PhysicalExpr, physical_exprs_equal,
+};
+
+use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties,
InvariantLevel};
+
+/// Distribution requirements for an [`ExecutionPlan`]'s inputs.
+///
+/// [`RequiredInputDistributions`] describes what distribution an operator
+/// requires from each child.
+///
+/// - [`Self::new`] describes independent per-child requirements.
+/// - [`Self::co_partitioned`] additionally requires child partitions with the
+/// same index to cover compatible key ranges.
+///
+/// For a single-input aggregate:
+///
+/// ```text
+/// AggregateExec
+/// child 0 requirement: KeyPartitioned(group_exprs)
+/// ```
+///
+/// each input partition can aggregate its own key domain independently.
+///
+/// For a partitioned join:
+///
+/// ```text
+/// HashJoinExec
+/// child 0 requirement: KeyPartitioned(left_keys)
+/// child 1 requirement: KeyPartitioned(right_keys)
+///
+/// partition 0: join(left partition 0, right partition 0)
+/// partition 1: join(left partition 1, right partition 1)
+/// partition 2: join(left partition 2, right partition 2)
+/// ```
+///
+/// each child must satisfy its own key requirement. In addition, matching
+/// partition indexes must be safe to process together.
+#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub struct RequiredInputDistributions {
Review Comment:
:+1: I like the API this is exposing, it feels intuitive and future-proof.
Regarding naming of the struct, unless someone has a better suggestion, I
think this one is fine.
##########
datafusion/physical-plan/src/distribution_requirements.rs:
##########
@@ -0,0 +1,484 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Input distribution requirements for physical execution plans.
+
+use std::sync::Arc;
+
+use datafusion_common::{Result, internal_err};
+use datafusion_physical_expr::{
+ Distribution, EquivalenceProperties, Partitioning,
PartitioningSatisfaction,
+ PhysicalExpr, physical_exprs_equal,
+};
+
+use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties,
InvariantLevel};
+
+/// Distribution requirements for an [`ExecutionPlan`]'s inputs.
+///
+/// [`RequiredInputDistributions`] describes what distribution an operator
+/// requires from each child.
+///
+/// - [`Self::new`] describes independent per-child requirements.
+/// - [`Self::co_partitioned`] additionally requires child partitions with the
+/// same index to cover compatible key ranges.
+///
+/// For a single-input aggregate:
+///
+/// ```text
+/// AggregateExec
+/// child 0 requirement: KeyPartitioned(group_exprs)
+/// ```
+///
+/// each input partition can aggregate its own key domain independently.
+///
+/// For a partitioned join:
+///
+/// ```text
+/// HashJoinExec
+/// child 0 requirement: KeyPartitioned(left_keys)
+/// child 1 requirement: KeyPartitioned(right_keys)
+///
+/// partition 0: join(left partition 0, right partition 0)
+/// partition 1: join(left partition 1, right partition 1)
+/// partition 2: join(left partition 2, right partition 2)
+/// ```
+///
+/// each child must satisfy its own key requirement. In addition, matching
+/// partition indexes must be safe to process together.
+#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub struct RequiredInputDistributions {
+ /// Per-child distribution requirements, indexed by child position.
+ children: Vec<ChildDistributionRequirement>,
+ /// Child indexes that must also have compatible partition layouts.
+ co_partitioned: Option<Vec<usize>>,
+}
+
+impl RequiredInputDistributions {
+ /// Create independent per-child requirements.
+ pub fn new(per_child: Vec<Distribution>) -> Self {
+ let children = per_child
+ .into_iter()
+ .map(|distribution| ChildDistributionRequirement {
+ distribution,
+ satisfaction: InputDistributionSatisfaction::Default,
+ })
+ .collect();
+
+ Self {
+ children,
+ co_partitioned: None,
+ }
+ }
+
+ /// Create a requirement that all children are co-partitioned.
+ ///
+ /// Each child must satisfy its own [`Distribution`]. Matching partition
+ /// indexes are processed together:
+ ///
+ /// ```text
+ /// left: Range(left.a ASC, split_points=[10, 20])
+ /// right: Range(right.x ASC, split_points=[10, 20])
+ ///
+ /// partition 0 from both sides contains keys before 10
+ /// partition 1 from both sides contains keys in [10, 20)
+ /// partition 2 from both sides contains keys at/after 20
+ /// ```
+ ///
+ /// If the split points differ, partition `i` from one side no longer
covers
+ /// the same key range as partition `i` from the other side.
+ pub fn co_partitioned(per_child: Vec<Distribution>) -> Self {
+ debug_assert!(
+ per_child.len() >= 2,
+ "co-partitioned distribution requirements need at least two
children"
+ );
+ let co_partitioned = (0..per_child.len()).collect();
+ let mut result = Self::new(per_child);
+ result.co_partitioned = Some(co_partitioned);
+ result
+ }
+
+ /// Return the per-child distribution requirements.
+ pub fn per_child_distributions(
+ &self,
+ ) -> impl ExactSizeIterator<Item = &Distribution> + '_ {
+ self.children.iter().map(|child| &child.distribution)
+ }
+
+ /// Return the distribution requirement for a child.
+ pub fn child_distribution(&self, child_idx: usize) ->
Option<&Distribution> {
+ self.children
+ .get(child_idx)
+ .map(|child| &child.distribution)
+ }
+
+ /// Return the per-child distribution requirements.
+ ///
+ /// WARNING: This intentionally drops any grouped relationship.
+ pub fn into_per_child(self) -> Vec<Distribution> {
+ self.children
+ .into_iter()
+ .map(|child| child.distribution)
+ .collect()
+ }
+
+ /// Returns how a child satisfies its distribution requirement using this
+ /// requirement set's satisfaction policy.
+ pub fn child_satisfaction(
+ &self,
+ child_idx: usize,
+ child: &dyn ExecutionPlan,
+ allow_subset: bool,
+ ) -> Result<PartitioningSatisfaction> {
+ let Some(requirement) = self.children.get(child_idx) else {
+ return internal_err!(
+ "missing distribution requirement for child {child_idx}"
+ );
+ };
+
+ Ok(requirement.satisfaction.satisfaction(
+ child.output_partitioning(),
+ &requirement.distribution,
+ child.equivalence_properties(),
+ allow_subset,
+ ))
+ }
+
+ /// Return child indexes whose co-partitioning requirements are
+ /// unsatisfied by the provided candidate children.
+ ///
+ /// Independent per-child requirements are intentionally ignored here, use
+ /// [`Self::child_satisfaction`] for those checks. An empty result means
all
+ /// co-partitioning requirements are satisfied.
+ #[doc(hidden)]
+ pub fn unsatisfied_co_partitioned_children(
+ &self,
+ plan_name: &str,
+ children: &[&dyn ExecutionPlan],
+ ) -> Result<Vec<usize>> {
+ self.validate_shape(plan_name, children.len())?;
+
+ let Some(co_partitioned) = &self.co_partitioned else {
+ return Ok(vec![]);
+ };
+ if self.co_partitioning_satisfied(co_partitioned, children) {
+ return Ok(vec![]);
+ }
+
+ Ok(co_partitioned.clone())
+ }
+
+ /// TODO: remove this temporary bridge once [`Partitioning::Range`]
+ /// generally satisfies [`Distribution::KeyPartitioned`] through
+ /// [`Partitioning::satisfaction`].
+ /// <https://github.com/apache/datafusion/issues/23266>.
+ ///
+ /// Also allow compatible [`Partitioning::Range`] to satisfy
+ /// [`Distribution::KeyPartitioned`].
+ #[expect(
+ deprecated,
+ reason = "HashPartitioned is accepted during the KeyPartitioned
migration"
+ )]
+ pub(crate) fn allow_range_satisfaction_for_key_partitioning(mut self) ->
Self {
+ for child in &mut self.children {
+ if matches!(
+ child.distribution,
+ Distribution::HashPartitioned(_) |
Distribution::KeyPartitioned(_)
+ ) {
+ child.satisfaction =
+ InputDistributionSatisfaction::AllowRangeKeyPartitioning;
+ }
+ }
+ self
+ }
+
+ /// Validate the requirements against a plan's children.
+ pub(crate) fn check_invariants<P: ExecutionPlan + ?Sized>(
+ &self,
+ plan: &P,
+ check: InvariantLevel,
+ ) -> Result<()> {
+ let children = plan.children();
+ self.validate_shape(plan.name(), children.len())?;
+
+ let children = children
+ .into_iter()
+ .map(|child| child.as_ref())
+ .collect::<Vec<_>>();
+ if matches!(check, InvariantLevel::Executable)
+ && let Some(co_partitioned) = &self.co_partitioned
+ && !self.co_partitioning_satisfied(co_partitioned, &children)
+ {
+ return internal_err!(
+ "{} requires children {:?} to be co-partitioned",
+ plan.name(),
+ co_partitioned
+ );
+ }
+
+ Ok(())
+ }
+
+ fn validate_shape(&self, plan_name: &str, children_len: usize) ->
Result<()> {
+ if self.children.len() != children_len {
+ return internal_err!(
+ "{plan_name}::input_distribution_requirements returned
incorrect child count: {} != {}",
+ self.children.len(),
+ children_len
+ );
+ }
+
+ if let Some(co_partitioned) = &self.co_partitioned {
+ if co_partitioned.len() < 2 {
+ return internal_err!(
+ "{plan_name} has invalid co-partitioning requirement: at
least two children are required"
+ );
+ }
+ let mut seen = vec![false; self.children.len()];
+ for &child in co_partitioned {
+ validate_child_index(plan_name, child, self.children.len(),
&mut seen)?;
+ if matches!(
+ self.children[child].distribution,
+ Distribution::UnspecifiedDistribution
+ ) {
+ return internal_err!(
+ "{plan_name} has invalid co-partitioning requirement:
child {child} has unspecified distribution"
+ );
+ }
+ }
+ }
+
+ Ok(())
+ }
+
+ fn co_partitioning_satisfied(
+ &self,
+ co_partitioned: &[usize],
+ children: &[&dyn ExecutionPlan],
+ ) -> bool {
+ let first_idx = co_partitioned[0];
+ let first_requirement = &self.children[first_idx];
+ let first = children[first_idx];
+ let first_partitioning = first.output_partitioning();
+
+ if !first_requirement
+ .satisfaction
+ .satisfaction(
+ first_partitioning,
+ &first_requirement.distribution,
+ first.equivalence_properties(),
+ false,
+ )
+ .is_satisfied()
+ {
+ return false;
+ }
+
+ for &child_idx in co_partitioned.iter().skip(1) {
+ let requirement = &self.children[child_idx];
+ let child = children[child_idx];
+ if !requirement
+ .satisfaction
+ .satisfaction(
+ child.output_partitioning(),
+ &requirement.distribution,
+ child.equivalence_properties(),
+ false,
+ )
+ .is_satisfied()
+ || !compatible_co_partitioning_layout(
+ first_requirement,
+ first_partitioning,
+ requirement,
+ child.output_partitioning(),
+ )
+ {
+ return false;
+ }
+ }
+
+ true
+ }
+}
+
+/// A distribution requirement for a single child.
+#[derive(Debug, Clone)]
+struct ChildDistributionRequirement {
+ distribution: Distribution,
+ satisfaction: InputDistributionSatisfaction,
+}
+
+/// TODO: remove this temporary bridge once [`Partitioning::Range`]
+/// generally satisfies [`Distribution::KeyPartitioned`] through
+/// [`Partitioning::satisfaction`].
+/// <https://github.com/apache/datafusion/issues/23266>.
+#[non_exhaustive]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+enum InputDistributionSatisfaction {
+ /// Use [`Partitioning::satisfaction`] as-is.
+ #[default]
+ Default,
+ /// Also allow [`Partitioning::Range`] to satisfy
+ /// [`Distribution::KeyPartitioned`].
+ AllowRangeKeyPartitioning,
+}
+
Review Comment:
:+1: really nice how you've managed to keep these intermediate temporary
structs private so that we can remove them in the future with no breaking api
change.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1148,35 +1238,44 @@ pub fn ensure_distribution(
.is_some_and(|join| join.mode == PartitionMode::Partitioned)
|| plan.is::<SortMergeJoinExec>();
+ let input_distributions = plan.input_distribution_requirements();
let repartition_status_flags =
get_repartition_requirement_status(&plan, batch_size,
should_use_estimates)?;
// This loop iterates over all the children to:
// - Increase parallelism for every child if it is beneficial.
// - Satisfy the distribution requirements of every child, if it is not
// already satisfied.
// We store the updated children in `new_children`.
- let children = izip!(
+ let mut children = izip!(
children.into_iter(),
plan.required_input_ordering(),
plan.maintains_input_order(),
repartition_status_flags.into_iter()
)
+ .enumerate()
.map(
|(
- mut child,
- required_input_ordering,
- maintains,
- RepartitionRequirementStatus {
- requirement,
- roundrobin_beneficial,
- roundrobin_beneficial_stats,
- hash_necessary,
- },
+ child_idx,
+ (
+ mut child,
+ required_input_ordering,
+ maintains,
+ RepartitionRequirementStatus {
+ requirement,
+ roundrobin_beneficial,
+ roundrobin_beneficial_stats,
+ hash_necessary,
+ },
+ ),
)| {
// Allow subset satisfaction when:
Review Comment:
A bit nicer to the eye:
```rust
let mut children = izip!(
0..children.len(),
children.into_iter(),
plan.required_input_ordering(),
plan.maintains_input_order(),
repartition_status_flags.into_iter()
)
.map(
|(
child_idx,
mut child,
required_input_ordering,
maintains,
RepartitionRequirementStatus {
requirement,
roundrobin_beneficial,
roundrobin_beneficial_stats,
hash_necessary,
},
)| {
```
##########
datafusion/physical-plan/src/distribution_requirements.rs:
##########
@@ -0,0 +1,484 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Input distribution requirements for physical execution plans.
+
+use std::sync::Arc;
+
+use datafusion_common::{Result, internal_err};
+use datafusion_physical_expr::{
+ Distribution, EquivalenceProperties, Partitioning,
PartitioningSatisfaction,
+ PhysicalExpr, physical_exprs_equal,
+};
+
+use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties,
InvariantLevel};
+
+/// Distribution requirements for an [`ExecutionPlan`]'s inputs.
+///
+/// [`RequiredInputDistributions`] describes what distribution an operator
+/// requires from each child.
+///
+/// - [`Self::new`] describes independent per-child requirements.
+/// - [`Self::co_partitioned`] additionally requires child partitions with the
+/// same index to cover compatible key ranges.
+///
+/// For a single-input aggregate:
+///
+/// ```text
+/// AggregateExec
+/// child 0 requirement: KeyPartitioned(group_exprs)
+/// ```
+///
+/// each input partition can aggregate its own key domain independently.
+///
+/// For a partitioned join:
+///
+/// ```text
+/// HashJoinExec
+/// child 0 requirement: KeyPartitioned(left_keys)
+/// child 1 requirement: KeyPartitioned(right_keys)
+///
+/// partition 0: join(left partition 0, right partition 0)
+/// partition 1: join(left partition 1, right partition 1)
+/// partition 2: join(left partition 2, right partition 2)
+/// ```
+///
+/// each child must satisfy its own key requirement. In addition, matching
+/// partition indexes must be safe to process together.
+#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub struct RequiredInputDistributions {
+ /// Per-child distribution requirements, indexed by child position.
+ children: Vec<ChildDistributionRequirement>,
+ /// Child indexes that must also have compatible partition layouts.
+ co_partitioned: Option<Vec<usize>>,
+}
+
+impl RequiredInputDistributions {
+ /// Create independent per-child requirements.
+ pub fn new(per_child: Vec<Distribution>) -> Self {
+ let children = per_child
+ .into_iter()
+ .map(|distribution| ChildDistributionRequirement {
+ distribution,
+ satisfaction: InputDistributionSatisfaction::Default,
+ })
+ .collect();
+
+ Self {
+ children,
+ co_partitioned: None,
+ }
+ }
+
+ /// Create a requirement that all children are co-partitioned.
+ ///
+ /// Each child must satisfy its own [`Distribution`]. Matching partition
+ /// indexes are processed together:
+ ///
+ /// ```text
+ /// left: Range(left.a ASC, split_points=[10, 20])
+ /// right: Range(right.x ASC, split_points=[10, 20])
+ ///
+ /// partition 0 from both sides contains keys before 10
+ /// partition 1 from both sides contains keys in [10, 20)
+ /// partition 2 from both sides contains keys at/after 20
+ /// ```
+ ///
+ /// If the split points differ, partition `i` from one side no longer
covers
+ /// the same key range as partition `i` from the other side.
+ pub fn co_partitioned(per_child: Vec<Distribution>) -> Self {
+ debug_assert!(
+ per_child.len() >= 2,
+ "co-partitioned distribution requirements need at least two
children"
+ );
+ let co_partitioned = (0..per_child.len()).collect();
+ let mut result = Self::new(per_child);
+ result.co_partitioned = Some(co_partitioned);
+ result
+ }
+
+ /// Return the per-child distribution requirements.
+ pub fn per_child_distributions(
+ &self,
+ ) -> impl ExactSizeIterator<Item = &Distribution> + '_ {
+ self.children.iter().map(|child| &child.distribution)
+ }
+
+ /// Return the distribution requirement for a child.
+ pub fn child_distribution(&self, child_idx: usize) ->
Option<&Distribution> {
+ self.children
+ .get(child_idx)
+ .map(|child| &child.distribution)
+ }
+
+ /// Return the per-child distribution requirements.
+ ///
+ /// WARNING: This intentionally drops any grouped relationship.
+ pub fn into_per_child(self) -> Vec<Distribution> {
+ self.children
+ .into_iter()
+ .map(|child| child.distribution)
+ .collect()
+ }
+
+ /// Returns how a child satisfies its distribution requirement using this
+ /// requirement set's satisfaction policy.
+ pub fn child_satisfaction(
+ &self,
+ child_idx: usize,
+ child: &dyn ExecutionPlan,
+ allow_subset: bool,
Review Comment:
If this is going to be part of the public API, it might be worth
encapsulating it into a `ChildSatisfactionOptions` object so that:
1. Adding new parameters does not imply a breaking change in the API
2. The meaning of the third param is known from just reading its usage:
`child_satisfaction(idx, child.as_ref(), ChildSatisfactionOptions {
allow_subset: true })?` vs `child_satisfaction(idx, child.as_ref(), true)?`
##########
datafusion/physical-plan/src/distribution_requirements.rs:
##########
@@ -0,0 +1,484 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Input distribution requirements for physical execution plans.
+
+use std::sync::Arc;
+
+use datafusion_common::{Result, internal_err};
+use datafusion_physical_expr::{
+ Distribution, EquivalenceProperties, Partitioning,
PartitioningSatisfaction,
+ PhysicalExpr, physical_exprs_equal,
+};
+
+use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties,
InvariantLevel};
+
+/// Distribution requirements for an [`ExecutionPlan`]'s inputs.
+///
+/// [`RequiredInputDistributions`] describes what distribution an operator
+/// requires from each child.
+///
+/// - [`Self::new`] describes independent per-child requirements.
+/// - [`Self::co_partitioned`] additionally requires child partitions with the
+/// same index to cover compatible key ranges.
+///
+/// For a single-input aggregate:
+///
+/// ```text
+/// AggregateExec
+/// child 0 requirement: KeyPartitioned(group_exprs)
+/// ```
+///
+/// each input partition can aggregate its own key domain independently.
+///
+/// For a partitioned join:
+///
+/// ```text
+/// HashJoinExec
+/// child 0 requirement: KeyPartitioned(left_keys)
+/// child 1 requirement: KeyPartitioned(right_keys)
+///
+/// partition 0: join(left partition 0, right partition 0)
+/// partition 1: join(left partition 1, right partition 1)
+/// partition 2: join(left partition 2, right partition 2)
+/// ```
+///
+/// each child must satisfy its own key requirement. In addition, matching
+/// partition indexes must be safe to process together.
+#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub struct RequiredInputDistributions {
Review Comment:
Given that this struct is returned by the `input_distribution_requirements`
method, maybe calling this `InputDistributionRequirements` is actually not a
bad call.
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -260,24 +260,263 @@ set datafusion.execution.target_partitions = 4;
statement ok
reset datafusion.optimizer.preserve_file_partitions;
+statement ok
+set datafusion.optimizer.prefer_hash_join = true;
+
+statement ok
+set datafusion.optimizer.repartition_joins = true;
+
+statement ok
+set datafusion.optimizer.preserve_file_partitions = 0;
+
##########
# TEST 9: Join on Range Partition Column
-# Both inputs expose Range partitioning on range_key. Join planning currently
-# reaches the unsupported Range output-partitioning path; later optimizer PRs
-# can replace this baseline with a successful plan and result test.
+# A partitioned inner hash join requires co-partitioned KeyPartitioned inputs.
+# Compatible Range layouts satisfy both the per-child key requirements and the
+# cross-child layout requirement, so no Hash repartitioning is inserted.
+##########
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+03)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+##########
+# TEST 10: Incompatible Range Join Repartitions
+# Both inputs are independently range partitioned on range_key, but their split
+# points differ. The per-child key requirements can be satisfied by Range, but
+# the co-partitioned layout requirement cannot, so Hash repartitioning repairs
+# both sides.
+##########
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned_shifted r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+05)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(15), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned_shifted r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+##########
+# TEST 11: Non-Range Join Repartitions
+# Range([range_key]) does not satisfy KeyPartitioned([non_range_key]), so
+# planning inserts Hash repartitioning on the actual join key.
+##########
+
+query TT
+EXPLAIN SELECT l.non_range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.non_range_key = r.non_range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(non_range_key@0,
non_range_key@0)], projection=[non_range_key@0, value@1, value@3]
+02)--RepartitionExec: partitioning=Hash([non_range_key@0], 4),
input_partitions=4
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4),
file_type=csv, has_header=false
+04)--RepartitionExec: partitioning=Hash([non_range_key@0], 4),
input_partitions=4
+05)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4),
file_type=csv, has_header=false
+
+query III
+SELECT l.non_range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.non_range_key = r.non_range_key
+ORDER BY l.non_range_key, l.value, r.value;
+----
+1 10 10
+1 10 100
+1 10 200
+1 10 300
+1 100 10
+1 100 100
+1 100 200
+1 100 300
+1 200 10
+1 200 100
+1 200 200
+1 200 300
+1 300 10
+1 300 100
+1 300 200
+1 300 300
+2 50 50
+2 50 150
+2 50 250
+2 50 350
+2 150 50
+2 150 150
+2 150 250
+2 150 350
+2 250 50
+2 250 150
+2 250 250
+2 250 350
+2 350 50
+2 350 150
+2 350 250
+2 350 350
+
+##########
+# TEST 12: Non-Inner Range Join Repartitions
+# Only inner partitioned hash joins opt in to Range satisfying KeyPartitioned
+# requirements in this PR. Non-inner joins keep using Hash repartitioning.
+##########
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+LEFT JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+05)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+LEFT JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+##########
+# TEST 13: Compatible Range Join Repartitions to Increase Parallelism
+# Co-partitioning satisfaction does not prevent a repartition that increases
+# parallelism. With target_partitions larger than the Range partition count,
+# both sides are hash repartitioned.
##########
-query error This feature is not implemented: Join output partitioning with
range partitioning is not implemented
+statement ok
+set datafusion.execution.target_partitions = 5;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=4
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=4
+05)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
SELECT l.range_key, l.value, r.value
FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+##########
+# TEST 14: Preserve File Partitions Preserves Range Join Inputs
+# preserve_file_partitions preserves compatible Range inputs for partitioned
+# joins even when target_partitions is higher than the input partition count.
+##########
+
+statement ok
+set datafusion.optimizer.preserve_file_partitions = 1;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+03)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.optimizer.preserve_file_partitions = 0;
+
+statement ok
+reset datafusion.optimizer.prefer_hash_join;
+
+statement ok
+reset datafusion.optimizer.repartition_joins;
+
+statement ok
+reset datafusion.optimizer.preserve_file_partitions;
##########
-# TEST 10: Union of Range Partitioned Inputs
-# Each input exposes Range partitioning on range_key. This records current
-# UNION ALL behavior before later PRs decide whether compatible range inputs
can
-# preserve Range partitioning across the union.
+# TEST 15: Union of Range Partitioned Inputs
+# Each input exposes Range partitioning on range_key. This PR does not add a
Review Comment:
I'd probably not say "This PR". This will lose meaning once this is merged
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -260,24 +260,263 @@ set datafusion.execution.target_partitions = 4;
statement ok
reset datafusion.optimizer.preserve_file_partitions;
+statement ok
+set datafusion.optimizer.prefer_hash_join = true;
+
+statement ok
+set datafusion.optimizer.repartition_joins = true;
+
+statement ok
+set datafusion.optimizer.preserve_file_partitions = 0;
+
Review Comment:
It would be cool to add some tests for two nested hash joins. One
immediately above the other one.
Probably it would also be cool to see how range partitioning can be
maintained across an aggregation -> join, and join -> aggregation.
##########
datafusion/physical-plan/src/distribution_requirements.rs:
##########
@@ -0,0 +1,480 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Input distribution requirements for physical execution plans.
+
+use std::sync::Arc;
+
+use datafusion_common::{Result, internal_err};
+use datafusion_physical_expr::{
+ Distribution, EquivalenceProperties, Partitioning,
PartitioningSatisfaction,
+ PhysicalExpr, physical_exprs_equal,
+};
+
+use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties,
InvariantLevel};
+
+/// Distribution requirements for an [`ExecutionPlan`]'s inputs.
+///
+/// [`RequiredInputDistributions`] describes what distribution an operator
+/// requires from each child.
+///
+/// - [`Self::new`] describes independent per-child requirements.
+/// - [`Self::co_partitioned`] additionally requires child partitions with the
+/// same index to cover compatible key ranges.
+///
+/// For a single-input aggregate:
+///
+/// ```text
+/// AggregateExec
+/// child 0 requirement: KeyPartitioned(group_exprs)
+/// ```
+///
+/// each input partition can aggregate its own key domain independently.
+///
+/// For a partitioned join:
+///
+/// ```text
+/// HashJoinExec
+/// child 0 requirement: KeyPartitioned(left_keys)
+/// child 1 requirement: KeyPartitioned(right_keys)
+///
+/// partition 0: join(left partition 0, right partition 0)
+/// partition 1: join(left partition 1, right partition 1)
+/// partition 2: join(left partition 2, right partition 2)
+/// ```
+///
+/// each child must satisfy its own key requirement. In addition, matching
+/// partition indexes must be safe to process together.
+#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub struct RequiredInputDistributions {
+ /// Per-child distribution requirements, indexed by child position.
+ children: Vec<ChildDistributionRequirement>,
+ /// Child indexes that must also have compatible partition layouts.
+ co_partitioned: Option<Vec<usize>>,
+}
+
+impl RequiredInputDistributions {
+ /// Create independent per-child requirements.
+ pub fn new(per_child: Vec<Distribution>) -> Self {
+ let children = per_child
+ .into_iter()
+ .map(|distribution| ChildDistributionRequirement {
+ distribution,
+ satisfaction: InputDistributionSatisfaction::Default,
+ })
+ .collect();
+
+ Self {
+ children,
+ co_partitioned: None,
+ }
+ }
+
+ /// Create a requirement that all children are co-partitioned.
+ ///
+ /// Each child must satisfy its own [`Distribution`]. Matching partition
+ /// indexes are processed together:
+ ///
+ /// ```text
+ /// left: Range(left.a ASC, split_points=[10, 20])
+ /// right: Range(right.x ASC, split_points=[10, 20])
+ ///
+ /// partition 0 from both sides contains keys before 10
+ /// partition 1 from both sides contains keys in [10, 20)
+ /// partition 2 from both sides contains keys at/after 20
+ /// ```
+ ///
+ /// If the split points differ, partition `i` from one side no longer
covers
+ /// the same key range as partition `i` from the other side.
+ pub fn co_partitioned(per_child: Vec<Distribution>) -> Self {
+ debug_assert!(
+ per_child.len() >= 2,
+ "co-partitioned distribution requirements need at least two
children"
+ );
+ let co_partitioned = (0..per_child.len()).collect();
+ let mut result = Self::new(per_child);
+ result.co_partitioned = Some(co_partitioned);
+ result
+ }
+
+ /// Return the per-child distribution requirements.
+ pub fn per_child_distributions(
+ &self,
+ ) -> impl ExactSizeIterator<Item = &Distribution> + '_ {
+ self.children.iter().map(|child| &child.distribution)
+ }
+
+ /// Return the distribution requirement for a child.
+ pub fn child_distribution(&self, child_idx: usize) ->
Option<&Distribution> {
+ self.children
+ .get(child_idx)
+ .map(|child| &child.distribution)
+ }
+
+ /// Return the per-child distribution requirements.
+ ///
+ /// WARNING: This intentionally drops any grouped relationship.
+ pub fn into_per_child(self) -> Vec<Distribution> {
+ self.children
+ .into_iter()
+ .map(|child| child.distribution)
+ .collect()
+ }
+
+ /// Returns how a child satisfies its distribution requirement using this
+ /// requirement set's satisfaction policy.
+ pub fn child_satisfaction(
+ &self,
+ child_idx: usize,
+ child: &dyn ExecutionPlan,
+ allow_subset: bool,
+ ) -> Result<PartitioningSatisfaction> {
+ let Some(requirement) = self.children.get(child_idx) else {
+ return internal_err!(
+ "missing distribution requirement for child {child_idx}"
+ );
+ };
+
+ Ok(requirement.satisfaction.satisfaction(
+ child.output_partitioning(),
+ &requirement.distribution,
+ child.equivalence_properties(),
+ allow_subset,
+ ))
+ }
+
+ /// Return child indexes whose co-partitioning requirements are
+ /// unsatisfied by the provided candidate children.
+ ///
+ /// Independent per-child requirements are intentionally ignored here, use
+ /// [`Self::child_satisfaction`] for those checks. An empty result means
all
+ /// co-partitioning requirements are satisfied.
+ #[doc(hidden)]
Review Comment:
Yeah, I think this is not too bad though. You can always extend the doc
comment saying that this is not expected to be used by people.
--
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]
