feat: add sort pushdown benchmark and SLT tests (apache#21213)

## Which issue does this PR close?

- Related to apache#17348
- Precursor to apache#21182

## Rationale for this change

Add benchmark and integration tests for sort pushdown optimization,
split out from apache#21182 per [reviewer
request](apache#21182 (comment)).
This allows comparing benchmark results before and after the
optimization lands, and the SLT diff in apache#21182 will clearly show which
test expectations changed due to the optimization.

## What changes are included in this PR?

### Benchmark

New `sort-pushdown` benchmark subcommand with 4 queries testing sort
elimination:

| Query | Description |
|-------|-------------|
| Q1 | `ORDER BY l_orderkey ASC` (full scan) |
| Q2 | `ORDER BY l_orderkey ASC LIMIT 100` |
| Q3 | `SELECT * ORDER BY l_orderkey ASC` (wide) |
| Q4 | `SELECT * ORDER BY l_orderkey ASC LIMIT 100` (wide) |

Usage:
\`\`\`bash
./bench.sh data sort_pushdown
./bench.sh run sort_pushdown          # baseline
./bench.sh run sort_pushdown_sorted   # with sort elimination
\`\`\`

### SLT Integration Tests (5 new groups)

- **Test A**: Non-overlapping files + WITH ORDER → Sort eliminated
(single partition)
- **Test B**: Overlapping files → SortExec retained (baseline, files in
original order)
- **Test C**: LIMIT queries (ASC sort elimination + DESC reverse scan)
- **Test D**: \`target_partitions=2\` → SPM + per-partition sort
elimination
- **Test E**: Inferred ordering from Parquet metadata (no WITH ORDER) —
single and multi partition

### Files Changed

| File | Change |
|------|--------|
| \`benchmarks/src/sort_pushdown.rs\` | New benchmark module |
| \`benchmarks/src/lib.rs\` | Register module |
| \`benchmarks/src/bin/dfbench.rs\` | Register subcommand |
| \`benchmarks/bench.sh\` | Add data/run entries |
| \`datafusion/sqllogictest/test_files/sort_pushdown.slt\` | 5 new SLT
test groups |

## Test plan

- [x] \`cargo clippy -p datafusion-benchmarks\` — 0 warnings
- [x] \`cargo test -p datafusion-sqllogictest -- sort_pushdown\` — all
tests pass

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

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 2 people

benchmarks/bench.sh

@@ -106,6 +106,10 @@ clickbench_partitioned: ClickBench queries against partitioned (100 files) parqu
 clickbench_pushdown:    ClickBench queries against partitioned (100 files) parquet w/ filter_pushdown enabled
 clickbench_extended:    ClickBench \"inspired\" queries against a single parquet (DataFusion specific)
 
+# 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
+
 # Sorted Data Benchmarks (ORDER BY Optimization)
 clickbench_sorted:     ClickBench queries on pre-sorted data using prefer_existing_sort (tests sort elimination optimization)
 
@@ -309,6 +313,10 @@ main() {
                     # same data as for tpch
                     data_tpch "1" "parquet"
                     ;;
+                sort_pushdown|sort_pushdown_sorted)
+                    # same data as for tpch
+                    data_tpch "1" "parquet"
+                    ;;
                 sort_tpch)
                     # same data as for tpch
                     data_tpch "1" "parquet"
@@ -509,6 +517,12 @@ main() {
                 external_aggr)
                     run_external_aggr
                     ;;
+                sort_pushdown)
+                    run_sort_pushdown
+                    ;;
+                sort_pushdown_sorted)
+                    run_sort_pushdown_sorted
+                    ;;
                 sort_tpch)
                     run_sort_tpch "1"
                     ;;
@@ -1070,6 +1084,22 @@ run_external_aggr() {
     debug_run $CARGO_COMMAND --bin external_aggr -- benchmark --partitions 4 --iterations 5 --path "${TPCH_DIR}" -o "${RESULTS_FILE}" ${QUERY_ARG}
 }
 
+# Runs the sort pushdown benchmark (without WITH ORDER)
+run_sort_pushdown() {
+    TPCH_DIR="${DATA_DIR}/tpch_sf1"
+    RESULTS_FILE="${RESULTS_DIR}/sort_pushdown.json"
+    echo "Running sort pushdown benchmark (no WITH ORDER)..."
+    debug_run $CARGO_COMMAND --bin dfbench -- sort-pushdown --iterations 5 --path "${TPCH_DIR}" -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
+}
+
+# Runs the sort pushdown benchmark with WITH ORDER (enables sort elimination)
+run_sort_pushdown_sorted() {
+    TPCH_DIR="${DATA_DIR}/tpch_sf1"
+    RESULTS_FILE="${RESULTS_DIR}/sort_pushdown_sorted.json"
+    echo "Running sort pushdown benchmark (with WITH ORDER)..."
+    debug_run $CARGO_COMMAND --bin dfbench -- sort-pushdown --sorted --iterations 5 --path "${TPCH_DIR}" -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
+}
+
 # Runs the sort integration benchmark
 run_sort_tpch() {
     SCALE_FACTOR=$1

benchmarks/queries/sort_pushdown/q1.sql

@@ -0,0 +1,6 @@
+-- Sort elimination: ORDER BY sort key ASC (full scan)
+-- With --sorted: SortExec removed, sequential scan in file order
+-- Without --sorted: full SortExec required
+SELECT l_orderkey, l_partkey, l_suppkey
+FROM lineitem
+ORDER BY l_orderkey

benchmarks/queries/sort_pushdown/q2.sql

@@ -0,0 +1,7 @@
+-- Sort elimination + limit pushdown
+-- With --sorted: SortExec removed + limit pushed to DataSourceExec
+-- Without --sorted: TopK sort over all data
+SELECT l_orderkey, l_partkey, l_suppkey
+FROM lineitem
+ORDER BY l_orderkey
+LIMIT 100

benchmarks/queries/sort_pushdown/q3.sql

@@ -0,0 +1,5 @@
+-- Sort elimination: wide projection (all columns)
+-- Tests sort elimination benefit with larger row payload
+SELECT *
+FROM lineitem
+ORDER BY l_orderkey

benchmarks/queries/sort_pushdown/q4.sql

@@ -0,0 +1,5 @@
+-- Sort elimination + limit: wide projection
+SELECT *
+FROM lineitem
+ORDER BY l_orderkey
+LIMIT 100

benchmarks/src/bin/dfbench.rs

@@ -34,7 +34,8 @@ static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
 static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
 use datafusion_benchmarks::{
-    cancellation, clickbench, h2o, hj, imdb, nlj, smj, sort_tpch, tpcds, tpch,
+    cancellation, clickbench, h2o, hj, imdb, nlj, smj, sort_pushdown, sort_tpch, tpcds,
+    tpch,
 };
 
 #[derive(Debug, Parser)]
@@ -53,6 +54,7 @@ enum Options {
     Imdb(imdb::RunOpt),
     Nlj(nlj::RunOpt),
     Smj(smj::RunOpt),
+    SortPushdown(sort_pushdown::RunOpt),
     SortTpch(sort_tpch::RunOpt),
     Tpch(tpch::RunOpt),
     Tpcds(tpcds::RunOpt),
@@ -72,6 +74,7 @@ pub async fn main() -> Result<()> {
         Options::Imdb(opt) => Box::pin(opt.run()).await,
         Options::Nlj(opt) => opt.run().await,
         Options::Smj(opt) => opt.run().await,
+        Options::SortPushdown(opt) => opt.run().await,
         Options::SortTpch(opt) => opt.run().await,
         Options::Tpch(opt) => Box::pin(opt.run()).await,
         Options::Tpcds(opt) => Box::pin(opt.run()).await,

benchmarks/src/lib.rs

@@ -23,6 +23,7 @@ pub mod hj;
 pub mod imdb;
 pub mod nlj;
 pub mod smj;
+pub mod sort_pushdown;
 pub mod sort_tpch;
 pub mod tpcds;
 pub mod tpch;