use std::time::{Duration, Instant, SystemTime};
use std::path::{Path, PathBuf};
use std::fs;
use crate::SiftDB;
use crate::locking::SWMRLockManager;
use crate::compaction::CollectionCompactor;
use crate::incremental::IncrementalUpdater;
use crate::ingest::{Ingester, IngestOptions};
use anyhow::Result;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BenchmarkResults {
pub name: String,
pub duration: Duration,
pub files_processed: u64,
pub bytes_processed: u64,
pub queries_per_second: Option<f64>,
pub throughput_mbps: Option<f64>,
}
#[derive(Serialize, Deserialize)]
pub struct BenchmarkSuite {
pub version: String,
pub timestamp: String,
pub git_commit: Option<String>,
pub test_environment: TestEnvironment,
pub benchmarks: Vec<BenchmarkResults>,
}
#[derive(Serialize, Deserialize)]
pub struct TestEnvironment {
pub os: String,
pub cpu: String,
pub memory: String,
}
impl BenchmarkResults {
pub fn print(&self) {
println!("=== {} ===", self.name);
println!("Duration: {:.2}s", self.duration.as_secs_f64());
if self.files_processed > 0 {
println!("Files processed: {}", self.files_processed);
println!("Files/sec: {:.1}", self.files_processed as f64 / self.duration.as_secs_f64());
}
if self.bytes_processed > 0 {
let mb = self.bytes_processed as f64 / (1024.0 * 1024.0);
println!("Data processed: {:.2} MB", mb);
if let Some(throughput) = self.throughput_mbps {
println!("Throughput: {:.2} MB/s", throughput);
}
}
if let Some(qps) = self.queries_per_second {
println!("Queries/sec: {:.1}", qps);
}
println!();
}
}
pub struct SiftDBBenchmark {
collection_path: std::path::PathBuf,
source_path: std::path::PathBuf,
lock_manager: SWMRLockManager,
}
impl SiftDBBenchmark {
pub fn new<P1: AsRef<Path>, P2: AsRef<Path>>(collection_path: P1, source_path: P2) -> Self {
let collection_path = collection_path.as_ref().to_path_buf();
let lock_manager = SWMRLockManager::new(&collection_path);
Self {
collection_path,
source_path: source_path.as_ref().to_path_buf(),
lock_manager,
}
}
pub fn run_all(&mut self) -> Vec<BenchmarkResults> {
println!("🚀 SiftDB Performance Benchmark");
println!("================================");
println!();
let mut results = Vec::new();
results.push(self.bench_init());
results.push(self.bench_import());
results.extend(self.bench_searches());
self.print_summary(&results);
if let Err(e) = self.save_results(&results) {
eprintln!("Warning: Failed to save benchmark results: {}", e);
}
results
}
pub fn run_all_quiet(&mut self) -> Vec<BenchmarkResults> {
let mut results = Vec::new();
results.push(self.bench_init_quiet());
results.push(self.bench_import_quiet());
results.extend(self.bench_searches_quiet());
results
}
fn bench_init_quiet(&self) -> BenchmarkResults {
let start = Instant::now();
SiftDB::init(&self.collection_path).expect("Failed to initialize collection");
let duration = start.elapsed();
BenchmarkResults {
name: "Collection Initialization".to_string(),
duration,
files_processed: 0,
bytes_processed: 0,
queries_per_second: None,
throughput_mbps: None,
}
}
fn bench_import_quiet(&mut self) -> BenchmarkResults {
let start = Instant::now();
let _db = SiftDB::open(&self.collection_path).expect("Failed to open collection");
let mut options = IngestOptions::default();
options.include_patterns = vec!["**/*.rs".to_string(), "**/*.md".to_string(), "**/*.toml".to_string(), "**/*.json".to_string()];
let mut ingester = Ingester::new(self.collection_path.clone(), options);
let (source_files, source_bytes) = self.count_source_files();
let stats = ingester.ingest_from_fs(&self.source_path).expect("Failed to ingest");
let duration = start.elapsed();
BenchmarkResults {
name: "File Import".to_string(),
duration,
files_processed: stats.ingested,
bytes_processed: source_bytes,
queries_per_second: None,
throughput_mbps: Some(source_bytes as f64 / (1024.0 * 1024.0) / duration.as_secs_f64()),
}
}
fn bench_searches_quiet(&self) -> Vec<BenchmarkResults> {
let mut results = Vec::new();
let db = SiftDB::open(&self.collection_path).expect("Failed to open collection");
let mut snapshot = db.snapshot().expect("Failed to create snapshot");
let queries = vec![
("fn", "Function definitions"),
("println", "Print statements"),
("use", "Import statements"),
("struct", "Struct definitions"),
("impl", "Implementation blocks"),
("pub", "Public items"),
("let", "Variable declarations"),
("match", "Pattern matching"),
("async", "Async code"),
("Result", "Result types"),
];
for (query, description) in queries {
results.push(self.bench_single_search_quiet(&mut snapshot, query, description));
}
results
}
fn bench_single_search_quiet(&self, snapshot: &mut crate::Snapshot, query: &str, description: &str) -> BenchmarkResults {
let iterations = 10;
let mut total_duration = Duration::new(0, 0);
let mut total_hits = 0;
snapshot.find(query, None, Some(1000)).ok();
for _ in 0..iterations {
let start = Instant::now();
if let Ok(hits) = snapshot.find(query, None, Some(1000)) {
total_hits = hits.len();
}
total_duration += start.elapsed();
}
let avg_duration = total_duration / iterations as u32;
let qps = if avg_duration.as_secs_f64() > 0.0 {
1.0 / avg_duration.as_secs_f64()
} else {
f64::INFINITY
};
BenchmarkResults {
name: format!("Search: '{}' ({})", query, description),
duration: avg_duration,
files_processed: total_hits as u64,
bytes_processed: 0,
queries_per_second: Some(qps),
throughput_mbps: None,
}
}
fn bench_init(&self) -> BenchmarkResults {
let start = Instant::now();
SiftDB::init(&self.collection_path).expect("Failed to initialize collection");
let duration = start.elapsed();
BenchmarkResults {
name: "Collection Initialization".to_string(),
duration,
files_processed: 0,
bytes_processed: 0,
queries_per_second: None,
throughput_mbps: None,
}
}
fn bench_import(&mut self) -> BenchmarkResults {
println!("📁 Starting import benchmark...");
let start = Instant::now();
let _db = SiftDB::open(&self.collection_path).expect("Failed to open collection");
let mut options = IngestOptions::default();
options.include_patterns = vec!["**/*.rs".to_string(), "**/*.md".to_string(), "**/*.toml".to_string(), "**/*.json".to_string()];
let mut ingester = Ingester::new(self.collection_path.clone(), options);
let (source_files, source_bytes) = self.count_source_files();
println!(" Source files found: {}", source_files);
println!(" Source data: {:.2} MB", source_bytes as f64 / (1024.0 * 1024.0));
let stats = ingester.ingest_from_fs(&self.source_path).expect("Failed to ingest");
let duration = start.elapsed();
let storage_bytes = self.calculate_total_bytes();
let compression_ratio = if source_bytes > 0 {
storage_bytes as f64 / source_bytes as f64
} else {
1.0
};
println!(" ✅ Import completed in {:.2}s", duration.as_secs_f64());
println!(" 📊 Files ingested: {} ({} skipped, {} errors)",
stats.ingested, stats.skipped, stats.errors);
println!(" 💾 Storage size: {:.2} MB (ratio: {:.2}x)",
storage_bytes as f64 / (1024.0 * 1024.0), compression_ratio);
BenchmarkResults {
name: "File Import".to_string(),
duration,
files_processed: stats.ingested,
bytes_processed: source_bytes,
queries_per_second: None,
throughput_mbps: Some(source_bytes as f64 / (1024.0 * 1024.0) / duration.as_secs_f64()),
}
}
fn bench_searches(&self) -> Vec<BenchmarkResults> {
let mut results = Vec::new();
let db = SiftDB::open(&self.collection_path).expect("Failed to open collection");
let mut snapshot = db.snapshot().expect("Failed to create snapshot");
let queries = vec![
("fn", "Function definitions"),
("println", "Print statements"),
("use", "Import statements"),
("struct", "Struct definitions"),
("impl", "Implementation blocks"),
("pub", "Public items"),
("let", "Variable declarations"),
("match", "Pattern matching"),
("async", "Async code"),
("Result", "Result types"),
];
for (query, description) in queries {
results.push(self.bench_single_search(&mut snapshot, query, description));
}
results
}
fn bench_single_search(&self, snapshot: &mut crate::Snapshot, query: &str, description: &str) -> BenchmarkResults {
let iterations = 10;
let mut total_duration = Duration::new(0, 0);
let mut total_hits = 0;
snapshot.find(query, None, Some(1000)).ok();
for _ in 0..iterations {
let start = Instant::now();
if let Ok(hits) = snapshot.find(query, None, Some(1000)) {
total_hits = hits.len();
}
total_duration += start.elapsed();
}
let avg_duration = total_duration / iterations as u32;
let qps = iterations as f64 / total_duration.as_secs_f64();
BenchmarkResults {
name: format!("Search: '{}' ({})", query, description),
duration: avg_duration,
files_processed: total_hits as u64,
bytes_processed: 0,
queries_per_second: Some(qps),
throughput_mbps: None,
}
}
fn count_source_files(&self) -> (u64, u64) {
let mut file_count = 0;
let mut byte_count = 0;
let walker = ignore::WalkBuilder::new(&self.source_path)
.hidden(false)
.git_ignore(true)
.build();
for entry in walker {
if let Ok(entry) = entry {
let path = entry.path();
if path.is_file() {
let path_str = path.to_string_lossy();
if path_str.ends_with(".rs") || path_str.ends_with(".md") ||
path_str.ends_with(".toml") || path_str.ends_with(".json") {
file_count += 1;
if let Ok(metadata) = path.metadata() {
byte_count += metadata.len();
}
}
}
}
}
(file_count, byte_count)
}
fn calculate_total_bytes(&self) -> u64 {
let mut total = 0;
if let Ok(entries) = fs::read_dir(&self.collection_path.join("store")) {
for entry in entries.flatten() {
if let Ok(metadata) = entry.metadata() {
total += metadata.len();
}
}
}
total
}
fn save_results(&self, results: &[BenchmarkResults]) -> Result<(), Box<dyn std::error::Error>> {
let suite = BenchmarkSuite {
version: env!("CARGO_PKG_VERSION").to_string(),
timestamp: SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)?
.as_secs().to_string(),
git_commit: self.get_git_commit(),
test_environment: TestEnvironment {
os: std::env::consts::OS.to_string(),
cpu: "unknown".to_string(), memory: "unknown".to_string(),
},
benchmarks: results.to_vec(),
};
let benchmarks_dir = self.collection_path.parent()
.unwrap_or(&self.collection_path)
.join("benchmarks/results");
std::fs::create_dir_all(&benchmarks_dir)?;
let filename = format!("benchmark-{}.json", suite.timestamp);
let filepath = benchmarks_dir.join(filename);
let json = serde_json::to_string_pretty(&suite)?;
std::fs::write(filepath, json)?;
Ok(())
}
fn get_git_commit(&self) -> Option<String> {
std::process::Command::new("git")
.arg("rev-parse")
.arg("--short")
.arg("HEAD")
.current_dir(self.collection_path.parent().unwrap_or(&self.collection_path))
.output()
.ok()
.and_then(|output| {
if output.status.success() {
String::from_utf8(output.stdout).ok()
.map(|s| s.trim().to_string())
} else {
None
}
})
}
fn print_summary(&self, results: &[BenchmarkResults]) {
println!("📊 Benchmark Summary");
println!("===================");
for result in results {
result.print();
}
let import_result = results.iter().find(|r| r.name.contains("Import"));
if let Some(import) = import_result {
println!("🎯 Key Performance Metrics:");
println!("- Import Rate: {:.0} files/sec", import.files_processed as f64 / import.duration.as_secs_f64());
if let Some(throughput) = import.throughput_mbps {
println!("- Import Throughput: {:.1} MB/s", throughput);
}
let search_results: Vec<_> = results.iter()
.filter(|r| r.name.contains("Search"))
.collect();
if !search_results.is_empty() {
let avg_qps: f64 = search_results.iter()
.filter_map(|r| r.queries_per_second)
.sum::<f64>() / search_results.len() as f64;
println!("- Average Search Rate: {:.1} queries/sec", avg_qps);
}
}
println!("✅ Benchmark completed successfully!");
}
}
pub struct AdvancedBenchmark {
collection_path: PathBuf,
source_path: PathBuf,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct AdvancedBenchmarkResults {
pub incremental_update: IncrementalUpdateBenchmark,
pub compaction: CompactionBenchmark,
pub overall_stats: OverallAdvancedStats,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct IncrementalUpdateBenchmark {
pub initial_import_time_ms: u64,
pub file_change_detection_time_ms: u64,
pub delta_application_time_ms: u64,
pub total_update_time_ms: u64,
pub files_changed: usize,
pub files_added: usize,
pub files_removed: usize,
pub changes_per_second: f64,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct CompactionBenchmark {
pub tombstone_analysis_time_ms: u64,
pub compaction_time_ms: u64,
pub total_time_ms: u64,
pub tombstones_removed: usize,
pub segments_compacted: usize,
pub space_reclaimed_bytes: u64,
pub compaction_throughput_mb_per_sec: f64,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct OverallAdvancedStats {
pub total_benchmark_time_ms: u64,
pub collection_size_bytes: u64,
pub source_files_count: usize,
pub features_tested: Vec<String>,
}
impl AdvancedBenchmark {
pub fn new(collection_path: &Path, source_path: &Path) -> Self {
Self {
collection_path: collection_path.to_path_buf(),
source_path: source_path.to_path_buf(),
}
}
pub fn run_all(&mut self) -> Result<AdvancedBenchmarkResults> {
let start_time = Instant::now();
println!("📋 Phase 1: Incremental Update Benchmark");
let incremental_results = self.benchmark_incremental_updates()?;
println!(" ✅ Completed in {}ms", incremental_results.total_update_time_ms);
println!();
println!("📋 Phase 2: Compaction Benchmark");
let compaction_results = self.benchmark_compaction()?;
println!(" ✅ Completed in {}ms", compaction_results.total_time_ms);
println!();
let total_time = start_time.elapsed().as_millis() as u64;
let collection_size = self.calculate_collection_size()?;
let source_files = self.count_source_files()?;
let overall_stats = OverallAdvancedStats {
total_benchmark_time_ms: total_time,
collection_size_bytes: collection_size,
source_files_count: source_files,
features_tested: vec![
"incremental_updates".to_string(),
"compaction".to_string(),
"delta_manifests".to_string(),
"file_timestamp_tracking".to_string(),
],
};
println!("🎯 Benchmark Summary");
println!(" Total time: {}ms", total_time);
println!(" Collection size: {} bytes", collection_size);
println!(" Source files: {}", source_files);
println!(" Incremental update performance: {:.2} changes/sec", incremental_results.changes_per_second);
println!(" Compaction throughput: {:.2} MB/sec", compaction_results.compaction_throughput_mb_per_sec);
Ok(AdvancedBenchmarkResults {
incremental_update: incremental_results,
compaction: compaction_results,
overall_stats,
})
}
pub fn run_all_quiet(&mut self) -> AdvancedBenchmarkResults {
self.run_all().unwrap_or_else(|_| AdvancedBenchmarkResults {
incremental_update: IncrementalUpdateBenchmark {
initial_import_time_ms: 0,
file_change_detection_time_ms: 0,
delta_application_time_ms: 0,
total_update_time_ms: 0,
files_changed: 0,
files_added: 0,
files_removed: 0,
changes_per_second: 0.0,
},
compaction: CompactionBenchmark {
tombstone_analysis_time_ms: 0,
compaction_time_ms: 0,
total_time_ms: 0,
tombstones_removed: 0,
segments_compacted: 0,
space_reclaimed_bytes: 0,
compaction_throughput_mb_per_sec: 0.0,
},
overall_stats: OverallAdvancedStats {
total_benchmark_time_ms: 0,
collection_size_bytes: 0,
source_files_count: 0,
features_tested: vec![],
},
})
}
fn benchmark_incremental_updates(&self) -> Result<IncrementalUpdateBenchmark> {
if self.collection_path.exists() {
fs::remove_dir_all(&self.collection_path).ok();
}
let initial_start = Instant::now();
let db = SiftDB::init(&self.collection_path)?;
let mut options = IngestOptions::default();
options.include_patterns = vec!["**/*.rs".to_string(), "**/*.md".to_string()];
let mut ingester = Ingester::new(self.collection_path.clone(), options);
ingester.ingest_from_fs(&self.source_path)?;
let initial_import_time = initial_start.elapsed().as_millis() as u64;
let temp_dir = self.source_path.join("temp_changes");
fs::create_dir_all(&temp_dir).ok();
for i in 0..5 {
let content = format!("New test file {} with timestamp", i);
fs::write(temp_dir.join(format!("new_file_{}.txt", i)), content)?;
}
let detection_start = Instant::now();
let updater = IncrementalUpdater::new(&self.collection_path);
let changes = updater.scan_for_changes(&self.source_path, &[], &[])?;
let detection_time = detection_start.elapsed().as_millis() as u64;
let application_start = Instant::now();
let _delta_manifest = updater.apply_changes(changes.clone(), &self.source_path)?;
let application_time = application_start.elapsed().as_millis() as u64;
let total_time = detection_time + application_time;
let total_changes = changes.len();
let changes_per_second = if total_time > 0 {
(total_changes as f64) / (total_time as f64 / 1000.0)
} else {
0.0
};
fs::remove_dir_all(&temp_dir).ok();
Ok(IncrementalUpdateBenchmark {
initial_import_time_ms: initial_import_time,
file_change_detection_time_ms: detection_time,
delta_application_time_ms: application_time,
total_update_time_ms: total_time,
files_changed: changes.iter().filter(|c| matches!(c.change_type, crate::incremental::ChangeType::Modified)).count(),
files_added: changes.iter().filter(|c| matches!(c.change_type, crate::incremental::ChangeType::Added)).count(),
files_removed: changes.iter().filter(|c| matches!(c.change_type, crate::incremental::ChangeType::Deleted)).count(),
changes_per_second,
})
}
fn benchmark_compaction(&self) -> Result<CompactionBenchmark> {
let db = SiftDB::open(&self.collection_path)?;
let tombstone_manager = crate::tombstone::TombstoneManager::new(&self.collection_path);
for i in 0..5 {
tombstone_manager.mark_file_deleted(
i as u32,
PathBuf::from(format!("test_file_{}.txt", i)),
1,
0,
0
)?;
}
let analysis_start = Instant::now();
let compactor = CollectionCompactor::new(&self.collection_path);
let needs_compaction = compactor.needs_compaction()?;
let analysis_time = analysis_start.elapsed().as_millis() as u64;
if !needs_compaction {
for i in 5..15 {
tombstone_manager.mark_file_deleted(
i as u32,
PathBuf::from(format!("test_file_{}.txt", i)),
1,
0,
0
)?;
}
}
let compaction_start = Instant::now();
let stats = compactor.compact()?;
let compaction_time = compaction_start.elapsed().as_millis() as u64;
let total_time = analysis_time + compaction_time;
let throughput_mb_per_sec = if compaction_time > 0 {
(stats.space_reclaimed_bytes as f64) / (1024.0 * 1024.0) / (compaction_time as f64 / 1000.0)
} else {
0.0
};
Ok(CompactionBenchmark {
tombstone_analysis_time_ms: analysis_time,
compaction_time_ms: compaction_time,
total_time_ms: total_time,
tombstones_removed: stats.tombstones_removed,
segments_compacted: stats.segments_compacted,
space_reclaimed_bytes: stats.space_reclaimed_bytes,
compaction_throughput_mb_per_sec: throughput_mb_per_sec,
})
}
fn calculate_collection_size(&self) -> Result<u64> {
let mut total_size = 0;
if self.collection_path.exists() {
for entry in fs::read_dir(&self.collection_path)? {
let entry = entry?;
if entry.path().is_file() {
total_size += entry.metadata()?.len();
}
}
}
Ok(total_size)
}
fn count_source_files(&self) -> Result<usize> {
let mut count = 0;
if self.source_path.exists() {
for entry in fs::read_dir(&self.source_path)? {
let entry = entry?;
if entry.path().is_file() {
count += 1;
}
}
}
Ok(count)
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
#[ignore] fn bench_test() {
let temp_dir = TempDir::new().unwrap();
let collection_path = temp_dir.path().join("test-bench.sift");
let mut benchmark = SiftDBBenchmark::new(&collection_path, ".");
let _results = benchmark.run_all();
}
}