diff --git a/content/blog/2026-07-20-sort-pushdown.md b/content/blog/2026-07-20-sort-pushdown.md new file mode 100644 index 00000000..3fdebe41 --- /dev/null +++ b/content/blog/2026-07-20-sort-pushdown.md @@ -0,0 +1,528 @@ +--- +layout: post +title: Optimizing for Almost Sorted Data: Sort Pushdown in Apache DataFusion +date: 2026-07-20 +author: Qi Zhu and Andrew Lamb +categories: [performance] +--- + + + +[TOC] + +*[Qi Zhu](https://github.com/zhuqi-lucas) ([Massive](https://www.massive.com/)); [Andrew Lamb](https://github.com/alamb) ([InfluxData](https://www.influxdata.com/))* + +[Apache DataFusion] uses sortedness automatically — even when data is only +partially sorted or when no ordering was declared. This post explains how +plan-time sort pushdown, runtime scan reordering, and row-group pruning driven +by [dynamic filters][dyn-filters-blog] make that possible. + +[Apache DataFusion]: https://datafusion.apache.org/ +[dyn-filters-blog]: https://datafusion.apache.org/blog/2025/09/10/dynamic-filters/ + +## Are Real Datasets Sorted? + +Sorting data is prohibitively expensive for many workloads, so maintaining it in fully +sorted order is often not practical. +However, many real datasets are at least partly sorted when stored: time-series files by +ingestion time, event logs by event id, partitioned tables by partition key, and +data lakes based on [Apache Iceberg] and similar formats in write order. + +Sortedness only helps if the query engine can detect and use it. Two common +cases make that hard: + +1. The ordering is undeclared in the file. For example, the writer did not set Parquet + [sorting_columns](https://github.com/apache/parquet-format/blob/8a5e04bdecf100e8e981daacfa117e8b5aadacb9/src/main/thrift/parquet.thrift#L1044), + or the table was not created with DataFusion's + [WITH ORDER](https://datafusion.apache.org/user-guide/sql/ddl.html#create-external-table) clause. +2. Files are individually sorted, but the engine is scanning multiple files + and does not know a global ordering at plan time. + +In both cases, queries with `ORDER BY` or `ORDER BY ... LIMIT N` pay for a full +scan and sometimes a blocking full sort, which buffers every row and can +dominate latency and peak memory on large scans. + +Using min/max statistics for *predicate* pushdown is well-known and widely +implemented across databases, as covered in [@XiangpengHao]'s +post on [Parquet pruning][parquet-pruning-blog]. Using the same +statistics to *reason about sort order* and to prune `ORDER BY ... LIMIT` +(top-k) queries is also increasingly common, for example in the +[Pruning in Snowflake: Working Smarter, Not Harder] paper (see [Related Work](#related-work) for more). +This post describes several such techniques for a general audience, +including the less-common case of *discovering* a +global sort order from per-file statistics, +deleting redundant sorts, and biasing scan order toward the +most-promising data. + +[Apache Iceberg]: https://iceberg.apache.org/ +[@XiangpengHao]: https://github.com/XiangpengHao +[parquet-pruning-blog]: https://datafusion.apache.org/blog/2025/03/20/parquet-pruning +[Pruning in Snowflake: Working Smarter, Not Harder]: https://dl.acm.org/doi/10.1145/3722212.3724447 + +## Exact vs. Inexact Ordering + +DataFusion has long skipped sorts when it knows the data is **exactly** sorted, as covered in +[@akurmustafa's earlier post][ordering-analysis]: +if the table declares an ordering (via `WITH ORDER` or Parquet +`sorting_columns`) and the file listing matches it, the redundant sort is removed. + +This post is about the messier real-world cases +where sortedness exists but is *inexact* or not provable up front: + +- Files listed in the "wrong" order on disk (each file is internally + sorted, but the files are not globally ordered). +- Ordering is known, but the sort key ranges **overlap** across files. +- **No** declared ordering at all. +- `ORDER BY ... DESC` on ASC-sorted data. + +This post describes two primary techniques: + +1. **Statistics-based sort elimination** (`Exact`): avoids sorts entirely by + reordering files by min/max statistics to create a global ordering. This + extends DataFusion's existing statistics-based file-group ordering + ([#9593]) and is similar to prior work on concatenating disjoint ranges. +2. **Runtime reorder and dynamic pruning** (`Inexact`): reorders the scan to read + the most-promising data first, then re-checks the `TopK` dynamic filter at file + and row-group boundaries. + +These techniques are implemented in DataFusion and together result in: + +- **Sort elimination**: 2×–49× faster on `ORDER BY` and `ORDER BY ... LIMIT` + queries where the file list was in the wrong disk order. +- **Runtime row-group pruning**: 5 of 11 benchmark queries run ≥2× faster + (up to ~4×) with zero regressions; total runtime drops by 44%. + +The rest of this post walks through each technique in detail. + +[#22450]: https://github.com/apache/datafusion/pull/22450 +[#9593]: https://github.com/apache/datafusion/pull/9593 +[ordering-analysis]: https://datafusion.apache.org/blog/2025/03/11/ordering-analysis/ + +## Tracking Ordering + + +The [PushdownSort](https://github.com/apache/datafusion/blob/main/datafusion/physical-optimizer/src/pushdown_sort.rs) +optimizer rule classifies each scan below a sort as either `Exact`, +`Inexact`, or `Unsupported`, and records this information on DataFusion's +[FileScanConfig](https://docs.rs/datafusion-datasource/latest/datafusion_datasource/file_scan_config/struct.FileScanConfig.html): + +- **`Exact`** — the optimizer is *certain* of the output order, + and removes redundant [SortExec](https://docs.rs/datafusion-physical-plan/latest/datafusion_physical_plan/sorts/sort/struct.SortExec.html) operators entirely. +- **`Inexact`** — the optimizer believes the output is probably ordered + but cannot prove it, so the sort is kept but the scan is biased to read the most-promising data first. +- **`Unsupported`** — the optimizer cannot determine the ordering, so no sort is removed. + +PushdownSort decision tree: Exact, Inexact, or Unsupported
+*Figure: the `PushdownSort` rule asks each scan whether it can satisfy the +required ordering, returning `Exact` (drop the sort), `Inexact` (bias the scan +and keep the sort), or `Unsupported` (the sort stays).* + +For example, given a query that returns the 10 most recent trades: + +```sql +SELECT ts, symbol, amount FROM trades ORDER BY ts DESC LIMIT 10; +``` + +- With **`Exact`** ordering knowledge, DataFusion drops the sort entirely and + stops reading after emitting 10 rows. +- With **`Inexact`** ordering, the `SortExec` stays but scans start + from the most-promising data, so the `TopK` threshold is more likely to + tighten quickly and prune the rest with statistics. +- With no ordering knowledge, DataFusion scans everything and uses a + `TopK` heap to keep the running best 10. + +## Exact: Sort Elimination via Statistics + +DataFusion can eliminate sorts entirely when it can use Parquet min/max statistics to reorder files and prove a global ordering +(relevant PRs: [#19064] and [#21182]) as shown below. + +EXPLAIN before / after: SortExec eliminated once ordering is Exact +*Figure: EXPLAIN output before and after `PushdownSort` eliminates the sort. The `SortExec` is removed and the scan's ordering claim is upgraded to `Exact`.* + + + +[#19064]: https://github.com/apache/datafusion/pull/19064 +[#21182]: https://github.com/apache/datafusion/pull/21182 + +For example, consider three files `a.parquet`, `b.parquet`, `c.parquet`. Each is +internally sorted by `ts` and the time ranges do not overlap, but the files were +written by different jobs. Without this optimization, a query engine is forced to +use a full sort for `ORDER BY ts` queries, even though it could simply read the +files one after another and produce the correct result. + +The `Exact` path requires each file to be +*individually* sorted on the key, and DataFusion knows this from the declared +ordering (`WITH ORDER` or Parquet `sorting_columns`). +Concatenating the files in min/max range order yields a globally sorted stream +only when that per-file precondition holds. When no per-file ordering is declared, the +`Inexact` path (described next) applies instead. + +`PushdownSort` fixes this in three steps at the file-scan node: + +1. **Sort the file list by per-file `min`** on the sort column. +2. **Check overlap**: does `file[i].max ≤ file[i+1].min` hold for + every adjacent pair? If yes, the sorted file list produces a globally + sorted stream. +3. **Upgrade the source's ordering claim to `Exact`** and remove the + surrounding `Sort`. Note that this requires some additional performance optimizations + described in [appendix on buffering without sorting](#appendix-buffering-without-sorting). + +This is an instance of what Graefe calls *virtual concatenation*, and is similar to +LSM "trivial move" compaction (see [Related Work](#related-work)). It is less common +to discover the disjoint ranges from per-file +min/max statistics rather than from a declared partitioning scheme. + +[graefe2006]: https://dl.acm.org/doi/10.1145/1132960.1132964 +[graefe1993]: https://dl.acm.org/doi/10.1145/152610.152611 + +File reorder: rearranging files within a partition by min/max statistics so the file list is in range order
+*Figure: reordering by per-file `min/max` puts the file list in range order.* + + +Detecting non-overlapping ranges via min/max statistics
+*Figure: `PushdownSort` sorts files by `min` and checks adjacency, +upgrading to `Exact` only when the ranges don't overlap. Left: non-overlapping +ranges are safe to upgrade to `Exact` and the sort is removed; right: +overlapping ranges keep the sort and fall through to the `Inexact` path described next.* + +We measured the single-core performance of statistics-based sort elimination with DataFusion's +[sort_pushdown](https://github.com/apache/datafusion/tree/main/benchmarks/queries/sort_pushdown) benchmark suite +by setting `--partitions 1`. The results are as follows: + +Sort pushdown benchmark: 2×–49× speedup across four queries
+ +| Query | Before | After | Speedup | +| ------------------------------------------- | -------:| -------:| -------: | +| Q1 — `ORDER BY key` (full scan) | 259 ms | 122 ms | **2.1×** | +| Q2 — `ORDER BY key LIMIT 100` | 80 ms | 3 ms | **27×** | +| Q3 — `SELECT * ORDER BY key` | 700 ms | 313 ms | **2.2×** | +| Q4 — `SELECT * ORDER BY key LIMIT 100` | 342 ms | 7 ms | **49×** | + +*Figure: `sort_pushdown` results. All four queries are `ASC` `ORDER BY` where the +file list on disk is in the reverse of sort-key order; a naive engine sorts even +though the files are already individually sorted.* + +While DataFusion can avoid sorting for all four queries, the benefit is most +dramatic for `LIMIT` queries: + +- **Full-scan queries (Q1, Q3)** result in a ~2× speedup as the scan is now a single-pass streaming read. +- **`LIMIT` queries (Q2, Q4)** result in a 27×–49× speedup because `LIMIT N` turns into a streaming read with early stopping. + +## Inexact: Scan Reordering Makes Dynamic Filters More Effective + +It is not possible to eliminate the sort when the files have +overlapping sort key ranges. However, it is still possible to use sorted or +partly sorted data to optimize `LIMIT` queries. By reordering the read, +data that is most likely to improve the `TopK` dynamic filter is seen first and +the filter is more effective at pruning files, row groups, and rows that cannot +possibly contribute to the final result. + +This `Inexact` path is used when Parquet min/max statistics are available +and the sort expression is supported by DataFusion's +[equivalence-properties][ordering-analysis], including plain columns, monotonic +functions[1](#footnote1), constants inferred from filters, +and multi-column orderings. The same dynamic filter is used to drive three layers of pruning: + +Three-layer pruning: file-level, row-group-level, row-level, all driven by the same TopK dynamic filter
+*Figure: the Parquet reader applies three pruning layers, all driven by the same `TopK` dynamic filter.* + +* **File-level pruning** ([file_pruner] + [EarlyStoppingStream]) — skips whole files before they are opened. +* **Row-group-level pruning** ([#22450]) — skips whole row groups at each boundary, before any pages are fetched. +* **Row-level filtering** ([RowFilter]) — skips decoding the remaining columns in a surviving row group. + +[file_pruner]: https://docs.rs/datafusion-pruning/latest/datafusion_pruning/struct.FilePruner.html +[EarlyStoppingStream]: https://github.com/apache/datafusion/blob/e104138b4d45d3acfb76223cd968385f6764477b/datafusion/datasource-parquet/src/opener/early_stop.rs +[RowFilter]: https://docs.rs/parquet/latest/parquet/arrow/arrow_reader/struct.RowFilter.html + +When `Inexact` applies, the Parquet opener reorders +the scan before and during execution using three steps: + +Runtime reorder pipeline: file reorder, RG reorder, then optional reverse
+ +*Figure: the Parquet opener applies file-level reordering, row-group-level reordering, and an optional iteration reverse.* + +1. **File-level reordering** — the file list is sorted by `min(col)`, + so the most-promising file is picked first across all partitions. +2. **Row-group-level reordering** — once a file is opened, its row groups + are sorted by `min(col)`. +3. **Iteration reverse** — for `DESC` queries, the file and row-group order is reversed. + +**File-Level Pruning**: once files are ordered "most-promising first", the +`TopK`'s heap fills quickly and its dynamic filter threshold tightens. The +[FilePruner] then cuts low-value files before opening them, as shown in the +following example. + +[FilePruner]: https://github.com/apache/datafusion/blob/e104138b4d45d3acfb76223cd968385f6764477b/datafusion/pruning/src/file_pruner.rs + +File-level reorder with early stop via file_pruner
+ +*Figure: after reordering files by their sort key, the low sort-key value files (`file_d` and `file_c`, +whose sort-key values are least promising for this `DESC LIMIT` query) +end up at the tail of the read queue and are pruned by the file-level pruner before they +are ever opened.* + +**Row-Group-Level Pruning**: When a file is first opened, DataFusion prunes +row groups based on predicates and statistics, and then determines +the order to scan the row groups. During the scan, immediately before DataFusion +reads the next row group, it checks whether the `TopK` dynamic filter has changed. If so, +it re-evaluates the remaining row groups against the new threshold and +prunes any that cannot contribute to the final result (added in [#22450]). See the +[Appendix: Decoder Loop and Decision Point](#appendix-decoder-loop-and-decision-point) for more details. + + +Cascading prune: one row group fills the heap, threshold snaps, all subsequent row groups are pruned in a single pass
+*Figure: for `ORDER BY x DESC LIMIT 10`, after reading the first row group, DataFusion prunes the rest of the row groups in a single pass (walkthrough below).* + +For the example shown above, the file has 10 row groups, each containing rows with values +`[0..10)`, `[10..20)`, ..., `[90..100)` and the query is `ORDER BY x DESC LIMIT 10`. + +1. The row groups are first reordered by their maximum values, so the highest-value row group is read first. +2. Row group 9 (values `[90..100)`) is opened and read, filling + the heap with 10 values (the minimum heap value is `90`). The `TopK` dynamic filter threshold is updated to `90`. +3. At the next row-group boundary, the pruner sees the dynamic filter has been updated + and re-evaluates the filter on all remaining row groups. Since row group 8 + through row group 0 have `max < 90`, they are all pruned and skipped in a single pass. +4. The scan ends early, having read only one row group instead of all 10. + +Note that for the row groups that were skipped, no data is fetched or decoded. +This example shows the core benefit provided by runtime row-group dynamic +pruning: reading a single row group can cascade-eliminate every remaining row +group. + +**Row-Level Early Stopping**: Even if the row group cannot be pruned with statistics, +the tightened dynamic filter often rules out all rows after evaluating just the +filter columns. When it does, arrow-rs's `RowFilter` skips fetching the projection +columns for that row group entirely. [#22450] makes this fire more often +by keeping the threshold up-to-date at every row-group boundary. + +Row-group-level reorder — filter column still read for every row group before row-group filter early stop
+ +*Figure: inside a file, the first row group tightens the threshold — +subsequent row groups have their projection columns short-circuited, +but the filter column still has to be read to discover that no rows +qualify.* + + +## Benchmark: sort_tpch + +To test the overall benefit of this work, we used DataFusion's +[sort_tpch](https://github.com/apache/datafusion/blob/main/benchmarks/src/sort_tpch.rs) +benchmark that runs 11 queries over the TPC-H SF1 dataset, each with +`ORDER BY ... LIMIT 100`. The data is stored in Parquet files, sorted by +`l_orderkey`. For the largest table, `lineitem`, `l_orderkey` is a `BIGINT` with ~1.5M +distinct values, so per-row-group `min/max` ranges are cleanly disjoint. +We compared DataFusion with the sort optimizations enabled (the default) and disabled. + +sort_tpch benchmark results: 5 of 11 queries ≥2× faster (up to ~4×), 0 regressions, total -44% + +| Metric | Value | +| ----------------------------------- | ---------------------------------- | +| Total wall-clock (sum of 11 queries) | 248.8 ms → 139.1 ms (**−44%**) | +| Queries with ≥2× speedup | **5 of 11** (Q2, Q4, Q8, Q9, Q10) | +| Queries with regression | **0** | +| Best single-query speedup | **~4×** | + +*Per-query raw numbers are available in the [#22450 benchmark run][topk-tpch-raw].* + +[topk-tpch-raw]: https://github.com/apache/datafusion/pull/22450#issuecomment-4765161078 + +The five queries that improved use `l_orderkey` as the **first** sort key column, so +row-group-level pruning can cascade-prune aggressively. The other queries have multi-column +sorts with +low-cardinality or unsorted columns (`l_linenumber`, `l_comment`, +`l_shipmode`), whose per-row-group ranges overlap heavily. Even when `l_orderkey` +appears later as a tie-breaker, the leading key controls row-group-level disjointness, +so row-level filtering is still partially effective. + + +## Conclusion and Future Work + +Use cases where the query sort key (e.g. `ORDER BY time DESC LIMIT 10`) is +aligned with the physical layout (e.g. the data is ordered by `time`) are common +in time-series, partitioned tables, and ingestion-ordered event logs. Over the +last few DataFusion releases, we implemented several optimizations that make +these workloads up to ~4× faster without slowing down queries for which they +don't apply. We hope you enjoy using them and welcome your feedback. + +Two follow-ups are open: page-level `Exact` reverse would let `DESC` queries drop the sort +but needs lower-level support in the Parquet reader ([arrow-rs#9937](https://github.com/apache/arrow-rs/pull/9937)) +for reading pages in reverse order. +We are also considering implementing page-level dynamic pruning at +row-group boundaries ([#23216](https://github.com/apache/datafusion/issues/23216)) +using the same pattern one level deeper in Parquet. + +## Acknowledgements + +Thank you to [@adriangb], [@xudong963], [@2010YOUY01], and +[@Dandandan] for reviewing the design and the patches across many +iterations. The DataFusion community's willingness to engage deeply +with optimizer changes, including ones that touch foundational +code like the Parquet inner decode loop, is what made this work +possible. + +We also thank [Massive](https://www.massive.com/) and [InfluxData](https://www.influxdata.com/) for sponsoring this +work. + +[1](#fn1) For example, if the source declares +`ts DESC`, reversing that ordering gives `ts ASC`, which can satisfy +`date_trunc('day', ts) ASC`. + +[@adriangb]: https://github.com/adriangb +[@xudong963]: https://github.com/xudong963 +[@2010YOUY01]: https://github.com/2010YOUY01 +[@Dandandan]: https://github.com/Dandandan + +## Get Involved + +- **Try it out**: Run your `ORDER BY` / `ORDER BY ... LIMIT N` queries on your own data and share what you see. +- **Work on a [good first issue](https://github.com/apache/datafusion/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22)**, or pick up one of the [follow-ups](https://github.com/apache/datafusion/issues/23036) listed in the umbrella issue. +- **File issues or join the conversation**: [GitHub](https://github.com/apache/datafusion/) for bugs and feature requests, [Slack or Discord](https://datafusion.apache.org/contributor-guide/communication.html) for discussion. +- Learn more by visiting the [DataFusion](https://datafusion.apache.org/index.html) project page. + +## Related Work + +The ideas in this blog build on a long line of research, and several other +systems implement closely related techniques. + +**Exploiting and proving sort order.** Reasoning about orderings to +eliminate redundant sorts goes back to "interesting orders" in System R +([Selinger et al., SIGMOD 1979][selinger1979]) and was formalized by +[Simmen, Shekita, and Malkemus (SIGMOD 1996)][simmen1996]. Those techniques +derive orderings from *schema* — declared keys, indexes, and functional +dependencies — whereas we derive them from per-file min/max *statistics*. +Concatenating disjoint key ranges rather than merging them is Graefe's +*virtual concatenation* ([ACM Computing Surveys 2006][graefe2006]) — an idea +rooted further back in range-partitioned parallel sort +([Graefe, ACM Computing Surveys 1993][graefe1993]). It also appears as +"trivial move" compaction in LSM trees ([RocksDB][rocksdb-trivial]). For +*declared* partition bounds, the "concatenate, don't merge" optimization +appears in [TimescaleDB's OrderedAppend][timescale-append] and +[PostgreSQL's ordered partition scans][pg-append]. + +**Top-k early termination and threshold pushdown.** Using a running +threshold to stop early is described in [Fagin's Threshold Algorithm +(PODS 2001)][fagin2001], with index choice for faster convergence +explored in [IO-Top-k (VLDB 2006)][iotopk2006]; see +[Ilyas et al. (ACM Computing Surveys 2008)][ilyas2008] for a survey. +Pushing a *live* top-k boundary value into a columnar scan to skip blocks +via min/max is described for Snowflake in +[Pruning in Snowflake: Working Smarter, Not Harder] and ships in +[ClickHouse][ch-topn] (granule-level top-N skipping), [DuckDB][duckdb-topn] +(dynamic Top-N table filters), [PolarDB-IMCI][polardb-topk] (self-sharpening +runtime filters), and InfluxDB IOx ([ProgressiveEvalExec][iox-progressive]). + +**Reverse scans and data skipping.** Reading physically-ordered data in +reverse to satisfy `DESC ... LIMIT` is a well-known technique in traditional +databases (backward/descending index scans in [PostgreSQL][pg-backward] and [Oracle][oracle-backward]). +Min/max block skipping itself originates with +[Small Materialized Aggregates (Moerkotte, VLDB 1998)][sma1998] and is +now ubiquitous across columnar storage formats and query engines. + +[selinger1979]: https://dl.acm.org/doi/10.1145/582095.582099 +[simmen1996]: https://dl.acm.org/doi/10.1145/233269.233320 +[fagin2001]: https://arxiv.org/abs/cs/0204046 +[iotopk2006]: https://www.vldb.org/conf/2006/p475-bast.pdf +[ilyas2008]: https://dl.acm.org/doi/10.1145/1391729.1391730 +[sma1998]: https://www.vldb.org/conf/1998/p476.pdf +[rocksdb-trivial]: https://github.com/facebook/rocksdb/wiki/Compaction-Trivial-Move +[timescale-append]: https://www.tigerdata.com/blog/ordered-append-postgresql-optimization +[pg-append]: https://www.postgresql.org/message-id/2401607.SfZhPQhbS4@ronan_laptop +[pg-backward]: https://www.postgresql.org/docs/current/indexes-ordering.html +[oracle-backward]: https://docs.oracle.com/en/database/oracle/oracle-database/19/tgsql/optimizer-access-paths.html +[ch-topn]: https://clickhouse.com/blog/clickhouse-top-n-queries-granule-level-data-skipping +[duckdb-topn]: https://duckdb.org/2024/10/25/topn +[polardb-topk]: https://www.alibabacloud.com/blog/how-does-the-imci-of-polardb-for-mysql-achieve-ultimate-topk-query-performance_600006 +[iox-progressive]: https://www.influxdata.com/blog/query-optimization-progressive-evaluation-influxdb/ + + +## References + +Umbrella issue tracking the entire effort: + +* **[EPIC] Sort Pushdown · skip sorts and skip IO for ORDER BY / TopK queries: [#23036](https://github.com/apache/datafusion/issues/23036)** — phase-by-phase status of all the PRs and follow-ups. + +Major PRs: + +* `MinMaxStatistics` foundation: [#9593](https://github.com/apache/datafusion/pull/9593) +* `PushdownSort` rule + row-group reverse: [#19064](https://github.com/apache/datafusion/pull/19064) +* Reverse-output redesign: [#19446](https://github.com/apache/datafusion/pull/19446), [#19557](https://github.com/apache/datafusion/pull/19557) +* Sort elimination via statistics: [#21182](https://github.com/apache/datafusion/pull/21182) +* `BufferExec` capacity for sort elimination: [#21426](https://github.com/apache/datafusion/pull/21426) +* Push-based Parquet decoder (DataFusion owns the loop): [#20839](https://github.com/apache/datafusion/pull/20839) +* Morsel-style work scheduling: [#21351](https://github.com/apache/datafusion/pull/21351) +* Runtime reorder for `TopK` convergence: [#21956](https://github.com/apache/datafusion/pull/21956) +* **Runtime row-group dynamic pruning ([#22450])** — the centerpiece of this post. +* `peek_next_row_group` API (arrow-rs): [arrow-rs#10158](https://github.com/apache/arrow-rs/pull/10158) — enables per-RG `fully_matched` RowFilter skip. + +In flight / open: + +* Page-level reverse (arrow-rs): [arrow-rs#9937](https://github.com/apache/arrow-rs/pull/9937), discussion in [arrow-rs#9934](https://github.com/apache/arrow-rs/issues/9934) +* Per-RG `fully_matched` RowFilter skip on top of [#22450] (uses arrow-rs#10158): [#23067](https://github.com/apache/datafusion/issues/23067) +* Page-level dynamic prune at RG boundary: [#23216](https://github.com/apache/datafusion/issues/23216) +* Multi-column / function-wrapped stats reorder follow-ups: [#22198](https://github.com/apache/datafusion/issues/22198) + +Benchmark suites: [sort_pushdown](https://github.com/apache/datafusion/tree/main/benchmarks/queries/sort_pushdown), [sort_tpch](https://github.com/apache/datafusion/blob/main/benchmarks/src/sort_tpch.rs). + + +## Appendix: Buffering Without Sorting + +SPM stalls when SortExec is removed in multi-partition plans
+*Figure: removing the per-partition `SortExec` leaves the top-of-plan +merge (`SortPreservingMergeExec`) directly consuming raw I/O; a stall +on any partition stalls the whole plan.* + +Removing `SortExec` was not always faster in multi-partition plans: the deleted +sort had also acted as an implicit buffer. The fix was explicit buffering in +some plans; see +[#21426](https://github.com/apache/datafusion/pull/21426) +for details, as shown in the following figure. + +BufferExec replaces the deleted SortExec with a bounded streaming buffer per partition
+*Figure: `BufferExec` is inserted where the `SortExec` used to live — +same greedy per-partition prefill, but no blocking sort.* + +The fix is [BufferExec](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/buffer.rs): +a bounded per-partition prefill buffer that restores the greedy parallel I/O +driver role without sorting. + + + +## Appendix: Decoder Loop and Decision Point + +transition() loop: drain, decide, drive — Step 2 is the #22450 addition
+*Figure: the decoder loop has three steps. Step 2 (DECIDE) is what +[#22450] adds — it only fires at row-group boundaries.* + +The loop body reads: **drain** the current row group's batches until +it's exhausted; **decide** at the boundary whether any of the +remaining row groups can be dropped based on the live threshold; then +**drive** the decoder into the next row group and repeat. Inside a +row group, only drain and drive run — no decision point. + +RowGroupPruner: watch (cheap), rebuild (expensive, only if changed), prune (cheap)
+*Figure: the pruner has a cheap "check if the filter changed" step, a +moderately expensive "rebuild the predicate if so" step, and a cheap +"apply the predicate to remaining row groups" step.* + +The pruner does expensive work only when it can help: a cheap epoch check +detects dynamic-filter changes, and only then rebuilds the pruning predicate. +The predicate is then applied to remaining row groups' min/max statistics using +metadata only. + diff --git a/content/images/sort-pushdown/arch_one_glance.png b/content/images/sort-pushdown/arch_one_glance.png new file mode 100644 index 00000000..7e8dafc3 Binary files /dev/null and b/content/images/sort-pushdown/arch_one_glance.png differ diff --git a/content/images/sort-pushdown/benchmark.svg b/content/images/sort-pushdown/benchmark.svg new file mode 100644 index 00000000..4eed0a7c --- /dev/null +++ b/content/images/sort-pushdown/benchmark.svg @@ -0,0 +1,78 @@ + + + + sort_pushdown benchmark (single partition, release, reversed-name data) + + + + + + + 1000ms + + 100ms + + 10ms + + 1ms + + + + + Q1 + ORDER BY full + + 259 + + 122 + 2.1× + + + Q2 + ORDER BY LIMIT + + 80 + + 3 + 27× + + + Q3 + SELECT * ORDER BY + + 700 + + 313 + 2.2× + + + Q4 + SELECT * ORDER BY LIMIT + + 342 + + 7 + 49× + + + latency (ms, log scale) + + + + main (before) + + sort pushdown + Lower is better + --partitions 1, release + diff --git a/content/images/sort-pushdown/buffer-exec-stall.svg b/content/images/sort-pushdown/buffer-exec-stall.svg new file mode 100644 index 00000000..bbcc9a24 --- /dev/null +++ b/content/images/sort-pushdown/buffer-exec-stall.svg @@ -0,0 +1,160 @@ + + + + + + + + + + + + + + + + ✗ Without BufferExec + SPM polls I/O directly — k-way merge stalls on any slow partition + + + + DataSource + (I/O bound) + + + + Partition 0 + ⏳ I/O in flight + no batch ready + + + + Partition 1 + + ready + + + + Partition 2 + + ready + + + + Partition N + + ready + + + + + + + + + + + + + + + + SPM + k-way merge + STALLED + waits for P0 + + + ↓ + no progress + + + + + + ✓ With BufferExec + Background prefill — SPM always has rows in hand + + + + DataSource + (I/O bound) + + + + BufferExec (bounded) + + + + Partition 0 + + + + + + ready + + + + Partition 1 + + + + + + ready + + + + Partition 2 + + + + + + ready + + + + Partition N + + + + + + ready + + + + + + + + + → SPM + unblocked + + + + + SPM polls partition (blocked when no batch) + + background prefill keeps queue warm + + buffered batch (ready to drain) + diff --git a/content/images/sort-pushdown/buffer-exec.svg b/content/images/sort-pushdown/buffer-exec.svg new file mode 100644 index 00000000..fe8a744d --- /dev/null +++ b/content/images/sort-pushdown/buffer-exec.svg @@ -0,0 +1,89 @@ + + + + + + + + + + + + BufferExec — per-partition bounded queues + + + + DataSource + (I/O bound) + + + + BufferExec + capacity: sort_pushdown_buffer_capacity · default 1 GB + + + + Partition 0 + + + + + + bounded queue + + + + Partition 1 + + + + + + + + … + + + + Partition N + + + + + + + + + + + + + + + + + + + SPM + k-way merge + + + + + background prefill + + consumer drains on demand (no I/O stall) + + buffered batch + diff --git a/content/images/sort-pushdown/desc_walk_file.svg b/content/images/sort-pushdown/desc_walk_file.svg new file mode 100644 index 00000000..7e411774 --- /dev/null +++ b/content/images/sort-pushdown/desc_walk_file.svg @@ -0,0 +1,80 @@ + + + + + + + + + File-level reorder (per partition) + + + + + + + BEFORE + + group 0 + + file_d + [10..20) + + file_b + [50..60) + + group 1 + + file_c + [0..10) + + file_a + [30..40) + + + + reorder files by + min/max statistics + + + AFTER + + group 0 + + file_b + [50..60) + + file_d + [10..20) + + group 1 + + file_a + [30..40) + + file_c + [0..10) + + + high-value files come first + → TopK heap fills fast + → threshold tightens + + + + + + diff --git a/content/images/sort-pushdown/desc_walk_rg.svg b/content/images/sort-pushdown/desc_walk_rg.svg new file mode 100644 index 00000000..945ce29c --- /dev/null +++ b/content/images/sort-pushdown/desc_walk_rg.svg @@ -0,0 +1,61 @@ + + + + + + + + + Row-group early stop (inside one file) + + + + + + + Original + + [10..20) + + [50..60) + + [0..10) + + [30..40) + + + + reorder by min/max + statistics, then reverse + + + Decoder reads + + [50..60) + + [30..40) + + [10..20) + + [0..10) + + + high-value row groups first → heap fills fast + ✓ later row groups short-circuit projection columns + (RowSelection empty → arrow-rs selects_any()) + ✗ but the filter column is still read for each row group + (the sort column is fetched + decoded to check the threshold) + diff --git a/content/images/sort-pushdown/future_page_level.png b/content/images/sort-pushdown/future_page_level.png new file mode 100644 index 00000000..e41f781b Binary files /dev/null and b/content/images/sort-pushdown/future_page_level.png differ diff --git a/content/images/sort-pushdown/phase1-file-reorder.svg b/content/images/sort-pushdown/phase1-file-reorder.svg new file mode 100644 index 00000000..00d8e24a --- /dev/null +++ b/content/images/sort-pushdown/phase1-file-reorder.svg @@ -0,0 +1,88 @@ + + + + + + + + + + File rearrangement by min/max statistics + + + Before — directory order: + + + + a.parquet + ts ∈ [200, 300] + + + b.parquet + ts ∈ [100, 200] + + + c.parquet + ts ∈ [0, 100] + + validated_output_ordering() = Unsupported + → Sort required + + + + PushdownSort + sort by min(ts) + + + After — sorted by stats: + + + c.parquet + ts ∈ [0, 100] + + + b.parquet + ts ∈ [100, 200] + + + a.parquet + ts ∈ [200, 300] + + validated_output_ordering() = Exact + → Sort removed + + + Range layout + + + + + + 0 + 100 + 200 + 300 + + + + c + + b + + a + + Non-overlapping → ordering provable + + + SELECT * FROM events ORDER BY ts + diff --git a/content/images/sort-pushdown/phase2-stats-overlap.svg b/content/images/sort-pushdown/phase2-stats-overlap.svg new file mode 100644 index 00000000..8580c210 --- /dev/null +++ b/content/images/sort-pushdown/phase2-stats-overlap.svg @@ -0,0 +1,89 @@ + + + + Using min/max statistics to prove non-overlap + + + + Non-overlapping ranges + + + file_c [0..100] + + + + file_b [100..200] + + + + file_a [200..300] + + + + + + + + + 0 + 100 + 200 + 300 + min(ts) / max(ts) + + + + + + Ordering: Exact ✓ + Sort can be removed + + + + Overlapping ranges + + + file_x [0..180] + + + + file_y [80..260] + + + + file_z [140..300] + + + + + + + + + + + + + 0 + 100 + 200 + 300 + min(ts) / max(ts) + + Ordering: Inexact (overlap) + Sort stays + diff --git a/content/images/sort-pushdown/plan-diff.svg b/content/images/sort-pushdown/plan-diff.svg new file mode 100644 index 00000000..8fdd8bc4 --- /dev/null +++ b/content/images/sort-pushdown/plan-diff.svg @@ -0,0 +1,53 @@ + + + + + + + + + EXPLAIN before / after sort pushdown (single-partition) + + + + Before — SortExec on top of scan + + + SortExec: TopK(fetch=3) + expr=[ts@0 ASC NULLS LAST] + preserve_partitioning=[false] + + + + + DataSourceExec + file_groups={a, b, c} + file_type=parquet + + blocking full sort across the scan + LIMIT applied after sort + + + + After — SortExec eliminated + + + DataSourceExec + file_groups={c, b, a} + limit=3 + output_ordering=[ts ASC NULLS LAST] + + files reordered by min/max stats + LIMIT becomes a static fetch on the source + reader stops the moment 3 rows are emitted + no blocking sort, no full scan + diff --git a/content/images/sort-pushdown/pr21956-decision.svg b/content/images/sort-pushdown/pr21956-decision.svg new file mode 100644 index 00000000..f03aa853 --- /dev/null +++ b/content/images/sort-pushdown/pr21956-decision.svg @@ -0,0 +1,87 @@ + + + + + + + + + + + + try_pushdown_sort — Exact / Inexact / Unsupported + + + + PushdownSort rule + source.try_pushdown_sort(req) + + + + + + + Natural ordering satisfies request? + eq.ordering_satisfy(req) + + + + yes + + + Exact + drop SortExec + + + + no + + + + Pushdown still applies? + column_in_file_schema + OR reversed_satisfies(req) + + + + yes + + + Inexact + set runtime-reorder flags + + + + no + + + + Unsupported + SortExec stays · full external sort + + + + Outcomes + + Exact · static limit on source + + Inexact · TopK + RG pruner + + Unsupported · no benefit + + diff --git a/content/images/sort-pushdown/pr21956-runtime-pipeline.svg b/content/images/sort-pushdown/pr21956-runtime-pipeline.svg new file mode 100644 index 00000000..1218af12 --- /dev/null +++ b/content/images/sort-pushdown/pr21956-runtime-pipeline.svg @@ -0,0 +1,71 @@ + + + + + + + + + + + ParquetSource carries the inexact-pushdown decision + set by try_pushdown_sort, read by the opener below + + + + + + + + 1 + File-level reorder + across the whole scan + Reorder files by min(col) so the most-promising file opens first + + + + for each opened file + + + + + 2 + Row-group-level reorder + per opened file + Sort row groups ASC by min(col) from the parquet column statistics + + + + if reverse_row_groups + + + + + 3 + Reverse iteration + DESC requests only + Iterate row-group list in reverse → "row groups DESC × rows ASC" + + + + + + + Decoder reads row groups in this order + approximate ordering — the Sort / TopK above the source still enforces final ordering + diff --git a/content/images/sort-pushdown/pruner_loop.png b/content/images/sort-pushdown/pruner_loop.png new file mode 100644 index 00000000..3852b0c3 Binary files /dev/null and b/content/images/sort-pushdown/pruner_loop.png differ diff --git a/content/images/sort-pushdown/pruning_stack.svg b/content/images/sort-pushdown/pruning_stack.svg new file mode 100644 index 00000000..f59decda --- /dev/null +++ b/content/images/sort-pushdown/pruning_stack.svg @@ -0,0 +1,84 @@ + + + + + + + + + Three-layer pruning · file + row group + row + + + + TopK heap threshold + pushes DynamicFilter to all three layers + + + + + File-level dynamic prune + file_pruner + EarlyStoppingStream + + + skipped file + + + Skipped File ✓ done + ✗ no metadata load + ✗ no row-group decode + + + surviving file + + + + + Row-group prune (#22450) + every row-group boundary + + + pruned RG + + + Pruned RG ✓ done + ✗ no byte fetch + ✗ no decode · ✗ no RowFilter eval + + + surviving RG + + + + + Row-level RowFilter + per-row · decode sort col + eval + + + reject row + + + Reject row ✓ partial savings + ✗ skip other projection cols + ✗ skip their column IO + (sort col was decoded to evaluate) + + + pass row + + + + + Pass row · full work + ✓ decode every projection col + ✓ materialize batch upward + diff --git a/content/images/sort-pushdown/reverse-scan.svg b/content/images/sort-pushdown/reverse-scan.svg new file mode 100644 index 00000000..443a0a1c --- /dev/null +++ b/content/images/sort-pushdown/reverse-scan.svg @@ -0,0 +1,100 @@ + + + + + + + + + + + + ORDER BY ts DESC LIMIT 10 — row-group reverse vs page reverse + + + + Row-group reverse (today, merged) + + + + RowGroup (last, ~128 MB) + + + + + + + + + + + + P0 + P1 + P2 + P3 + P4 + P5 + P6 + P7 + + + Decode the entire row group, reverse in memory, take 10. + Peak buffer: ~128 MB + Pages decoded: 8 + Time-to-first-N: ~29 µs + + + + + + + + + + + + + + Page reverse (upstream POC, arrow-rs #9937) + + + RowGroup (last) + + + + + + + + + + + P0 + P1 + P2 + P3 + P4 + P5 + P6 + P7 + + + + + + Seek to last page only via OffsetIndex, decode, reverse, return. + Peak buffer: ~1 MB + Pages decoded: 1 + Time-to-first-N: ~565 ns (≈ 50× faster) + diff --git a/content/images/sort-pushdown/rg_cascade.svg b/content/images/sort-pushdown/rg_cascade.svg new file mode 100644 index 00000000..dc13616b --- /dev/null +++ b/content/images/sort-pushdown/rg_cascade.svg @@ -0,0 +1,113 @@ + + + + + + + + + Cascading prune · how the heap prunes row groups + ORDER BY x DESC LIMIT 10 · one parquet file, 10 row groups + + + + ① heap fills + + ② threshold climbs + + ③ cascading prune at next row-group boundary + + + + + + + + + + value range ↑ + scan time → + + + + threshold + + + + + Row group 9 + [90..100) + + + + + Row group 8 + [80..90) + + + + + Row group 7 + [70..80) + + + + + Row group 6 + [60..70) + + + + + Row group 5 + [50..60) + + + + + Row group 4 + [40..50) + + + + + Row group 3 + [30..40) + + + + + Row group 2 + [20..30) + + + + + Row group 1 + [10..20) + + + + + Row group 0 + [0..10) + + + Row group 9 alone fills the heap of size 10, so the threshold rises into its range (≥ 90). + At the next row-group boundary, every remaining row group with max < 90 is pruned in one pass — bytes never fetched. + diff --git a/content/images/sort-pushdown/topk_tpch_bench.svg b/content/images/sort-pushdown/topk_tpch_bench.svg new file mode 100644 index 00000000..da65fd17 --- /dev/null +++ b/content/images/sort-pushdown/topk_tpch_bench.svg @@ -0,0 +1,121 @@ + + + + topk_tpch (TPC-H SF1, ORDER BY … LIMIT 100) — main vs runtime row-group pruning + + + + + + + + + + + + 0 + 10 + 20 + 30 + 40 + 50 + 60 + 70 + Median time (ms, lower is better) + + + main + this PR + this PR — ≥2× faster + + + + + 2.6 + Q1 + + + + + 11.3 + + 3.15× + Q2 + + + + + 32.1 + Q3 + + + + + 12.2 + + 3.78× + Q4 + + + + + 10.1 + Q5 + + + + + 17.9 + Q6 + + + + + 38.1 + Q7 + + + + + 28.6 + + 3.99× + Q8 + + + + + 36.9 + + 4.34× + Q9 + + + + + 55.3 + + 4.25× + Q10 + + + + + 3.7 + Q11 + + + 5 of 11 queries 3–4× faster · 0 regressions · total 248.8 → 139.1 ms (−44%) + diff --git a/content/images/sort-pushdown/transition_anatomy.png b/content/images/sort-pushdown/transition_anatomy.png new file mode 100644 index 00000000..37a38b52 Binary files /dev/null and b/content/images/sort-pushdown/transition_anatomy.png differ