Add preferred_ordering field to TableScan

This commit adds a new optional field `preferred_ordering` to the `TableScan`
logical plan node to support sort pushdown optimizations.

Changes include:
- Add `preferred_ordering: Option<Vec<SortExpr>>` field to `TableScan` struct
- Add `try_new_with_preferred_ordering` constructor method
- Update all `TableScan` constructors throughout the codebase to include the new field
- Update `Debug`, `PartialEq`, `Hash`, and `PartialOrd` implementations
- Update pattern matching in optimizer and other modules

The preferred_ordering field is currently not used by any optimization rules
but provides the foundation for future sort pushdown implementations.

This is part 2 of 2 PRs split from apache#17273 as requested in
apache#17273 (comment)

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: adriangb

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,22 @@ impl TableScan {
             projected_schema,
             filters,
             fetch,
+            preferred_ordering: None,
         })
     }
+
+    pub fn try_new_with_preferred_ordering(
+        table_name: impl Into<TableReference>,
+        table_source: Arc<dyn TableSource>,
+        projection: Option<Vec<usize>>,
+        filters: Vec<Expr>,
+        fetch: Option<usize>,
+        preferred_ordering: Option<Vec<SortExpr>>,
+    ) -> Result<Self> {
+        let mut table_scan = Self::try_new(table_name, table_source, projection, filters, fetch)?;
+        table_scan.preferred_ordering = preferred_ordering;
+        Ok(table_scan)
+    }
 }
 
 // Repartition the plan based on a partitioning scheme.
@@ -4814,6 +4837,7 @@ mod tests {
             projected_schema: Arc::clone(&schema),
             filters: vec![],
             fetch: None,
+            preferred_ordering: None,
         }));
         let col = schema.field_names()[0].clone();
 
@@ -4844,6 +4868,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/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)
@@ -250,12 +251,13 @@ fn optimize_projections(
                 Some(projection) => indices.into_mapped_indices(|idx| projection[idx]),
                 None => indices.into_inner(),
             };
-            return TableScan::try_new(
+            return TableScan::try_new_with_preferred_ordering(
                 table_name,
                 source,
                 Some(projection),
                 filters,
                 fetch,
+                preferred_ordering,
             )
             .map(LogicalPlan::TableScan)
             .map(Transformed::yes);

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))

datafusion/proto/src/logical_plan/mod.rs

@@ -271,6 +271,7 @@ fn from_table_source(
         projected_schema,
         filters: vec![],
         fetch: None,
+        preferred_ordering: None,
     });
 
     LogicalPlanNode::try_from_logical_plan(&r, extension_codec)