feat: stage ORDER BY expression evaluation

datafusion/common/src/config.rs

@@ -998,6 +998,11 @@ config_namespace! {
         /// batches and merged.
         pub sort_in_place_threshold_bytes: usize, default = 1024 * 1024
 
+        /// Enables staged sorting for eligible in-memory sorts.
+        /// Later keys may be skipped for rows already in order
+        /// TopK and spill sorting do not use staged sorting.
+        pub enable_staged_sort: bool, default = false
+
         /// Maximum buffer capacity (in bytes) per partition for BufferExec
         /// inserted during sort pushdown optimization.
         ///

datafusion/physical-expr/src/expressions/binary.rs

@@ -1443,7 +1443,7 @@ fn rows_to_filter_before(
 
 /// Whether evaluating `expr` on all rows is cheap and cannot fail.
 /// Inspired by `reorder_predicates`' cheap forms, with fallible forms excluded.
-fn is_cheap_and_infallible(expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
+pub fn is_cheap_and_infallible(expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
     use crate::expressions::{
         CastExpr, Column, InListExpr, IsNotNullExpr, IsNullExpr, Literal, NotExpr,
     };

datafusion/physical-expr/src/expressions/mod.rs

@@ -28,7 +28,7 @@ pub use crate::PhysicalSortExpr;
 /// Module with some convenient methods used in expression building
 pub use crate::aggregate::stats::StatsType;
 
-pub use binary::{BinaryExpr, binary, similar_to};
+pub use binary::{BinaryExpr, binary, is_cheap_and_infallible, similar_to};
 pub use case::{CaseExpr, case};
 pub use cast::{CastExpr, cast};
 pub use column::{Column, col, with_new_schema};