feat: generate reversed-name data for sort pushdown benchmark

Generate benchmark data with file names that don't match sort key order:
  c_high.parquet: l_orderkey 1-200k     (c sorts last, but has lowest keys)
  b_mid.parquet:  l_orderkey 200k-400k
  a_low.parquet:  l_orderkey 400k+      (a sorts first, but has highest keys)

This ensures the sort pushdown optimizer must reorder files by statistics
to achieve sort elimination. Without the optimization, SortExec stays.

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

Author: zhuqi-lucas

benchmarks/bench.sh

@@ -314,8 +314,7 @@ main() {
                     data_tpch "1" "parquet"
                     ;;
                 sort_pushdown|sort_pushdown_sorted)
-                    # same data as for tpch
-                    data_tpch "1" "parquet"
+                    data_sort_pushdown
                     ;;
                 sort_tpch)
                     # same data as for tpch
@@ -1085,19 +1084,80 @@ run_external_aggr() {
 }
 
 # Runs the sort pushdown benchmark (without WITH ORDER)
-run_sort_pushdown() {
+# Generates sort pushdown benchmark data: multiple sorted parquet files with
+# reversed naming so alphabetical order does NOT match sort key order.
+# This forces the sort pushdown optimizer to reorder files by statistics.
+#
+# Files created (l_orderkey ranges):
+#   c_high.parquet: 1-200000       (c sorts last alphabetically, but has lowest keys)
+#   b_mid.parquet:  200001-400000
+#   a_low.parquet:  400001-600001  (a sorts first alphabetically, but has highest keys)
+data_sort_pushdown() {
+    SORT_PUSHDOWN_DIR="${DATA_DIR}/sort_pushdown/lineitem"
+    if [ -d "${SORT_PUSHDOWN_DIR}" ] && [ "$(ls -A ${SORT_PUSHDOWN_DIR}/*.parquet 2>/dev/null)" ]; then
+        echo "Sort pushdown data already exists at ${SORT_PUSHDOWN_DIR}"
+        return
+    fi
+
+    echo "Generating sort pushdown benchmark data..."
+
+    # First ensure we have TPC-H data to work with
+    data_tpch "1" "parquet"
+
     TPCH_DIR="${DATA_DIR}/tpch_sf1"
+    mkdir -p "${SORT_PUSHDOWN_DIR}"
+
+    # Build datafusion-cli if needed
+    $CARGO_COMMAND --bin datafusion-cli
+
+    DATAFUSION_CLI="./target/release/datafusion-cli"
+    if [ ! -f "$DATAFUSION_CLI" ]; then
+        DATAFUSION_CLI="./target/debug/datafusion-cli"
+    fi
+
+    echo "Creating sorted parquet files with reversed naming..."
+
+    $DATAFUSION_CLI --command "
+        CREATE EXTERNAL TABLE lineitem
+        STORED AS PARQUET
+        LOCATION '${TPCH_DIR}/lineitem/';
+
+        COPY (
+            SELECT * FROM lineitem
+            WHERE l_orderkey <= 200000
+            ORDER BY l_orderkey ASC
+        ) TO '${SORT_PUSHDOWN_DIR}/c_high.parquet';
+
+        COPY (
+            SELECT * FROM lineitem
+            WHERE l_orderkey > 200000 AND l_orderkey <= 400000
+            ORDER BY l_orderkey ASC
+        ) TO '${SORT_PUSHDOWN_DIR}/b_mid.parquet';
+
+        COPY (
+            SELECT * FROM lineitem
+            WHERE l_orderkey > 400000
+            ORDER BY l_orderkey ASC
+        ) TO '${SORT_PUSHDOWN_DIR}/a_low.parquet';
+    "
+
+    echo "Sort pushdown data generated at ${SORT_PUSHDOWN_DIR}"
+    ls -la "${SORT_PUSHDOWN_DIR}"
+}
+
+run_sort_pushdown() {
+    SORT_PUSHDOWN_DIR="${DATA_DIR}/sort_pushdown"
     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}
+    debug_run $CARGO_COMMAND --bin dfbench -- sort-pushdown --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 (enables sort elimination)
 run_sort_pushdown_sorted() {
-    TPCH_DIR="${DATA_DIR}/tpch_sf1"
+    SORT_PUSHDOWN_DIR="${DATA_DIR}/sort_pushdown"
     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}
+    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 integration benchmark