Splitting File Groups by Statistics: Giving Each Partition a Contiguous Slice of the Sort Key

TL;DR: When a table’s files are sorted by a key, you can cut the file list into N partitions by the key ranges rather than arbitrarily. Each partition then owns a contiguous slice of the key space, the partition boundaries are meaningful, and the scan can declare an output ordering — which lets the optimizer delete a downstream sort. DataFusion does this with a greedy best-fit assignment on the files’ min/max statistics.

Consider a table stored as 1,000 Parquet files, each holding a contiguous range of event_date. A query does SELECT ... WHERE event_date BETWEEN '2024-01-01' AND '2024-06-30'.

The obvious plan: list the files, assign them to 16 partitions somehow, scan them, and sort the result. But the files are already sorted by event_date, and if the partition assignment is done by key range rather than arbitrarily, the scan’s output is already sorted — and the sort disappears.

That is the motivation for split_file_groups_by_statistics, a config option that defaults to false because it only applies when the table declares an output ordering.

The entry point

In datafusion/catalog-listing/src/table.rs:

let split_file_groups_by_statistics = declared_output_partitioning.is_none()
    && state
        .config_options()
        .execution
        .split_file_groups_by_statistics;
match split_file_groups_by_statistics
    .then(|| {
        output_ordering.first().map(|output_ordering| {
            FileScanConfig::split_groups_by_statistics_with_target_partitions(
                &self.table_schema,
                &partitioned_file_lists,
                output_ordering,
                file_group_count,
            )
        })
    })
    .flatten()
{
    Some(Err(e)) => log::debug!("failed to split file groups by statistics: {e}"),
    Some(Ok(new_groups)) => {
        if new_groups.len() <= file_group_count {
            partitioned_file_lists = new_groups;
        } else {
            log::debug!(
                "attempted to split file groups by statistics, but there were more file groups than target_partitions; falling back to unordered"
            )
        }
    }
    None => {} // no ordering required
}

Note the conditions:

• declared_output_partitioning.is_none() — if the table already declares a partitioning (say, hash-partitioned on a join key), the file assignment is not free to choose.
• output_ordering.first() — there must be an ordering to split by. The first one is used, because the primary sort key is the one the ranges are cut along.
• The result must not exceed the target partition count, or the whole thing is abandoned and the table falls back to unordered groups. The failure is a log::debug!, not an error — the scan is still correct, it just does not get the ordering benefit.

The algorithm

FileScanConfig::split_groups_by_statistics_with_target_partitions (datafusion/datasource/src/file_scan_config/mod.rs):

let statistics = MinMaxStatistics::new_from_files(
    sort_order,
    table_schema,
    None,
    files.iter().copied(),
)?;

let indices_sorted_by_min = statistics.min_values_sorted();

// Initialize with target_partitions empty groups
let mut file_groups_indices: Vec<Vec<usize>> = vec![vec![]; target_partitions];

for (idx, min) in indices_sorted_by_min {
    if let Some((_, group)) = file_groups_indices
        .iter_mut()
        .enumerate()
        .filter(|(_, group)| {
            group.is_empty()
                || min
                    > statistics
                        .max(*group.last().expect("groups should not be empty"))
        })
        .min_by_key(|(_, group)| group.len())
    {
        group.push(idx);
    } else {
        // Create a new group if no existing group fits
        file_groups_indices.push(vec![idx]);
    }
}

Three steps:

1. Sort the files by their minimum value. min_values_sorted() returns the files ordered by the min of the sort key. This is what makes the groups contiguous: a group is built from files that are adjacent in this order.

2. For each file, find a group it can join. The filter is:

group.is_empty()
    || min > statistics.max(*group.last().expect("groups should not be empty"))

The group’s last element is the file with the largest min, because the files were processed in increasing min order. And min > max_of_group means the file’s entire range lies strictly above the group’s range. So the group stays a contiguous, non-overlapping range of the key space.

3. Among the groups that fit, pick the smallest.

.min_by_key(|(_, group)| group.len())

That is best-fit: put the file in the group that currently has the fewest files. It is the classic bin-packing heuristic, and it keeps the file counts balanced.

4. If no group fits, create a new one. The else branch. The file’s range overlaps every existing group, so it cannot be added to any of them without breaking the contiguity. A new group is created — and this is where the group count can exceed target_partitions, which is exactly why the caller checks the count and bails out.

Why contiguity matters

The whole point is that the output of the scan is ordered. If partition 0 owns the files for 2024-01 and partition 1 owns 2024-02, and the partitions are merged with a SortPreservingMergeExec, the merged output is sorted by event_date — and the merge does not need to sort, it just needs to compare the partition heads.

That is the EquivalenceProperties machinery from the ordering article: the scan can declare an output ordering, and the optimizer can use that declaration to remove a sort, to skip a repartition, or to satisfy a join’s ordering requirement.

And the min/max ranges of the groups are meaningful: partition 0 covers [2024-01-01, 2024-01-31]. That is a fact about the data that the planner can use.

Empty files

One preliminary step:

let (files, empty_files) = partition_known_empty_files(flattened_files);
if files.is_empty() {
    return Ok(file_groups.to_vec());
}

Files that are known to be empty — their statistics say zero rows — are pulled out and handled separately. They would otherwise waste a slot in the best-fit assignment, and a group containing only empty files is a partition that does no work.

The catch

• It only applies to an ordered table. output_ordering.first() must exist. An unordered table — the common case — gets nothing.
• Overlapping file ranges break it. If two files cover overlapping key ranges (which happens when a table is not strictly sorted, or when files were written out of order), no group can accept the second file and the group count grows. Past target_partitions, the whole thing is abandoned.
• It is off by default. The config option split_file_groups_by_statistics defaults to false, which suggests the DataFusion developers found the benefit situational.
• The partition count can shrink. The retain(|group| !group.is_empty()) at the end drops empty groups, so you can end up with fewer partitions than target_partitions — and hence less parallelism than you asked for. That is a deliberate trade: correct ordering over maximum parallelism.
• The sort order must be compatible with the file layout. The algorithm assumes the files’ min values are meaningfully ordered with respect to the key — if the files’ contents are not actually sorted by the key, the resulting groups are not sorted ranges and the declared output ordering is a lie.

Why it matters

This is a small feature with a disproportionate effect when it applies. The difference between “the scan’s output is sorted” and “the scan’s output is not” is a SortExec — which for a large table is the single most expensive operator in the plan.

And the implementation is a nice piece of engineering: a greedy best-fit over min/max statistics that respects contiguity, handles the empty-file case, and bails out cleanly when the data does not fit the model. The bail-out is the important part — an optimisation that cannot apply must degrade to the old behaviour, not produce a wrong plan.