#!/usr/bin/env python3 """ DuckDB Delta Lake S3 Query Performance Benchmark This script benchmarks DuckDB's performance when querying Delta Lake tables from S3. It includes sample queries and measures execution time, memory usage, and other metrics. """ import time import argparse import duckdb import logging import os from typing import List, Dict, Any, Tuple import psutil import statistics import json # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class DuckDBDeltaBenchmark: def __init__(self, region: str, tenant_id: str, o11y_id: str): """Initialize the benchmark with Delta Lake S3 connection.""" self.connection = duckdb.connect(database=':memory:', read_only=False) self.region = region self.tenant_id = tenant_id self.o11y_id = o11y_id self.setup_delta_connection() def setup_delta_connection(self): """Configure Delta Lake S3 connection using credential chain.""" try: # Install and load required extensions for ext in ['aws', 'httpfs', 'delta']: self.connection.execute(f"INSTALL {ext};") self.connection.execute(f"LOAD {ext};") # Configure S3 access using credential chain self.connection.execute( f"CREATE OR REPLACE SECRET secret (TYPE s3, PROVIDER credential_chain, ENDPOINT 's3.{self.region}.amazonaws.com');" ) # Enable caching for better performance self.connection.execute("SET enable_object_cache=true;") self.connection.execute("SET enable_http_metadata_cache=true;") logger.info("Delta Lake S3 connection configured successfully") except Exception as e: logger.error(f"Failed to configure Delta Lake S3 connection: {e}") raise def get_memory_usage(self) -> Dict[str, float]: """Get current memory usage in MB.""" process = psutil.Process(os.getpid()) mem_info = process.memory_info() return { 'rss_mb': mem_info.rss / (1024 * 1024), 'vms_mb': mem_info.vms / (1024 * 1024) } def run_query(self, query: str, params: Dict = None) -> Tuple[float, Any]: """Execute a query and return execution time and result.""" try: start_time = time.time() result = self.connection.execute(query, params or {}).fetchall() execution_time = time.time() - start_time return execution_time, result except Exception as e: logger.error(f"Query failed: {e}") raise def benchmark_query(self, query_name: str, query: str, params: Dict = None, warmup_runs: int = 1, benchmark_runs: int = 5) -> Dict: """Run a query multiple times and collect performance metrics.""" logger.info(f"Benchmarking query: {query_name}") # Warmup runs for i in range(warmup_runs): self.run_query(query, params) # Benchmark runs execution_times = [] memory_usages = [] for i in range(benchmark_runs): # Clear caches between runs self.connection.execute("PRAGMA clear_cache;") # Get memory before mem_before = self.get_memory_usage() # Run query execution_time, result = self.run_query(query, params) # Get memory after mem_after = self.get_memory_usage() execution_times.append(execution_time) memory_usages.append(mem_after['rss_mb'] - mem_before['rss_mb']) logger.info(f"Run {i + 1}/{benchmark_runs} - Time: {execution_time:.3f}s, " f"Memory: {memory_usages[-1]:.2f}MB, " f"Result rows: {len(result)}") # Calculate statistics stats = { 'query_name': query_name, 'query': query, 'runs': benchmark_runs, 'time_avg': statistics.mean(execution_times), 'time_min': min(execution_times), 'time_max': max(execution_times), 'time_stddev': statistics.stdev(execution_times) if len(execution_times) > 1 else 0, 'memory_avg': statistics.mean(memory_usages), 'memory_max': max(memory_usages), 'execution_times': execution_times, 'memory_usages': memory_usages, # Include sample of results 'result_sample': str(result[:3]) if result else "[]" } return stats def register_table(self, table_name: str, s3_path: str): """Register a Delta table from S3.""" delta_path = f"delta_scan('{s3_path}')" self.connection.execute(f""" CREATE OR REPLACE VIEW {table_name} AS SELECT * FROM {delta_path} """) logger.info(f"Registered Delta table: {table_name} from {s3_path}") def get_sample_queries(self, start_ts: int, end_ts: int, point_count: int = 60) -> List[Dict[str, Any]]: """Return a list of sample queries to benchmark. Args: start_ts: Start timestamp in seconds since epoch end_ts: End timestamp in seconds since epoch point_count: Number of points to return in the time series """ # Calculate time bucket size in seconds time_range = end_ts - start_ts bucket_size = max(1, time_range // point_count) return [ { 'name': 'slowlogs_count', 'query': f""" SELECT to_timestamp(time_bucket({bucket_size}, time, {start_ts})) as time_bucket, COUNT(*) as count FROM slowlogs WHERE time >= {start_ts} AND time <= {end_ts} GROUP BY time_bucket ORDER BY time_bucket; """, 'description': f'Count of slowlog entries in {point_count} time buckets between {start_ts} and {end_ts}' }, { 'name': 'top_queries_by_digest', 'query': """ SELECT s.digest, q.query, COUNT(*) as execution_count, AVG(s.query_time) as avg_duration, MAX(s.query_time) as max_duration, SUM(s.query_time) as total_duration, COUNT(DISTINCT s.plan_digest) as distinct_plans FROM slowlogs s JOIN queries q ON s.digest = q.digest WHERE s.time >= {start_ts} AND s.time <= {end_ts} GROUP BY s.digest, q.query ORDER BY execution_count DESC LIMIT 10; """, 'description': 'Top 10 queries by execution count with digest aggregation' }, { 'name': 'recent_slow_queries', 'query': f""" SELECT q.query, to_timestamp(s.time) as timestamp, s.query_time, s.memory_max, s.process_keys, s.total_keys, s.cop_wait_avg, s.cop_wait_max, s.cop_wait_p90, s.index_names, s.request_count FROM slowlogs s JOIN queries q ON s.digest = q.digest WHERE s.time >= {start_ts} AND s.time <= {end_ts} ORDER BY s.time DESC LIMIT 100; """, 'description': 'List of 100 most recent slow queries with detailed metrics' } ] def main(): parser = argparse.ArgumentParser( description='Benchmark DuckDB Delta Lake Query Performance') parser.add_argument('--region', required=True, help='AWS region (e.g., us-east-1)') parser.add_argument('--tenant-id', required=True, help='Tenant ID') parser.add_argument('--o11y-id', required=True, help='Observability ID') parser.add_argument('--slowlogs-path', help='S3 path to the Delta table for slowlogs (e.g., s3://bucket/path/to/slowlogs)') parser.add_argument('--start-ts', type=int, default=int(time.time()) - 3600, help='Start timestamp in seconds since epoch (default: 1 hour ago)') parser.add_argument('--end-ts', type=int, default=int(time.time()), help='End timestamp in seconds since epoch (default: now)') parser.add_argument('--point-count', type=int, default=60, help='Number of points to return in time series (default: 60)') parser.add_argument('--warmup-runs', type=int, default=1, help='Number of warmup runs per query (default: 1)') parser.add_argument('--benchmark-runs', type=int, default=5, help='Number of benchmark runs per query (default: 5)') parser.add_argument( '--output', help='Output file for benchmark results (JSON)') args = parser.parse_args() try: # Initialize benchmark benchmark = DuckDBDeltaBenchmark( region=args.region, tenant_id=args.tenant_id, o11y_id=args.o11y_id ) # Register tables if args.slowlogs_path: benchmark.register_table('slowlogs', args.slowlogs_path) else: # Default path if not specified benchmark.register_table( 'slowlogs', f's3://o11y-prod-shared-{args.region}/datalake/{args.tenant_id}/{args.o11y_id}/slowlogs' ) # Get sample queries with time range parameters queries = benchmark.get_sample_queries( start_ts=args.start_ts, end_ts=args.end_ts, point_count=args.point_count ) # Run benchmarks results = [] for query_info in queries: try: stats = benchmark.benchmark_query( query_name=query_info['name'], query=query_info['query'], warmup_runs=args.warmup_runs, benchmark_runs=args.benchmark_runs ) stats['description'] = query_info['description'] results.append(stats) # Print summary print(f"\n=== {query_info['name']} ===") print(f"Description: {query_info['description']}") print( f"Average time: {stats['time_avg']:.3f}s (±{stats['time_stddev']:.3f}s)") print( f"Min/Max time: {stats['time_min']:.3f}s / {stats['time_max']:.3f}s") print(f"Avg memory usage: {stats['memory_avg']:.2f}MB") print(f"Result sample: {stats['result_sample']}") except Exception as e: logger.error(f"Error running query {query_info['name']}: {e}") continue # Save results if output file specified if args.output: with open(args.output, 'w') as f: json.dump({ 'timestamp': time.time(), 'parameters': vars(args), 'results': results }, f, indent=2) logger.info(f"Benchmark results saved to {args.output}") except Exception as e: logger.error(f"Benchmark failed: {e}", exc_info=True) return 1 return 0 if __name__ == "__main__": exit(main())