Establish the high level API for sort pushdown and the optimizer rule and support reverse files and row groups

datafusion/common/src/config.rs

@@ -831,6 +831,23 @@ config_namespace! {
         /// writing out already in-memory data, such as from a cached
         /// data frame.
         pub maximum_buffered_record_batches_per_stream: usize, default = 2
+
+       /// Enable sort pushdown optimization for Parquet files.
+       ///
+       /// When enabled, optimizes queries with ORDER BY:
+       /// - Reordering files based on statistics
+       /// - Reversing row group read order when beneficial
+       ///
+       /// Returns **inexact ordering**: Sort operator is kept for correctness,
+       /// but can terminate early for TopK queries (ORDER BY ... LIMIT N),
+       /// providing huge speedup.
+       ///
+       /// Memory: No additional overhead (only changes read order).
+       ///
+       /// Future TODO: Will add option to support detect perfectly sorted data and eliminate Sort completely.
+       ///
+       /// Default: true
+       pub enable_sort_pushdown: bool, default = true
     }
 }
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -208,6 +208,7 @@ impl ParquetOptions {
             coerce_int96: _,     // not used for writer props
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
+            enable_sort_pushdown: _,
         } = self;
 
         let mut builder = WriterProperties::builder()
@@ -472,6 +473,7 @@ mod tests {
             skip_arrow_metadata: defaults.skip_arrow_metadata,
             coerce_int96: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
+            enable_sort_pushdown: true,
         }
     }
 
@@ -585,6 +587,7 @@ mod tests {
                 binary_as_string: global_options_defaults.binary_as_string,
                 skip_arrow_metadata: global_options_defaults.skip_arrow_metadata,
                 coerce_int96: None,
+                enable_sort_pushdown: true,
             },
             column_specific_options,
             key_value_metadata,

datafusion/core/tests/physical_optimizer/mod.rs

@@ -32,6 +32,7 @@ mod limit_pushdown;
 mod limited_distinct_aggregation;
 mod partition_statistics;
 mod projection_pushdown;
+mod pushdown_sort;
 mod replace_with_order_preserving_variants;
 mod sanity_checker;
 #[expect(clippy::needless_pass_by_value)]