Refactor PushdownSort to avoid some clones

datafusion/expr/src/logical_plan/plan.rs

@@ -2525,6 +2525,8 @@ pub struct TableScan {
     pub filters: Vec<Expr>,
     /// Optional number of rows to read
     pub fetch: Option<usize>,
+    /// Optional preferred ordering for the scan
+    pub preferred_ordering: Option<Vec<SortExpr>>,
 }
 
 impl Debug for TableScan {
@@ -2536,6 +2538,7 @@ impl Debug for TableScan {
             .field("projected_schema", &self.projected_schema)
             .field("filters", &self.filters)
             .field("fetch", &self.fetch)
+            .field("preferred_ordering", &self.preferred_ordering)
             .finish_non_exhaustive()
     }
 }
@@ -2547,6 +2550,7 @@ impl PartialEq for TableScan {
             && self.projected_schema == other.projected_schema
             && self.filters == other.filters
             && self.fetch == other.fetch
+            && self.preferred_ordering == other.preferred_ordering
     }
 }
 
@@ -2566,18 +2570,22 @@ impl PartialOrd for TableScan {
             pub filters: &'a Vec<Expr>,
             /// Optional number of rows to read
             pub fetch: &'a Option<usize>,
+            /// Optional preferred ordering for the scan
+            pub preferred_ordering: &'a Option<Vec<SortExpr>>,
         }
         let comparable_self = ComparableTableScan {
             table_name: &self.table_name,
             projection: &self.projection,
             filters: &self.filters,
             fetch: &self.fetch,
+            preferred_ordering: &self.preferred_ordering,
         };
         let comparable_other = ComparableTableScan {
             table_name: &other.table_name,
             projection: &other.projection,
             filters: &other.filters,
             fetch: &other.fetch,
+            preferred_ordering: &other.preferred_ordering,
         };
         comparable_self.partial_cmp(&comparable_other)
     }
@@ -2590,6 +2598,7 @@ impl Hash for TableScan {
         self.projected_schema.hash(state);
         self.filters.hash(state);
         self.fetch.hash(state);
+        self.preferred_ordering.hash(state);
     }
 }
 
@@ -2643,8 +2652,68 @@ impl TableScan {
             projected_schema,
             filters,
             fetch,
+            preferred_ordering: None,
         })
     }
+
+    /// Sets the preferred ordering for this table scan using the builder pattern.
+    ///
+    /// The preferred ordering serves as a hint to table providers about the desired
+    /// sort order for the data. Table providers can use this information to optimize
+    /// data access patterns, choose appropriate indexes, or leverage existing sort
+    /// orders in the underlying storage.
+    ///
+    /// # Parameters
+    ///
+    /// * `preferred_ordering` - An optional vector of sort expressions representing
+    ///   the desired ordering. `None` indicates no specific ordering preference.
+    ///
+    /// # Returns
+    ///
+    /// Returns `self` to enable method chaining in the builder pattern.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use datafusion_expr::{col, SortExpr};
+    /// # use datafusion_expr::logical_plan::{TableScan, builder::table_source};
+    /// # use std::sync::Arc;
+    /// # use datafusion_common::{TableReference, DFSchema};
+    /// # use arrow::datatypes::{Schema, Field, DataType};
+    ///
+    /// // Create a table scan with preferred ordering by column 'a' ascending
+    /// # let table_name = TableReference::bare("test");
+    /// # let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+    /// # let source = table_source(&schema);
+    /// # let projection = None;
+    /// # let projected_schema = Arc::new(datafusion_common::DFSchema::empty());
+    /// # let filters = vec![];
+    /// # let fetch = None;
+    /// let table_scan = TableScan {
+    ///     table_name,
+    ///     source,
+    ///     projection,
+    ///     projected_schema,
+    ///     filters,
+    ///     fetch,
+    ///     preferred_ordering: None,
+    /// }.with_preferred_ordering(Some(vec![
+    ///     SortExpr::new(col("a"), true, false) // ASC NULLS LAST
+    /// ]));
+    /// ```
+    ///
+    /// # Notes
+    ///
+    /// This is purely an optimization hint. The table provider may choose to ignore
+    /// the preferred ordering if it cannot be efficiently satisfied, and the query
+    /// execution engine should not rely on the data being returned in this order.
+    pub fn with_preferred_ordering(
+        mut self,
+        preferred_ordering: Option<Vec<SortExpr>>,
+    ) -> Self {
+        self.preferred_ordering = preferred_ordering;
+        self
+    }
 }
 
 // Repartition the plan based on a partitioning scheme.
@@ -4814,6 +4883,7 @@ mod tests {
             projected_schema: Arc::clone(&schema),
             filters: vec![],
             fetch: None,
+            preferred_ordering: None,
         }));
         let col = schema.field_names()[0].clone();
 
@@ -4844,6 +4914,7 @@ mod tests {
             projected_schema: Arc::clone(&unique_schema),
             filters: vec![],
             fetch: None,
+            preferred_ordering: None,
         }));
         let col = schema.field_names()[0].clone();
 

datafusion/expr/src/logical_plan/tree_node.rs

@@ -599,6 +599,7 @@ impl LogicalPlan {
                 projected_schema,
                 filters,
                 fetch,
+                preferred_ordering,
             }) => filters.map_elements(f)?.update_data(|filters| {
                 LogicalPlan::TableScan(TableScan {
                     table_name,
@@ -607,6 +608,7 @@ impl LogicalPlan {
                     projected_schema,
                     filters,
                     fetch,
+                    preferred_ordering,
                 })
             }),
             LogicalPlan::Distinct(Distinct::On(DistinctOn {

datafusion/optimizer/src/lib.rs

@@ -58,6 +58,7 @@ pub mod optimizer;
 pub mod propagate_empty_relation;
 pub mod push_down_filter;
 pub mod push_down_limit;
+pub mod push_down_sort;
 pub mod replace_distinct_aggregate;
 pub mod scalar_subquery_to_join;
 pub mod simplify_expressions;

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -242,6 +242,7 @@ fn optimize_projections(
                 filters,
                 fetch,
                 projected_schema: _,
+                preferred_ordering,
             } = table_scan;
 
             // Get indices referred to in the original (schema with all fields)
@@ -257,6 +258,7 @@ fn optimize_projections(
                 filters,
                 fetch,
             )
+            .map(|s| s.with_preferred_ordering(preferred_ordering))
             .map(LogicalPlan::TableScan)
             .map(Transformed::yes);
         }

datafusion/optimizer/src/optimizer.rs

@@ -51,6 +51,7 @@ use crate::plan_signature::LogicalPlanSignature;
 use crate::propagate_empty_relation::PropagateEmptyRelation;
 use crate::push_down_filter::PushDownFilter;
 use crate::push_down_limit::PushDownLimit;
+use crate::push_down_sort::PushDownSort;
 use crate::replace_distinct_aggregate::ReplaceDistinctWithAggregate;
 use crate::scalar_subquery_to_join::ScalarSubqueryToJoin;
 use crate::simplify_expressions::SimplifyExpressions;
@@ -242,6 +243,8 @@ impl Optimizer {
             Arc::new(EliminateOuterJoin::new()),
             // Filters can't be pushed down past Limits, we should do PushDownFilter after PushDownLimit
             Arc::new(PushDownLimit::new()),
+            // Sort pushdown should happen before filter pushdown to maximize optimization opportunities
+            Arc::new(PushDownSort::new()),
             Arc::new(PushDownFilter::new()),
             Arc::new(SingleDistinctToGroupBy::new()),
             // The previous optimizations added expressions and projections,

datafusion/optimizer/src/push_down_filter.rs

@@ -3055,6 +3055,7 @@ mod tests {
             projection,
             source: Arc::new(test_provider),
             fetch: None,
+            preferred_ordering: None,
         });
 
         Ok(LogicalPlanBuilder::from(table_scan))