use anyhow::{bail, Context, Result};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use sysinfo::{get_current_pid, ProcessesToUpdate, System};
use tecton_compute::{execute_query, QueryResult};
use tracing::info;
const BENCH_SQL: &str =
"SELECT gender, COUNT(*) AS count FROM data GROUP BY gender ORDER BY count DESC";
#[derive(Debug)]
struct RunMetrics {
wall_ms: u64,
query_ms: u64,
peak_rss_bytes: u64,
rows_returned: usize,
total_rows_affected: u64,
optimization: Option<String>,
}
pub async fn run_benchmark(file: PathBuf, keep_parquet: bool) -> Result<()> {
let file = resolve_bench_file(&file)?;
if !file
.extension()
.and_then(|e| e.to_str())
.is_some_and(|e| e.eq_ignore_ascii_case("csv"))
{
bail!("benchmark expects a .csv file (got {})", file.display());
}
let meta = std::fs::metadata(&file)?;
let size_mb = meta.len() as f64 / (1024.0 * 1024.0);
let parquet = file.with_extension("parquet");
println!();
println!("╔══════════════════════════════════════════════════════════════╗");
println!("║ Tecton Performance Benchmark Report ║");
println!("╚══════════════════════════════════════════════════════════════╝");
println!();
println!("Dataset : {}", file.display());
println!("Size : {size_mb:.2} MiB ({:.3} GiB)", size_mb / 1024.0);
println!("Query : {BENCH_SQL}");
println!();
if parquet.exists() {
std::fs::remove_file(&parquet)
.with_context(|| format!("failed to remove {}", parquet.display()))?;
println!("Cleared existing Parquet cache for cold-run measurement.");
}
println!("── Run 1: Cold (CSV → Parquet conversion + query) ──");
let cold = timed_query(&file, "cold").await?;
print_run(&cold);
if !parquet.exists() && size_mb > 100.0 {
println!(
"Note: Parquet cache not found after cold run (file may be ≤100 MiB; conversion skipped)."
);
} else if parquet.exists() {
let pq_mb = std::fs::metadata(&parquet)?.len() as f64 / (1024.0 * 1024.0);
println!("Parquet cache : {} ({pq_mb:.2} MiB)", parquet.display());
}
println!();
println!("── Run 2: Warm (Parquet cache / repeat query) ──");
let warm = timed_query(&file, "warm").await?;
print_run(&warm);
println!();
print_comparison(&cold, &warm, size_mb);
if !keep_parquet && parquet.exists() {
}
let _ = keep_parquet;
Ok(())
}
async fn timed_query(file: &Path, label: &str) -> Result<RunMetrics> {
let sampling = Arc::new(AtomicBool::new(true));
let peak = Arc::new(AtomicU64::new(0));
let sampling_flag = Arc::clone(&sampling);
let peak_bytes = Arc::clone(&peak);
let sampler = tokio::spawn(async move {
sample_peak_rss(sampling_flag, peak_bytes).await;
});
tokio::time::sleep(Duration::from_millis(80)).await;
let wall = Instant::now();
let result: QueryResult = execute_query(file.to_string_lossy().as_ref(), BENCH_SQL)
.await
.map_err(|e| anyhow::anyhow!(e))?;
let wall_ms = wall.elapsed().as_millis() as u64;
sampling.store(false, Ordering::SeqCst);
let _ = sampler.await;
let peak_rss_bytes = peak.load(Ordering::SeqCst);
info!(
label,
wall_ms,
query_ms = result.execution_time_ms,
peak_rss_bytes,
"benchmark run complete"
);
let _ = label;
Ok(RunMetrics {
wall_ms,
query_ms: result.execution_time_ms,
peak_rss_bytes,
rows_returned: result.data.len(),
total_rows_affected: result.total_rows_affected,
optimization: result.optimization_applied,
})
}
async fn sample_peak_rss(active: Arc<AtomicBool>, peak: Arc<AtomicU64>) {
let Ok(pid) = get_current_pid() else {
return;
};
let mut sys = System::new();
while active.load(Ordering::Relaxed) {
sys.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
if let Some(proc) = sys.process(pid) {
peak.fetch_max(proc.memory(), Ordering::Relaxed);
}
tokio::time::sleep(Duration::from_millis(40)).await;
}
sys.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
if let Some(proc) = sys.process(pid) {
peak.fetch_max(proc.memory(), Ordering::Relaxed);
}
}
fn print_run(m: &RunMetrics) {
println!(" Wall clock : {:>10} ms", m.wall_ms);
println!(" Engine time : {:>10} ms", m.query_ms);
println!(
" Peak RSS : {:>10} ({})",
format_bytes(m.peak_rss_bytes),
m.peak_rss_bytes
);
println!(" Rows returned : {:>10}", m.rows_returned);
println!(" Rows affected : {:>10}", m.total_rows_affected);
println!(
" Optimization : {}",
m.optimization.as_deref().unwrap_or("(none)")
);
}
fn print_comparison(cold: &RunMetrics, warm: &RunMetrics, size_mb: f64) {
let speedup = if warm.wall_ms == 0 {
f64::INFINITY
} else {
cold.wall_ms as f64 / warm.wall_ms as f64
};
let saved = cold.wall_ms.saturating_sub(warm.wall_ms);
let saved_pct = if cold.wall_ms == 0 {
0.0
} else {
(saved as f64 / cold.wall_ms as f64) * 100.0
};
println!("── Summary ──");
println!("┌────────────────────┬──────────────┬──────────────┐");
println!("│ Metric │ Cold (CSV) │ Warm (Parq.) │");
println!("├────────────────────┼──────────────┼──────────────┤");
println!(
"│ Wall clock (ms) │ {:>12} │ {:>12} │",
cold.wall_ms, warm.wall_ms
);
println!(
"│ Engine time (ms) │ {:>12} │ {:>12} │",
cold.query_ms, warm.query_ms
);
println!(
"│ Peak RSS │ {:>12} │ {:>12} │",
format_bytes(cold.peak_rss_bytes),
format_bytes(warm.peak_rss_bytes)
);
println!("└────────────────────┴──────────────┴──────────────┘");
println!();
println!("Dataset size : {size_mb:.2} MiB");
println!("Speedup (cold/warm) : {speedup:.2}×");
println!("Time saved : {saved} ms ({saved_pct:.1}%)");
println!();
println!("Interpretation:");
println!(" • Cold run includes CSV scan + optional Parquet conversion.");
println!(" • Warm run reuses the sibling .parquet cache when present.");
println!(" • Auto-optimization triggers for CSV files larger than 100 MiB.");
println!();
}
fn resolve_bench_file(file: &Path) -> Result<PathBuf> {
if file.exists() {
return Ok(if file.is_absolute() {
file.to_path_buf()
} else {
std::env::current_dir()?.join(file)
});
}
let under_data = PathBuf::from("apps/tecton-ui/src-tauri/data").join(file);
if under_data.exists() {
return Ok(std::env::current_dir()?.join(under_data));
}
let under_dot = PathBuf::from("data").join(file);
if under_dot.exists() {
return Ok(std::env::current_dir()?.join(under_dot));
}
bail!("benchmark file not found: {}", file.display())
}
fn format_bytes(bytes: u64) -> String {
const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
let mut value = bytes as f64;
let mut unit = 0usize;
while value >= 1024.0 && unit < UNITS.len() - 1 {
value /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{bytes} {}", UNITS[unit])
} else {
format!("{value:.2} {}", UNITS[unit])
}
}