Why Sorted Parquet Files Do Not Guarantee an Ordered Scan

Suppose we keep a purchases table in Parquet, one file per day, each file sorted by customer. Suppose also that a view over the data reports, at the end of the month, how many purchases each customer made: in SQL, a COUNT(*) grouped by customer.

Since every file is sorted by customer, we would expect the query engine to use that order to optimize the GROUP BY. Yet EXPLAIN, or EXPLAIN VERBOSE, sometimes shows the aggregation running without using the order that is already there. Why? Where along the way does the ordering of the rows in the files get lost?

To see how a query engine propagates ordering, we will follow it from the file scan to the aggregate. We focus on Parquet scans in Apache DataFusion, but the principles apply to other relational query engines as well. The distinction between the order of a file and the order of a partition is general; the steps and settings described are those of DataFusion 55.1.0 and its listing tables.

Two things to settle before we set off. Asking for the result in customer order, with ORDER BY, is a different matter. It concerns the order of the rows the query returns, not the order an operator receives. And “sorted” here means that the rows inside each file follow the sort key used by whoever wrote them. Directory partitioning and clustering schemes also shape the data layout, but on their own they do not guarantee that row order.

1. The example

The example is simple: three days of purchases, one Parquet file per day.

/data/purchases/monday.parquet
/data/purchases/tuesday.parquet
/data/purchases/wednesday.parquet

Every file is sorted by customer. These are the rows in each file, in the order in which they are stored:

/data/purchases/monday.parquet

/data/purchases/tuesday.parquet

/data/purchases/wednesday.parquet

The table built on these files declares the order in which the rows were written:

CREATE EXTERNAL TABLE purchases
STORED AS PARQUET
WITH ORDER (customer ASC)
LOCATION '/data/purchases/';

And the query counts purchases per customer (to keep the example small, we configure the engine to use two partitions; by default it uses as many as the machine’s cores):

SET datafusion.execution.target_partitions = 2;

SELECT customer, COUNT(*)
FROM purchases
GROUP BY customer;

Every file is sorted by customer, and every file contains Anna, Bruno and Zeno: their customer ranges overlap, so no file ends before another begins. This is the layout we are probing. The table declares the order by customer, and the query groups by the same column. Yet this is what EXPLAIN gives us:

EXPLAIN FORMAT INDENT
SELECT customer, COUNT(*)
FROM purchases
GROUP BY customer;

ProjectionExec: expr=[customer@0 as customer, count(Int64(1))@1 as count(*)]
  AggregateExec: mode=FinalPartitioned, gby=[customer@0 as customer], aggr=[count(Int64(1))]
    RepartitionExec: partitioning=Hash([customer@0], 2), input_partitions=2
      AggregateExec: mode=Partial, gby=[customer@0 as customer], aggr=[count(Int64(1))]
        DataSourceExec: file_groups={2 groups: [[data/purchases/monday.parquet, data/purchases/tuesday.parquet], [data/purchases/wednesday.parquet]]}, projection=[customer], file_type=parquet

Nothing in this plan says the order was preserved. If it had been, the plan would mark it on the scan with output_ordering and on the aggregate with ordering_mode; here neither appears. Even with the experimental setting that groups files by their statistics turned on (datafusion.execution.split_file_groups_by_statistics = true), the plan stays the same.

What is going on? Why is the order gone?

2. The premise: order starts inside a file

Let’s start at the beginning.

We assumed that the files were sorted internally when they were written. The engine does not know this on its own: it has to be told, and in DataFusion there are two ways to tell it.

The first is the WITH ORDER clause in the definition of the external table, the one in our example, which informs the planner of the existing order.

The second is the file’s metadata. In Parquet, rows are divided into row groups, and the metadata of each row group has a field, sorting_columns, where the writer can record which columns its rows are sorted by. DataFusion reads this field only from the first row group of each file and takes it as the order of the whole file. If every file has this field, the engine derives the table’s ordering from their common prefix, even without a declaration. If even one file lacks it, no ordering is inferred. DataFusion reads this metadata alongside the file statistics, so this happens only when datafusion.execution.collect_statistics is on, as it is by default.

Some words of caution are due here. If the table declares WITH ORDER and the files carry the metadata too, the declaration wins. Furthermore, the field is not always filled in: DataFusion writes it when it saves sorted rows, for example with COPY … ORDER BY, but other writers may leave it out. Finally, the engine takes the order on trust: it does not check it. If Monday’s file had in fact been written by time of purchase, declaring the order by customer, or recording it in the metadata, would hand the planner a false premise, and an optimization built on that premise could produce wrong results. With the metadata there is one more risk: sorting_columns only says that each row group is sorted internally. For the whole file to be sorted, the boundary between consecutive row groups has to respect the order too, and DataFusion does not check that. [1]

3. The scan operator produces partitions

Sorted files and an order known to the engine are just a promising starting point. The question now is what happens to them. The answer lies in the physical plan. [2]

A physical plan is a tree of operators, and each operator produces its output as a number of partitions, parallel streams of batches. Partition i of an operator usually reads from partition i of its input, so each partition is a lane that runs up the tree, and the engine executes each lane as an independent unit of work (this is also called intra-operator parallelism). [3] Most operators keep the number of lanes they receive: a projection, a filter or a partial aggregate neither adds nor removes any. The number is set at the bottom by the scan, and it changes only where the plan says so: a repartition operator moves rows between lanes and may change their number, and a merge or coalesce operator brings them down to one.

The scan is the operator at the bottom of the tree, the one with no input. Its partitions are decided by how it organizes the files into groups: each partition is the stream that reads one group, one file after the other, while different partitions run in parallel. The order within a partition therefore depends on reading the files of its group in sequence. Two sorted files read one after the other stay sorted only if the first ends no later than the second begins. The scan concatenates the files of a group, it does not merge them.

By default the engine aims for as many partitions, and so as many groups, as the processor has cores. We can set this number explicitly by assigning a value N to target_partitions.

How are the files assigned to those groups? It takes up to six steps. [4]

1. Statistics. With statistics collection enabled, the engine reads the column minima and maxima from each file footer, along with sorting_columns if present, as we saw in section 2.
2. Default grouping. It sorts the files by path and cuts them into contiguous blocks of ⌈number of files / N⌉ files each. Each block is a group. The values inside the files play no part in this step: only the file path and its position in the list matter.
3. Candidate order. It is the one from WITH ORDER, if there is one; otherwise the one common to the sorting_columns of all the files; otherwise none. Without a candidate order the scan has nothing to preserve, and it goes straight to step 6.
4. Statistics-based grouping. If it is on, the engine tries to build different groups. It sorts the files by ascending minimum and starts from N empty groups. It places each file in the least full group among those that are empty and those whose last file ends strictly before the new one begins. If no group fits, it opens one more. At the end, if there are no more than N groups, it uses them; otherwise it falls back to the groups of step 2 and notes it in the debug log.
5. Validation. It checks the candidate order on the groups obtained. In every group with more than one file, the maximum of each file must be less than or equal to the minimum of the next. If all the groups pass, the scan advertises its ordering. If even one fails, or if statistics are missing, the scan does not advertise it.
6. Splitting. If the table has more rows than one batch (8192 by default), the engine may split the files into byte ranges to fill more partitions. With an advertised ordering it does so only if the groups are fewer than N, and it splits only groups made of a single file: each piece becomes a group of its own, and the order remains. Without an order, if the files together exceed repartition_file_min_size (1 MiB by default), it redistributes all the bytes into N groups of equal size, even across file boundaries.

In our example N is 2, and the two file groups in the plan are the two partitions of the scan. [5] Step 1 reads the same statistics from every file: minimum Anna, maximum Zeno. Step 2 puts Monday and Tuesday in the same group and leaves Wednesday on its own. Step 3 takes the candidate order from the declaration: by customer. Step 4, when it is on, would want one group per file, because no file can follow another, but three groups exceed N, and the proposal is discarded. Step 5 finds Zeno before Anna in the group of Monday and Tuesday, and the scan does not advertise its ordering. Step 6 does not apply because there are too few rows. So an order known for individual files may not survive this stage. [6]

With N equal to 3, or with the default on a machine with more cores, every file would get a group of its own at step 2, and every partition would stay sorted. The three sorted partitions still overlap, though: all three contain Anna. For a GROUP BY customer this is not a problem. The partial aggregate works on each partition separately and only needs that partition to be sorted; Anna’s counts from the different partitions are combined later, as section 4 shows.

But take thirty daily files and N = 8: step 2 fills blocks of a fixed size until the files run out, and the last block takes what is left.

The higher parallelism does not help: the engine still forms exactly N groups, so with thirty files most of them hold several files each, and the order is lost in the same way as before.

To summarize, what matters is which files end up in the same group: by path at step 2, by range at step 4. With overlapping files like ours no grouping can put two files together and keep them sorted, so the order survives only while the files are no more than N. Files with disjoint ranges can do better, as section 6.2 shows.

One last point about groups and partitions. The six steps fix the groups at planning time, but execution could still move files between them: by default, a partition that has run out of work can take files assigned to another. When the table has an order, declared or derived from the metadata, DataFusion switches this mechanism off, and every file stays in its group. [5]

4. How the order climbs the plan

Suppose the scan operator really does deliver sorted partitions. What happens then? The order now has to climb the plan, one operator at a time.

The planner makes its decisions before execution, so it cannot inspect the rows the operators will produce: it relies on what each operator declares about its own output, including its ordering. Each operator, when it is built, works out the order of its output from the order of its input; since the plan is built from the scan upward, the order propagates in the same direction. Different operators treat the order differently. A filter drops rows but does not move them, so its output keeps the order of its input. An operator that shuffles rows loses it, unless it does something to preserve it.

For a partial aggregate (the aggregation that runs within a partition), the question is how much of that order concerns its keys. The plan shows it with ordering_mode: Sorted if the incoming order begins with all the keys of the GROUP BY, in any sequence; PartiallySorted if it begins with only some of them. Only the leading run of keys counts: as soon as the order reaches a column outside the GROUP BY, the keys after it no longer help, unless a filter has fixed that column to a single value, in which case the planner drops it from the order. In both modes the aggregate emits its groups in the order in which their keys arrive, so its output is sorted by the same keys and the next operator receives that order. If none of the keys forms a useful ordering prefix, the plan shows no ordering_mode: the aggregate does not use the input ordering, its output is unsorted, and it passes no ordering on. [7]

In our example, with two partitions, the order is already lost at the scan. To see how it travels along the plan we need a scan that preserves it and passes it on. Raising the example’s target partitions to three is enough: every file then gets a group of its own, as we saw in section 3.

SET datafusion.execution.target_partitions = 3;

ProjectionExec: expr=[customer@0 as customer, count(Int64(1))@1 as count(*)]
  AggregateExec: mode=FinalPartitioned, gby=[customer@0 as customer], aggr=[count(Int64(1))], ordering_mode=Sorted
    RepartitionExec: partitioning=Hash([customer@0], 3), input_partitions=3, preserve_order=true, sort_exprs=customer@0 ASC NULLS LAST
      AggregateExec: mode=Partial, gby=[customer@0 as customer], aggr=[count(Int64(1))], ordering_mode=Sorted
        DataSourceExec: file_groups={3 groups: [[data/purchases/monday.parquet], [data/purchases/tuesday.parquet], [data/purchases/wednesday.parquet]]}, projection=[customer], output_ordering=[customer@0 ASC NULLS LAST], file_type=parquet

Read from the bottom up: the scan advertises its ordering (output_ordering), the partial aggregate receives it and works in Sorted mode, the repartition operator preserves it (preserve_order=true), and the final aggregate works in Sorted mode too.

So far we have seen the order reach the partial aggregate and get passed on. What remains is how the repartition operator manages to preserve it. The final aggregate requires all the partial states for a customer to land in the same partition, and to make that possible the planner inserts a repartition operator that redistributes the states by a hash of the key. Every output partition thus receives states from all the scan partitions, and even if every incoming stream is sorted, it could receive Zeno’s state from Monday before Anna’s state from Tuesday: taking the batches as they arrive would break the order. [7]

To solve this, the repartition operator has an order-preserving variant, which merges the incoming streams for each output partition: it compares the next customer in each stream and emits the smallest, so Anna comes before Zeno.

The planner inserts this variant whenever the repartition operator’s input is sorted and has more than one partition, then checks whether the operators above make use of it. It stays only if the operator above preserves the order, or if the order lets that operator emit results before the end of its input, as our final aggregate does. Otherwise the plain repartition operator takes its place, because the merge has a runtime cost.

The fallback shows up, for example, in a partitioned hash join between two tables sorted by customer with nothing but a projection above it: nothing downstream uses the order, so the repartition operators below the join stay plain. The join itself passes on the order of its second input, so a GROUP BY customer above the join brings the order-preserving variant back on that side. [8]

5. What the aggregate gains

The climb ends at the aggregate, the operator that uses the order: the partial one inside each partition and the final one after the repartition. Now we can look at what the order means during execution, and whether it helps.

Back to the count per customer:

SELECT customer, COUNT(*)
FROM purchases
GROUP BY customer;

To group the rows by customer, DataFusion hashes the key of each row and keeps one entry per customer in a hash table, where the count accumulates. The order does not change how the rows are counted; it changes when an entry can leave the table.

Inside a partition sorted by customer, once the partial aggregate reaches Bruno, no more rows for Anna can arrive in that partition: her count is complete and her entry can leave. Without a useful order, Anna could turn up again further on, and her entry would have to stay until the end of the input. After each incoming batch, the partial aggregate emits the completed groups and removes their entries from the table. [9]

What the partial aggregate emits is Anna’s count in that partition, not her total. The totals are computed by the final aggregate, which receives Anna’s states in order from the repartition operator, as we saw in section 4. When Bruno turns up in its stream, the final aggregate knows in turn that no more partial states for Anna can arrive, and emits Anna’s total. This is why the planner keeps the order-preserving variant of the repartition operator. [10]

With the order on all the keys, as here, each aggregate keeps only the last group open after each batch, and the state it holds between batches does not grow with the number of customers. If the order covers only some of the keys, as in GROUP BY customer, product, the advantage shrinks: a group closes only when the customer changes, and many product groups can accumulate for a single customer. [9]

If the order covers none of the keys of the GROUP BY, there is no gain at all: when we count purchases by city, Rome appears both under Anna and under Zeno. Ordering by customer gives us no boundary at which to close the group for Rome. In fact, in this query the scan does not even advertise the order: it reads only city, and an ordering on a column that is not read cannot be expressed on the scan’s output. [6]

6. Diagnosis and remedies

So far we have looked at the mechanism. But in practice, how can we tell where the order is lost, and how do we keep it?

6.1. Where the order stops

The first question is answered by the plan. Read EXPLAIN from the bottom up, as in section 4. If the aggregate shows no ordering_mode, it is not using an ordering useful to its keys. Trace its input back toward the scan to find the last operator that still advertised one. From there, these questions retrace the chain from the files upward:

• Are the rows inside each file really sorted as declared, or as recorded in the metadata? (section 2)
• If the table does not declare the order, do all the files report sorting_columns in the first row group? (section 2)
• Are file statistics being collected (datafusion.execution.collect_statistics)? (section 3, statistics)
• If statistics-based grouping is on, do the proposed groups fit within the target? (section 3, statistics-based grouping)
• Can the files of each group be concatenated in the assigned sequence? (section 3, validation)
• Does the query read the sort column? An ordering on a column the scan does not read disappears at the scan. (section 5)
• Does the scan advertise its ordering, and does the repartition operator preserve it? (section 4)
• Does the aggregate show ordering_mode for its keys? (sections 4 and 5)

One caveat. The plan does not say why an order was discarded. The rejection of statistics-based grouping, for example, appears only in the debug log; with datafusion-cli you can see it by setting RUST_LOG=datafusion_catalog_listing=debug.

6.2. How to keep it

When the diagnosis shows that the order is lost in the scan, because the files of a group cannot be concatenated or because statistics-based grouping is discarded, there are a few things you can do to keep it.

The first is to change the layout of the files before querying them: for example, files covering separate customer ranges can be concatenated in range order.

The layout alone, however, is not enough: the scan groups have to respect it too. [6] Suppose each day is written into two files, one for customers from A to L and one for those from M to Z, each sorted by customer:

/data/purchases/a-l-monday.parquet      Anna, Bruno, Carla
/data/purchases/a-l-tuesday.parquet     Anna, Bruno, Carla
/data/purchases/a-l-wednesday.parquet   Anna, Bruno, Carla
/data/purchases/m-z-monday.parquet      Marco, Sara, Zeno
/data/purchases/m-z-tuesday.parquet     Marco, Sara, Zeno
/data/purchases/m-z-wednesday.parquet   Marco, Sara, Zeno

The table and the query are the same, and the engine works with three partitions:

SET datafusion.execution.target_partitions = 3;

The default grouping cuts the list by path into blocks of two files. The first group puts Monday’s and Tuesday’s A–L files together, which overlap, so the order is lost:

ProjectionExec: expr=[customer@0 as customer, count(Int64(1))@1 as count(*)]
  AggregateExec: mode=FinalPartitioned, gby=[customer@0 as customer], aggr=[count(Int64(1))]
    RepartitionExec: partitioning=Hash([customer@0], 3), input_partitions=3
      AggregateExec: mode=Partial, gby=[customer@0 as customer], aggr=[count(Int64(1))]
        DataSourceExec: file_groups={3 groups: [[data/purchases/a-l-monday.parquet, data/purchases/a-l-tuesday.parquet], [data/purchases/a-l-wednesday.parquet, data/purchases/m-z-monday.parquet], [data/purchases/m-z-tuesday.parquet, data/purchases/m-z-wednesday.parquet]]}, projection=[customer], file_type=parquet

A solution is to turn on statistics-based grouping:

SET datafusion.execution.split_file_groups_by_statistics = true;

Off by default, this setting builds groups whose files can be concatenated according to their statistics. [11]

ProjectionExec: expr=[customer@0 as customer, count(Int64(1))@1 as count(*)]
  AggregateExec: mode=FinalPartitioned, gby=[customer@0 as customer], aggr=[count(Int64(1))], ordering_mode=Sorted
    RepartitionExec: partitioning=Hash([customer@0], 3), input_partitions=3, preserve_order=true, sort_exprs=customer@0 ASC NULLS LAST
      AggregateExec: mode=Partial, gby=[customer@0 as customer], aggr=[count(Int64(1))], ordering_mode=Sorted
        DataSourceExec: file_groups={3 groups: [[data/purchases/a-l-monday.parquet, data/purchases/m-z-monday.parquet], [data/purchases/a-l-tuesday.parquet, data/purchases/m-z-tuesday.parquet], [data/purchases/a-l-wednesday.parquet, data/purchases/m-z-wednesday.parquet]]}, projection=[customer], output_ordering=[customer@0 ASC NULLS LAST], file_type=parquet

There are still three groups, but each pairs a day’s A–L file with its M–Z file. Those two files can be concatenated in order. With the same number of partitions, the order reaches the aggregate.

Turning the setting on, however, does not guarantee that the proposed groups are used: as we saw at the statistics-based grouping step in section 3, the engine discards them if they exceed the target. [12]

Raising the target can make the grouping acceptable, and sometimes it is enough on its own. With six partitions the default grouping gives every file a group of its own, even without statistics:

SET datafusion.execution.target_partitions = 6;

ProjectionExec: expr=[customer@0 as customer, count(Int64(1))@1 as count(*)]
  AggregateExec: mode=FinalPartitioned, gby=[customer@0 as customer], aggr=[count(Int64(1))], ordering_mode=Sorted
    RepartitionExec: partitioning=Hash([customer@0], 6), input_partitions=6, preserve_order=true, sort_exprs=customer@0 ASC NULLS LAST
      AggregateExec: mode=Partial, gby=[customer@0 as customer], aggr=[count(Int64(1))], ordering_mode=Sorted
        DataSourceExec: file_groups={6 groups: [[data/purchases/a-l-monday.parquet], [data/purchases/a-l-tuesday.parquet], [data/purchases/a-l-wednesday.parquet], [data/purchases/m-z-monday.parquet], [data/purchases/m-z-tuesday.parquet], ...]}, projection=[customer], output_ordering=[customer@0 ASC NULLS LAST], file_type=parquet

The order reaches the aggregate this way too, but with twice as many partitions. The target controls more than the number of scan groups: in this two-stage plan, it also sets the number of partitions the repartition operator creates for the final aggregate.

It pays not to overdo it, though. A higher target can mean more concurrent operators and more pressure on memory, and preserving the order already has a cost, the merge inside the repartition operator.

7. Conclusions

The order that matters to our GROUP BY is the one guaranteed within each partition it receives. To exploit the order stored in Parquet, DataFusion must be told that the order exists, establish that guarantee at the scan and preserve it through the intervening operators. In the original example, the guarantee fails at the scan: reading Monday and Tuesday in sequence takes the customer order from Zeno back to Anna.

File layout, the grouping of the files and the partition target determine whether the scan can produce sorted partitions. Adjusting them can preserve the order, but may also change parallelism and resource use. Further up the plan, carrying that order through a repartition requires merging streams.

EXPLAIN shows whether the aggregate receives and uses a useful ordering. Measurement tells us whether the aggregation work it saves outweighs the cost of preserving it. [13]

The examples can be reproduced with datafusion-cli 55.1.0; the cost of order is measured in a small reproducible lab.

References

The settings and scan behavior described here refer to DataFusion’s listing-table path. Implementation references are pinned to the 55.1.0 release (tag 55.1.0).

1. WITH ORDER and its correctness requirements, precedence of the declaration over the metadata and order read from sorting_columns.
2. Goetz Graefe and William J. McKenna, The Volcano Optimizer Generator: Extensibility and Efficient Search, ICDE 1993: logical and physical algebra, with sort order as a physical property; in DataFusion, the logical scan becomes a physical operator.
3. Andy Pavlo, Query Execution II, lecture notes 14 of 15-445/645 Database Systems, Carnegie Mellon University, Fall 2024, section 5: intra-operator parallelism.
4. Collection of file statistics and order, default grouping, statistics-based grouping, conditions for splitting files and splitting with and without order.
5. File groups and sequential reads, when files can change partition and when that does not happen.
6. Ordering validation, boundary comparison and orderings projected onto the columns the scan reads.
7. Using Ordering for Better Plans in Apache DataFusion, 11 March 2025, aggregate mode and the order of its output and distribution required by the final aggregate.
8. Order-preserving repartition and its runtime cost, when the variant is enabled, insertion with the ordered variant and when the planner keeps it.
9. Ordered partial aggregate emission.
10. Aggregation with ordered grouping keys, emission of totals in the final aggregate.
11. Statistics-based grouping configuration.
12. Group-count guard and debug log.
13. Tuning memory-limited queries.