Redesign the try_reverse_output to support more cases (apache#19446)

## Which issue does this PR close?
- Closes apache#19403

## Rationale for this change

After apache#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]>

Author: zhuqi-lucas

datafusion/core/tests/physical_optimizer/pushdown_sort.rs

@@ -41,6 +41,7 @@ use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use datafusion_expr::{WindowFrame, WindowFunctionDefinition};
 use datafusion_functions_aggregate::count::count_udaf;
+use datafusion_physical_expr::EquivalenceProperties;
 use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
 use datafusion_physical_expr::expressions::{self, col};
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
@@ -54,6 +55,7 @@ use datafusion_physical_plan::aggregates::{
 };
 use datafusion_physical_plan::coalesce_batches::CoalesceBatchesExec;
 use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
+use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion_physical_plan::filter::FilterExec;
 use datafusion_physical_plan::joins::utils::{JoinFilter, JoinOn};
 use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode, SortMergeJoinExec};
@@ -68,7 +70,7 @@ use datafusion_physical_plan::union::UnionExec;
 use datafusion_physical_plan::windows::{BoundedWindowAggExec, create_window_expr};
 use datafusion_physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, InputOrderMode, Partitioning,
-    PlanProperties, displayable,
+    PlanProperties, SortOrderPushdownResult, displayable,
 };
 
 /// Create a non sorted parquet exec
@@ -774,3 +776,206 @@ pub fn format_execution_plan(plan: &Arc<dyn ExecutionPlan>) -> Vec<String> {
 fn format_lines(s: &str) -> Vec<String> {
     s.trim().split('\n').map(|s| s.to_string()).collect()
 }
+
+/// Create a simple ProjectionExec with column indices (simplified version)
+pub fn simple_projection_exec(
+    input: Arc<dyn ExecutionPlan>,
+    columns: Vec<usize>,
+) -> Arc<dyn ExecutionPlan> {
+    let schema = input.schema();
+    let exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = columns
+        .iter()
+        .map(|&i| {
+            let field = schema.field(i);
+            (
+                Arc::new(expressions::Column::new(field.name(), i))
+                    as Arc<dyn PhysicalExpr>,
+                field.name().to_string(),
+            )
+        })
+        .collect();
+
+    projection_exec(exprs, input).unwrap()
+}
+
+/// Create a ProjectionExec with column aliases
+pub fn projection_exec_with_alias(
+    input: Arc<dyn ExecutionPlan>,
+    columns: Vec<(usize, &str)>,
+) -> Arc<dyn ExecutionPlan> {
+    let schema = input.schema();
+    let exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = columns
+        .iter()
+        .map(|&(i, alias)| {
+            (
+                Arc::new(expressions::Column::new(schema.field(i).name(), i))
+                    as Arc<dyn PhysicalExpr>,
+                alias.to_string(),
+            )
+        })
+        .collect();
+
+    projection_exec(exprs, input).unwrap()
+}
+
+/// Create a sort expression with custom name and index
+pub fn sort_expr_named(name: &str, index: usize) -> PhysicalSortExpr {
+    PhysicalSortExpr {
+        expr: Arc::new(expressions::Column::new(name, index)),
+        options: SortOptions::default(),
+    }
+}
+
+/// A test data source that can display any requested ordering
+/// This is useful for testing sort pushdown behavior
+#[derive(Debug, Clone)]
+pub struct TestScan {
+    schema: SchemaRef,
+    output_ordering: Vec<LexOrdering>,
+    plan_properties: PlanProperties,
+    // Store the requested ordering for display
+    requested_ordering: Option<LexOrdering>,
+}
+
+impl TestScan {
+    /// Create a new TestScan with the given schema and output ordering
+    pub fn new(schema: SchemaRef, output_ordering: Vec<LexOrdering>) -> Self {
+        let eq_properties = if !output_ordering.is_empty() {
+            // Convert Vec<LexOrdering> to the format expected by new_with_orderings
+            // We need to extract the inner Vec<PhysicalSortExpr> from each LexOrdering
+            let orderings: Vec<Vec<PhysicalSortExpr>> = output_ordering
+                .iter()
+                .map(|lex_ordering| {
+                    // LexOrdering implements IntoIterator, so we can collect it
+                    lex_ordering.iter().cloned().collect()
+                })
+                .collect();
+
+            EquivalenceProperties::new_with_orderings(Arc::clone(&schema), orderings)
+        } else {
+            EquivalenceProperties::new(Arc::clone(&schema))
+        };
+
+        let plan_properties = PlanProperties::new(
+            eq_properties,
+            Partitioning::UnknownPartitioning(1),
+            EmissionType::Incremental,
+            Boundedness::Bounded,
+        );
+
+        Self {
+            schema,
+            output_ordering,
+            plan_properties,
+            requested_ordering: None,
+        }
+    }
+
+    /// Create a TestScan with a single output ordering
+    pub fn with_ordering(schema: SchemaRef, ordering: LexOrdering) -> Self {
+        Self::new(schema, vec![ordering])
+    }
+}
+
+impl DisplayAs for TestScan {
+    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+        match t {
+            DisplayFormatType::Default | DisplayFormatType::Verbose => {
+                write!(f, "TestScan")?;
+                if !self.output_ordering.is_empty() {
+                    write!(f, ": output_ordering=[")?;
+                    // Format the ordering in a readable way
+                    for (i, sort_expr) in self.output_ordering[0].iter().enumerate() {
+                        if i > 0 {
+                            write!(f, ", ")?;
+                        }
+                        write!(f, "{sort_expr}")?;
+                    }
+                    write!(f, "]")?;
+                }
+                // This is the key part - show what ordering was requested
+                if let Some(ref req) = self.requested_ordering {
+                    write!(f, ", requested_ordering=[")?;
+                    for (i, sort_expr) in req.iter().enumerate() {
+                        if i > 0 {
+                            write!(f, ", ")?;
+                        }
+                        write!(f, "{sort_expr}")?;
+                    }
+                    write!(f, "]")?;
+                }
+                Ok(())
+            }
+            DisplayFormatType::TreeRender => {
+                write!(f, "TestScan")
+            }
+        }
+    }
+}
+
+impl ExecutionPlan for TestScan {
+    fn name(&self) -> &str {
+        "TestScan"
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn properties(&self) -> &PlanProperties {
+        &self.plan_properties
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        if children.is_empty() {
+            Ok(self)
+        } else {
+            internal_err!("TestScan should have no children")
+        }
+    }
+
+    fn execute(
+        &self,
+        _partition: usize,
+        _context: Arc<TaskContext>,
+    ) -> Result<SendableRecordBatchStream> {
+        internal_err!("TestScan is for testing optimizer only, not for execution")
+    }
+
+    fn partition_statistics(&self, _partition: Option<usize>) -> Result<Statistics> {
+        Ok(Statistics::new_unknown(&self.schema))
+    }
+
+    // This is the key method - implement sort pushdown
+    fn try_pushdown_sort(
+        &self,
+        order: &[PhysicalSortExpr],
+    ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
+        // For testing purposes, accept ANY ordering request
+        // and create a new TestScan that shows what was requested
+        let requested_ordering = LexOrdering::new(order.to_vec());
+
+        let mut new_scan = self.clone();
+        new_scan.requested_ordering = requested_ordering;
+
+        // Always return Inexact to keep the Sort node (like Phase 1 behavior)
+        Ok(SortOrderPushdownResult::Inexact {
+            inner: Arc::new(new_scan),
+        })
+    }
+}
+
+/// Helper function to create a TestScan with ordering
+pub fn test_scan_with_ordering(
+    schema: SchemaRef,
+    ordering: LexOrdering,
+) -> Arc<dyn ExecutionPlan> {
+    Arc::new(TestScan::with_ordering(schema, ordering))
+}

datafusion/core/tests/physical_optimizer/test_utils.rs

@@ -38,8 +38,8 @@ use datafusion_common::config::TableParquetOptions;
 use datafusion_datasource::TableSchema;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
-use datafusion_physical_expr::conjunction;
 use datafusion_physical_expr::projection::ProjectionExprs;
+use datafusion_physical_expr::{EquivalenceProperties, conjunction};
 use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
@@ -54,7 +54,7 @@ use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 
 #[cfg(feature = "parquet_encryption")]
 use datafusion_execution::parquet_encryption::EncryptionFactory;
-use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use itertools::Itertools;
 use object_store::ObjectStore;
 #[cfg(feature = "parquet_encryption")]
@@ -288,9 +288,6 @@ pub struct ParquetSource {
     pub(crate) projection: ProjectionExprs,
     #[cfg(feature = "parquet_encryption")]
     pub(crate) encryption_factory: Option<Arc<dyn EncryptionFactory>>,
-    /// The ordering of data within the files
-    /// This is set by FileScanConfig when it knows the file ordering
-    file_ordering: Option<LexOrdering>,
     /// If true, read files in reverse order and reverse row groups within files.
     /// But it's not guaranteed that rows within row groups are in reverse order,
     /// so we still need to sort them after reading, so the reverse scan is inexact.
@@ -320,7 +317,6 @@ impl ParquetSource {
             metadata_size_hint: None,
             #[cfg(feature = "parquet_encryption")]
             encryption_factory: None,
-            file_ordering: None,
             reverse_row_groups: false,
         }
     }
@@ -397,12 +393,6 @@ impl ParquetSource {
         self
     }
 
-    /// If set, indicates the ordering of data within the files being read.
-    pub fn with_file_ordering(mut self, ordering: Option<LexOrdering>) -> Self {
-        self.file_ordering = ordering;
-        self
-    }
-
     /// Return the value described in [`Self::with_pushdown_filters`]
     pub(crate) fn pushdown_filters(&self) -> bool {
         self.table_parquet_options.global.pushdown_filters
@@ -769,44 +759,61 @@ impl FileSource for ParquetSource {
     fn try_reverse_output(
         &self,
         order: &[PhysicalSortExpr],
+        eq_properties: &EquivalenceProperties,
     ) -> datafusion_common::Result<SortOrderPushdownResult<Arc<dyn FileSource>>> {
-        // Check if we have file ordering information
-        let file_ordering = match &self.file_ordering {
-            Some(ordering) => ordering,
-            None => return Ok(SortOrderPushdownResult::Unsupported),
-        };
-
-        // Create a LexOrdering from the requested order to use the is_reverse method
-        let Some(requested_ordering) = LexOrdering::new(order.to_vec()) else {
-            // Empty ordering requested, cannot optimize
+        if order.is_empty() {
             return Ok(SortOrderPushdownResult::Unsupported);
-        };
+        }
 
-        // Check if reversing the file ordering would satisfy the requested ordering
-        if file_ordering.is_reverse(&requested_ordering) {
-            // Phase 1: Enable reverse row group scanning
-            let new_source = self.clone().with_reverse_row_groups(true);
+        // Build new equivalence properties with the reversed ordering.
+        // This allows us to check if the reversed ordering satisfies the request
+        // by leveraging:
+        // - Function monotonicity (e.g., extract_year_month preserves ordering)
+        // - Constant columns (from filters)
+        // - Other equivalence relationships
+        //
+        // Example flow:
+        // 1. File ordering: [extract_year_month(ws) DESC, ws DESC]
+        // 2. After reversal: [extract_year_month(ws) ASC, ws ASC]
+        // 3. Requested: [ws ASC]
+        // 4. Through extract_year_month's monotonicity property, the reversed
+        //    ordering satisfies [ws ASC] even though it has additional prefix
+        let reversed_eq_properties = {
+            let mut new = eq_properties.clone();
+            new.clear_orderings();
+
+            // Reverse each ordering in the equivalence properties
+            let reversed_orderings = eq_properties
+                .oeq_class()
+                .iter()
+                .map(|ordering| {
+                    ordering
+                        .iter()
+                        .map(|expr| expr.reverse())
+                        .collect::<Vec<_>>()
+                })
+                .collect::<Vec<_>>();
+
+            new.add_orderings(reversed_orderings);
+            new
+        };
 
-            // Return Inexact because we're only reversing row group order,
-            // not guaranteeing perfect row-level ordering
-            return Ok(SortOrderPushdownResult::Inexact {
-                inner: Arc::new(new_source) as Arc<dyn FileSource>,
-            });
+        // Check if the reversed ordering satisfies the requested ordering
+        if !reversed_eq_properties.ordering_satisfy(order.iter().cloned())? {
+            return Ok(SortOrderPushdownResult::Unsupported);
         }
 
+        // Return Inexact because we're only reversing row group order,
+        // not guaranteeing perfect row-level ordering
+        let new_source = self.clone().with_reverse_row_groups(true);
+        Ok(SortOrderPushdownResult::Inexact {
+            inner: Arc::new(new_source) as Arc<dyn FileSource>,
+        })
+
         // TODO Phase 2: Add support for other optimizations:
         // - File reordering based on min/max statistics
         // - Detection of exact ordering (return Exact to remove Sort operator)
         // - Partial sort pushdown for prefix matches
-
-        Ok(SortOrderPushdownResult::Unsupported)
-    }
-
-    fn with_file_ordering_info(
-        &self,
-        ordering: Option<LexOrdering>,
-    ) -> datafusion_common::Result<Arc<dyn FileSource>> {
-        Ok(Arc::new(self.clone().with_file_ordering(ordering)))
     }
 }