Repository navigation
Establish the high level API for sort pushdown and the optimizer rule and support reverse files and row groups - #19064
Conversation
…. Only re-arrange files and row groups and return Inexact.
There was a problem hiding this comment.
Pull request overview
This PR establishes the high-level API for sort pushdown optimization in DataFusion, specifically targeting Parquet files. The optimization improves TopK query performance by reordering files and row groups based on statistics, though it returns inexact ordering (keeping the Sort operator for correctness while enabling early termination).
Key Changes:
- Added
enable_sort_pushdownconfiguration option (default: true) to control the optimization - Implemented the optimizer rule
PushdownSortthat detects SortExec nodes and pushes sort requirements down to data sources - Extended ExecutionPlan, DataSource, and FileSource traits with
try_pushdown_sort()methods to support the optimization - Added
reverse_scan_inexactflag to ParquetSource to reverse row group read order when beneficial
Reviewed changes
Copilot reviewed 25 out of 30 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
docs/source/user-guide/configs.md |
Added documentation for the new enable_sort_pushdown configuration option |
datafusion/sqllogictest/test_files/topk.slt |
Updated test expectation to show reverse_scan_inexact=true in explain output |
datafusion/sqllogictest/test_files/spark/bitwise/bit_count.slt |
Fixed line number formatting in test file |
datafusion/sqllogictest/test_files/slt_features.slt |
Fixed line number formatting in test file |
datafusion/sqllogictest/test_files/information_schema.slt |
Added enable_sort_pushdown to configuration schema output |
datafusion/sqllogictest/test_files/explain.slt |
Updated test expectations to show PushdownSort optimizer phase |
datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt |
Added comprehensive integration tests for sort pushdown with various scenarios |
datafusion/sqllogictest/test_files/create_external_table.slt |
Updated test expectation to show reverse_scan_inexact=true |
datafusion/proto/src/logical_plan/file_formats.rs |
Added enable_sort_pushdown field to protobuf serialization (with hardcoded value bug) |
datafusion/proto/src/generated/datafusion_proto_common.rs |
Added enable_sort_pushdown field to generated protobuf structures |
datafusion/proto-common/src/to_proto/mod.rs |
Implemented protobuf serialization for enable_sort_pushdown (with hardcoded value bug) |
datafusion/proto-common/src/generated/prost.rs |
Added enable_sort_pushdown field to generated protobuf structures |
datafusion/proto-common/src/generated/pbjson.rs |
Added JSON serialization support for enable_sort_pushdown |
datafusion/proto-common/src/from_proto/mod.rs |
Implemented protobuf deserialization for enable_sort_pushdown (with hardcoded value bug) |
datafusion/proto-common/proto/datafusion_common.proto |
Added enable_sort_pushdown field to protobuf schema |
datafusion/physical-plan/src/execution_plan.rs |
Added try_pushdown_sort() trait method to ExecutionPlan |
datafusion/physical-optimizer/src/pushdown_sort.rs |
Implemented the core PushdownSort optimizer rule |
datafusion/physical-optimizer/src/optimizer.rs |
Registered PushdownSort optimizer in the optimization pipeline |
datafusion/physical-optimizer/src/lib.rs |
Exported the new pushdown_sort module |
datafusion/execution/src/config.rs |
Added session config method (with naming and documentation issues) |
datafusion/datasource/src/source.rs |
Added try_pushdown_sort() trait method to DataSource and DataSourceExec |
datafusion/datasource/src/file_scan_config.rs |
Implemented sort pushdown logic for FileScanConfig including file reversal |
datafusion/datasource/src/file.rs |
Added SortOrderPushdownResult enum and try_pushdown_sort() trait method to FileSource |
datafusion/datasource-parquet/src/source.rs |
Implemented sort pushdown for ParquetSource with reverse_scan_inexact flag |
datafusion/datasource-parquet/src/opener.rs |
Implemented row group reversal logic in ParquetOpener |
datafusion/core/tests/physical_optimizer/test_utils.rs |
Added OptimizationTest harness for testing physical optimizers |
datafusion/core/tests/physical_optimizer/pushdown_sort.rs |
Added comprehensive unit tests for the PushdownSort optimizer |
datafusion/core/tests/physical_optimizer/mod.rs |
Registered the pushdown_sort test module |
datafusion/common/src/file_options/parquet_writer.rs |
Added enable_sort_pushdown field to ParquetOptions destructuring |
datafusion/common/src/config.rs |
Added enable_sort_pushdown configuration field (with indentation issue) |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>
| order: &[PhysicalSortExpr], | ||
| ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> { | ||
| // RepartitionExec only maintains input order if preserve_order is set | ||
| // or if there's only one partition |
There was a problem hiding this comment.
Is the single partition case handled?
There was a problem hiding this comment.
Yes - maintains_input_order_helper returns true for single partitions, so the check passes and sort pushdown works correctly.
| // Default: clone self without modification | ||
| // ParquetSource will override this | ||
| not_impl_err!("with_file_ordering_info not implemented for this FileSource") |
There was a problem hiding this comment.
The comment doesn't match the implementation
There was a problem hiding this comment.
Good catch @adriangb , fixed the comments in latest PR, thanks!
Let me know if you want me to double check anything specific. Otherwise don't feel like you have to wait for me -- I am not likely to be able to get back here for the next few days BTW I also added this to the list of features to highlight for 52 n |
Thank you @alamb and @adriangb for the review and discussion! I've merged the PR now. If any issues come up, I'll address them promptly. Thanks again! |
|
Thank you for this amazing piece of work! Very excited to see where this goes next. |
|
100% agree -- this is so cool |
| match sort_input.try_pushdown_sort(required_ordering)? { | ||
| SortOrderPushdownResult::Exact { inner } => { | ||
| // Data source guarantees perfect ordering - remove the Sort operator | ||
| Ok(Transformed::yes(inner)) |
There was a problem hiding this comment.
If the sort has fetch, can we directly remove the sort?
There was a problem hiding this comment.
Yes @xudong963 , we will implement the remove sort part in phase 2, here is the ticket:
in #19329
There was a problem hiding this comment.
This PR implemented the inexact mode, which has similar performance from sorted data clickbench.
There was a problem hiding this comment.
I think the sort operator will be removed in the enforce_sorting if the required ordering is matched.
There was a problem hiding this comment.
Yes agreed. I think what we ultimately need to do is return an ExecutionPlan that claims to satisfy the ordering and let the existing optimizer rule remove the sort for us.
There was a problem hiding this comment.
Yes, i was testing the exact reverse cases, the enforce_sorting is not enough for the exact reverse, but it's ok for current main branch.
## Which issue does this PR close? - Closes #19403 ## Rationale for this change After #19064 was completed, I found it couldn't meet our internal project requirements: 1. **Redesign try_reverse_output using EquivalenceProperties** - The previous implementation used a simple `is_reverse` method that could only handle basic reverse matching - Now leveraging `EquivalenceProperties` can handle more complex scenarios, including constant column elimination and monotonic functions 2. **Switch to ordering_satisfy method for ordering matching** - This method internally: - Normalizes orderings (removes constant columns) - Checks monotonic functions (like `date_trunc`, `CAST`, `CEIL`) - Handles prefix matching 3. **Extend sort pushdown support to more operators** - Added `try_pushdown_sort` implementation for `ProjectionExec`, `FilterExec`, `CooperativeExec` - These operators can now pass sort requests down to their children ## What changes are included in this PR? ### Core Changes: 1. **ParquetSource::try_reverse_output** (datasource-parquet/src/source.rs) - Added `eq_properties` parameter - Reverses all orderings in equivalence properties - Uses `ordering_satisfy` to check if reversed ordering satisfies the request - Removed `file_ordering` field and `with_file_ordering_info` method 2. **FileSource trait** (datasource/src/file.rs) - Updated `try_reverse_output` signature with `eq_properties` parameter - Added detailed documentation explaining parameter usage and examples 3. **FileScanConfig::try_pushdown_sort** (datasource/src/file_scan_config.rs) - Simplified logic to directly call `file_source.try_reverse_output` - No longer needs to pre-check ordering satisfaction or set file ordering info 4. **New operator support** - `FilterExec::try_pushdown_sort` - Pushes sort below filters - `ProjectionExec::try_pushdown_sort` - Pushes sort below projections - `CooperativeExec::try_pushdown_sort` - Supports sort pushdown in cooperative execution 5. **Removed obsolete methods** - Deleted `LexOrdering::is_reverse` - replaced by `ordering_satisfy` ### Feature Enhancements: **Supported optimization scenarios:** 1. **Constant column elimination** (Test 7) ```sql -- File ordering: [timeframe ASC, period_end ASC] -- Query: WHERE timeframe = 'quarterly' ORDER BY period_end DESC -- Effect: After timeframe becomes constant, reverse scan is enabled ``` 2. **Monotonic function support** (Test 8) ```sql -- File ordering: [ts ASC] -- Query: ORDER BY date_trunc('month', ts) DESC -- Effect: date_trunc is monotonic, reverse scan satisfies the request ``` ## Are these changes tested? Yes, comprehensive tests have been added: - **Test 7 (237 lines)**: Constant column elimination scenarios - Single constant column filter - Multi-value IN clauses (doesn't trigger optimization) - Literal constants in sort expressions - Non-leading column filters (edge cases) - **Test 8 (355 lines)**: Monotonic function scenarios - `date_trunc` (date truncation) - `CAST` (type conversion) - `CEIL` (ceiling) - `ABS` (negative case - not monotonic over mixed positive/negative range) All tests verify: - Presence of `reverse_row_groups=true` in physical plans - Correctness of query results ## Are there any user-facing changes? **API Changes:** - `FileSource::try_reverse_output` signature changed (added `eq_properties` parameter) - Removed `FileSource::with_file_ordering_info` method - Removed `LexOrdering::is_reverse` public method **User-visible improvements:** - More queries can leverage reverse row group scanning for optimization - Especially queries with `WHERE` clauses that make certain columns constant - Queries using monotonic functions (like date functions, type conversions) **Note:** This PR returns `Inexact` results because only row group order is reversed, not row order within row groups. Future enhancements could include: - File reordering based on statistics (returning `Exact`) - Partial sort pushdown for prefix matches --------- Co-authored-by: Adrian Garcia Badaracco <[email protected]>
## Which issue does this PR close? - Part of apache#17172 (Part 1, order inversion). Follows apache#19064 (`Inexact` reverse scan) and apache#19329 (`Exact` for the natural ordering). - Related: apache#19048 (row group size and memory for the reverse scan). - Closes apache#27194. ## Rationale for this change When a query asks for the reverse of the files' declared ordering (for example `ORDER BY ts DESC LIMIT 10` on files written in `ts ASC`), the parquet scan already reverses the row group order, but rows inside each row group stay in on-disk order. The pushdown is therefore `Inexact` and a `SortExec` (usually `TopK`) stays above the scan, which has to consume at least one full row group per file before it can emit anything. Reading each row group back to front makes the scan exactly ordered, so the `SortExec` can go and the `LIMIT` can be pushed into the scan. ## What changes are included in this PR? - New option `datafusion.execution.parquet.enable_exact_reverse_scan` (default `false`). - `ParquetSource::try_pushdown_sort` returns `Exact` when the reversed declared ordering satisfies the request and the option is set. The source sets a new `reverse_rows` flag next to `reverse_row_groups`, and does not reorder row groups by statistics, since that is not order-preserving for row groups that share a `min`. - The parquet stream drains each row group's reader, reverses the rows of every batch, and emits the batches last to first. At most one decoded row group (after projection and filtering) is held at a time. - A limit is counted in decode order, which the reversal does not follow, so a reversing scan does not pass it to the decoder or use it for row group / page pruning; the file stream applies it to the reversed output. Under a limit the scan reads less instead: - a row group whose output is known up front (no predicate, or fully matched by statistics) is cut to its last `limit` rows with a row selection, and the offset index is loaded so the pages ahead of them are not fetched; - for other row groups only the last `limit` decoded rows are kept, so at most `limit` rows plus one batch are held. - The buffered rows are charged to the task's memory pool through a new `FileSource::with_memory_pool` hook (default: no accounting; `FileScanConfig` calls it when the scan is opened). An oversized row group fails the scan with a resources error instead of growing the process. - `FileScanConfig::rebuild_with_source` reverses the files of each group, declares the reversed orderings, and re-validates them, falling back to `Inexact` when they do not hold (for example overlapping files). For a source that is `Exact` but whose request is not literally the reverse of the declared ordering, the forward `Exact` path now also checks that the kept ordering satisfies the request, so such a claim is downgraded instead of keeping an ordering that does not describe the scan. - `reverse_rows` is carried in `ParquetScanExecNode` and the option in `ParquetOptions` proto. - `PushdownSort` and `FilterPushdown` treat a node with a fetch as a barrier: a fetch keeps the first rows in its input's current order, so the sort rule does not reorder the input below it, and the filter rule does not move a filter from above into or below it (the node's own filters still push down). This replaces scan-specific limit checks and fixes apache#27194. ## Are these changes tested? - Opener: rows reversed within row groups spanning several batches, limit not applied in decode order, pushed-down filter before reversal, limit with a filter (selected tail for the fully matched row group, kept tail for the other), fewer bytes fetched under a limit, resources error from a small pool, reservation released after the scan. - `PreparedAccessPlan::select_tail_for_limit`: tail selection, spanning row groups, stopping at a row group whose output is unknown or that already has a selection. - `ParquetSource::try_pushdown_sort`: `Exact` with the option, `Inexact` without. - `FileScanConfig`: files and ordering reversed, overlapping files downgraded, an `Exact` claim without file reversal downgraded. - Proto round trip of a plan with `reverse_rows`. - `sort_pushdown_exact_reverse.slt`: single and multiple files, filter, overlapping files, multi-column prefix, NULLs, NOT NULL `DESC NULLS LAST`, two partitions, option off. - The three queries from apache#27194 in `sort_pushdown_exact_reverse.slt` (test 16), with the option off. ### Benchmark `sort_pushdown_sorted` (TPC-H SF1 `lineitem`, 3 files sorted by `l_orderkey`, `WITH ORDER`), median of 5 iterations, this branch with the option off vs on (run with `dfbench sort-pushdown --sorted` as in `bench.sh run sort_pushdown_sorted`): | Query | | off | on | | |---|---|---:|---:|---:| | q1 | 3 cols, `ORDER BY l_orderkey` | 97.62 ms | 96.48 ms | 1.01x | | q2 | 3 cols, `ORDER BY l_orderkey LIMIT 100` | 2.20 ms | 2.11 ms | 1.04x | | q3 | `*`, `ORDER BY l_orderkey` | 313.01 ms | 313.18 ms | 1.00x | | q4 | `*`, `ORDER BY l_orderkey LIMIT 100` | 4.74 ms | 4.76 ms | 1.00x | | q5 | 3 cols, `DESC LIMIT 100` | 7.65 ms | 2.08 ms | 3.68x | | q6 | 3 cols, `DESC LIMIT 1000` | 10.50 ms | 2.18 ms | 4.82x | | q7 | `*`, `DESC LIMIT 100` | 13.61 ms | 3.99 ms | 3.41x | | q8 | `*`, `DESC LIMIT 1000` | 15.54 ms | 4.04 ms | 3.85x | The reversed `LIMIT` queries now cost about the same as their forward counterparts (q5 vs q2, q7 vs q4). The `sort_pushdown_inexact*` benchmarks are unaffected: their files overlap, so the exact path does not apply. ## Are there any user-facing changes? At the default configuration, a query whose `ORDER BY ... LIMIT` over a sorted Parquet file ends up as a scan limit, with a reversing sort or a dynamic filter above it, now returns different (correct) rows (apache#27194). Two `filter_pushdown.rs` snapshots change accordingly: an outer filter is no longer merged into a `FilterExec` with a fetch. A new config option, off by default, and a new `FileSource::with_memory_pool` method with a default implementation. With the option on, plans for reversed requests drop the `SortExec` and show `reverse_rows=true` on the scan. A scan without a limit still holds one decoded row group per partition (charged to the memory pool), which is why the option is off by default; page-level reverse decoding (apache#19486) would bound it further.
Which issue does this PR close?
Establish the high level API for sort pushdown and the optimizer rule. Only re-arrange files and row groups and return Inexact, now support reverse order case, and we don't need to cache anything for this implementation, so it's no memory overhead.
It will have huge performance improvement with dynamic topk pushdown to skip row groups.
Details:
Performance results on ClickBench sorted data: 13ms vs 300ms baseline (23x faster), close to aggressive caching approach (9.8ms) but with much better memory stability details
Rationale for this change
What changes are included in this PR?
Are these changes tested?
Are there any user-facing changes?