Skip to content

Push down sort order for ungrouped MIN/MAX aggregates and through FilterExec - #27450

Draft
galaxytemple wants to merge 4 commits into
apache:mainfrom
galaxytemple:feature-26809-filter-sort-pushdown
Draft

galaxytemple wants to merge 4 commits into
apache:mainfrom
galaxytemple:feature-26809-filter-sort-pushdown

Conversation

@galaxytemple

@galaxytemple galaxytemple commented Nov 15, 2026 •

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Rationale for this change

Ungrouped MIN(col) / MAX(col) queries with a filter (e.g., SELECT MAX(ts) FROM t WHERE val > 0) and TopK queries over a FilterExec currently scan Parquet files and row groups in their original listing order rather than statistics order:

  1. FilterExec did not implement try_pushdown_sort: Even though FilterExec preserves input ordering, it did not forward sort pushdown requests to its input, blocking sort pushdown whenever a FilterExec remained above the scan.
  2. PushdownSort only matched SortExec: Ungrouped MIN/MAX AggregateExec nodes emit a dynamic filter (AggrDynFilter, introduced in optimizer: Support dynamic filter in MIN/MAX aggregates #18644), but without a scan-order hint from PushdownSort, files and row groups were scanned in listing order rather than statistics order, preventing the dynamic filter bound from tightening early.

Another possible approach was rewriting SELECT MAX(col) into ORDER BY col DESC LIMIT 1 in the logical optimizer. However, LIMIT 1 returns 0 rows on empty input (while MAX returns a [NULL] row). And using the existing AggregateExec dynamic filter and try_pushdown_sort felt much simpler.

What changes are included in this PR?

  • Implement FilterExec::try_pushdown_sort in datafusion/physical-plan/src/filter.rs:
    • Forward the sort request to the input and rebuild FilterExec via FilterExecBuilder::from(self).with_input(new_input).
    • Remap sort columns through self.projection using FilterRemapper::with_column_mapping (same as FilterExec::handle_child_pushdown_result).
    • Return Unsupported if self.fetch.is_some().
  • In PushdownSort, push [col DESC NULLS LAST] for ungrouped MAX(col) and [col ASC NULLS LAST] for MIN(col) down to the aggregate input when dynamic filter pushdown is enabled and the input has no fetch.
  • Updated existing sqllogictest EXPLAIN snapshots where WHERE + ORDER BY queries now push sort order down to Parquet scans through FilterExec (sort_order_for_reorder=[...]).

What is the testing strategy for this PR?

  • Added snapshot tests in pushdown_sort.rs using TestScan:
    • Sort pushdown through FilterExec (TopK).
    • Sort pushdown through FilterExec with column projection remapping.
    • Sort pushdown for ungrouped MAX and MIN aggregates.
  • Tested locally by generating dummy 50 Parquet files (10M rows total, 200k rows per file in ascending ts order) and running EXPLAIN ANALYZE:
    • SELECT MAX(ts) FROM t WHERE val > 0: 2.78s → 29ms (reads file_050 first and skips the remaining 49 files via dynamic filter).
    • SELECT ts FROM t WHERE val > 0 ORDER BY ts DESC LIMIT 1: 2.75s → 34ms.

Are there any user-facing changes?

No API changes.

Notes

  1. Existing sqllogictests:
    Several existing .slt tests (e.g., push_down_filter_regression.slt) had their EXPLAIN physical plans updated to include sort_order_for_reorder=[...].
    Previously, when a FilterExec was placed above the scan, sort pushdown was blocked. But now that FilterExec::try_pushdown_sort is implemented, sort requirements are forwarded to the scan.

  2. AggregateMode::Single
    Initially, I also considered enabling dynamic filter for AggregateMode::Single as well (since single-partition plans can still scan across multiple Parquet files or row groups).
    However, seeing that test_aggregate_dynamic_filter_not_created_for_single_mode explicitly asserts against creating dynamic filters in Single mode, and allowing Single is slightly out of scope for the original issue. I kept init_dynamic_filter restricted to AggregateMode::Partial to preserve existing design.

@github-actions github-actions Bot added optimizer Optimizer rules core Core DataFusion crate physical-plan Changes to the physical-plan crate labels Nov 15, 2026
@galaxytemple
galaxytemple force-pushed the feature-26809-filter-sort-pushdown branch from 25afd40 to 55d3b82 Compare November 15, 2026 03:28
@galaxytemple
galaxytemple force-pushed the feature-26809-filter-sort-pushdown branch from 55d3b82 to 0985821 Compare November 15, 2026 03:35
@galaxytemple
galaxytemple force-pushed the feature-26809-filter-sort-pushdown branch from 0985821 to 335a608 Compare November 15, 2026 03:41
@github-actions github-actions Bot added the sqllogictest SQL Logic Tests (.slt) label Nov 15, 2026
@galaxytemple
galaxytemple force-pushed the feature-26809-filter-sort-pushdown branch from a3249a2 to 00c1dde Compare November 15, 2026 04:48
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Core DataFusion crate optimizer Optimizer rules physical-plan Changes to the physical-plan crate sqllogictest SQL Logic Tests (.slt)

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Push sort order into parquet scans for ungrouped MIN/MAX (same visit order as TopK)

1 participant