Repository navigation
API: Partitioning::DynamicRange for runtime-discovered split points #23093
Description
Status: Implementation PR open at #23094. This issue stays open as the design home — discuss the API here, code review there.
What is the problem the feature is trying to solve?
Partitioning::Range(RangePartitioning { ordering, split_points }) landed in #22207 (@gene-bordegaray, Datadog) as the declarative form of range partitioning — split points are known at plan time, either declared by a TableProvider or computed by a planner from statistics.
What is missing is the symmetric runtime-discovered form: a sibling variant where the boundary set is only known once an upstream operator has observed its actual data range. Two concrete needs:
- Parallelizing single-partition window functions (RANGE frames, no PARTITION BY). A planner can declare "I want N output partitions of this sorted stream, split by sort-key range" without knowing the actual range yet. The implementing operator discovers the range at execute time (typically from its input's runtime extrema, see API: runtime partition extrema for range-aware operators #23089) and computes interior split points before routing rows. The motivating use case from the spike in Parallel bounded RANGE-frame window functions without PARTITION BY (draft) #23026.
- Future range-aware operators where boundaries are data-dependent. Anything that wants to bucket a single stream into N value-ranges without forcing statistics-based estimates to be precise. The existing declarative form is the right answer when split points are known a priori; the dynamic form is the right answer when they're not.
These two cases share the partitioning semantics (lexicographic ordering, half-open intervals) but differ on when the boundaries are known. Keeping them as two variants of Partitioning lets downstream distribution requirements have a stable answer at plan time (partition count is fixed) while leaving the boundary discovery to the implementing operator.
Describe the solution you'd like
A small addition to Partitioning:
pub enum Partitioning {
RoundRobinBatch(usize),
Hash(Vec<Arc<dyn PhysicalExpr>>, usize),
Range(RangePartitioning),
DynamicRange(DynamicRangePartitioning), // <- new
UnknownPartitioning(usize),
}
pub struct HaloSpec {
preceding: ScalarValue,
following: ScalarValue,
}
pub struct DynamicRangePartitioning {
ordering: LexOrdering,
partition_count: usize,
halo: Option<HaloSpec>,
}partition_count is fixed at plan time so downstream distribution requirements have a stable answer. Only the split point values are runtime-discovered.
halo is the plan-time contract for "this bucket carries extra rows beyond its primary [min, max] range by distance D on each side". Distances are in the leading sort key's domain — e.g. for RANGE BETWEEN 5 PRECEDING AND 3 FOLLOWING over an Int64 key, the halo is preceding=5, following=3. With halo unset, the partitioning produces disjoint buckets and a downstream consumer sees [ExtremaKind::Observed] extrema (per #23089). With halo set, the routing operator publishes [ExtremaKind::Expanded] extrema, and a downstream halo-strip filter is contracted to read the primary range and discard the extra rows.
ROWS-frame halo (a count of neighbor rows rather than a domain distance) is intentionally not represented; a separate variant can be added later if motivated.
Variant introduction only — no execution slot in this PR. RepartitionExec returns not_impl_err! for DynamicRange at every site it does for Range. Proto serialization returns not_impl_err! (proto plumbing follows the same incremental cadence as #22207 → #22787). FFI bridges to UnknownPartitioning(n) (same path Range takes per #22394).
Implementation in PR #23094 (+470 / -14 LoC, 6 unit tests covering construction, compatible_with, project, halo metadata + display, halo affecting compatibility, and halo preserved through projection).
Design points worth debating
- Is
Partitioning::DynamicRangethe right shape, or should this be a flag on the existingRangePartitioning? AOption<Vec<SplitPoint>>field onRangePartitioning(None = "discover at runtime") is the alternative. Two variants is preferred here because the operator contract is materially different — declared partitioning is a static property the planner can reason about; dynamic partitioning is a runtime contract whose split points only become defined once an upstream operator publishes them. Keeping them separate lets the type system enforce the distinction at every match site. Open to be talked out of this. partition_countat plan time vs. dynamic. Pinning the partition count at plan time keeps the rest of the optimizer/planner simple — every distribution decision still has a stable answer. The alternative (dynamic partition count too) would require deeper planner changes and is not motivated by any current use case.- How does the execution slot discover boundaries? That's the follow-up: the implementing operator reads its input's runtime extrema via the API proposed in API: runtime partition extrema for range-aware operators #23089 / implemented in feat: PartitionExtrema + SortExec observer + BoundedWindowAggExec passthrough #23090. This issue scopes only the partitioning variant itself; the execution slot will be filed separately once both this and feat: PartitionExtrema + SortExec observer + BoundedWindowAggExec passthrough #23090 settle.
HaloSpecon the partitioning vs. on the operator. Halo could alternatively live as a config field on the routing operator (RepartitionExec's new dynamic-range mode) rather than on the partitioning type. Putting it on the partitioning means a downstream halo-strip filter can read the halo distance directly from the upstream's declared output partitioning — no operator-internal handshake needed. Open to be pushed back on.- Naming.
DynamicRangeis straightforward but other options exist (DiscoveredRange,RuntimeRange). Open to the reviewer's preference.
Describe alternatives you've considered
- A flag on
RangePartitioning. Discussed above (point 1). - A new dedicated operator (not a
Partitioningvariant) that routes by runtime-discovered boundaries. Rejected because that would duplicateRepartitionExec's mechanics and bypass the unifiedEnforceDistributionstory the existingPartitioningenum gives. - Inferring "dynamic" from absence of split points in
RangePartitioning. Same problem as MINOR: Set GitHub description and labels #1 — collapses two contracts into one type at the cost of every consumer having to guard against the absent case.
Coexistence with existing work
- Add Physical
Partitioning::Rangeenum variant #22207, Add range partitioning sqllogictest fixture #22607, Add logical range partitioning representation #22777 — Gene Bordegaray'sPartitioning::Rangework.DynamicRangeis the runtime-discovered sibling; the declarative path is unchanged. - API: runtime partition extrema for range-aware operators #23089 —
PartitionExtremaAPI proposal.DynamicRange's execution slot is the canonical consumer. - feat: PartitionExtrema + SortExec observer + BoundedWindowAggExec passthrough #23090 —
PartitionExtremaimplementation PR. Lands the primitive the dynamic execution slot will read. - Parallel bounded RANGE-frame window functions without PARTITION BY (draft) #23026 — original parallel-window draft PR, the spike that surfaced both this and
PartitionExtrema. Kept open for context; the rewrite plan splits it into small focused PRs of which this is one. - [EPIC] Implement Range Partitioning #22395 (epic), Implement Range Repartitioning #22397 (RepartitionExec execution slot for Range) — the umbrella issues for
Partitioning::Rangeend-to-end landing. ADynamicRangeexecution slot will share infrastructure with the eventualRangerow-routing implementation.
Additional context
This issue is the discussion home for the variant. The implementation is PR #23094 — pure addition, behavior mirrors Range at every match site (every not_impl_err! site for Range gets the same treatment for DynamicRange). Execution-slot, proto, logical representation, and substrait round-trip land as separate follow-ups on top, mirroring the cadence Gene used for Range.
Activity
alamb commentedon Jun 22, 2026
This is an interesting idea -- thank you for filing it @avantgardnerio
One question I have is how this will work with specific operators? For example, since DataFusion's
ExecutionPlanis streaming, most operators will not have seen all of their input when their parents start executing. When they start they won't know their actual output partitioningThus how would the parent operators know when they could rely on the partitioning declaration of their children?
I do see that
single-partition window functionsis an example of an operator that might buffer its entire input before producing output so in that case the parent could ask before the first row is produced.But if the point is that this mechanism only works for "pipeline breaking operators" maybe it is something we could handle at a higher level -- for example building some sort of more "adaptive query processing" into DataFusion (a way to re-optimize the plan after some blocking operator has run and we know more about the statistics
avantgardnerio commentedon Jun 22, 2026
Exactly. Pipeline breakers can populate, other operators either pass-through, or drop. If present we know they are accurate.
Admittedly, this is a way to get 90% of the benefit of AQE, with 10% of the effort.
alamb commentedon Aug 1, 2026
FWIW I heard a talk by Thomas Neumann recently on a proposal to parallelize window functions, and he basically proposed using this approach (pick ranges based on actual distribution knowledge after a pipeline blocker operation like sort)
avantgardnerio commentedon Aug 1, 2026
Good to know Neumann converged on the same shape.
FWIW, this is how it is incubating in Ballista today:
If we are open to dynamic things, I think
OrderedRangeRepartitionExecwould just collapse into the existingRepartitionExec::Rangewhich has nearly identical code.alamb commentedon Aug 3, 2026
I think figuring out how to pass more dynamic information between plan nodes is definitely something to explore. The key will be to invest some API that is general enough to be used for some mechanism other than dynamic range partitioning 🤔
I wonder if we could add something like
DynamicStatisticsor something to mirror the DynamicFilter and statistics APIs but be clear it woudl update during runtime 🤔