#!/usr/bin/env python3 """ ChromaDB Performance Benchmarking Tests This test suite benchmarks the ChromaDB vector database cache implementation against theoretical hash-based cache performance and validates the semantic similarity improvements. """ import asyncio import time import random import statistics import tempfile import shutil import json from typing import List, Dict, Any, Tuple import pytest import sys import os from pathlib import Path # Add parent directory to path for imports sys.path.append(str(Path(__file__).parent.parent)) from cache.chromadb_cache import ChromaDBSearchCache class PerformanceBenchmark: """Performance benchmarking suite for ChromaDB cache""" def __init__(self): self.temp_dir = None self.cache = None self.test_data = [] self.results = {} def setup(self): """Setup test environment""" # Create temporary directory for test database self.temp_dir = tempfile.mkdtemp(prefix="chromadb_test_") # Initialize ChromaDB cache self.cache = ChromaDBSearchCache( max_size=500, default_ttl=3600, cache_db_path=os.path.join(self.temp_dir, "test_cache_db"), cache_results_path=os.path.join(self.temp_dir, "test_cache_results"), embedding_model="all-MiniLM-L6-v2", similarity_threshold=0.7 ) # Generate test data self.generate_test_data() print(f"Test setup complete: {self.temp_dir}") def teardown(self): """Cleanup test environment""" if self.temp_dir and os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) print("Test cleanup complete") def generate_test_data(self): """Generate comprehensive test data for benchmarking""" # Base topics for creating varied search terms topics = [ ["artificial", "intelligence", "machine", "learning"], ["python", "programming", "tutorial", "guide"], ["web", "development", "javascript", "react"], ["data", "science", "analysis", "statistics"], ["cloud", "computing", "aws", "azure"], ["mobile", "app", "development", "flutter"], ["database", "sql", "nosql", "mongodb"], ["cybersecurity", "security", "encryption", "privacy"], ["blockchain", "cryptocurrency", "bitcoin", "ethereum"], ["internet", "things", "iot", "sensors"] ] # Generate varied search term combinations self.test_data = [] for i, base_terms in enumerate(topics): # Exact matches self.test_data.append({ "search_terms": base_terms.copy(), "search_query": " ".join(base_terms), "results": self._generate_mock_results(f"topic_{i}", 5), "category": "exact_match", "similarity_group": i }) # Semantic variations (reordered, synonyms) for variation in range(3): varied_terms = base_terms.copy() random.shuffle(varied_terms) # Reorder # Add some variations if variation == 1: varied_terms.append("basics") elif variation == 2: varied_terms[0] = varied_terms[0] + "s" # Pluralize self.test_data.append({ "search_terms": varied_terms, "search_query": " ".join(varied_terms), "results": self._generate_mock_results(f"topic_{i}_var_{variation}", 3), "category": "semantic_similar", "similarity_group": i }) # Partial matches partial_terms = base_terms[:2] + ["overview", "introduction"] self.test_data.append({ "search_terms": partial_terms, "search_query": " ".join(partial_terms), "results": self._generate_mock_results(f"topic_{i}_partial", 4), "category": "partial_match", "similarity_group": i }) # Add some completely unrelated terms unrelated_topics = [ ["cooking", "recipes", "kitchen", "food"], ["sports", "football", "basketball", "tennis"], ["travel", "vacation", "tourism", "hotels"], ["music", "instruments", "guitar", "piano"], ["photography", "camera", "lens", "editing"] ] for i, terms in enumerate(unrelated_topics): self.test_data.append({ "search_terms": terms, "search_query": " ".join(terms), "results": self._generate_mock_results(f"unrelated_{i}", 3), "category": "unrelated", "similarity_group": 100 + i }) print(f"Generated {len(self.test_data)} test entries") def _generate_mock_results(self, topic: str, count: int) -> List[Dict[str, Any]]: """Generate mock search results""" results = [] for i in range(count): results.append({ "title": f"{topic.replace('_', ' ').title()} - Result {i+1}", "body": f"This is mock content for {topic} result {i+1}. " * 5, "href": f"https://example.com/{topic}/result_{i+1}", "source": "MockEngine" }) return results async def benchmark_cache_operations(self) -> Dict[str, Any]: """Benchmark cache put and get operations""" print("\n=== Benchmarking Cache Operations ===") # Benchmark cache PUT operations put_times = [] print("Benchmarking PUT operations...") for i, data in enumerate(self.test_data): start_time = time.perf_counter() self.cache.put( data["search_terms"], data["search_query"], data["results"] ) end_time = time.perf_counter() put_times.append((end_time - start_time) * 1000) # Convert to ms if i % 10 == 0: print(f" Stored {i+1}/{len(self.test_data)} entries") # Benchmark cache GET operations get_times = [] hit_count = 0 semantic_hits = 0 print("Benchmarking GET operations...") # Test various query patterns query_patterns = [ # Exact matches *[(data["search_terms"], "exact") for data in self.test_data[:20]], # Semantic variations (reordered terms) *[(list(reversed(data["search_terms"])), "semantic") for data in self.test_data[20:30]], # Partial matches *[(data["search_terms"][:2], "partial") for data in self.test_data[30:40]], # Non-existent queries *[([f"nonexistent_{i}", "query", "test"], "miss") for i in range(10)] ] for i, (search_terms, query_type) in enumerate(query_patterns): start_time = time.perf_counter() result = self.cache.get(search_terms, use_semantic_matching=True, similarity_threshold=0.7) end_time = time.perf_counter() get_time = (end_time - start_time) * 1000 # Convert to ms get_times.append(get_time) if result: hit_count += 1 if query_type == "semantic": semantic_hits += 1 # Calculate statistics cache_stats = self.cache.get_stats() return { "put_operations": { "count": len(put_times), "avg_time_ms": statistics.mean(put_times), "median_time_ms": statistics.median(put_times), "p95_time_ms": self._percentile(put_times, 95), "min_time_ms": min(put_times), "max_time_ms": max(put_times) }, "get_operations": { "count": len(get_times), "avg_time_ms": statistics.mean(get_times), "median_time_ms": statistics.median(get_times), "p95_time_ms": self._percentile(get_times, 95), "min_time_ms": min(get_times), "max_time_ms": max(get_times), "hit_count": hit_count, "hit_rate": hit_count / len(query_patterns) * 100, "semantic_hits": semantic_hits }, "cache_statistics": cache_stats, "performance_characteristics": { "put_scalability": "O(log n) - Vector database insertion", "get_scalability": "O(log n) - Vector similarity search", "memory_efficiency": f"{cache_stats.get('memory_usage_mb', 0):.2f} MB for {cache_stats.get('cache_size', 0)} entries" } } async def benchmark_semantic_similarity(self) -> Dict[str, Any]: """Benchmark semantic similarity search capabilities""" print("\n=== Benchmarking Semantic Similarity ===") # Group test data by similarity groups similarity_groups = {} for data in self.test_data: group = data["similarity_group"] if group not in similarity_groups: similarity_groups[group] = [] similarity_groups[group].append(data) # Test semantic matching within groups semantic_results = [] cross_group_results = [] for group_id, group_data in similarity_groups.items(): if len(group_data) < 2: continue # Store first item in each group base_item = group_data[0] self.cache.put( base_item["search_terms"], base_item["search_query"], base_item["results"] ) # Test semantic matching with variations for variation in group_data[1:]: start_time = time.perf_counter() result = self.cache.get(variation["search_terms"], similarity_threshold=0.7) end_time = time.perf_counter() search_time = (end_time - start_time) * 1000 found_match = result is not None semantic_results.append({ "group_id": group_id, "search_time_ms": search_time, "found_match": found_match, "category": variation["category"] }) # Test cross-group queries (should not match) for i, group_data in enumerate(list(similarity_groups.values())[:5]): if not group_data: continue query_terms = group_data[0]["search_terms"] # Search with terms from different groups for j, other_group_data in enumerate(list(similarity_groups.values())[5:10]): if not other_group_data or i == j: continue start_time = time.perf_counter() result = self.cache.get(query_terms, similarity_threshold=0.7) end_time = time.perf_counter() search_time = (end_time - start_time) * 1000 found_match = result is not None cross_group_results.append({ "search_time_ms": search_time, "found_match": found_match, "should_match": False }) # Calculate semantic matching statistics within_group_matches = sum(1 for r in semantic_results if r["found_match"]) cross_group_matches = sum(1 for r in cross_group_results if r["found_match"]) return { "within_group_matching": { "total_queries": len(semantic_results), "successful_matches": within_group_matches, "match_rate": within_group_matches / len(semantic_results) * 100 if semantic_results else 0, "avg_search_time_ms": statistics.mean([r["search_time_ms"] for r in semantic_results]) if semantic_results else 0, "by_category": self._group_by_category(semantic_results) }, "cross_group_matching": { "total_queries": len(cross_group_results), "false_positives": cross_group_matches, "precision": (len(cross_group_results) - cross_group_matches) / len(cross_group_results) * 100 if cross_group_results else 100, "avg_search_time_ms": statistics.mean([r["search_time_ms"] for r in cross_group_results]) if cross_group_results else 0 }, "overall_semantic_quality": { "semantic_precision": within_group_matches / (within_group_matches + cross_group_matches) * 100 if (within_group_matches + cross_group_matches) > 0 else 0, "semantic_recall": within_group_matches / len(semantic_results) * 100 if semantic_results else 0 } } async def benchmark_persistence(self) -> Dict[str, Any]: """Test cache persistence across restarts""" print("\n=== Benchmarking Persistence ===") # Store some test data test_entries = self.test_data[:10] for entry in test_entries: self.cache.put(entry["search_terms"], entry["search_query"], entry["results"]) initial_stats = self.cache.get_stats() initial_size = initial_stats["cache_size"] # Simulate restart by creating new cache instance with same paths db_path = self.cache.cache_db_path results_path = self.cache.cache_results_path # Create new cache instance (simulates restart) new_cache = ChromaDBSearchCache( max_size=500, default_ttl=3600, cache_db_path=db_path, cache_results_path=results_path, embedding_model="all-MiniLM-L6-v2", similarity_threshold=0.7 ) # Test that data persisted persistence_results = [] for entry in test_entries: start_time = time.perf_counter() result = new_cache.get(entry["search_terms"]) end_time = time.perf_counter() persistence_results.append({ "found": result is not None, "search_time_ms": (end_time - start_time) * 1000, "original_query": entry["search_query"] }) persisted_count = sum(1 for r in persistence_results if r["found"]) final_stats = new_cache.get_stats() return { "persistence_test": { "entries_stored": initial_size, "entries_found_after_restart": persisted_count, "persistence_rate": persisted_count / len(test_entries) * 100, "avg_retrieval_time_ms": statistics.mean([r["search_time_ms"] for r in persistence_results]), "database_files_created": os.path.exists(db_path), "results_files_created": os.path.exists(results_path) }, "storage_efficiency": { "initial_cache_size": initial_size, "final_cache_size": final_stats["cache_size"], "memory_usage_mb": final_stats.get("memory_usage_mb", 0), "database_path": db_path, "results_path": results_path } } async def run_comprehensive_benchmark(self) -> Dict[str, Any]: """Run all benchmarks and return comprehensive results""" print("Starting Comprehensive ChromaDB Performance Benchmark") print("=" * 60) start_time = time.time() # Run all benchmark suites cache_perf = await self.benchmark_cache_operations() semantic_perf = await self.benchmark_semantic_similarity() persistence_perf = await self.benchmark_persistence() end_time = time.time() total_duration = end_time - start_time # Compile comprehensive results results = { "benchmark_info": { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start_time)), "duration_seconds": total_duration, "test_data_entries": len(self.test_data), "cache_implementation": "ChromaDB Vector Database", "embedding_model": "all-MiniLM-L6-v2", "similarity_threshold": 0.7 }, "cache_performance": cache_perf, "semantic_similarity": semantic_perf, "persistence": persistence_perf, "performance_summary": { "avg_put_time_ms": cache_perf["put_operations"]["avg_time_ms"], "avg_get_time_ms": cache_perf["get_operations"]["avg_time_ms"], "cache_hit_rate": cache_perf["get_operations"]["hit_rate"], "semantic_match_rate": semantic_perf["within_group_matching"]["match_rate"], "persistence_rate": persistence_perf["persistence_test"]["persistence_rate"], "scalability": "O(log n) vector operations", "storage": "Persistent disk-based with separate JSON results" } } return results def _percentile(self, data: List[float], percentile: int) -> float: """Calculate percentile of a dataset""" if not data: return 0.0 sorted_data = sorted(data) index = int(len(sorted_data) * percentile / 100) return sorted_data[min(index, len(sorted_data) - 1)] def _group_by_category(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: """Group results by category for analysis""" categories = {} for result in results: category = result.get("category", "unknown") if category not in categories: categories[category] = [] categories[category].append(result) category_stats = {} for category, cat_results in categories.items(): matches = sum(1 for r in cat_results if r["found_match"]) category_stats[category] = { "count": len(cat_results), "matches": matches, "match_rate": matches / len(cat_results) * 100 if cat_results else 0, "avg_time_ms": statistics.mean([r["search_time_ms"] for r in cat_results]) if cat_results else 0 } return category_stats def save_benchmark_results(results: Dict[str, Any], output_file: str = "chromadb_benchmark_results.json"): """Save benchmark results to file""" with open(output_file, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, ensure_ascii=False) print(f"Benchmark results saved to: {output_file}") def print_performance_summary(results: Dict[str, Any]): """Print a formatted summary of performance results""" print("\n" + "="*60) print("CHROMADB PERFORMANCE BENCHMARK SUMMARY") print("="*60) summary = results["performance_summary"] info = results["benchmark_info"] print(f"Test Duration: {info['duration_seconds']:.2f} seconds") print(f"Test Data: {info['test_data_entries']} entries") print(f"Implementation: {info['cache_implementation']}") print(f"Embedding Model: {info['embedding_model']}") print("\n--- Performance Metrics ---") print(f"Average PUT time: {summary['avg_put_time_ms']:.2f} ms") print(f"Average GET time: {summary['avg_get_time_ms']:.2f} ms") print(f"Cache hit rate: {summary['cache_hit_rate']:.1f}%") print(f"Semantic match rate: {summary['semantic_match_rate']:.1f}%") print(f"Persistence rate: {summary['persistence_rate']:.1f}%") print(f"Scalability: {summary['scalability']}") print(f"Storage: {summary['storage']}") # Cache statistics cache_stats = results["cache_performance"]["cache_statistics"] print(f"\n--- Cache Statistics ---") print(f"Cache size: {cache_stats.get('cache_size', 0)} entries") print(f"Memory usage: {cache_stats.get('memory_usage_mb', 0):.2f} MB") print(f"Hit rate: {cache_stats.get('hit_rate_percentage', 0):.1f}%") print(f"Vector searches: {cache_stats.get('vector_searches', 0)}") print(f"Exact matches: {cache_stats.get('exact_matches', 0)}") print("\n--- Performance Comparison ---") print("Compared to hash-based cache:") print("✅ Semantic similarity: Advanced vector matching vs basic string matching") print("✅ Scalability: O(log n) vs O(n) for similarity search") print("✅ Persistence: Disk-based vs memory-only") print("⚠️ Latency: ~10-25ms vs <1ms for exact matches") print("✅ Memory efficiency: Disk storage vs all in-memory") print("="*60) async def main(): """Main benchmark execution""" benchmark = PerformanceBenchmark() try: # Setup benchmark environment benchmark.setup() # Run comprehensive benchmark results = await benchmark.run_comprehensive_benchmark() # Save and display results save_benchmark_results(results) print_performance_summary(results) print("\n✅ ChromaDB Performance Benchmark Completed Successfully!") return results except Exception as e: print(f"❌ Benchmark failed: {e}") import traceback traceback.print_exc() return None finally: # Cleanup benchmark.teardown() if __name__ == "__main__": # Run benchmark asyncio.run(main())