Comparison to baseline (#468)

bench-vortex/benches/tpch_benchmark.rs

@@ -1,5 +1,5 @@
 use bench_vortex::tpch::dbgen::{DBGen, DBGenOptions};
-use bench_vortex::tpch::{load_datasets, Format};
+use bench_vortex::tpch::{load_datasets, tpch_queries, Format};
 use criterion::{criterion_group, criterion_main, Criterion};
 use tokio::runtime::Builder;
 
@@ -32,14 +32,7 @@ fn benchmark(c: &mut Criterion) {
         .block_on(load_datasets(&data_dir, Format::Parquet))
         .unwrap();
 
-    for q in 1..=22 {
-        if q == 15 {
-            // DataFusion does not support query 15 since it has multiple SQL statements.
-            continue;
-        }
-
-        let query = bench_vortex::tpch::tpch_query(q);
-
+    for (q, query) in tpch_queries() {
         let mut group = c.benchmark_group(format!("tpch_q{q}"));
         group.sample_size(10);
 

bench-vortex/src/bin/tpch_benchmark.rs

@@ -2,7 +2,7 @@ use std::sync;
 use std::time::SystemTime;
 
 use bench_vortex::tpch::dbgen::{DBGen, DBGenOptions};
-use bench_vortex::tpch::{load_datasets, tpch_query, Format};
+use bench_vortex::tpch::{load_datasets, tpch_queries, Format};
 use futures::future::join_all;
 use indicatif::ProgressBar;
 use itertools::Itertools;
@@ -52,17 +52,13 @@ async fn main() {
 
     // Send back a channel with the results of Row.
     let (rows_tx, rows_rx) = sync::mpsc::channel();
-    for i in 1..=22 {
-        if i == 15 {
-            continue;
-        }
+    for (q, query) in tpch_queries() {
         let _ctxs = ctxs.clone();
         let _tx = rows_tx.clone();
         let _progress = progress.clone();
         rayon::spawn_fifo(move || {
-            let query = tpch_query(i);
             let mut cells = Vec::with_capacity(formats.len());
-            cells.push(Cell::new(&format!("Q{}", i)));
+            cells.push(Cell::new(&format!("Q{}", q)));
 
             let mut elapsed_us = Vec::new();
             let rt = tokio::runtime::Builder::new_current_thread()
@@ -75,25 +71,25 @@ async fn main() {
                     rt.block_on(async {
                         ctx.sql(&query)
                             .await
-                            .map_err(|e| println!("Failed to run {} {:?}: {}", i, format, e))
+                            .map_err(|e| println!("Failed to run {} {:?}: {}", q, format, e))
                             .unwrap()
                             .collect()
                             .await
-                            .map_err(|e| println!("Failed to collect {} {:?}: {}", i, format, e))
+                            .map_err(|e| println!("Failed to collect {} {:?}: {}", q, format, e))
                             .unwrap();
                     })
                 }
                 let mut measure = Vec::new();
-                for _ in 0..20 {
+                for _ in 0..10 {
                     let start = SystemTime::now();
                     rt.block_on(async {
                         ctx.sql(&query)
                             .await
-                            .map_err(|e| println!("Failed to run {} {:?}: {}", i, format, e))
+                            .map_err(|e| println!("Failed to run {} {:?}: {}", q, format, e))
                             .unwrap()
                             .collect()
                             .await
-                            .map_err(|e| println!("Failed to collect {} {:?}: {}", i, format, e))
+                            .map_err(|e| println!("Failed to collect {} {:?}: {}", q, format, e))
                             .unwrap();
                     });
                     let elapsed = start.elapsed().unwrap();
@@ -119,11 +115,17 @@ async fn main() {
                 } else {
                     "bFdBG"
                 };
-                cells
-                    .push(Cell::new(&format!("{} us", measure.as_micros())).style_spec(style_spec));
+                cells.push(
+                    Cell::new(&format!(
+                        "{} us ({:.2})",
+                        measure.as_micros(),
+                        measure.as_micros() as f64 / baseline.as_micros() as f64
+                    ))
+                    .style_spec(style_spec),
+                );
             }
 
-            _tx.send((i, Row::new(cells))).unwrap();
+            _tx.send((q, Row::new(cells))).unwrap();
         });
     }
 

bench-vortex/src/tpch/mod.rs

@@ -206,6 +206,15 @@ async fn register_vortex(
     Ok(())
 }
 
+pub fn tpch_queries() -> impl Iterator<Item = (usize, String)> {
+    (1..=22)
+        .filter(|q| {
+            // Query 15 has multiple SQL statements so doesn't yet run in DataFusion.
+            *q != 15
+        })
+        .map(|q| (q, tpch_query(q)))
+}
+
 pub fn tpch_query(query_idx: usize) -> String {
     let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"))
         .join("tpch")