Support co-partitioned range inner equi joins by gene-bordegaray · Pull Request #23184 · apache/datafusion

gene-bordegaray · 2026-06-25T12:16:08Z

Which issue does this PR close?

Rationale for this change

DataFusion can represent source-declared range partitioning, but partitioned hash joins still required hash partitioned inputs. So an inner join on compatible range-partitioned keys would insert unnecessary hash repartitions, even when each left/right partition already covered the same key domain.

This PR adds a partitioning requirement that means "equal key values are co-located" . I was calling this "compatibility" but found we can satisfy the requirement with looser conditions. Other systems call this "co-location" or "co-partitioning" (trino, spark). Which they (and now I am proposing) define as when both sides of a join are already partitioned so matching key values appear in corresponding partitions, so we can join partition pairs directly without repartitioning the sides.

This lets "co-partitioned" range inputs satisfy inner partitioned hash joins. This will also be applicable to other join types and operators but kept the first PR thin to keep scope more reviewable.

What changes are included in this PR?

Adds Distribution::KeyPartitioned(Vec<Arc<dyn PhysicalExpr>>) as a public distribution requirement.
- HashPartitioned([a]) means rows must be partitioned by hash on a.
- KeyPartitioned([a]) means rows with equal a values must be co-located, but the partitioning algorithm may be hash, range, or another compatible scheme.
- Example:
```
Hash([left.a], 3) satisfies KeyPartitioned([left.a])
Range([right.b ASC], [(10), (20)], 3) satisfies KeyPartitioned([right.b])
```
Adds Partitioning::co_partitioned_with(...) to validate that two independently satisfying partitionings also can be paired by partition index.
- Examples:
  - Accepted: both sides satisfy their own key requirement and have matching range boundaries.
```
left:  Range([a ASC], [(10), (20)], 3), required KeyPartitioned([a])
right: Range([b ASC], [(10), (20)], 3), required KeyPartitioned([b])
```
  - Accepted: both sides satisfy their own key requirement and have matching hash partition counts.
```
left:  Hash([a], 3), required KeyPartitioned([a])
right: Hash([b], 3), required KeyPartitioned([b])
```
  - Rejected: both sides satisfy their own key requirement, but range boundaries differ.
```
left:  Range([a ASC], [(10), (20)], 3), required KeyPartitioned([a])
right: Range([b ASC], [(15), (20)], 3), required KeyPartitioned([b])
```
  - Rejected: both sides satisfy their own key requirement, but partition counts differ.
```
left:  Hash([a], 3), required KeyPartitioned([a])
right: Hash([b], 4), required KeyPartitioned([b])
```
Changes inner partitioned HashJoinExec requirements from HashPartitioned to KeyPartitioned.
- All other hash joins still require HashPartitioned for now.

Updates EnforceDistribution so co-partitioned range inner joins avoid repartitioning.

Examples:

Compatible range partitioning: no repartition is inserted because partitions can be joined by index.

HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a, b)]
  DataSourceExec: output_partitioning=Range([a ASC], [(10), (20)], 3)
  DataSourceExec: output_partitioning=Range([b ASC], [(10), (20)], 3)

Incompatible range boundaries: both sides are repartitioned by hash because partition i does not represent the same key domain on both sides.

HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a, b)]
  RepartitionExec: partitioning=Hash([a], target_partitions)
    DataSourceExec: output_partitioning=Range([a ASC], [(10), (20)], 3)
  RepartitionExec: partitioning=Hash([b], target_partitions)
    DataSourceExec: output_partitioning=Range([b ASC], [(15), (20)], 3)

Mismatched hash partition counts: both sides are forced to the target hash partition count so partition indexes line up.

HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a, b)]
  RepartitionExec: partitioning=Hash([a], target_partitions)
    DataSourceExec: output_partitioning=Hash([a], 11)
  RepartitionExec: partitioning=Hash([b], target_partitions)
    DataSourceExec: output_partitioning=Hash([b], 12)

Non-inner joins: range inputs still get hash repartitioning because only inner partitioned hash joins use KeyPartitioned in this PR.

HashJoinExec: mode=Partitioned, join_type=Left, on=[(a, b)]
  RepartitionExec: partitioning=Hash([a], target_partitions)
    DataSourceExec: output_partitioning=Range([a ASC], [(10), (20)], 3)
  RepartitionExec: partitioning=Hash([b], target_partitions)
    DataSourceExec: output_partitioning=Range([b ASC], [(10), (20)], 3)

Keeps partitioned dynamic filter pushdown restricted to hash-compatible routing.
- Compatible range partitioning can satisfy the join, but dynamic filters still route by hash, so range/range partitioned joins disable dynamic filters.
Degrades range join output partitioning to UnknownPartitioning(n) rather than erroring. Adding this behavior would need more tests and careful thought about, I think its safert o just degrade for first PR.

Are these changes tested?

Yes.

KeyPartitioned satisfaction for hash and range partitioning.
co_partitioned_with for compatible and incompatible range/hash partitioning.
EnforceDistribution behavior for:
- compatible range joins avoiding hash repartitioning
- incompatible range bounds rehashing
- mismatched hash partition counts rehashing
- non-inner range joins rehashing
sanity checking for invalid partitioned hash joins.
dynamic filter rejection for range partitioning, preserved file partitions, and mismatched hash counts.
sqllogictest coverage for range-partitioned joins avoiding hash repartitioning and non-range joins still repartitioning.

Are there any user-facing changes?

Yes.

This PR changes public physical planning APIs:

Adds Distribution::KeyPartitioned.
Adds Partitioning::co_partitioned_with.
- NOTE: This replaces the previous partition compatibility API with the new co-partitioning API. Since the compatibility API was never in a release I believe this is ok to do (lesson learned to not make API change until ew have definitive consumer).
Affects users matching exhaustively on Distribution.

github-actions · 2026-06-25T12:30:29Z

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion v54.0.0 (current)
       Built [ 111.451s] (current)
     Parsing datafusion v54.0.0 (current)
      Parsed [   0.037s] (current)
    Building datafusion v54.0.0 (baseline)
       Built [ 103.109s] (baseline)
     Parsing datafusion v54.0.0 (baseline)
      Parsed [   0.037s] (baseline)
    Checking datafusion v54.0.0 -> v54.0.0 (no change; assume patch)
     Checked [   0.936s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [ 217.097s] datafusion
    Building datafusion-physical-expr v54.0.0 (current)
       Built [  27.126s] (current)
     Parsing datafusion-physical-expr v54.0.0 (current)
      Parsed [   0.048s] (current)
    Building datafusion-physical-expr v54.0.0 (baseline)
       Built [  27.406s] (baseline)
     Parsing datafusion-physical-expr v54.0.0 (baseline)
      Parsed [   0.049s] (baseline)
    Checking datafusion-physical-expr v54.0.0 -> v54.0.0 (no change; assume patch)
     Checked [   0.502s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip

--- failure inherent_method_missing: pub method removed or renamed ---

Description:
A publicly-visible method or associated fn is no longer available under its prior name. It may have been renamed or removed entirely.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#item-remove
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.48.0/src/lints/inherent_method_missing.ron

Failed in:
  RangePartitioning::compatible_with, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/712c515214a083057f77287848d3c22e8903c1dd/datafusion/physical-expr/src/partitioning.rs:253
  Partitioning::compatible_with, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/712c515214a083057f77287848d3c22e8903c1dd/datafusion/physical-expr/src/partitioning.rs:416

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  56.011s] datafusion-physical-expr
    Building datafusion-physical-optimizer v54.0.0 (current)
       Built [  36.958s] (current)
     Parsing datafusion-physical-optimizer v54.0.0 (current)
      Parsed [   0.023s] (current)
    Building datafusion-physical-optimizer v54.0.0 (baseline)
       Built [  36.865s] (baseline)
     Parsing datafusion-physical-optimizer v54.0.0 (baseline)
      Parsed [   0.022s] (baseline)
    Checking datafusion-physical-optimizer v54.0.0 -> v54.0.0 (no change; assume patch)
     Checked [   0.164s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  75.178s] datafusion-physical-optimizer
    Building datafusion-physical-plan v54.0.0 (current)
       Built [  35.047s] (current)
     Parsing datafusion-physical-plan v54.0.0 (current)
      Parsed [   0.143s] (current)
    Building datafusion-physical-plan v54.0.0 (baseline)
       Built [  34.608s] (baseline)
     Parsing datafusion-physical-plan v54.0.0 (baseline)
      Parsed [   0.144s] (baseline)
    Checking datafusion-physical-plan v54.0.0 -> v54.0.0 (no change; assume patch)
     Checked [   0.904s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  72.067s] datafusion-physical-plan
    Building datafusion-sqllogictest v54.0.0 (current)
       Built [ 179.911s] (current)
     Parsing datafusion-sqllogictest v54.0.0 (current)
      Parsed [   0.022s] (current)
    Building datafusion-sqllogictest v54.0.0 (baseline)
       Built [ 180.024s] (baseline)
     Parsing datafusion-sqllogictest v54.0.0 (baseline)
      Parsed [   0.024s] (baseline)
    Checking datafusion-sqllogictest v54.0.0 -> v54.0.0 (no change; assume patch)
     Checked [   0.115s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [ 363.142s] datafusion-sqllogictest

gene-bordegaray · 2026-06-25T12:47:51Z

+/// This is optimizer policy: partitioned joins require children that can be
+/// paired by partition index. Inner hash joins can reuse compatible range
+/// partitioning; otherwise the existing hash repartitioning policy applies.
+fn partitioned_join_distribution(


I noticed that simple checking logic is slightly duplicated here, here, and here

There may be a good way to extract this out but didnt want to premptively do a public change on speculation but something I am noting. Let me know if any one has suggestsion

gene-bordegaray · 2026-06-25T12:48:00Z

+    /// left:  Range(left.a ASC,  [10, 20]), required KeyPartitioned(left.a)
+    /// right: Range(right.x ASC, [15, 20]), required KeyPartitioned(right.x)
+    /// ```
+    pub fn co_partitioned_with(


I realized the concept was a bit off. I think this is ok if we haven't had a release... 😅

I linked the mirroring concepts that are used in trino in spark in the PR description

Lesson learned to have consumer of the public API first

👍 I don't think it's a big deal, this is anyways not the typical method external consumers rely one.

Without taking a look at the usages of this method, I do notice that the signature seems a bit off: why would it need the required Distribution for checking if two Partitionings are co-partitioned? it seems unrelated.

I'll keep reading though

Ya I thought similarly but this matters for multi-children operators. For joins, each side has a different may have a diff schema:

left: Partitioning::Range(left.a) right: Partitioning::Range(right.x) join: left.a = right.x

So the expressions left.a and right.x are not equal since they are from different children. Because of this we need to check each sides req Distribution to know what each partitioning has to satisfy:

left requirement: key partitioned on left.a right requirement: key partitioned on right.x

So we need to check:

left partitioning satisfies left requirement

right partitioning satisfies right requirement

the two partition mappings can be paired

I could do something like this:

struct RequiredPartitioning<'a> { partitioning: &'a Partitioning, requirement: &'a Distribution, eq_properties: &'a EquivalenceProperties, } impl RequiredPartitioning<'_> { fn co_partitioned_with(&self, other: &RequiredPartitioning<'_>) -> bool }

or derive the EquivalenceProperties in the method if we pass the children directly

gene-bordegaray · 2026-06-25T12:51:04Z

cc: @gabotechs @stuhood

stuhood · 2026-06-25T17:01:24Z

Amazing timing. Will get you some feedback on this by early next week! Thank you!

gene-bordegaray · 2026-06-25T17:02:24Z

Amazing timing. Will get you some feedback on this by early next week!

@stuhood great, thank you! I also have a huge line of follow up issues to support more join types that we can split up 👍

I am trying to make smaller tickets as more of the plumbing gets in to get more people involved

gene-bordegaray · 2026-06-26T19:45:18Z

I have added a stacked PR that shows how I imagine Aggregations will consume the KeyPartitioned API: gene-bordegaray#6 (not done yet 😄)

gabotechs

This is looking good! left a first round of comments.

The main one is about trying to collapse Distribution::KeyPartitioned and Distribution::HashPartitioned into just a single Distribution variant, I do see that in pretty much all places the code that handles both is the same, and I get the feeling that Distribution::HashPartitioned is just an unfortunate name that was choosen a while back, but it's not really coupled to hashes.

gabotechs · 2026-06-27T12:56:38Z

+    /// left:  Range(left.a ASC,  [10, 20]), required KeyPartitioned(left.a)
+    /// right: Range(right.x ASC, [15, 20]), required KeyPartitioned(right.x)
+    /// ```
+    pub fn co_partitioned_with(


👍 I don't think it's a big deal, this is anyways not the typical method external consumers rely one.

gabotechs · 2026-06-27T12:57:52Z

+    /// left:  Range(left.a ASC,  [10, 20]), required KeyPartitioned(left.a)
+    /// right: Range(right.x ASC, [15, 20]), required KeyPartitioned(right.x)
+    /// ```
+    pub fn co_partitioned_with(


Without taking a look at the usages of this method, I do notice that the signature seems a bit off: why would it need the required Distribution for checking if two Partitionings are co-partitioned? it seems unrelated.

I'll keep reading though

2010YOUY01 · 2026-06-28T04:20:47Z

I think KeyPartitioned is a really clever idea. I have some thoughts on simplifying the implementation. This is basically the same point as @gabotechs, but expressed slightly differently.

Key idea

KeyPartitioned(expr) is the logical generalization of HashPartitioned(expr): it means that rows with equal values for expr are guaranteed to be in the same partition.

Hash partitioning and range partitioning are two concrete ways to satisfy this requirement.

More precisely:

A plan is KeyPartitioned(k) iff:

for any two rows r1 and r2 in the same input:
    if r1.k = r2.k
    then partition(r1) = partition(r2)

Equivalently, different output partitions contain disjoint sets of key values.

This seems to cover the logical requirement that HashPartitioned(expr) is trying to express today, and it also covers the non-overlapping range partitioning cases. Unifying them should make the implementation simpler.

Example 1: aggregation has one input

select k, avg(v)
from t1
group by k

AggregateExec requires its input to be partitioned by the group keys. Conceptually, that requirement is KeyPartitioned(k).

That means all rows with the same k value must be in the same input partition, so each partition can compute its local groups independently.

For example, both of the following satisfy this requirement:

hash partitioned by k
range partitioned by k, assuming the ranges are non-overlapping

In either case, we do not need to insert an extra hash repartition before the aggregation.

Example 2: hash join has two inputs

select *
from t1
join t2
on t1.v1 = t2.v1

For a partitioned hash join, it is not enough for each side to be independently KeyPartitioned.

The two sides must also be co-partitioned with respect to the join keys. That means equal join-key values from the left and right inputs must be assigned to the same partition id.

The physical execution pattern is:

Union of:

join(t1_partition_0, t2_partition_0)
join(t1_partition_1, t2_partition_1)
join(t1_partition_2, t2_partition_2)
...

So the required co-partitioning rule is:

left and right are co-partitioned on left.a = right.b iff:

for any left row l and right row r:
    if l.a = r.b
    then partition(l) = partition(r)

This implies some extra compatibility requirements across the two inputs:

If both sides are hash partitioned, they must use compatible hash semantics: the same hash function, seed/salt, null handling, and modulo / partition count.
If both sides are range partitioned, they must use compatible range boundaries. The simplest case is the same partition count and the same split points. In the future, we may be able to relax this to support more general compatible range layouts.

So I think there are two related concepts:

KeyPartitioned(k):
    a per-input property:
    equal keys within this input go to the same partition

CoPartitioned(left.k, right.k):
    a cross-input property:
    equal join keys across both inputs go to the same partition id

Implementation plan

Here is my initial thoughts on a implementation plan (note I haven't read the related code carefully, so it's just some rough ideas)

Collapse Distribution::HashPartitioned into Distribution::KeyPartitioned conceptually, or rename HashPartitioned to KeyPartitioned, and update the docs to describe the more general semantics.
Update and refine the existing test coverage for hash partitioning under the new semantics.
Support aggregation first. Since aggregation only has one input, it only needs the unary KeyPartitioned property. If the input is already range partitioned by the group keys, we should not insert an extra hash repartition.
Support hash join next. Since join has two inputs, we also need to validate the cross-input co-partitioning requirement described above.

gene-bordegaray · 2026-06-28T21:03:16Z

@2010YOUY01 thank you for the comments and suggestions. I do think that KeyPartitioned and HashPartitioned can probably collapse into a single distribution type (I prefer KeyPartitioned as hash is / has been histroically misleading).

I also believe my current implementation of co-partitioning is satisfying the same contract as described here. Since Distribution was not designed for multiple children I like the current co_partitioned_with() API as it can accept any Distributions and check them without a need to redesign the API.

I would be ok with supporting aggregations before multi-input operators. Something I would like to point out though, is if we are going to collapse KeyPartitioned and HashPartitioned, and have RangePartitioning satisfy the KeyPartitioned type in the public satisfaction() method, this will apply to the checks in EnforceDistribution which applies repartitioning across many operators.

If we take this approach we could add explicit checks of what operator we are checking distribuion for in EnforceDistribution and just start with aggregations or add private helpers for each before having RangePartitioning saisfy KeyPartitioned

After thinking about it for a bit I would prefer adding private helpers for each operator and then have general satisfaction for RangePartitioning once a good amount of operators are supported. Let me know what you think 😄

PS: also see my linked aggregation pR in my fork 👍 it will give a good idea of what this looks like using this brnach as a base: gene-bordegaray#6

cc: @gabotechs

2010YOUY01 · 2026-06-29T01:31:00Z

After thinking about it for a bit I would prefer adding private helpers for each operator and then have general satisfaction for RangePartitioning once a good amount of operators are supported. Let me know what you think 😄

Thank you for the extra context! This approach makes sense if it's easier to implement.

I think we should start with aggregation first, since single-input cases are simpler.

Two-input/co-partitioned operators are trickier. For example, when a hash join requires its inputs to be [HashPartitioned(left_key), HashPartitioned(right_key)], it's really enforcing the co-partitioning property we're discussing here. We'll need to figure out how to incrementally migrate that to the new API to avoid duplicated implementations in the long term.

2010YOUY01 · 2026-06-29T03:51:09Z

Since Distribution was not designed for multiple children I like the current co_partitioned_with() API as it can accept any Distributions and check them without a need to redesign the API.

Making this co-partition requirement part of the ExecutionPlan API seems like a better long-term solution. The existing co_partitioned_with adds a hidden assumption to ExecutionPlan::required_input_distribution: if both sides are KeyPartitioned, then left.co_partitioned_with(right) will be checked somewhere else. This creates a leaky abstraction somehow.

If we can instead make this an explicit requirement at the ExecutionPlan level, the implementation is likely to become simpler and easier to understand.

// Before
    fn required_input_distribution(&self) -> Vec<Distribution> {
        vec![Distribution::UnspecifiedDistribution; self.children().len()]
    }

// After
pub struct RequiredInputDistributions {
    pub per_child: Vec<Distribution>,
    pub cross_child: Vec<CrossChildDistribution>,
}

trait ExecutionPlan {
    fn required_input_distributions(&self) -> RequiredInputDistributions {
        ...
    }
}

The tradeoff is that this would require a large refactor upfront. The should not be a hard requirement for now, and I'm also unsure how to carry this out incrementally, but I would still love to see us move towards this direction sooner.

gene-bordegaray · 2026-06-29T11:12:31Z

If we can instead make this an explicit requirement at the ExecutionPlan level, the implementation is likely to become simpler and easier to understand.
// Before
    fn required_input_distribution(&self) -> Vec<Distribution> {
        vec![Distribution::UnspecifiedDistribution; self.children().len()]
    }
// After
pub struct RequiredInputDistributions {
    pub per_child: Vec<Distribution>,
    pub cross_child: Vec<CrossChildDistribution>,
}

trait ExecutionPlan {
    fn required_input_distributions(&self) -> RequiredInputDistributions {
        ...
    }
}
The tradeoff is that this would require a large refactor upfront. The should not be a hard requirement for now, and I'm also unsure how to carry this out incrementally, but I would still love to see us move towards this direction sooner.

@2010YOUY01

This is an intersting idea, thank you. Let me cherry pick my aggregation commit to be stacked on main and can open something up for a unary operator first. I think something like this would be worth exploring 👍

gene-bordegaray · 2026-06-29T11:22:42Z

Also, I would prefer to support aggregations and joins (and maybe a few more operators) before deciding to rename or replace HashPartitioned with KeyPartitioned. I do not want to over commit to a large public API change before working through kinks with a few differently shaped use cases.

For something like the aggregations, I can just introduce a private helper in EnforceDistribution to have RangePartitioning satisfy HashPartitioned.

stuhood

Thanks a lot!

I was able to incorporate this branch into datafusion-distributed and paradedb, and observe partitioned hash joins both with and without df-d. With df-d, stage boundaries were placed appropriately above the join.

I agree with collapsing Key and Hash in #23236.

stuhood · 2026-06-29T21:18:10Z

+                    vec![
+                        Distribution::HashPartitioned(left_expr),
+                        Distribution::HashPartitioned(right_expr),
+                    ]


Worth a TODO pointing at followup?

## Which issue does this PR close?  - Closes apache#23191. - Related discussion: apache#23184, apache#23236. ## Rationale for this change  Range partitioning can satisfy aggregate hash partitioning: equal group keys are already partitioned, even though the partitioning is not hash-based. This is the first unary-operator implementation from the range partitioning discussion before making broader public API changes around `HashPartitioned` / `KeyPartitioned`. ## What changes are included in this PR?  - Let compatible range partitioning satisfy aggregate hash distribution requirements in `EnforceDistribution` - Keep this private to aggregate planning for now to not make public API changes to `Distribution` enum variants yet until more operators are supported ## Are these changes tested?  Yes. ## Are there any user-facing changes?  Yes. Range-partitioned aggregate plans can now avoid hash repartitioning.

gene-bordegaray · 2026-07-06T20:36:10Z

@stuhood @2010YOUY01 @gabotechs

I have explored this idea:

// Before
    fn required_input_distribution(&self) -> Vec<Distribution> {
        vec![Distribution::UnspecifiedDistribution; self.children().len()]
    }

// After
pub struct RequiredInputDistributions {
    pub per_child: Vec<Distribution>,
    pub cross_child: Vec<CrossChildDistribution>,
}

trait ExecutionPlan {
    fn required_input_distributions(&self) -> RequiredInputDistributions {
        ...
    }
}

I implemented this idea through RequiredInputDistributions in distribution_requirements.rs I am not done but wanted to flesh out this rough draft just to get feedback at a high level if you guys think this is worth purusing in this PR 👍

stuhood

I implemented this idea through RequiredInputDistributions in distribution_requirements.rs I am not done but wanted to flesh out this rough draft just to get feedback at a high level if you guys think this is worth purusing in this PR

I like this! I don't have much to say about the exact shape of struct RequiredInputDistributions, but the ability to expose requirements across multiple children makes a lot of sense.

stuhood · 2026-07-06T21:06:33Z

+    /// Designates whether hash repartitioning is required.
+    hash_repartition_required: bool,


Since you're renaming this, why isn't this repartition_required or aligned_repartition_required? Or is the idea that this flag would be set in cases where Hash in particular would be beneficial (to balance out skew, for example), and then you'd eventually have another flag for Range?

the idea was that we are specifically having logic for hash repartitioning right now in the rule, then if / when we decide to add logic for range repartitioning there will be distinction

gabotechs · 2026-07-07T07:21:33Z

I have explored this idea:

It sounds a bit... complex. I wonder if there's a simpler way of modeling this. I'll try to make some suggestions there.

gabotechs

Flushing some [not super useful] comments, I'll need to invest some time in understanding what's happening in enforce_distribution before actually making some better suggestions.

gabotechs · 2026-07-07T07:43:12Z

+    children: &mut [(
+        DistributionContext,
+        Option<OrderingRequirements>,
+        bool,
+        Distribution,
+    )],
+    target_partitions: usize,


This type is pretty messed up, I'm surprised that clippy allows this. It would be nice to avoid these wide unnamed tuple arguments, as reading the signature is impossible to figure out what's any of that.

I can pull it into some type of struct, its really trying to model children's dtribution state

maybe a good chance for a refactor of this logic

gabotechs · 2026-07-07T07:48:35Z

+                // 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:


The amount of if statement branching happening here is most likely going to take a huge hit in maintainability.

This was already a difficult optimizer rule to understand, but I fear now it's worst.

I'll likely need to invest some time in understanding this before I can contribute any meaningful suggestion. In the mean time, if you see opportunities for simplyfing this code, that's more than welcome.

😂 damn, I'm realizing this is not even new code...

Hahaha yes I'm trying my hardest to make it more readable jnremently 😭. But please if it's harder now let me know

gabotechs · 2026-07-07T07:51:11Z

+    /// Specifies data distribution requirements for all children, including
+    /// non-default satisfaction policies or cross-child relationships.
+    fn required_input_distributions(&self) -> RequiredInputDistributions {
+        RequiredInputDistributions::new(self.required_input_distribution())
+    }
+
    /// Specifies the data distribution requirements for all the


Is there any chance we can not maintain these two methods?

required_input_distributions

required_input_distribution

Having the two at the same time does not seem like a nice public API design

We would either deprecate the old one or just flat out change the existing. Didn't know if we wanted the churn

gabotechs · 2026-07-07T07:53:06Z

+            AggregateMode::FinalPartitioned | AggregateMode::SinglePartitioned
+                if !self.group_by.has_grouping_set() =>
+            {
+                requirements.with_range_key_partitioning(0)


Without looking at the specific implementation, this reads:

"The required input distribution must be range key partitioned"

Which I assume is wrong right?

gabotechs · 2026-07-07T07:54:58Z

+        match (
+            self.left.output_partitioning(),
+            self.right.output_partitioning(),
+        ) {
+            (
+                Partitioning::Hash(_, left_partition_count),
+                Partitioning::Hash(_, right_partition_count),
+            ) => left_partition_count == right_partition_count,
+            (left_partitioning, right_partitioning) => {


I imagine that dynamic filters should work at some point with co-partitioned joins on a range key right? probably another issue worth adding to your batch?

gabotechs · 2026-07-07T09:06:40Z

+    /// TODO: remove this field once [`Partitioning::Range`] generally
+    /// satisfies [`Distribution::KeyPartitioned`] through
+    /// [`Partitioning::satisfaction`]. See
+    /// <https://github.com/apache/datafusion/issues/23266>.
+    per_child_satisfaction: Vec<InputDistributionSatisfaction>,
+    /// Cross-child distribution relationships required by the plan.


I see there's a big amount of code shipped in this PR just because of the fact that is happening before #23266.

How about shipping #23266 first? would that avoid introducing all this intermediate toil?

There is a good amount, its really just tradeoff between if we want to test lots of operators upfront in a single PR or have this bridge and do one by one

This is centralizing the opt in so from this point on we just need to add the allow_range_partitioning()

gabotechs · 2026-07-07T09:09:00Z

+    /// <https://github.com/apache/datafusion/issues/23266>.
+    per_child_satisfaction: Vec<InputDistributionSatisfaction>,
+    /// Cross-child distribution relationships required by the plan.
+    relationships: Vec<InputDistributionRelationship>,


Is there any scenario where this Vec would have a length greater than 1?

unions or intervleave execs

I don't fully understand how more than 1 would be necessary for unions or interleave execs.

InterleaveExec only works if all the inputs have the same hash partitioning:

https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/union.rs#L457-L457

So I don't think there's any scenario where there'll be more than 1 relationship between a subset of the InterleaveExec children

For UnionExec, how could this operator try to enforce a specific input distribution requirement to its children? unions don't really care what was the partitioning scheme of children, and I cannot think of a situation it'll want to enforce any specific distribution requirement

github-actions Bot added physical-expr Changes to the physical-expr crates optimizer Optimizer rules core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels Jun 25, 2026

gene-bordegaray changed the title ~~Support co-partitioned range hash joins~~ [WIP] Support co-partitioned range hash joins Jun 25, 2026

github-actions Bot added the auto detected api change Auto detected API change label Jun 25, 2026

gene-bordegaray commented Jun 25, 2026

View reviewed changes

gene-bordegaray changed the title ~~[WIP] Support co-partitioned range hash joins~~ Support co-partitioned range hash joins Jun 25, 2026

gene-bordegaray force-pushed the gene.bordegaray/2026/06/range-partitioned-joins branch from 8583303 to 45d598b Compare June 25, 2026 12:59

gene-bordegaray marked this pull request as ready for review June 25, 2026 13:04

gene-bordegaray mentioned this pull request Jun 25, 2026

Allow co-partitioned range inputs to satisfy inner partitioned hash joins #23183

Open

gene-bordegaray commented Jun 25, 2026

View reviewed changes

Comment thread datafusion/sqllogictest/test_files/range_partitioning.slt

gene-bordegaray mentioned this pull request Jun 25, 2026

Allow Partitioning::Range to satisfy aggregation Distribution::KeyPartitioned requirements #23191

Closed

gene-bordegaray changed the title ~~Support co-partitioned range hash joins~~ Support co-partitioned range inner equi joins Jun 25, 2026

gene-bordegaray force-pushed the gene.bordegaray/2026/06/range-partitioned-joins branch from 45d598b to 2fca2bb Compare June 26, 2026 15:21

gabotechs reviewed Jun 27, 2026

View reviewed changes

gene-bordegaray mentioned this pull request Jun 29, 2026

Replace / rename HashPartitioned distribution as KeyPartitioned #23236

Closed

gene-bordegaray mentioned this pull request Jun 29, 2026

Aggregations Support Partitioning::Range #23239

Merged

stuhood approved these changes Jun 29, 2026

View reviewed changes

gene-bordegaray mentioned this pull request Jul 2, 2026

Allow Partitioning::Range to satisfy Distribution::KeyPartitioned generally #23266

Open

gene-bordegaray force-pushed the gene.bordegaray/2026/06/range-partitioned-joins branch 3 times, most recently from 2642d4d to f20e0bf Compare July 6, 2026 20:32

gene-bordegaray force-pushed the gene.bordegaray/2026/06/range-partitioned-joins branch 2 times, most recently from c3f4319 to 2dbbdf7 Compare July 6, 2026 20:48

stuhood approved these changes Jul 6, 2026

View reviewed changes

stuhood mentioned this pull request Jul 6, 2026

Allow Partitioning::Range to satisfy PartitionedTopK Distribution::KeyPartitioned requirements #23290

Open

Support co-partitioned range inner equi joins

001b10c

gene-bordegaray force-pushed the gene.bordegaray/2026/06/range-partitioned-joins branch from 2dbbdf7 to 001b10c Compare July 7, 2026 01:37

gabotechs reviewed Jul 7, 2026

View reviewed changes

gene-bordegaray marked this pull request as draft July 7, 2026 10:24

		/// Designates whether hash repartitioning is required.
		hash_repartition_required: bool,

Uh oh!

Conversation

gene-bordegaray commented Jun 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

github-actions Bot commented Jun 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gene-bordegaray commented Jun 25, 2026

Uh oh!

Uh oh!

stuhood commented Jun 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gene-bordegaray commented Jun 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gene-bordegaray commented Jun 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

2010YOUY01 commented Jun 28, 2026

Key idea

Example 1: aggregation has one input

Example 2: hash join has two inputs

Implementation plan

Uh oh!

gene-bordegaray commented Jun 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

2010YOUY01 commented Jun 29, 2026

Uh oh!

2010YOUY01 commented Jun 29, 2026

Uh oh!

gene-bordegaray commented Jun 29, 2026

Uh oh!

gene-bordegaray commented Jun 29, 2026

Uh oh!

stuhood left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gene-bordegaray commented Jul 6, 2026

Uh oh!

stuhood left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gabotechs commented Jul 7, 2026

Uh oh!

gabotechs left a comment

gene-bordegaray commented Jun 25, 2026 •

edited

Loading

github-actions Bot commented Jun 25, 2026 •

edited

Loading

stuhood commented Jun 25, 2026 •

edited

Loading

gene-bordegaray commented Jun 25, 2026 •

edited

Loading

gene-bordegaray commented Jun 26, 2026 •

edited

Loading

gene-bordegaray commented Jun 28, 2026 •

edited

Loading

gene-bordegaray Jul 7, 2026 •

edited

Loading