Extract merge benchmark to CLI

abhiaagarwal · abhiaagarwal · commit 2cca8c90e6ec · 2025-10-18T19:24:46.000-04:00
Signed-off-by: Abhi Agarwal &lt;abhiaagarwal01@gmail.com&gt;
diff --git a/crates/benchmarks/Cargo.toml b/crates/benchmarks/Cargo.toml
@@ -14,6 +14,7 @@ clap = { version = "4", features = ["derive"] }
 tokio = { workspace = true, features = ["fs", "macros", "rt", "io-util"] }
 url = { workspace = true }
 tempfile = { workspace = true }
+anyhow = "1"
 
 [dependencies.deltalake-core]
 path = "../core"
@@ -26,3 +27,7 @@ divan = "0.1"
 [[bench]]
 name = "merge"
 harness = false
+
+[[bench]]
+name = "smoke"
+harness = false
diff --git a/crates/benchmarks/README.md b/crates/benchmarks/README.md
@@ -48,13 +48,34 @@ A simple CLI is available to run a single merge with configurable parameters (us
 
 Run (from repo root):
 ```bash
-cargo run --profile profiling -p delta-benchmarks -- upsert --matched 0.01 --not-matched 0.10
+cargo run --profile profiling -p delta-benchmarks -- merge --op upsert --matched 0.01 --not-matched 0.10
 ```
 
 Options:
-- `upsert | delete | insert`: operation to benchmark
+- `--op <upsert|delete|insert>`: operation to benchmark
 - `--matched <fraction>`: fraction of rows that match existing keys (default 0.01)
 - `--not-matched <fraction>`: fraction of rows that do not match (default 0.10)
+- `--case <name>`: run one of the predefined merge scenarios mirrored from the Delta Spark suite
+
+List cases with:
+```bash
+cargo run --release -p delta-benchmarks -- merge --case single_insert_only_filesMatchedFraction_0.05_rowsNotMatchedFraction_0.05
+```
+
+## TPC-DS query helper
+
+All 99 TPC-DS SQL statements (matching the Spark benchmark suite) are stored under `queries/tpcds`. The CLI can list or print them:
+
+```bash
+cargo run --release -p delta-benchmarks -- tpcds --list
+cargo run --release -p delta-benchmarks -- tpcds --case q1
+```
+
+There is also a micro-benchmark that iterates over every query string to ensure the include paths stay wired correctly:
+
+```bash
+cargo bench -p delta-benchmarks --bench tpcds
+```
 
 ### Flamegraphs using `samply`
 
@@ -66,4 +87,4 @@ To start,
 cargo install samply --locked
 cargo build --profile profiling -p delta-benchmarks
 samply record ./target/profiling/delta-benchmarks upsert
-```
+```
diff --git a/crates/benchmarks/benches/merge.rs b/crates/benchmarks/benches/merge.rs
@@ -1,7 +1,7 @@
 use std::path::PathBuf;
 
 use delta_benchmarks::{
-    merge_delete, merge_insert, merge_upsert, prepare_source_and_table, MergeOp, MergePerfParams,
+    delete_only_cases, insert_only_cases, prepare_source_and_table, upsert_cases, MergeTestCase,
 };
 
 use divan::{AllocProfiler, Bencher};
@@ -13,8 +13,11 @@ fn main() {
 #[global_allocator]
 static ALLOC: AllocProfiler = AllocProfiler::system();
 
-fn bench_merge(bencher: Bencher, op: MergeOp, params: &MergePerfParams) {
+fn bench_merge_case(bencher: Bencher, case: &MergeTestCase) {
     let rt = tokio::runtime::Runtime::new().unwrap();
+    let case_copy = *case;
+    let params = case_copy.params;
+
     bencher
         .with_inputs(|| {
             let tmp_dir = tempfile::tempdir().unwrap();
@@ -23,74 +26,33 @@ fn bench_merge(bencher: Bencher, op: MergeOp, params: &MergePerfParams) {
                     .unwrap_or_else(|_| "data/tpcds_parquet".to_string()),
             );
             rt.block_on(async move {
-                let (source, table) = prepare_source_and_table(params, &tmp_dir, &parquet_dir)
+                let (source, table) = prepare_source_and_table(&params, &tmp_dir, &parquet_dir)
                     .await
-                    .unwrap();
-                (source, table, tmp_dir)
+                    .expect("prepare inputs");
+                (case, source, table, tmp_dir)
             })
         })
-        .bench_local_values(|(source, table, tmp_dir)| {
+        .bench_local_values(|(case, source, table, tmp_dir)| {
             rt.block_on(async move {
-                let _ = divan::black_box(op(source, table).unwrap().await.unwrap());
+                let (_, metrics) = case.execute(source, table).await.expect("execute merge");
+                case.validate(&metrics).expect("validate merge");
+                divan::black_box(metrics.num_target_rows_inserted);
             });
             drop(tmp_dir);
         });
 }
 
-#[divan::bench(args = [
-    MergePerfParams {
-        sample_matched_rows: 0.05,
-        sample_not_matched_rows: 0.0,
-    }
-])]
-fn delete_only(bencher: Bencher, params: &MergePerfParams) {
-    bench_merge(bencher, merge_delete, params);
+#[divan::bench(args = insert_only_cases())]
+fn insert_only(bencher: Bencher, case: &MergeTestCase) {
+    bench_merge_case(bencher, case);
 }
 
-#[divan::bench(args = [
-    MergePerfParams {
-        sample_matched_rows: 0.00,
-        sample_not_matched_rows: 0.05,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.00,
-        sample_not_matched_rows: 0.50,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.00,
-        sample_not_matched_rows: 1.0,
-    },
-])]
-fn multiple_insert_only(bencher: Bencher, params: &MergePerfParams) {
-    bench_merge(bencher, merge_insert, params);
+#[divan::bench(args = delete_only_cases())]
+fn delete_only(bencher: Bencher, case: &MergeTestCase) {
+    bench_merge_case(bencher, case);
 }
 
-#[divan::bench(args = [
-    MergePerfParams {
-        sample_matched_rows: 0.01,
-        sample_not_matched_rows: 0.1,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.1,
-        sample_not_matched_rows: 0.0,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.1,
-        sample_not_matched_rows: 0.01,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.5,
-        sample_not_matched_rows: 0.001,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.99,
-        sample_not_matched_rows: 0.001,
-    },
-    MergePerfParams {
-        sample_matched_rows: 0.001,
-        sample_not_matched_rows: 0.001,
-    },
-])]
-fn upsert_file_matched(bencher: Bencher, params: &MergePerfParams) {
-    bench_merge(bencher, merge_upsert, params);
+#[divan::bench(args = upsert_cases())]
+fn upsert(bencher: Bencher, case: &MergeTestCase) {
+    bench_merge_case(bencher, case);
 }
diff --git a/crates/benchmarks/src/lib.rs b/crates/benchmarks/src/lib.rs
@@ -2,6 +2,8 @@ pub mod merge;
 pub mod smoke;
 
 pub use merge::{
-    merge_delete, merge_insert, merge_upsert, prepare_source_and_table, MergeOp, MergePerfParams,
+    delete_only_cases, insert_only_cases, merge_case_by_name, merge_case_names, merge_delete,
+    merge_insert, merge_test_cases, merge_upsert, prepare_source_and_table, upsert_cases, MergeOp,
+    MergePerfParams, MergeScenario, MergeTestCase,
 };
 pub use smoke::{run_smoke_once, SmokeParams};
diff --git a/crates/benchmarks/src/main.rs b/crates/benchmarks/src/main.rs
@@ -3,8 +3,8 @@ use std::{path::PathBuf, time::Instant};
 use clap::{Parser, Subcommand, ValueEnum};
 
 use delta_benchmarks::{
-    merge_delete, merge_insert, merge_upsert, prepare_source_and_table, run_smoke_once, MergeOp,
-    MergePerfParams, SmokeParams,
+    merge_case_by_name, merge_case_names, merge_delete, merge_insert, merge_upsert,
+    prepare_source_and_table, run_smoke_once, MergeOp, MergePerfParams, MergeTestCase, SmokeParams,
 };
 use deltalake_core::ensure_table_uri;
 
@@ -28,7 +28,7 @@ enum Command {
     Merge {
         /// Operation to benchmark
         #[arg(value_enum)]
-        op: OpKind,
+        op: Option<OpKind>,
 
         /// Fraction of rows that match an existing key (0.0-1.0)
         #[arg(long, default_value_t = 0.01)]
@@ -37,6 +37,10 @@ enum Command {
         /// Fraction of rows that do not match (0.0-1.0)
         #[arg(long, default_value_t = 0.10)]
         not_matched: f32,
+
+        /// Named test case to run (overrides manual parameters)
+        #[arg(long)]
+        case: Option<String>,
     },
 
     /// Run the smoke workload to validate delta-rs read/write operations
@@ -60,36 +64,36 @@ async fn main() -> anyhow::Result<()> {
             op,
             matched,
             not_matched,
+            case,
         } => {
-            let op_fn: MergeOp = match op {
-                OpKind::Upsert => merge_upsert,
-                OpKind::Delete => merge_delete,
-                OpKind::Insert => merge_insert,
-            };
-
-            let params = MergePerfParams {
-                sample_matched_rows: matched,
-                sample_not_matched_rows: not_matched,
-            };
-
-            let tmp_dir = tempfile::tempdir()?;
-
-            let parquet_dir = PathBuf::from(
-                std::env::var("TPCDS_PARQUET_DIR")
-                    .unwrap_or_else(|_| "crates/benchmarks/data/tpcds_parquet".to_string()),
-            );
-
-            let (source, table) = prepare_source_and_table(&params, &tmp_dir, &parquet_dir).await?;
-
-            let start = Instant::now();
-            let (_table, metrics) = op_fn(source, table)?.await?;
-            let elapsed = start.elapsed();
-
-            println!(
-                "merge_duration_ms={} metrics={:?}",
-                elapsed.as_millis(),
-                metrics
-            );
+            if let Some(case_name) = case.as_deref() {
+                let merge_case = merge_case_by_name(case_name).ok_or_else(|| {
+                    anyhow::anyhow!(
+                        "unknown merge case '{}'. Available cases: {}",
+                        case_name,
+                        merge_case_names().join(", ")
+                    )
+                })?;
+
+                run_merge_case(merge_case).await?;
+            } else {
+                let op = op.ok_or_else(|| {
+                    anyhow::anyhow!("specify an operation (upsert/delete/insert) or provide --case")
+                })?;
+
+                let op_fn: MergeOp = match op {
+                    OpKind::Upsert => merge_upsert,
+                    OpKind::Delete => merge_delete,
+                    OpKind::Insert => merge_insert,
+                };
+
+                let params = MergePerfParams {
+                    sample_matched_rows: matched,
+                    sample_not_matched_rows: not_matched,
+                };
+
+                run_merge_with_params(op_fn, &params).await?;
+            }
         }
         Command::Smoke { rows, table_path } => {
             let params = SmokeParams { rows };
@@ -116,3 +120,49 @@ async fn main() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+async fn run_merge_with_params(op_fn: MergeOp, params: &MergePerfParams) -> anyhow::Result<()> {
+    let tmp_dir = tempfile::tempdir()?;
+    let parquet_dir = PathBuf::from(
+        std::env::var("TPCDS_PARQUET_DIR")
+            .unwrap_or_else(|_| "crates/benchmarks/data/tpcds_parquet".to_string()),
+    );
+
+    let (source, table) = prepare_source_and_table(params, &tmp_dir, &parquet_dir).await?;
+
+    let start = Instant::now();
+    let (_table, metrics) = op_fn(source, table)?.await?;
+    let elapsed = start.elapsed();
+
+    println!(
+        "merge_duration_ms={} metrics={:?}",
+        elapsed.as_millis(),
+        metrics
+    );
+
+    Ok(())
+}
+
+async fn run_merge_case(case: &MergeTestCase) -> anyhow::Result<()> {
+    let tmp_dir = tempfile::tempdir()?;
+    let parquet_dir = PathBuf::from(
+        std::env::var("TPCDS_PARQUET_DIR")
+            .unwrap_or_else(|_| "crates/benchmarks/data/tpcds_parquet".to_string()),
+    );
+
+    let (source, table) = prepare_source_and_table(&case.params, &tmp_dir, &parquet_dir).await?;
+
+    let start = Instant::now();
+    let (_table, metrics) = case.execute(source, table).await?;
+    case.validate(&metrics)?;
+    let elapsed = start.elapsed();
+
+    println!(
+        "merge_case={} merge_duration_ms={} metrics={:?}",
+        case.name,
+        elapsed.as_millis(),
+        metrics
+    );
+
+    Ok(())
+}
diff --git a/crates/benchmarks/src/merge.rs b/crates/benchmarks/src/merge.rs
