Skip to content

Establish the high level API for sort pushdown and the optimizer rule and support reverse files and row groups - #19064

Merged
zhuqi-lucas merged 33 commits into
apache:mainfrom
zhuqi-lucas:issue_19059
Dec 17, 2025
Merged

zhuqi-lucas merged 33 commits into
apache:mainfrom
zhuqi-lucas:issue_19059

Conversation

@zhuqi-lucas

@zhuqi-lucas zhuqi-lucas commented Dec 3, 2025 •

Copy link
Copy Markdown
Contributor

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?

…. Only re-arrange files and row groups and return Inexact.
@github-actions github-actions Bot added documentation Improvements or additions to documentation optimizer Optimizer rules core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) common Related to common crate execution Related to the execution crate proto Related to proto crate datasource Changes to the datasource crate physical-plan Changes to the physical-plan crate labels Dec 3, 2025
@zhuqi-lucas
zhuqi-lucas marked this pull request as ready for review December 3, 2025 09:53
Copilot AI review requested due to automatic review settings December 3, 2025 09:53
@zhuqi-lucas zhuqi-lucas changed the title Draft Establish the high level API for sort pushdown and the optimizer rule. Only re-arrange files and row groups and return Inexact. Establish the high level API for sort pushdown and the optimizer rule and support reverse files and row groups Dec 3, 2025

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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_pushdown configuration option (default: true) to control the optimization
  • Implemented the optimizer rule PushdownSort that 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_inexact flag 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.

Comment thread datafusion/execution/src/config.rs Resolved
Comment thread datafusion/execution/src/config.rs Resolved
Comment thread datafusion/common/src/config.rs Resolved
Comment thread datafusion/common/src/config.rs Resolved
Comment thread datafusion/datasource/src/file.rs Resolved
@zhuqi-lucas

Copy link
Copy Markdown
Contributor Author

I addressed all comments now, cc @alamb @adriangb.

And the latest run benchmark sql_planner shows no regression.

order: &[PhysicalSortExpr],
) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
// RepartitionExec only maintains input order if preserve_order is set
// or if there's only one partition

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the single partition case handled?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes - maintains_input_order_helper returns true for single partitions, so the check passes and sort pushdown works correctly.

Comment on lines +214 to +216
// Default: clone self without modification
// ParquetSource will override this
not_impl_err!("with_file_ordering_info not implemented for this FileSource")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The comment doesn't match the implementation

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch @adriangb , fixed the comments in latest PR, thanks!

@zhuqi-lucas

Copy link
Copy Markdown
Contributor Author

Thank you @adriangb for review! I will wait for @alamb to double check before merging it, thanks!

@alamb alamb mentioned this pull request Dec 16, 2025
32 of 41 tasks
@alamb

alamb commented Dec 16, 2025

Copy link
Copy Markdown
Contributor

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

@zhuqi-lucas
zhuqi-lucas added this pull request to the merge queue Dec 17, 2025
Merged via the queue into apache:main with commit 3a41cc6 Dec 17, 2025
14 of 15 checks passed
@zhuqi-lucas

Copy link
Copy Markdown
Contributor Author

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!

@adriangb

Copy link
Copy Markdown
Contributor

Thank you for this amazing piece of work! Very excited to see where this goes next.

@alamb

alamb commented Dec 17, 2025

Copy link
Copy Markdown
Contributor

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))

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the sort has fetch, can we directly remove the sort?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes @xudong963 , we will implement the remove sort part in phase 2, here is the ticket:
in #19329

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This PR implemented the inexact mode, which has similar performance from sorted data clickbench.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the sort operator will be removed in the enforce_sorting if the required ordering is matched.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

github-merge-queue Bot pushed a commit that referenced this pull request Dec 29, 2025
## 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]>
alamb pushed a commit to alamb/datafusion that referenced this pull request Nov 11, 2026
## 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.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

common Related to common crate core Core DataFusion crate datasource Changes to the datasource crate documentation Improvements or additions to documentation execution Related to the execution crate optimizer Optimizer rules physical-expr Changes to the physical-expr crates physical-plan Changes to the physical-plan crate proto Related to proto crate sqllogictest SQL Logic Tests (.slt)

Projects

None yet

7 participants