feat: add ProgressiveEval and rule to convert from SortPreservingMerge

benchmarks/README.md

@@ -1043,6 +1043,7 @@ Benchmarks for sort pushdown optimizations on TPC-H lineitem data (SF=1).
 |-----------|-------------|
 | `sort_pushdown` | Baseline — no `WITH ORDER`, tests standard sort behavior |
 | `sort_pushdown_sorted` | With `WITH ORDER` — tests sort elimination on sorted files |
+| `sort_pushdown_100` | Variation with the same data partitioned into many more (100) files |
 | `sort_pushdown_inexact` | Inexact path (`--sorted` DESC) — multi-file with scrambled RGs, tests reverse scan + RG reorder |
 | `sort_pushdown_inexact_unsorted` | No `WITH ORDER` — same data, tests Unsupported path + RG reorder |
 | `sort_pushdown_inexact_overlap` | Multi-file scrambled RGs — streaming data scenario |
@@ -1067,9 +1068,11 @@ adjacent chunks at RG boundaries.
 ```bash
 # Generate data and run all sort pushdown benchmarks
 ./bench.sh data sort_pushdown
+./bench.sh data sort_pushdown_100
 ./bench.sh data sort_pushdown_inexact
 ./bench.sh run sort_pushdown
 ./bench.sh run sort_pushdown_sorted
+./bench.sh run sort_pushdown_100
 ./bench.sh run sort_pushdown_inexact
 ./bench.sh run sort_pushdown_inexact_overlap
 ```

benchmarks/bench.sh

@@ -122,6 +122,7 @@ clickbench_extended:    ClickBench \"inspired\" queries against a single parquet
 # Sort Pushdown Benchmarks
 sort_pushdown:          Sort pushdown baseline (no WITH ORDER) on TPC-H data (SF=1)
 sort_pushdown_sorted:   Sort pushdown with WITH ORDER — tests sort elimination on non-overlapping files
+sort_pushdown_100:      Sort pushdown variation with 100 non-overlapping sorted files
 sort_pushdown_inexact:  Sort pushdown Inexact path (--sorted DESC) — multi-file with scrambled RGs, tests reverse scan + RG reorder
 sort_pushdown_inexact_unsorted: Sort pushdown Inexact path (no WITH ORDER) — same data, tests Unsupported path + RG reorder
 sort_pushdown_inexact_overlap: Sort pushdown Inexact path — multi-file scrambled RGs (streaming data scenario)
@@ -344,6 +345,9 @@ main() {
                 sort_pushdown|sort_pushdown_sorted)
                     data_sort_pushdown
                     ;;
+                sort_pushdown_100)
+                    data_sort_pushdown_100
+                    ;;
                 sort_pushdown_inexact|sort_pushdown_inexact_unsorted|sort_pushdown_inexact_overlap)
                     data_sort_pushdown_inexact
                     ;;
@@ -571,6 +575,9 @@ main() {
                 sort_pushdown_sorted)
                     run_sort_pushdown_sorted
                     ;;
+                sort_pushdown_100)
+                    run_sort_pushdown_100
+                    ;;
                 sort_pushdown_inexact)
                     run_sort_pushdown_inexact
                     ;;
@@ -1321,6 +1328,33 @@ data_sort_pushdown() {
     ls -la "${SORT_PUSHDOWN_DIR}"
 }
 
+# Generates the 100-file variation of the sort pushdown data: TPCH lineitem
+# (SF=1) as 100 sorted, non-overlapping parquet files.
+#
+# No renaming is needed to exercise reordering, the lexical file order already
+# differs from key order.
+data_sort_pushdown_100() {
+    SORT_PUSHDOWN_100_DIR="${DATA_DIR}/sort_pushdown_100/lineitem"
+    if [ -d "${SORT_PUSHDOWN_100_DIR}" ] && [ "$(ls -A ${SORT_PUSHDOWN_100_DIR}/*.parquet 2>/dev/null)" ]; then
+        echo "Sort pushdown 100-file data already exists at ${SORT_PUSHDOWN_100_DIR}"
+        return
+    fi
+
+    echo "Generating sort pushdown benchmark data (100 parts)..."
+
+    TEMP_DIR="${DATA_DIR}/sort_pushdown_100_temp"
+    mkdir -p "${TEMP_DIR}" "${SORT_PUSHDOWN_100_DIR}"
+
+    tpchgen-cli --scale-factor 1 --format parquet --parquet-compression='ZSTD(1)' --parts=100 --output-dir "${TEMP_DIR}"
+
+    mv "${TEMP_DIR}"/lineitem/*.parquet "${SORT_PUSHDOWN_100_DIR}/"
+
+    rm -rf "${TEMP_DIR}"
+
+    echo "Sort pushdown 100-file data generated at ${SORT_PUSHDOWN_100_DIR}"
+    ls "${SORT_PUSHDOWN_100_DIR}" | wc -l
+}
+
 run_sort_pushdown() {
     SORT_PUSHDOWN_DIR="${DATA_DIR}/sort_pushdown"
     RESULTS_FILE="${RESULTS_DIR}/sort_pushdown.json"
@@ -1336,6 +1370,14 @@ run_sort_pushdown_sorted() {
     debug_run $CARGO_COMMAND --bin dfbench -- sort-pushdown --sorted --iterations 5 --path "${SORT_PUSHDOWN_DIR}" --queries-path "${SCRIPT_DIR}/queries/sort_pushdown" -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
 }
 
+# Runs the sort pushdown benchmark with WITH ORDER on the 100 file variation
+run_sort_pushdown_100() {
+    SORT_PUSHDOWN_DIR="${DATA_DIR}/sort_pushdown_100"
+    RESULTS_FILE="${RESULTS_DIR}/sort_pushdown_100.json"
+    echo "Running sort pushdown benchmark (with 100 files)..."
+    debug_run $CARGO_COMMAND --bin dfbench -- sort-pushdown --sorted --iterations 5 --path "${SORT_PUSHDOWN_DIR}" --queries-path "${SCRIPT_DIR}/queries/sort_pushdown" -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
+}
+
 # Generates data for sort pushdown Inexact benchmark.
 #
 # Produces multiple parquet files where each file has MULTIPLE row groups

datafusion/common/src/config.rs

@@ -1162,6 +1162,17 @@ config_namespace! {
         ///
         /// Disabled by default, set to a number greater than 0 for enabling it.
         pub hash_join_buffering_capacity: usize, default = 0
+
+        /// Number of input streams to prefetch ahead-of-time for `ProgressiveEvalExec`.
+        /// Since `ProgressiveEvalExec` only polls one stream at a time in order,
+        /// we do not need to prefetch all streams at once, saving resources. However, if the
+        /// streams' IO time is much greater than their CPU/processing time, prefetching them will
+        /// help improve performance.
+        /// Default is 1 which means we will prefetch one extra stream before it is polled.
+        /// 0 means streams are only fetched immediately before they are required.
+        /// Increase this value if IO time to read a stream is often much more than CPU time to
+        /// process the previous one.
+        pub progressive_eval_num_prefetch_input_streams: usize, default = 1
     }
 }
 
@@ -1840,6 +1851,12 @@ config_namespace! {
         /// Default: true
         pub enable_sort_pushdown: bool, default = true
 
+        /// When set to true, the physical plan optimizer will replace
+        /// `SortPreservingMergeExec` with `ProgressiveEvalExec` when the input
+        /// partitions are non-overlapping ranges of the merge ordering,
+        /// avoiding a merge by emitting the partitions sequentially.
+        pub sequence_sorted_inputs: bool, default = false
+
         /// When set to true, the optimizer will extract leaf expressions
         /// (such as `get_field`) from filter/sort/join nodes into projections
         /// closer to the leaf table scans, and push those projections down

datafusion/core/src/optimizer_rule_reference.md

@@ -88,6 +88,7 @@ in multiple phases.
 | 16    | `TopKRepartition`              | -                       | Pushes TopK below hash repartition when the partition key is a prefix of the sort key.                       |
 | 17    | `ProjectionPushdown`           | late pass               | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities.               |
 | 18    | `PushdownSort`                 | -                       | Pushes sort requirements into data sources that can already return sorted output.                            |
-| 19    | `EnsureCooperative`            | -                       | Wraps non-cooperative plan parts so long-running tasks yield fairly.                                         |
-| 20    | `FilterPushdown(Post)`         | post-optimization phase | Pushes dynamic filters at the end of optimization, after plan references stop moving.                        |
-| 21    | `SanityCheckPlan`              | -                       | Validates that the final physical plan meets ordering, distribution, and infinite-input safety requirements. |
+| 19    | `SequenceSortedInputs`         | -                       | Replaces SortPreservingMergeExec with ProgressiveEvalExec when partitions don't overlap in the sort columns. |
+| 20    | `EnsureCooperative`            | -                       | Wraps non-cooperative plan parts so long-running tasks yield fairly.                                         |
+| 21    | `FilterPushdown(Post)`         | post-optimization phase | Pushes dynamic filters at the end of optimization, after plan references stop moving.                        |
+| 22    | `SanityCheckPlan`              | -                       | Validates that the final physical plan meets ordering, distribution, and infinite-input safety requirements. |

datafusion/datasource/src/file_scan_config/mod.rs

@@ -29,10 +29,14 @@ mod proto;
 
 use crate::file_groups::FileGroup;
 use crate::{
-    PartitionedFile, display::FileGroupsDisplay, file::FileSource,
-    file_compression_type::FileCompressionType, file_stream::FileStreamBuilder,
-    file_stream::work_source::SharedWorkSource, source::DataSource,
-    statistics::MinMaxStatistics,
+    PartitionedFile,
+    display::FileGroupsDisplay,
+    file::FileSource,
+    file_compression_type::FileCompressionType,
+    file_stream::FileStreamBuilder,
+    file_stream::work_source::SharedWorkSource,
+    source::DataSource,
+    statistics::{MinMaxStatistics, compute_file_group_statistics},
 };
 use arrow::datatypes::Fields;
 use arrow::datatypes::{DataType, Schema, SchemaRef};
@@ -1404,10 +1408,13 @@ impl FileScanConfig {
     ///
     /// The method distributes files across a target number of partitions while ensuring
     /// files within each partition maintain sort order based on their min/max statistics.
+    /// If possible, it also tries to make groups that are non-overlapping with respect
+    /// to the sort order.
     ///
     /// The algorithm works by:
     /// 1. Takes files sorted by minimum values
-    /// 2. For each file:
+    /// 2. If all files are non-overlapping with respect to the sort order, partition them into
+    ///    groups in order. Otherwise, for each file:
     ///   - Finds eligible groups (empty or where file's min > group's last max)
     ///   - Selects the smallest eligible group
     ///   - Creates a new group if needed
@@ -1456,43 +1463,72 @@ impl FileScanConfig {
 
         let indices_sorted_by_min = statistics.min_values_sorted();
 
-        // Initialize with target_partitions empty groups
-        let mut file_groups_indices: Vec<Vec<usize>> = vec![vec![]; target_partitions];
-
-        for (idx, min) in indices_sorted_by_min {
-            if let Some((_, group)) = file_groups_indices
-                .iter_mut()
-                .enumerate()
-                .filter(|(_, group)| {
-                    group.is_empty()
-                        || min
-                            > statistics
-                                .max(*group.last().expect("groups should not be empty"))
+        // When the files are all non-overlapping with respect to the sort order,
+        // chunk them contiguously so the resulting groups are also non-overlapping.
+        // This could allow the `sequence_sorted_inputs` optimization to be applied.
+        let non_overlapping_files = indices_sorted_by_min
+            .windows(2)
+            .all(|pair| pair[1].1 >= statistics.max(pair[0].0));
+
+        let mut file_groups_indices: Vec<Vec<usize>> = if non_overlapping_files {
+            let chunk_count = target_partitions.min(indices_sorted_by_min.len());
+            let chunk_size = indices_sorted_by_min.len() / chunk_count;
+            let remainder = indices_sorted_by_min.len() % chunk_count;
+            let mut files = indices_sorted_by_min.iter().map(|(idx, _)| *idx);
+            (0..chunk_count)
+                .map(|chunk_idx| {
+                    let len = chunk_size + usize::from(chunk_idx < remainder);
+                    files.by_ref().take(len).collect()
                 })
-                .min_by_key(|(_, group)| group.len())
-            {
-                group.push(idx);
-            } else {
-                // Create a new group if no existing group fits
-                file_groups_indices.push(vec![idx]);
+                .collect()
+        } else {
+            // Initialize with target_partitions empty groups
+            let mut file_groups_indices: Vec<Vec<usize>> =
+                vec![vec![]; target_partitions];
+
+            for (idx, min) in indices_sorted_by_min {
+                if let Some((_, group)) = file_groups_indices
+                    .iter_mut()
+                    .enumerate()
+                    .filter(|(_, group)| {
+                        group.is_empty()
+                            || min
+                                > statistics.max(
+                                    *group.last().expect("groups should not be empty"),
+                                )
+                    })
+                    .min_by_key(|(_, group)| group.len())
+                {
+                    group.push(idx);
+                } else {
+                    // Create a new group if no existing group fits
+                    file_groups_indices.push(vec![idx]);
+                }
             }
-        }
+            file_groups_indices
+        };
 
         // Remove any empty groups
         file_groups_indices.retain(|group| !group.is_empty());
 
-        // Assemble indices back into groups of PartitionedFiles
+        // Assemble indices back into groups of PartitionedFiles, recomputing
+        // group-level statistics.
         let mut file_groups = file_groups_indices
             .into_iter()
             .map(|file_group_indices| {
-                FileGroup::new(
+                let group = FileGroup::new(
                     file_group_indices
                         .into_iter()
                         .map(|idx| files[idx].clone())
                         .collect(),
-                )
+                );
+                if group.iter().all(|file| file.statistics.is_some()) {
+                    compute_file_group_statistics(group, Arc::clone(table_schema), true)
+                } else {
+                    Ok(group)
+                }
             })
-            .collect::<Vec<_>>();
+            .collect::<Result<Vec<_>>>()?;
 
         append_known_empty_files(&mut file_groups, empty_files);
         Ok(file_groups)

datafusion/datasource/src/file_scan_config/sort_pushdown.rs

@@ -28,7 +28,7 @@ use super::{FileScanConfig, partition_known_empty_files};
 use crate::file::FileSource;
 use crate::file_groups::FileGroup;
 use crate::source::DataSource;
-use crate::statistics::MinMaxStatistics;
+use crate::statistics::{MinMaxStatistics, compute_file_group_statistics};
 
 use arrow::datatypes::SchemaRef;
 use datafusion_common::Result;
@@ -124,6 +124,7 @@ impl FileScanConfig {
                     &sort_order,
                     &projected_schema,
                     projection_indices.as_deref(),
+                    new_config.file_source.table_schema().table_schema(),
                 );
                 new_config.file_groups = result.file_groups;
                 result.all_non_overlapping
@@ -251,6 +252,7 @@ impl FileScanConfig {
             &sort_order,
             &projected_schema,
             projection_indices.as_deref(),
+            self.file_source.table_schema().table_schema(),
         );
 
         if !result.any_reordered {
@@ -316,6 +318,7 @@ pub(crate) fn sort_files_within_groups_by_statistics(
     sort_order: &LexOrdering,
     projected_schema: &SchemaRef,
     projection_indices: Option<&[usize]>,
+    table_schema: &SchemaRef,
 ) -> SortedFileGroups {
     let mut any_reordered = false;
     let mut confirmed_non_overlapping: usize = 0;
@@ -367,11 +370,29 @@ pub(crate) fn sort_files_within_groups_by_statistics(
             group.clone()
         } else {
             any_reordered = true;
-            sorted_files
+            let reordered: FileGroup = sorted_files
                 .iter()
                 .map(|file| (**file).clone())
                 .chain(empty_files.iter().map(|file| (**file).clone()))
-                .collect()
+                .collect();
+            // Recompute group-level statistics.
+            if reordered.iter().all(|file| file.statistics.is_some()) {
+                match compute_file_group_statistics(
+                    reordered.clone(),
+                    Arc::clone(table_schema),
+                    true,
+                ) {
+                    Ok(with_stats) => with_stats,
+                    Err(e) => {
+                        log::trace!(
+                            "Failed to recompute file group statistics after reorder: {e}"
+                        );
+                        reordered
+                    }
+                }
+            } else {
+                reordered
+            }
         };
 
         let is_non_overlapping = match MinMaxStatistics::new_from_files(
@@ -690,6 +711,7 @@ mod tests {
             &sort_order,
             &schema,
             None,
+            &schema,
         );
 
         assert!(sorted.any_reordered);

datafusion/datasource/src/mod.rs

@@ -514,13 +514,13 @@ pub fn verify_sort_integrity(file_groups: &[FileGroup]) -> bool {
             let prev_file = files[i - 1];
             let curr_file = files[i];
 
-            // Check if the min value of current file is greater than max value of previous file
+            // Check if the min value of current file is greater or equal to max value of previous file
             if let (Some(prev_stats), Some(curr_stats)) =
                 (&prev_file.statistics, &curr_file.statistics)
             {
                 let prev_max = &prev_stats.column_statistics[0].max_value;
                 let curr_min = &curr_stats.column_statistics[0].min_value;
-                if curr_min.get_value().unwrap() <= prev_max.get_value().unwrap() {
+                if curr_min.get_value().unwrap() < prev_max.get_value().unwrap() {
                     return false;
                 }
             }

datafusion/physical-optimizer/src/lib.rs

@@ -45,6 +45,7 @@ pub use datafusion_pruning as pruning;
 pub mod hash_join_buffering;
 pub mod pushdown_sort;
 pub mod sanity_checker;
+pub mod sequence_sorted_inputs;
 pub mod topk_aggregation;
 pub mod topk_repartition;
 pub mod update_aggr_exprs;

datafusion/physical-optimizer/src/optimizer.rs

@@ -38,6 +38,7 @@ use crate::update_aggr_exprs::OptimizeAggregateOrder;
 use crate::hash_join_buffering::HashJoinBuffering;
 use crate::limit_pushdown_past_window::LimitPushPastWindows;
 use crate::pushdown_sort::PushdownSort;
+use crate::sequence_sorted_inputs::SequenceSortedInputs;
 use crate::window_topn::WindowTopN;
 use datafusion_common::config::ConfigOptions;
 
@@ -174,6 +175,10 @@ impl PhysicalOptimizer {
             Arc::new(ProjectionPushdown::new()),
             // PushdownSort: Detect sorts that can be pushed down to data sources.
             Arc::new(PushdownSort::new()),
+            // SequenceSortedInputs: Replace SortPreservingMergeExec with ProgressiveEvalExec
+            // for partitions that don't overlap in the sort columns.
+            // Runs after PushdownSort which might introduce a SortPreservingMergeExec.
+            Arc::new(SequenceSortedInputs::new()),
             Arc::new(EnsureCooperative::new()),
             // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan.
             // Therefore, it should be run at the end of the optimization process since any changes to the plan may break the dynamic filter's references.