Range Partitioning in DataFusion: Replacing the Hash Shuffle with Boundaries Chosen from the Data

TL;DR: A hash shuffle sends every key to hash(key) % N. If the key is skewed — say, 90% of the rows have the same customer_id — then one partition gets 90% of the data and the other N-1 partitions idle. DataFusion now has a second kind of shuffle: range partitioning. Instead of hashing the key, you cut the key space into N contiguous ranges at sampled boundaries, so the rows are balanced by construction and each partition owns a contiguous, ordered slice of the key space. Routing a row is then a binary search instead of a hash.

Every distributed query engine has the same weak point: the shuffle.

The standard shuffle is a hash. Both sides of a join compute hash(join_key) % num_partitions and send rows to the matching partition. Both sides agree without communicating, the join becomes a per-partition local operation, and everything is parallel.

The weakness is that hash is blind to the distribution of the data. A perfectly uniform hash of a perfectly skewed key is still skewed. If 90% of the orders rows have customer_id = 42, then hash-partitioning on customer_id sends 90% of the rows to one partition, and that partition takes 90% of the time while the other N-1 partitions finish instantly and wait. The query takes as long as the slowest partition, so the query takes as long as the skewed partition.

DataFusion’s answer is a different decomposition of the same problem.

Two kinds of partitioning

The physical partitioning enum has three relevant cases:

// datafusion/physical-expr/src/partitioning.rs
pub enum Partitioning {
    /// Allocate batches using a round-robin algorithm and the specified number of partitions
    RoundRobinBatch(usize),
    /// Allocate rows based on a hash of one of more expressions and the specified number
    /// of partitions.
    Hash(Vec<Expr>, usize),
    /// Partition rows by ranges.
    Range(RangePartitioning),
    /// The DISTRIBUTE BY clause is used to repartition the data based on the input expressions
    DistributeBy(Vec<Expr>),
}

Hash is the workhorse. Range is the interesting one.

What a range partition looks like

RangePartitioning describes an ordered key space cut at specific points:

/// For a single range key:
///
/// ```text
/// ordering = [date ASC NULLS LAST]
/// split_points = [
///   (2022-01-01),
///   (2023-01-01),
/// ]
///
/// partition 0: date before 2022-01-01
/// partition 1: date between 2022-01-01 (inclusive) and 2023-01-01 (exclusive)
/// partition 2: date at/after 2023-01-01
/// ```
///
/// The same model extends to compound keys.
/// For `ordering = [time ASC, city ASC]`, split points are ordered
/// lexicographically by `(time, city)`.

The ordering is a lexicographic ordering — an ASC/DESC and a null-ordering per key — and the split_points are the boundaries. This is not a new idea; it is what every “sort-merge” and “range-partitioned” system does, and what Spark’s repartitionByRange does. What is new here is how the boundaries are chosen and how the metadata is carried around the system.

Sampling: the trick that makes it work

The critical question is: where do the split points come from? If you split a key space into N equal-width ranges, you have no better chance of balance than a hash — the data may not be uniform in the key. If you split at arbitrary points, you may be lucky or not.

The answer is to sample the data and pick the boundaries from the sample’s quantiles. If the sample is representative, the boundaries are approximately the true quantiles, and each partition gets approximately 1/N of the rows.

DataFusion keeps the sample and the derived boundaries separately:

pub struct RangePartitioning {
    /// Ordered partitioning key.
    ordering: LexOrdering,
    /// Caller-supplied maximum-resolution split points used to derive the
    /// effective boundaries.
    samples: Arc<[SplitPoint]>,
    /// Effective boundaries for the current partition count.
    split_points: Arc<[SplitPoint]>,
}

Two levels of resolution. samples is the finest set of boundaries known. split_points is the set actually in use for the current partition count. The point of keeping both is rescaling:

/// Returns this range partitioning scaled to `target_partitions`.
///
/// Scaling retains the original samples, so a range partitioning that was
/// scaled down can later be scaled back up to `max_partition_count`.
/// Returns `None` when `target_partitions` is zero or the retained samples do
/// not support that many partitions.
pub fn scale(&self, target_partitions: usize) -> Option<Self> {
    if target_partitions == 0 || target_partitions > self.max_partition_count() {
        return None;
    }
    ...
    Some(Self {
        ordering: self.ordering.clone(),
        samples: Arc::clone(&self.samples),
        split_points: downsample_split_points(&self.samples, target_partitions),
    })
}

The down-sampling is an even stride over the sample array:

fn downsample_split_points(
    samples: &Arc<[SplitPoint]>,
    partition_count: usize,
) -> Arc<[SplitPoint]> {
    if partition_count == samples.len() + 1 {
        return Arc::clone(samples);
    }

    let sample_count = samples.len();
    (1..partition_count)
        .map(|partition| {
            // Use a wider intermediate so valid slice lengths cannot overflow
            // when calculating the evenly spaced sample index.
            let sample_index = ((partition as u128 * sample_count as u128)
                / partition_count as u128) as usize;
            samples[sample_index].clone()
        })
        .collect::<Vec<_>>()
        .into()
}

So if you sampled 1000 quantiles and want 8 partitions, you take every 125th sample. If later you want 16 partitions, you can go back up, because the 1000 samples are still there. The u128 intermediate is a small correctness detail — partition * sample_count can overflow a usize on 32-bit, and an overflow here would silently produce a wrong split point.

The max_partition_count accessor and the “insufficient samples are an expected condition” comment tell you the design intent: the sample resolution is a budget, and the planner is allowed to spend it.

Routing: binary search instead of hashing

Here is where the performance difference shows up. A hash routing computes hash(key) and a modulo. A range routing computes a binary search over the split points.

For the common case — a single primitive column with no nulls — DataFusion has a specialised router:

fn route_with<A: ArrowPrimitiveType<Native = T>, E: FnMut(usize, usize)>(
    &self,
    array: &PrimitiveArray<A>,
    mut emit: E,
) {
    let split_points = &self.split_points;
    let descending = self.sort_options.descending;
    let nulls_first = self.sort_options.nulls_first;

    if array.null_count() == 0 {
        let values = array.values().as_ref();
        if !descending {
            for (idx, &val) in values.iter().enumerate() {
                let p = split_points.partition_point(|&sp| sp <= val);
                emit(idx, p);
            }
        } else {
            ...

slice::partition_point is the standard library’s binary search: it returns the number of elements for which the predicate is true, i.e. the index of the first split point greater than the value. That is exactly the partition index. O(log S) per row, with S the number of split points.

The fast path is guarded twice: the array must have no nulls, and the values must be a contiguous slice with no offset. array.values().as_ref() gives the contiguous &[T] — and iterating it directly, rather than calling array.value(i), avoids a per-element null-check and bounds check. That is a real difference: the null-containing path below uses array.value(idx) and array.is_null(idx), which is two separate operations per row.

Nulls get their own partition:

} else {
    let null_partition = if nulls_first { 0 } else { split_points.len() };
    for idx in 0..array.len() {
        if array.is_null(idx) {
            emit(idx, null_partition);
        } else {
            let val = array.value(idx);
            let p = if !descending {
                split_points.partition_point(|&sp| sp <= val)
            } else {
                split_points.partition_point(|&sp| sp >= val)
            };
            emit(idx, p);
        }
    }
}

All the NULLs go to one partition — the first or the last, depending on the null ordering. That is a deliberate choice: it concentrates the nulls rather than spreading them, which is fine because nulls are usually a small fraction, and it keeps the partition boundaries a clean total order.

Floats get their own router, and it is a nice detail:

let p = split_points.partition_point(|&sp| {
    sp.total_cmp(&val) != Ordering::Greater
});

total_cmp is Rust’s total order for floats, where NaN sorts below everything and -0.0 < 0.0. Using partial_cmp would be a bug: NaN compares to nothing, so the binary search’s monotonicity assumption breaks and it can return an arbitrary index.

The general case: Arrow’s RowConverter

For composite keys, or strings, or any type Arrow cannot compare natively, DataFusion falls back to Arrow’s RowConverter — the same machinery used for row-format sorting:

enum RangeRouterInner {
    /// Specialized fast path for a single primitive column with non-null split points.
    Primitive(PrimitiveRangeRouter),
    /// Universal fast path using Arrow's RowConverter for arbitrary types and composite keys.
    Row(RowConverterRangeRouter),
}

The idea: encode the split points into Arrow’s row format — the same byte-oriented, order-preserving encoding used by the sort kernels — and then binary-search the encoded rows. Because the row encoding is order-preserving for the given sort options, a lexicographic byte comparison of two encoded rows gives the same answer as comparing the original values.

let split_point_rows = if split_points.is_empty() {
    None
} else {
    let split_point_arrays = (0..num_cols)
        .map(|col_idx| {
            let col_scalars =
                split_points.iter().map(|sp| sp.values()[col_idx].clone());
            ScalarValue::iter_to_array(col_scalars)
        })
        .collect::<Result<Vec<_>>>()?;

    Some(row_converter.convert_columns(&split_point_arrays)?)
};

This is a genuinely clever reuse: the row format is already an order-preserving serialisation, so it doubles as a search index over the key space. And the split points are encoded once, at router construction time, not per row.

The RowConverter::supports_fields check is the guard:

if !RowConverter::supports_fields(&sort_fields) {
    return not_impl_err!(
        "Range partitioning is not supported for data types: {:?}",
        data_types
    );
}

Metadata that travels with the plan

The other interesting part is that range partitioning is not just a runtime mechanism — it is a plan property, carried in the logical plan and propagated through the physical plan.

The logical plan has its own RangePartitioning:

/// Logical range partitioning.
///
/// - `ordering` defines the partitioning key and ordering using logical
///   [`SortExpr`]s.
/// - `split_points` define the boundaries between adjacent partitions.
///
/// The expressions are resolved against the declaring plan's schema. This
/// constructor does not validate split point value types against the resolved
/// expression types. Like other user-specified data properties such as
/// sortedness, if a source declares range partitioning, it is responsible for
/// placing each row in the partition described by the split points. DataFusion
/// will not validate this is upheld.
///
/// NOTE: Range-aware optimizer and execution behavior will be introduced
/// incrementally. See
/// <https://github.com/apache/datafusion/issues/22395>.

Note the honesty of the last paragraph. Range partitioning is new and still being rolled out, and the doc says so rather than pretending the feature is complete.

And the physical RangePartitioning has to survive being projected — a projection pushdown renames and drops columns, and the range metadata references columns by expression. Hence the project and adapt methods:

/// Calculates the range partitioning after applying the given projection.
///
/// Returns `None` if any range key cannot be projected or if projection
/// collapses distinct range keys into duplicate output expressions.
fn project(
    &self,
    mapping: &ProjectionMapping,
    input_eq_properties: &EquivalenceProperties,
) -> Option<Self> {

and the has_same_layout method with a very careful doc:

/// Whether two range partitionings have the same current key ordering and
/// effective boundaries, irrespective of their retained samples.
///
/// This does not imply equal scaling capacity. In particular, a plan that
/// combines inputs must not use this comparison to inherit one input's
/// samples for all inputs. Structural equality is required for that use case.
pub fn has_same_layout(&self, other: &Self) -> bool {
    self.ordering == other.ordering && self.split_points == other.split_points
}

That is a warning about a real bug: two partitions with the same current boundaries may have been built from different samples, and therefore have different scaling capacity. If you union them, you cannot keep both sets of samples.

The catch

• The boundaries have to be right, and nobody checks. The doc says it plainly: “if a source declares range partitioning, it is responsible for placing each row in the partition described by the split points. DataFusion will not validate this is upheld.” This is the same contract as “sortedness” — a declared property, not a verified one. A wrong declaration silently produces wrong results.
• Range partitioning needs a sample, and sampling costs a pass. The whole balance property depends on the boundaries being drawn from the data’s actual quantiles.
• Range partitioning is only useful for ordered joins. For a non-equi join, hash partitioning is still the right answer. The value of range partitioning is that it lets both sides of a join cut the key space at the same boundaries, so the ranges are co-partitioned and a sort-merge-style join becomes possible. That is what the co_partitioned distribution requirement (datafusion/physical-plan/src/distribution_requirements.rs) is for: “matching partition indexes must be safe to process together”.
• The implementation is in flux. The tracking issue (#22395) and the “introduced incrementally” note mean the feature’s surface area is still growing. Articles like this one age fast.
• Binary search is not free. A hash is O(1); a binary search is O(log S). For a well-distributed key, hashing wins. Range partitioning wins when the key is skewed and the boundaries are correct — and, in the co-partitioned case, when the ordering itself enables a cheaper join algorithm.

The bigger idea

The interesting thing about range partitioning is not the binary search. It is the decomposition: the shuffle is no longer “agree on a hash function” but “agree on a partition of the key space”. That is a strictly richer notion, and it is what makes co-partitioned range joins, skew-aware scheduling, and order-preserving merges possible.

Hash partitioning throws away the order of the key. Range partitioning keeps it. A hash shuffle is a lossy compression of the key space into N buckets; a range shuffle is a piecewise-linear map that preserves the order of the keys within each bucket.