Method try_pushdown_sort

datafusion/datasource-parquet/src/source.rs:830–962

Try to optimize the scan to produce data in the requested sort order. Inputs: 1. The query's required ordering (`order` parameter) 2. The source's equivalence properties (`eq_properties`) # Returns - `Exact`: the source's natural ordering already satisfies the request. The surrounding `SortExec` can be eliminated provided files within each group are non-overlapping (verified by `FileScanConfig`)

(
        &self,
        order: &[PhysicalSortExpr],
        eq_properties: &EquivalenceProperties,
    )

    ///   the request (function-wrapped case), this is always `true`
    ///   because we're flipping the file's natural order.
    fn try_pushdown_sort(
        &self,
        order: &[PhysicalSortExpr],
        eq_properties: &EquivalenceProperties,
    ) -> datafusion_common::Result<SortOrderPushdownResult<Arc<dyn FileSource>>> {
        if order.is_empty() {
            return Ok(SortOrderPushdownResult::Unsupported);
        }

        // Check if the natural (non-reversed) ordering already satisfies the request.
        // Parquet metadata guarantees within-file ordering, so if the ordering matches
        // we can return Exact. FileScanConfig will verify that files within each group
        // are non-overlapping before declaring the entire scan as Exact.
        if eq_properties.ordering_satisfy(order.iter().cloned())? {
            return Ok(SortOrderPushdownResult::Exact {
                inner: Arc::new(self.clone()) as Arc<dyn FileSource>,
            });
        }

        // If the source's declared ordering is a non-empty *proper* prefix
        // of the request (e.g. source `[a DESC, b ASC]`, request
        // `[a DESC, b ASC, c DESC]`), decline pushdown so the outer
        // `SortExec`'s `sort_prefix` optimisation — prefix-aware early
        // termination in `TopK` — can still fire. Firing the Inexact
        // pipeline below would invalidate the source's `output_ordering`
        // (the runtime row-group reorder is approximate, so we can't
        // honour the declared ordering anymore), which is exactly what
        // `EnforceSorting` needs to derive `sort_prefix`. On data that
        // is already in prefix order the stats-based reorder is mostly
        // a no-op anyway, so the trade-off is plainly bad.
        for prefix_len in 1..order.len() {
            let prefix = order[..prefix_len].to_vec();
            if eq_properties.ordering_satisfy(prefix.iter().cloned())? {
                return Ok(SortOrderPushdownResult::Unsupported);
            }
        }

        // Inexact pushdown. Two independent signals; either is enough
        // to produce an approximate ordering, and they compose when
        // both apply:
        //
        // 1. `column_in_file_schema`: the request's leading sort key is
        //    a plain `Column` present in the file schema. The opener
        //    can sort row groups by that column's `min` via parquet
        //    statistics. Drives `sort_order_for_reorder`'s actual use.
        //
        // 2. `reversed_satisfies`: the source's declared ordering,
        //    when reversed, satisfies the request. This is strictly
        //    more powerful than the column-in-schema check because it
        //    runs the request through `EquivalenceProperties`'s full
        //    reasoning machinery:
        //
        //    - Function monotonicity: e.g. file declares
        //      `[extract_year_month(ws) DESC, ws DESC]`, request is
        //      `[ws ASC]`; the reversed ordering still satisfies the
        //      request via `extract_year_month`'s monotonicity even
        //      though parquet has no stats keyed by the function
        //      expression itself.