feat: make sort pushdown BufferExec capacity configurable, default 1GB

Replace hardcoded 64MB BUFFER_CAPACITY_AFTER_SORT_ELIMINATION with
configurable `datafusion.execution.sort_pushdown_buffer_capacity`.

Default 1GB — large enough to hold wide-row data without I/O stalls.
This is strictly less memory than the SortExec it replaces, and actual
usage is bounded by partition size and global memory pool limits.

Closes #21417

Author: zhuqi-lucas

datafusion/common/src/config.rs

@@ -557,6 +557,19 @@ config_namespace! {
         /// batches and merged.
         pub sort_in_place_threshold_bytes: usize, default = 1024 * 1024
 
+        /// Maximum buffer capacity (in bytes) per partition for BufferExec
+        /// inserted during sort pushdown optimization.
+        ///
+        /// When PushdownSort eliminates a SortExec under SortPreservingMergeExec,
+        /// a BufferExec is inserted to replace SortExec's buffering role. This
+        /// prevents I/O stalls by allowing the scan to run ahead of the merge.
+        ///
+        /// This uses strictly less memory than the SortExec it replaces (which
+        /// buffers the entire partition). The buffer respects the global memory
+        /// pool limit. Setting this to a large value is safe — actual memory
+        /// usage is bounded by partition size and global memory limits.
+        pub sort_pushdown_buffer_capacity: usize, default = 1024 * 1024 * 1024
+
         /// Maximum size in bytes for individual spill files before rotating to a new file.
         ///
         /// When operators spill data to disk (e.g., RepartitionExec), they write

datafusion/physical-optimizer/src/pushdown_sort.rs

@@ -65,20 +65,6 @@ use datafusion_physical_plan::sorts::sort::SortExec;
 use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
 use std::sync::Arc;
 
-/// Per-partition buffer capacity (in bytes) inserted between SPM and
-/// DataSourceExec when sort elimination removes the buffering SortExec.
-///
-/// SortExec buffers all input data in memory (potentially GB per partition)
-/// before outputting sorted results. When we eliminate SortExec, SPM reads
-/// directly from I/O-bound sources. BufferExec compensates with bounded
-/// buffering, allowing I/O to pipeline with merge computation.
-///
-/// This is strictly less memory than the SortExec it replaces, and only
-/// inserted when PushdownSort eliminates a SortExec — no impact on other
-/// query plans. BufferExec also integrates with MemoryPool, so it respects
-/// the global memory limit and won't cause OOM.
-const BUFFER_CAPACITY_AFTER_SORT_ELIMINATION: usize = 64 * 1024 * 1024; // 64 MB
-
 /// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources.
 ///
 /// See module-level documentation for details.
@@ -102,6 +88,8 @@ impl PhysicalOptimizerRule for PushdownSort {
             return Ok(plan);
         }
 
+        let buffer_capacity = config.execution.sort_pushdown_buffer_capacity;
+
         // Use transform_down to find and optimize all SortExec nodes (including nested ones)
         // Also handles SPM → SortExec pattern to insert BufferExec when sort is eliminated
         plan.transform_down(|plan: Arc<dyn ExecutionPlan>| {
@@ -124,10 +112,8 @@ impl PhysicalOptimizerRule for PushdownSort {
                         // Insert BufferExec to replace SortExec's buffering role.
                         // SortExec buffered all data in memory; BufferExec provides
                         // bounded buffering so SPM doesn't stall on I/O.
-                        let buffered: Arc<dyn ExecutionPlan> = Arc::new(BufferExec::new(
-                            inner,
-                            BUFFER_CAPACITY_AFTER_SORT_ELIMINATION,
-                        ));
+                        let buffered: Arc<dyn ExecutionPlan> =
+                            Arc::new(BufferExec::new(inner, buffer_capacity));
                         let new_spm =
                             SortPreservingMergeExec::new(spm.expr().clone(), buffered)
                                 .with_fetch(spm.fetch());

datafusion/sqllogictest/test_files/sort_pushdown.slt

@@ -2221,7 +2221,7 @@ logical_plan
 02)--TableScan: tg_buffer projection=[id, value]
 physical_plan
 01)SortPreservingMergeExec: [id@0 ASC NULLS LAST]
-02)--BufferExec: capacity=67108864
+02)--BufferExec: capacity=1073741824
 03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/a_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/c_low.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
 
 # Verify correctness
@@ -2248,7 +2248,7 @@ logical_plan
 02)--TableScan: tg_buffer projection=[id, value]
 physical_plan
 01)SortPreservingMergeExec: [id@0 ASC NULLS LAST], fetch=3
-02)--BufferExec: capacity=67108864
+02)--BufferExec: capacity=1073741824
 03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/a_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/c_low.parquet]]}, projection=[id, value], limit=3, output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
 
 query II