Adding more granular data to benchmark results

Author: shefeek-jinnah

benchmark/Cargo.toml

@@ -18,9 +18,20 @@ name = "generate-tpcds"
 path = "src/bin/generate_tpcds.rs"
 
 [dependencies]
-# Use the parent crate
+# DataFusion-DuckLake: Uncomment ONE of the following options
+
+# Option 1: Local development (default)
 datafusion-ducklake = { path = "..", features = ["metadata-duckdb"] }
 
+# Option 2: Specific git commit
+# datafusion-ducklake = { git = "https://github.com/hotdata-dev/datafusion-ducklake", rev = "COMMIT_HASH", features = ["metadata-duckdb"] }
+
+# Option 3: Git branch
+# datafusion-ducklake = { git = "https://github.com/hotdata-dev/datafusion-ducklake", branch = "main", features = ["metadata-duckdb"] }
+
+# Option 4: Published crates.io version
+# datafusion-ducklake = { version = "0.1.0", features = ["metadata-duckdb"] }
+
 # Core dependencies
 datafusion = "50.1.0"
 duckdb = { version = "1.4.1", features = ["bundled"] }

benchmark/src/datafusion_runner.rs

@@ -1,11 +1,12 @@
-use crate::runner::QueryResult;
+use crate::runner::{PhaseTiming, QueryResult};
 use anyhow::{Context, Result};
 use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::prelude::*;
 use datafusion_ducklake::{DuckLakeCatalog, DuckdbMetadataProvider};
 use futures::StreamExt;
 use std::path::Path;
 use std::sync::Arc;
+use std::time::Instant;
 
 pub struct DataFusionRunner {
     ctx: SessionContext,
@@ -39,26 +40,43 @@ impl DataFusionRunner {
     }
 
     pub async fn execute(&self, sql: &str) -> Result<QueryResult> {
+        // Phase 1: SQL parsing + logical planning + optimization
+        let plan_start = Instant::now();
         let df = self
             .ctx
             .sql(sql)
             .await
             .with_context(|| format!("Failed to parse SQL: {}", &sql[..sql.len().min(100)]))?;
+        let plan_ms = plan_start.elapsed().as_secs_f64() * 1000.0;
 
-        // Use streaming to count rows without loading all data into memory
-        let mut stream = df
-            .execute_stream()
+        // Phase 2: Physical plan creation
+        let physical_start = Instant::now();
+        let task_ctx = self.ctx.task_ctx();
+        let physical_plan = df
+            .create_physical_plan()
             .await
+            .context("Failed to create physical plan")?;
+        let physical_ms = physical_start.elapsed().as_secs_f64() * 1000.0;
+
+        // Phase 3: Execution
+        let exec_start = Instant::now();
+        let mut stream = datafusion::physical_plan::execute_stream(physical_plan, task_ctx)
             .context("Failed to execute query stream")?;
 
         let mut row_count = 0;
         while let Some(batch_result) = stream.next().await {
             let batch = batch_result?;
             row_count += batch.num_rows();
         }
+        let exec_ms = exec_start.elapsed().as_secs_f64() * 1000.0;
 
         Ok(QueryResult {
             row_count,
+            phases: Some(PhaseTiming {
+                plan_ms,
+                physical_ms,
+                exec_ms,
+            }),
         })
     }
 }

benchmark/src/duckdb_runner.rs

@@ -92,6 +92,7 @@ impl DuckDbRunner {
 
         Ok(QueryResult {
             row_count,
+            phases: None, // DuckDB doesn't provide phase breakdown
         })
     }
 }

benchmark/src/main.rs

@@ -10,14 +10,16 @@ use anyhow::{Context, Result};
 use clap::Parser;
 use runner::assert_results_match;
 use std::path::PathBuf;
+use std::time::Instant;
 
 #[derive(Parser)]
 #[command(name = "ducklake-benchmark")]
 #[command(about = "Benchmark comparing DuckDB-DuckLake vs DataFusion-DuckLake performance")]
+#[command(version)]
 struct Args {
     /// Path to the DuckLake catalog database
     #[arg(short, long)]
-    catalog: PathBuf,
+    catalog: Option<PathBuf>,
 
     /// Use TPC-H queries from DuckDB's tpch extension (22 queries)
     #[arg(long)]
@@ -46,24 +48,57 @@ struct Args {
     /// Skip queries that fail and continue (useful for TPC-DS where some queries may not be supported)
     #[arg(long)]
     skip_errors: bool,
+
+    /// Skip queries that take longer than this many milliseconds during warmup (e.g., 500)
+    #[arg(long)]
+    timeout_ms: Option<u64>,
+
+    /// Show version information for all components
+    #[arg(long)]
+    show_versions: bool,
+}
+
+fn print_versions() {
+    // Get DuckDB version by querying it
+    let duckdb_version = duckdb::Connection::open_in_memory()
+        .and_then(|conn| conn.query_row("SELECT version()", [], |row| row.get::<_, String>(0)))
+        .unwrap_or_else(|_| "unknown".to_string());
+
+    println!("Component Versions:");
+    println!("  datafusion-ducklake: {}", env!("CARGO_PKG_VERSION"));
+    println!("  datafusion:          {}", datafusion::DATAFUSION_VERSION);
+    println!("  duckdb:              {}", duckdb_version);
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
     let args = Args::parse();
 
+    // Handle --show-versions flag
+    if args.show_versions {
+        print_versions();
+        return Ok(());
+    }
+
+    // Validate catalog is provided for benchmarking
+    let catalog = args.catalog.ok_or_else(|| {
+        anyhow::anyhow!("--catalog is required. Use --show-versions to check component versions.")
+    })?;
+
     println!("╔═══════════════════════════════════════════════════════════════╗");
     println!("║   DuckLake Query Engine Comparison                            ║");
     println!("║   DuckDB-DuckLake vs DataFusion-DuckLake                      ║");
     println!("╚═══════════════════════════════════════════════════════════════╝");
     println!();
-    println!("Catalog:    {:?}", args.catalog);
+    print_versions();
+    println!();
+    println!("Catalog:    {:?}", catalog);
     println!("Iterations: {}", args.iterations);
     println!("Warmup:     {}", !args.no_warmup);
 
     // Initialize runners
-    let duckdb_runner = duckdb_runner::DuckDbRunner::new(&args.catalog)?;
-    let datafusion_runner = datafusion_runner::DataFusionRunner::new(&args.catalog).await?;
+    let duckdb_runner = duckdb_runner::DuckDbRunner::new(&catalog)?;
+    let datafusion_runner = datafusion_runner::DataFusionRunner::new(&catalog).await?;
 
     let mut all_results = Vec::new();
 
@@ -143,10 +178,18 @@ async fn main() -> Result<()> {
             let datafusion_metrics =
                 metrics::benchmark_async(|| datafusion_runner.execute(&query.sql), args.iterations)
                     .await?;
-            println!(
+            print!(
                 "avg={:>8.2}ms  min={:>8.2}ms  max={:>8.2}ms",
                 datafusion_metrics.avg_ms, datafusion_metrics.min_ms, datafusion_metrics.max_ms
             );
+            if let Some(phases) = &datafusion_metrics.phases {
+                println!(
+                    "  (plan={:.1}ms, phys={:.1}ms, exec={:.1}ms)",
+                    phases.plan_ms, phases.physical_ms, phases.exec_ms
+                );
+            } else {
+                println!();
+            }
 
             // Show ratio
             let ratio = datafusion_metrics.avg_ms / duckdb_metrics.avg_ms;
@@ -193,12 +236,27 @@ async fn main() -> Result<()> {
         // Separate warmup phase
         if !args.no_warmup {
             println!("───────────────────────────────────────────────────────────────");
-            println!("Warmup phase: running all queries once and verifying row counts...");
+            if let Some(timeout) = args.timeout_ms {
+                println!("Warmup phase: running all queries once (timeout={}ms)...", timeout);
+            } else {
+                println!("Warmup phase: running all queries once and verifying row counts...");
+            }
             for (i, query) in queries.iter().enumerate() {
                 print!("  [{:>2}/{}] {}... ", i + 1, queries.len(), query.name);
 
+                let start = Instant::now();
                 let duckdb_result = duckdb_runner.execute(&query.sql);
                 let datafusion_result = datafusion_runner.execute(&query.sql).await;
+                let elapsed_ms = start.elapsed().as_millis() as u64;
+
+                // Check timeout
+                if let Some(timeout) = args.timeout_ms {
+                    if elapsed_ms > timeout {
+                        println!("SKIP (too slow: {}ms > {}ms)", elapsed_ms, timeout);
+                        skipped_queries.push((query.name.clone(), format!("Timeout: {}ms", elapsed_ms)));
+                        continue;
+                    }
+                }
 
                 match (&duckdb_result, &datafusion_result) {
                     (Ok(duck), Ok(df)) => {
@@ -212,7 +270,7 @@ async fn main() -> Result<()> {
                                 return Err(e);
                             }
                         } else {
-                            println!("done ({} rows)", duck.row_count);
+                            println!("done ({} rows, {}ms)", duck.row_count, elapsed_ms);
                         }
                     },
                     (Err(e), _) => {
@@ -239,7 +297,7 @@ async fn main() -> Result<()> {
                 }
             }
             if !skipped_queries.is_empty() {
-                println!("\nSkipped {} queries due to errors", skipped_queries.len());
+                println!("\nSkipped {} queries due to errors/timeout", skipped_queries.len());
             }
             println!("Warmup complete.\n");
         }
@@ -277,10 +335,18 @@ async fn main() -> Result<()> {
             let datafusion_metrics =
                 metrics::benchmark_async(|| datafusion_runner.execute(&query.sql), args.iterations)
                     .await?;
-            println!(
+            print!(
                 "avg={:>8.2}ms  min={:>8.2}ms  max={:>8.2}ms",
                 datafusion_metrics.avg_ms, datafusion_metrics.min_ms, datafusion_metrics.max_ms
             );
+            if let Some(phases) = &datafusion_metrics.phases {
+                println!(
+                    "  (plan={:.1}ms, phys={:.1}ms, exec={:.1}ms)",
+                    phases.plan_ms, phases.physical_ms, phases.exec_ms
+                );
+            } else {
+                println!();
+            }
 
             // Show ratio
             let ratio = datafusion_metrics.avg_ms / duckdb_metrics.avg_ms;
@@ -362,10 +428,18 @@ async fn main() -> Result<()> {
             let datafusion_metrics =
                 metrics::benchmark_async(|| datafusion_runner.execute(sql), args.iterations)
                     .await?;
-            println!(
+            print!(
                 "avg={:>8.2}ms  min={:>8.2}ms  max={:>8.2}ms",
                 datafusion_metrics.avg_ms, datafusion_metrics.min_ms, datafusion_metrics.max_ms
             );
+            if let Some(phases) = &datafusion_metrics.phases {
+                println!(
+                    "  (plan={:.1}ms, phys={:.1}ms, exec={:.1}ms)",
+                    phases.plan_ms, phases.physical_ms, phases.exec_ms
+                );
+            } else {
+                println!();
+            }
 
             // Show ratio
             let ratio = datafusion_metrics.avg_ms / duckdb_metrics.avg_ms;