You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "gene-bordegaray (via GitHub)" <gi...@apache.org> on 2026/09/18 18:46:19 UTC
[I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
gene-bordegaray opened a new issue, #25483:
URL: https://github.com/apache/datafusion/issues/25483
Epic: #25421
`RangePartitioning` provides a separated, ordered key space using an ordering and split points.
Currently, DF can determine that two range layouts are co-partitioned when they have the same partition count, split points, and sort options ( see `compatible_co_partitioning_layout` in `datafusion/physical-plan/src/distribution_requirements.rs`).
Similarly, `InterleaveExec` requires exact partitioning equality. DF does not describe
how two different but compatible layouts relate.
Lets look at a source and target layout to understand the relationships between the two:
```text
Source layout Target layout
S0: (-inf, 10) ───┐
├─────────────────▶ T0: (-inf, 20)
S1: [10, 20) ─────┘
S2: [20, 30) ─────┐
├─────────────────▶ T1: [20, +inf)
S3: [30, +inf) ───┘
```
The two layouts define the space at different granularities: source partitions define the space with 4 while the target defines it with 2. What is important to not is that in order to transform the souce layout to the target, it is not necessary for every source partition to have a relationship with every target partition. Rather, every source partition has exactly one target partition. Thus, rows do not need to be evaluated or redistributed individually and adjacent source partitions can be grouped.
Nevertheless, the layouts are not equal, so the optimizer today cannot use this relationship.
This missing information affects range repartitioning. `RepartitionExec` constructs channels from every input partition to every output partition (an all-to-all relationship). Given six inputs and three outputs this looks like:
```text
┌──────────────> C0
P0 ───────────────┼──────────────▶ C1
└──────────────> C2
┌──────────────> C0
P1 ───────────────┼──────────────▶ C1
└──────────────> C2
...
┌──────────────> C0
P5 ───────────────┼──────────────▶ C1
└──────────────> C2
6 inputs * 3 outputs = 18 relationships
```
When input range metadata can that only a subset of these edges can carry rows the remaining channels still exist. Furthermore, an output will keep channels open from inputs that can never produce rows for it and can't complete until evey unrelated producer finishes.
## Desired behavior
DF should be able to identify the relationships between two range layouts and return the corresponding partition mapping. This can be used to reduce repartition overhead and make cleaner, more comprehensible optimizer decisions.
I propose the initial relationships should be:
1. **Equal**: Every source partition matches one target partition.
2. **SourceFiner**: The source is more granular of the same space. Adjacent source partitions can be grouped to form target partitions.
3. **TargetFiner**: The target is more granular than the source. Only source partitions crossed by a target range boundary require row splitting.
4. **Unproven**: Cannot prove a useful relationship.
I would prefer to start by being very strict here. Starting with this focused set, proving and testing them thoroughly and if we cannot prove a relationship fallback.
## Why this is good
This will help single node DF by eliminating lots of overhead when dealing with range layouts by skipping repartitions and row level comparisions in many scenarios.
This is only amplified in a distributed context for projects like distributed datafusion and Ballista where minimixaing these relationships turns into eliminating lots of expensive network coordination and transfers.
## Proposed solution
### 1. Add a range-layout relationship API
The API should return both classification and a mapping:
```rust
enum RangeLayoutRelationshipKind {
Equal,
SourceFiner,
TargetFiner,
Unproven,
}
struct RangeLayoutRelationship {
kind: RangeLayoutRelationshipKind,
source_to_targets: Vec<Range<>>,
target_to_sources: Vec<Range<>>,
}
**NOTE: range here is the range of partitions blogining to mapping at index i not range partitioning**
```
The mapping can be computed via the two split-point lists.
### 2. Use the relationship during distribution enforcement
Distribution enforcement should use the relationship like this:
1. **Equal**: Already co-partitioned, reuse existing layout.
2. **SourceFiner**: Group adjacent source partitions (or for first PR could just still insert repartitions on to limit scope).
3. **Target Finer**: Split across crossed partitions
4. **Unproven**: Not co-partitioned, fallback to repartition.
### 3. Make `RepartitionExec` use channel mapping
When a proven input-to-output mapping is available:
1. Create channels and spill state only for mapped `(input, output)` pairs
2. If an input only has one destination just send whole batches without computing per-row partition indices.
3. If an input has multiple destinations use the range router.
--
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5735805577
Just adding a couple examples where I think this change will help.
Aggregations
```sql
SELECT day, count(log)
FROM logs.range_partitioned_data_on_day_hour
GROUP by day
```
Map N partitions on `day, hour` to M partitions partitioned by day where M < N, requiring fewer than N * M relationships.
Partitioned Joins
```sql
SELECT *
FROM logs.range_partitioned_data_on_day_hour AS logs
JOIN logs.range_partitioned_data_on_day_only AS daily_logs
ON daily_logs.day = logs.day
```
We would be able to repartition the `day, hour` partition data to be partitioned on `day` only and perform a partition join with data that is partitioned by `day`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5734656891
@stuhood @gabotechs @jayshrivastava
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "stuhood (via GitHub)" <gi...@apache.org>.
stuhood commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5736071613
@jayshrivastava : I believe that these mappings can also be used to re-partition dynamic filters?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5734657492
take
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "stuhood (via GitHub)" <gi...@apache.org>.
stuhood commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5735243909
@gene-bordegaray: Nice! It seems like this potentially unifies `CoalescePartitionsExec` and `RepartitionExec`, which is awesome?:
1. `CoalescePartitionsExec` is just an `N -> 1` mapping with whole-batch forwarding.
2. It gives us `N -> M` coalescing (`1 < M < N`), which we don't have now (today, reducing partition count forces an all-to-all hash or round-robin repartition).
For `SourceFiner`, do you envision executing this via `RepartitionExec` (spawning background tasks with channels), or via direct in-thread stream concatenation (like a generalized `CoalescePartitionsExec`)?
If adjacent partitions `S0: (-inf, 10)` and `S1: [10, 20)` are internally sorted, streaming `S0` then `S1` sequentially in-thread preserves total sort order for free without channels, background tasks, or heaps. Routing through `RepartitionExec` channels would interleave them across threads and lose ordering.
----
We ran into this exact distinction between shuffle (`NetworkShuffleExec`) and disjoint task coalescing (`NetworkCoalesceExec`) in DFD: see https://github.com/paradedb/datafusion-distributed/issues/68.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "stuhood (via GitHub)" <gi...@apache.org>.
stuhood commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5766489785
> [@stuhood](https://github.com/stuhood) Makes sense. I think the only situation that happens is in joins, which I tried to cover in [this comment](https://github.com/apache/datafusion/issues/25483#issuecomment-5735805577). Ie. 2 differently range partitioned inputs or 1 range partitioned input + 1 unpartitioned input.
Yea. I guess that I just meant that rather than needing to degrade to a `case` statement when being pushed down through a `RepartionExec` (which would still be the case in #23817 I think), there are a few cases where dynamic filters would be able to instead re-map themselves using these mappings, and to remain partitioned.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "stuhood (via GitHub)" <gi...@apache.org>.
stuhood commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5735747860
Another thought here: can Hash repartitions also take advantage of these mapping types in any cases? If not, can the proposed mapping types be tweaked to allow that?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add Range Layout Replationships to Avoid Repartitions [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #25483:
URL: https://github.com/apache/datafusion/issues/25483#issuecomment-5764519515
@stuhood Makes sense. I think the only situation that happens is in joins, which I tried to cover in [this comment](https://github.com/apache/datafusion/issues/25483#issuecomment-5735805577). Ie. 2 differently range partitioned inputs or 1 range partitioned input + 1 unpartitioned input.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org