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

Is your feature request related to a problem or challenge?

Ungrouped MIN/MAX already push a dynamic filter into parquet scans (enable_aggregate_dynamic_filter_pushdown), but they do not ask the scan to visit files / row groups in best-first order.

ORDER BY col DESC LIMIT 1 does, via PushdownSort → ParquetSource::try_pushdown_sort (sort_order_for_reorder, reverse_row_groups, dynamic RG pruning). PushdownSort only runs when a SortExec is present, so SELECT MAX(col) FROM t WHERE ... never takes that path.

On unsorted parquet (no WITH ORDER / Exact output ordering), that visit-order gap dominates. A real workload (timestamp column, extra equality WHERE, Inexact file stats) measured:

MAX returned fewer rows and was still ~3× slower because it walked listing order. Dynamic prune cannot help until a large value has been seen.

A residual FilterExec makes this worse: FilterExec does not implement try_pushdown_sort (default Unsupported), so even TopK loses reorder if the predicate is not fully absorbed into the scan. Filters are order-preserving; Projection / CoalesceBatches already delegate.

Describe the solution you'd like

Keep AggregateExec. Do not SQL-rewrite MAX to LIMIT 1.

1. Ungrouped single-column MIN/MAX (optional WHERE, no other aggs that would make a single order illegal): attach an Inexact sort request to the input scan, using the same try_pushdown_sort pipeline parquet already has.
   • MAX(col) → [col DESC NULLS LAST] (SQL MAX skips nulls)
   • MIN(col) → [col ASC NULLS LAST]
2. Combined with the existing aggregate dynamic filter, the scan should see high-max (or low-min) row groups first and prune the rest.
3. FilterExec should be transparent for try_pushdown_sort (delegate to children), so a residual predicate does not drop reorder for TopK or min/max.

AggregateStatistics (Exact min/max → literal, no scan) should remain preferred when stats are Exact. This request is for the Inexact parquet / Delta path.

Describe alternatives you've considered

Rewrite MAX(col) to SELECT col ... ORDER BY col DESC NULLS LAST LIMIT 1. Rejected as the primary approach:

• Empty input: MAX returns one NULL row; LIMIT 1 returns zero rows.
• Bare ORDER BY col DESC is NULLS FIRST under default nulls_max; MAX is skip-nulls / NULLS LAST.
• SELECT MAX(a), MIN(b), COUNT(*) cannot share one TopK.

A rewrite could still be used internally if empty-input and null-ordering are preserved (e.g. keep AggregateExec on top of a fetch-1 scan). Pushing sort into the scan under AggregateExec is the smaller, equivalent change.

Rely only on aggregate dynamic filters. Already implemented; insufficient without visit order, as in the table above.

Document “write ORDER BY ... LIMIT 1 instead of MAX.” Works only when the user can change SQL and accept LIMIT empty-set semantics.

Additional context

Relevant code:

• PushdownSort only walks SortExec (datafusion/physical-optimizer/src/pushdown_sort.rs)
• ParquetSource::try_pushdown_sort Inexact reorder (datafusion/datasource-parquet/src/source.rs)
• Ungrouped min/max dynamic filter init (AggregateExec::init_dynamic_filter in datafusion/physical-plan/src/aggregates/mod.rs)
• FilterExec has no try_pushdown_sort override (datafusion/physical-plan/src/filter.rs; default in ExecutionPlan is Unsupported)

COUNT(*) / mixed aggregates should stay out of scope unless every aggregate is compatible with one order.

Willingness to contribute

I would like to contribute to this feature, but request guidance from the DataFusion community