feat: verify duckdb tpch results (#3365)

Signed-off-by: Alexander Droste <[email protected]>

Author: 0ax1

Cargo.lock

@@ -152,6 +152,7 @@ rustc-hash = "2"
 serde = "1.0.203"
 serde_json = "1.0.116"
 serde_test = "1.0.176"
+similar = "2.7.0"
 simplelog = "0.12"
 sketches-ddsketch = "0.3.0"
 static_assertions = "1.1"

Cargo.toml

@@ -51,6 +51,7 @@ regex = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+similar = { workspace = true }
 tabled = { workspace = true, features = ["std"] }
 tar = { workspace = true }
 tempfile = { workspace = true }

bench-vortex/Cargo.toml

@@ -181,7 +181,7 @@ async fn bench_main(
                     .path()
                     .join(format!("duckdb-file-{}.db", format.name()));
 
-                let executor = DuckDBExecutor::new(duckdb_resolved_path.to_owned(), duckdb_file);
+                let executor = DuckDBExecutor::new(duckdb_resolved_path, duckdb_file);
                 register_tables(&executor, &url, format, BenchmarkDataset::TpcDS)?;
 
                 for (query_idx, sql_query) in tpch_queries.clone() {

bench-vortex/src/bin/tpcds.rs

@@ -1,6 +1,7 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
+use std::{env, fs};
 
 use anyhow::anyhow;
 use bench_vortex::ddb::{DuckDBExecutor, register_tables};
@@ -25,6 +26,7 @@ use datafusion::physical_plan::metrics::{Label, MetricsSet};
 use indicatif::ProgressBar;
 use itertools::Itertools;
 use log::{info, warn};
+use similar::{ChangeTag, TextDiff};
 use tempfile::tempdir;
 use url::Url;
 use vortex::aliases::hash_map::HashMap;
@@ -455,10 +457,78 @@ async fn bench_main(
     }
 
     if verify_row_counts(&row_counts, expected_row_counts, &queries, &exclude_queries) {
-        Err(anyhow!("Mismatched row counts. See logs for details."))
-    } else {
-        anyhow::Ok(())
+        return Err(anyhow!("Mismatched row counts. See logs for details."));
+    }
+
+    if targets.iter().any(|t| t.engine() == Engine::DuckDB) {
+        verify_duckdb_tpch_results(scale_factor, duckdb_resolved_path)?;
+    }
+
+    anyhow::Ok(())
+}
+
+fn verify_duckdb_tpch_results(scale_factor: u8, duckdb_path: PathBuf) -> anyhow::Result<()> {
+    let query_dir = PathBuf::from("duckdb-vortex/duckdb/extension/tpch/dbgen/queries");
+    let tmp_dir = format!(
+        "{}/spiral-tpch",
+        // $RUNNER_TEMP is defined by GitHub Actions.
+        env::var("TMPDIR").unwrap_or(env::var("RUNNER_TEMP")?)
+    );
+    if PathBuf::from(&tmp_dir).exists() {
+        fs::remove_dir_all(&tmp_dir)?;
+    }
+    fs::create_dir(&tmp_dir)?;
+    let db_path = format!("{tmp_dir}/tpch_results_sf.db");
+
+    let executor = DuckDBExecutor::new(duckdb_path, &db_path);
+    ddb::execute_tpch_query(&[format!("CALL dbgen(sf={})", scale_factor)], &executor)?;
+
+    let query_files = fs::read_dir(query_dir)?
+        .filter_map(Result::ok)
+        .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "sql"))
+        .collect::<Vec<_>>();
+
+    for query_file in &query_files {
+        let query_file_path = query_file.path();
+        let query_name = query_file_path
+            .file_stem()
+            .and_then(|stem| stem.to_str())
+            .ok_or_else(|| anyhow!("Invalid query filename"))?;
+
+        let create_table = format!(
+            "CREATE OR REPLACE TABLE {query_name}_result AS {};",
+            fs::read_to_string(&query_file_path)?
+        );
+
+        let csv_actual = format!("{tmp_dir}/{query_name}.csv");
+        let write_csv =
+            format!("COPY {query_name}_result TO '{csv_actual}' (HEADER, DELIMITER '|');",);
+
+        ddb::execute_tpch_query(&[create_table, write_csv], &executor)?;
+
+        let csv_expected = format!("bench-vortex/tpch_results/duckdb/{query_name}.csv");
+        let expected = fs::read_to_string(csv_expected)?;
+        let actual = fs::read_to_string(csv_actual)?;
+
+        if expected != actual {
+            let diff = TextDiff::from_lines(&expected, &actual);
+
+            for change in diff.iter_all_changes() {
+                let sign = match change.tag() {
+                    ChangeTag::Delete => "-",
+                    ChangeTag::Insert => "+",
+                    ChangeTag::Equal => " ",
+                };
+                print!("{}{}", sign, change);
+            }
+
+            return Err(anyhow!(format!(
+                "query output does not match the reference for {query_name}"
+            )));
+        }
     }
+
+    Ok(())
 }
 
 fn validate_args(engines: &[Engine], args: &Args) {

bench-vortex/src/bin/tpch.rs

@@ -30,10 +30,10 @@ impl DuckDBExecutor {
         command
     }
 
-    pub fn new(duckdb_path: PathBuf, duckdb_file: PathBuf) -> Self {
+    pub fn new(duckdb_path: impl AsRef<Path>, duckdb_file: impl AsRef<Path>) -> Self {
         Self {
-            duckdb_path,
-            duckdb_file,
+            duckdb_path: duckdb_path.as_ref().to_path_buf(),
+            duckdb_file: duckdb_file.as_ref().to_path_buf(),
         }
     }
 }

bench-vortex/src/engines/ddb/mod.rs

@@ -0,0 +1,5 @@
+l_returnflag|l_linestatus|sum_qty|sum_base_price|sum_disc_price|sum_charge|avg_qty|avg_price|avg_disc|count_order
+A|F|37734107.00|56586554400.73|53758257134.8700|55909065222.827692|25.522005853257337|38273.129734621674|0.049985295838397614|1478493
+N|F|991417.00|1487504710.38|1413082168.0541|1469649223.194375|25.516471920522985|38284.4677608483|0.0500934266742163|38854
+N|O|74476040.00|111701729697.74|106118230307.6056|110367043872.497010|25.50222676958499|38249.11798890827|0.04999658605370408|2920374
+R|F|37719753.00|56568041380.90|53741292684.6040|55889619119.831932|25.50579361269077|38250.85462609966|0.05000940583012706|1478870