# Project export: Applovin Query Challenge Submission This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: Cal Hacks 12.0 - Tagline: A fast implementation for the query planner challenege - Devpost: https://devpost.com/software/applovin-query-challenge-submission - GitHub: https://github.com/jetpham/calhacks25# - Team: 1 GitHub contributor(s) — Jet Pham (12 commits) ## Devpost submission (written by the team) No Devpost description available. ## README (from the GitHub repository)
RuckDB Logo

CalHacks Database Query Optimizer

My attempt at the CalHacks 2025 query planner challenge using DuckDB and Rust. ## Performance **Database preprocessing**: 4m 18.9s Benchmark results on the full dataset (245M rows) with 1000 runs: | Query / Metric | judges.json | queries.json | |----------------|-------------|--------------| | Query 1 | 0.24ms | 0.77ms | | Query 2 | 0.40ms | 0.90ms | | Query 3 | 1.63ms | 0.96ms | | Query 4 | 1.66ms | 1.96ms | | Query 5 | 1.56ms | 1.74ms | | Query 6 | 0.19ms | - | | Query 7 | 0.20ms | - | | Query 8 | 4.49ms | - | | Query 9 | 0.88ms | - | | Query 10 | 0.77ms | - | | Query 11 | 1.23ms | - | | Query 12 | 0.86ms | - | | Query 13 | 2.51ms | - | | Query 14 | 0.84ms | - | | Query 15 | 0.75ms | - | | **Sum of averages** | **18.20ms** | **6.31ms** | | Query preparation and warmup | 1.4s | 1.3s | | Total execution time (1000 runs) | 1.8s | 6.5s | ## Usage ### Command Line Arguments | Argument | Description | Default | |----------|-------------|---------| | `--input-dir DIR` | Directory containing CSV files | Required | | `--output-dir DIR` | Output directory for query results | Required with `--run` | | `--queries FILE` | JSON file with query definitions | Required | | `--run` | Execute queries (required flag) | - | | `--runs N` | Number of times to run each query (for averaging) | 1 | | `--use-existing FILE` | Use existing database file (specify path) | None | | `--baseline-dir DIR` | Compare results against baseline | None | | `--profile` | Enable EXPLAIN ANALYZE profiling | False | ## Building ### Prerequisites - **Rust** - **DuckDB** ### Build Release Binary ```bash # Build optimized release binary cargo build --release ``` ### Using Nix Flake (Optional) For dev environment: ```bash # Enter development shell nix develop # Build cargo build --release ``` Or with `direnv`: ```bash direnv allow cargo build --release ``` ## Quick Start Create the database and run queries: ```bash ./calhacks \ --input-dir ../calhacks-applovin-query-planner-challenge/data/data \ --run \ --queries ../calhacks-applovin-query-planner-challenge/queries.json \ --output-dir results/ ``` ## Detected evidence (automated analysis) Indexed codebase: 10 recognized source files, 79 KB. - Rust (language) — detected in the code ## Codebase structure (from repository index) ### Files (18 of 18) ``` .cargo/config.toml .envrc .gitignore Cargo.lock Cargo.toml flake.lock flake.nix README.md rust-toolchain.toml src/data_loader.rs src/hardware.rs src/main.rs src/mv.rs src/planner.rs src/preprocessor.rs src/query_executor.rs src/query_handler.rs src/result_checker.rs ``` ### Dependencies - Cargo.toml: anyhow@1.0.100, chrono@0.4.42, clap@4.5, csv@1.3, duckdb@1.4, indicatif@0.18, num_cpus@1.16, serde@1.0.228, serde_json@1.0.145 ### Recent commits (newest first) - feat: update to remove unused options and logo - feat: update readme with new baseline - feat: use progress bars for simpler progress - feat: update to use the mv - fix: remove direnv - feat: clean up and update readme - feat: update ignores and checks - feat: copy first place methods - feat: clean up useless files - feat: persistant sequencial - feta: broom - feat: ;lkm - feat: move - feat readme rollups - log - feat: check and basic indexing - feat: back to basics - feat: too much ops - feat: optimized types and relationship - feat: super aggressive index permutation engine ## Key source files (fetched from GitHub, selected and truncated for size) ### Cargo.toml ``` [package] name = "calhacks" version = "0.1.0" edition = "2024" [profile.release] # Maximum optimization level opt-level = 3 # Enable link-time optimization (LTO) lto = "thin" # Reduce codegen units for better optimization codegen-units = 1 # Enable panic = "abort" for smaller binaries and better performance panic = "abort" # Strip debug symbols for smaller binaries strip = true # Enable overflow checks in debug, disable in release for performance overflow-checks = false [profile.profiling] # Inherit from release but with profiling-friendly settings inherits = "release" # Keep debug symbols for readable flamegraphs strip = false # Disable LTO to preserve function boundaries lto = false # Use more codegen units to preserve function boundaries codegen-units = 16 # Keep maximum optimization level opt-level = 3 [dependencies] anyhow = "1.0.100" chrono = "0.4.42" clap = { version = "4.5", features = ["derive"] } csv = "1.3" duckdb = "1.4" indicatif = "0.18" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" num_cpus = "1.16" ``` ### src/main.rs ```rust use anyhow::Result; use clap::Parser; use std::path::{PathBuf, Path}; use std::time::{Instant, Duration}; use duckdb::Connection; use indicatif::{ProgressBar, ProgressStyle}; fn format_duration_seconds(duration: Duration) -> String { let total_secs = duration.as_secs_f64(); format!("{:.1}s", total_secs) } fn format_duration_ms_ns(duration: Duration) -> String { let total_ns = duration.as_nanos(); let total_ms = total_ns as f64 / 1_000_000.0; format!("{:.2}ms", total_ms) } mod data_loader; mod preprocessor; mod query_executor; mod query_handler; mod result_checker; mod mv; mod planner; mod hardware; use data_loader::load_data; use preprocessor::{create_materialized_views, compute_mv_stats, warmup_cache, create_indexes, create_type_partitioned_materialized_views, load_all_mvs_from_db}; use query_executor::{prepare_query, write_single_result_to_csv, explain_query}; use query_handler::parse_queries_from_file; use result_checker::compare_results; use planner::Planner; #[derive(Parser, Debug)] #[command(version, about, long_about = None)] struct Args { #[arg(long, value_name = "DIR", default_value = "data/data")] input_dir: PathBuf, #[arg(long)] run: bool, #[arg(long, value_name = "DIR")] output_dir: Option, #[arg( long, value_name = "FILE", default_value = "queries.json", requires = "run" )] queries: PathBuf, #[arg(long, value_name = "FILE")] use_existing: Option, #[arg(long, value_name = "DIR")] baseline_dir: Option, #[arg(long)] profile: bool, #[arg(long, default_value = "1")] runs: usize, } fn find_next_db_filename() -> Result { // First check if any duckN.db files exist if !Path::new("duck1.db").exists() { return Ok(PathBuf::from("duck1.db")); } // Find the highest existing number let mut max_num = 0; for i in 1..=100 { let path = format!("duck{}.db", i); if Path::new(&path).exists() { max_num = i; } } // Return the next available number Ok(PathBuf::from(format!("duck{}.db", max_num + 1))) } fn main() -> Result<()> { let args = Args::parse(); if let Some(baseline_dir) = &args.baseline_dir { if !args.run { let Some(output_dir) = &args.output_dir else { anyhow::bail!("--output-dir required when using --baseline-dir"); }; return compare_results(baseline_dir, output_dir); } } let db_path = if let Some(existing_path) = &args.use_existing { existing_path.clone() } else { find_next_db_filename()? }; // Part 1: Print DB file status if db_path.exists() && args.use_existing.is_some() { println!("Using existing database: {}", db_path.display()); } else { println!("Creating new database: {}", db_path.display()); let preprocess_start = Instant::now(); // Part 2: Preprocessing progress bar let pb = ProgressBar::new(6); pb.set_style( ProgressStyle::with_template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} {msg}") .unwrap() .progress_chars("#>-") ); pb.set_message("Preprocessing database"); if db_path.exists() { std::fs::remove_file(&db_path)?; } let file_con = Connection::open(&db_path)?; pb.set_message("Loading data..."); load_data(&file_con, &args.input_dir)?; pb.inc(1); pb.set_message("Creating materialized views..."); let mut mvs = create_materialized_views(&file_con)?; pb.inc(1); pb.set_message("Computing MV statistics..."); compute_mv_stats(&file_con, &mut mvs)?; pb.inc(1); pb.set_message("Creating type-partitioned MVs..."); let mut partitioned_mvs = create_type_partitioned_materialized_views(&file_con, &mvs)?; pb.inc(1); let partitioned_count = partitioned_mvs.len(); // Combine base and partitioned MVs mvs.append(&mut partitioned_mvs); pb.set_message("Computing partitioned MV statistics..."); let total_mvs = mvs.len(); compute_mv_stats(&file_con, &mut mvs[total_mvs - partitioned_count..])?; pb.inc(1); pb.set_message("Creating indexes..."); create_indexes(&file_con, &mvs)?; pb.inc(1); pb.finish_and_clear(); let preprocess_duration = preprocess_start.elapsed(); println!("Database preprocessing completed in {}", format_duration_seconds(preprocess_duration)); } let con = Connection::open(&db_path)?; if args.run { let Some(output_dir) = &args.output_dir else { anyhow::bail!("--output-dir required with --run"); }; // Part 3: Query prep progress bar let prep_start = Instant::now(); let prep_pb = ProgressBar::new(4); prep_pb.set_style( ProgressStyle::with_template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} {msg}") .unwrap() .progress_chars("#>-") ); prep_pb.set_message("Preparing queries"); prep_pb.set_message("Parsing queries..."); let queries = parse_queries_from_file(&args.queries)?; prep_pb.inc(1); prep_pb.set_message("Loading materialized views..."); // Load all MVs from database (base + type-partitioned) let mut mvs = load_all_mvs_from_db(&con)?; if mvs.is_empty() { // Fallback: create base MVs if none exist mvs = create_materialized_views(&con)?; } prep_pb.inc(1); prep_pb.set_message("Computing statistics..."); compute_mv_stats [truncated — 3499 more characters] ``` ### src/hardware.rs ```rust use std::sync::OnceLock; pub struct HardwareInfo { pub num_threads: usize, pub available_memory_gb: f64, } static HARDWARE_INFO: OnceLock = OnceLock::new(); impl HardwareInfo { pub fn detect() -> Self { let num_threads = num_cpus::get(); // Try to detect memory (fallback to reasonable defaults) let available_memory_gb = if cfg!(target_os = "linux") { // Try to read from /proc/meminfo if let Ok(content) = std::fs::read_to_string("/proc/meminfo") { let mut available_kb = 0; for line in content.lines() { if line.starts_with("MemAvailable:") { if let Some(kb_str) = line.split_whitespace().nth(1) { available_kb = kb_str.parse().unwrap_or(0); } } } available_kb as f64 / 1_048_576.0 // KB to GB } else { 16.0 // Fallback } } else { 16.0 // Fallback for non-Linux }; Self { num_threads, available_memory_gb, } } pub fn get() -> &'static HardwareInfo { HARDWARE_INFO.get_or_init(|| Self::detect()) } /// Calculate optimal row group size for Parquet /// Winner's approach: aim for ~20 row groups per thread /// With 10 threads: ROW_GROUP_SIZE = 1M /// Formula: target ~20 row groups per thread pub fn optimal_row_group_size(&self, total_rows: usize) -> usize { // Winner's approach: with 10 threads, use 1M row groups // This gives ~20 row groups per thread (245M rows / 10 threads / 1M = ~24.5 groups) // For our hardware, adjust based on thread count let target_groups_per_thread = 20; let rows_per_thread = total_rows / self.num_threads.max(1); let optimal_size = rows_per_thread / target_groups_per_thread; // Clamp to reasonable values optimal_size.max(500_000).min(2_000_000) } /// Calculate cost function weights based on hardware /// More RAM = can scan more rows efficiently /// More threads = rollup is cheaper (parallel aggregation) pub fn cost_weights(&self) -> (f64, f64) { // Base weights (winner's values) let base_scan_weight = 1.0; let base_rollup_weight = 32.0; // Adjust based on available memory // Winner had 18GB, we have more - can afford to scan more let memory_factor = (self.available_memory_gb / 18.0).min(2.0).max(0.5); let scan_weight = base_scan_weight / memory_factor; // Lower weight = prefer scanning // Adjust rollup weight based on thread count // More threads = parallel aggregation is cheaper let thread_factor = (self.num_threads as f64 / 10.0).min(2.0).max(0.5); let rollup_weight = base_rollup_weight / thread_factor; // Lower weight = rollup cheaper (scan_weight, rollup_weight) } } pub fn get_hardware_info() -> &'static HardwareInfo { HardwareInfo::get() } ``` ### src/result_checker.rs ```rust use anyhow::Result; use std::path::{Path, PathBuf}; use std::fs; pub fn compare_results(baseline_dir: &Path, output_dir: &Path) -> Result<()> { let baseline_files = get_query_files(baseline_dir)?; let mut failed = Vec::new(); for (qnum, baseline_file) in baseline_files.iter().enumerate() { let query_num = qnum + 1; let output_file = output_dir.join(format!("q{}.csv", query_num)); if !output_file.exists() { failed.push((query_num, "Missing output file".to_string())); continue; } match compare_csv_files(baseline_file, &output_file) { Ok(()) => { } Err(e) => { failed.push((query_num, e.to_string())); } } } if !failed.is_empty() { anyhow::bail!("Some queries failed comparison") } Ok(()) } fn get_query_files(dir: &Path) -> Result> { let entries = fs::read_dir(dir)?; let mut files: Vec = Vec::new(); for entry in entries { let entry = entry?; let path = entry.path(); if let Some(filename) = path.file_name().and_then(|n| n.to_str()) { if filename.starts_with("q") && filename.ends_with(".csv") { files.push(path); } } } files.sort_by(|a, b| { let a_num = extract_query_number(a); let b_num = extract_query_number(b); a_num.cmp(&b_num) }); Ok(files) } fn extract_query_number(path: &Path) -> usize { if let Some(filename) = path.file_name().and_then(|n| n.to_str()) { if let Some(stripped) = filename.strip_suffix(".csv") { if let Some(num_str) = stripped.strip_prefix("q") { if let Ok(num) = num_str.parse::() { return num; } } } } 0 } fn compare_csv_files(baseline_file: &Path, output_file: &Path) -> Result<()> { let (baseline_header, baseline_rows) = parse_csv(baseline_file)?; let (output_header, output_rows) = parse_csv(output_file)?; if baseline_header != output_header { anyhow::bail!("Headers don't match"); } if baseline_rows.len() != output_rows.len() { anyhow::bail!("Row count mismatch (baseline: {}, output: {})", baseline_rows.len(), output_rows.len()); } let mut baseline_used = vec![false; baseline_rows.len()]; for output_row in &output_rows { let mut found = false; for (i, baseline_row) in baseline_rows.iter().enumerate() { if !baseline_used[i] && rows_match_with_tolerance(baseline_row, output_row) { baseline_used[i] = true; found = true; break; } } if !found { anyhow::bail!("Row not found in baseline or already matched (duplicate mismatch)"); } } Ok(()) } fn parse_csv(file: &Path) -> Result<(Vec, Vec>)> { let content = fs::read_to_string(file)?; let lines: Vec<&str> = content.lines().filter(|l| !l.trim().is_empty()).collect(); if lines.is_empty() { anyhow::bail!("Empty CSV file"); } let header = lines[0].split(',') .map(|s| s.trim().to_string()) .collect(); let data_rows: Vec> = lines[1..].iter() .map(|line| { line.split(',') .map(|s| s.trim().to_string()) .collect() }) .collect(); Ok((header, data_rows)) } fn rows_match_with_tolerance(row1: &Vec, row2: &Vec) -> bool { if row1.len() != row2.len() { return false; } for (cell1, cell2) in row1.iter().zip(row2.iter()) { if !cells_match_with_tolerance(cell1, cell2) { return false; } } true } fn cells_match_with_tolerance(cell1: &str, cell2: &str) -> bool { if cell1 == cell2 { return true; } if let (Ok(val1), Ok(val2)) = (cell1.parse::(), cell2.parse::()) { (val1 - val2).abs() < 0.1 } else { false } } ``` ### src/query_executor.rs ```rust use anyhow::Result; use duckdb::Connection; use std::path::PathBuf; use std::fs; fn extract_value_as_string(row: &duckdb::Row, col_index: usize) -> String { let value = row.get_ref::(col_index).unwrap(); match value { duckdb::types::ValueRef::Null => String::from("NULL"), duckdb::types::ValueRef::Boolean(b) => b.to_string(), duckdb::types::ValueRef::TinyInt(i) => i.to_string(), duckdb::types::ValueRef::SmallInt(i) => i.to_string(), duckdb::types::ValueRef::Int(i) => i.to_string(), duckdb::types::ValueRef::BigInt(i) => i.to_string(), duckdb::types::ValueRef::HugeInt(i) => i.to_string(), duckdb::types::ValueRef::UTinyInt(u) => u.to_string(), duckdb::types::ValueRef::USmallInt(u) => u.to_string(), duckdb::types::ValueRef::UInt(u) => u.to_string(), duckdb::types::ValueRef::UBigInt(u) => u.to_string(), duckdb::types::ValueRef::Float(f) => trim_float(f as f64), duckdb::types::ValueRef::Double(d) => trim_float(d), duckdb::types::ValueRef::Decimal(d) => d.to_string(), duckdb::types::ValueRef::Timestamp(_, ts) => format!("{}", ts), duckdb::types::ValueRef::Text(bytes) => { match std::str::from_utf8(bytes) { Ok(s) => s.to_string(), Err(_) => format!("{:?}", bytes), } }, duckdb::types::ValueRef::Blob(bytes) => format!("{:?}", bytes), duckdb::types::ValueRef::Date32(i) => { use chrono::{NaiveDate, Datelike}; let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).unwrap(); if let Some(date) = epoch.checked_add_signed(chrono::Duration::days(i as i64)) { format!("{:04}-{:02}-{:02}", date.year(), date.month(), date.day()) } else { i.to_string() } }, duckdb::types::ValueRef::Time64(_, i) => i.to_string(), duckdb::types::ValueRef::Interval { months, days, nanos } => format!("{}-{}-{}", months, days, nanos), _ => "".to_string(), } } fn trim_float(v: f64) -> String { let s = v.to_string(); if s.contains('.') { s.trim_end_matches('0').trim_end_matches('.').to_string() } else { s } } pub fn explain_query(con: &Connection, sql: &str, query_num: usize) -> Result<()> { use std::path::PathBuf; let profile_dir = PathBuf::from("profiling"); std::fs::create_dir_all(&profile_dir)?; let profile_file = profile_dir.join(format!("q{}.json", query_num)); let temp_file = format!("/tmp/duckdb_profile_{}.json", query_num); con.execute("PRAGMA enable_profiling = 'json'", [])?; con.execute(&format!("PRAGMA profiling_output = '{}'", temp_file), [])?; con.execute( r#"PRAGMA custom_profiling_settings = '{"OPERATOR_TIMING": "true", "OPERATOR_CARDINALITY": "true", "CPU_TIME": "true", "EXTRA_INFO": "true"}'"#, [], )?; let mut stmt = con.prepare(sql)?; let _rows = stmt.query([])?; match std::fs::read_to_string(&temp_file) { Ok(json_content) => { std::fs::write(&profile_file, &json_content)?; let _ = std::fs::remove_file(&temp_file); } Err(_e) => { } } Ok(()) } pub fn prepare_query<'a>(con: &'a Connection, sql: &str) -> Result> { let stmt = con.prepare(sql)?; Ok(stmt) } pub fn write_single_result_to_csv( query_num: usize, mut rows: duckdb::Rows, output_dir: &PathBuf, ) -> Result<()> { fs::create_dir_all(output_dir)?; let out_path = output_dir.join(format!("q{}.csv", query_num)); let mut file = std::fs::File::create(&out_path)?; let mut wtr = csv::Writer::from_writer(&mut file); let stmt_ref = rows.as_ref().ok_or_else(|| anyhow::anyhow!("Failed to get statement reference"))?; let column_count = stmt_ref.column_count(); let mut columns: Vec = (0..column_count) .map(|i| stmt_ref.column_name(i).map(|s| s.to_string())) .collect::, _>>()?; // Fix column name normalization: DuckDB Rust bindings may return count(*) instead of count_star() // Check if we have COUNT(*) and normalize it for col in &mut columns { if col == "count(*)" { *col = "count_star()".to_string(); } } wtr.write_record(&columns)?; while let Some(row) = rows.next()? { let mut record = Vec::new(); for i in 0..column_count { let value = extract_value_as_string(&row, i); record.push(value); } wtr.write_record(&record)?; } wtr.flush()?; Ok(()) } ``` ### src/query_handler.rs ```rust use anyhow::Result; use serde_json::Value; use std::path::PathBuf; use std::fs; pub fn parse_queries_from_file(queries_path: &PathBuf) -> Result> { let content = fs::read_to_string(queries_path)?; let queries: Vec = serde_json::from_str(&content)?; Ok(queries) } pub fn assemble_sql(q: &Value) -> String { let select = select_to_sql(q.get("select").unwrap_or(&Value::Array(vec![]))); let from_tbl = q["from"].as_str().unwrap_or("events_table"); let where_clause = where_to_sql(q.get("where")); let group_by = group_by_to_sql(q.get("group_by")); let order_by = order_by_to_sql(q.get("order_by")); let mut sql = format!("SELECT {} FROM {}", select, from_tbl); if !where_clause.is_empty() { sql.push_str(&format!(" {}", where_clause)); } if !group_by.is_empty() { sql.push_str(&format!(" {}", group_by)); } if !order_by.is_empty() { sql.push_str(&format!(" {}", order_by)); } if let Some(limit) = q.get("limit") { sql.push_str(&format!(" LIMIT {}", limit)); } sql } fn where_to_sql(where_clause: Option<&Value>) -> String { if where_clause.is_none() { return String::new(); } let Some(conditions) = where_clause.and_then(|w| w.as_array()) else { return String::new(); }; let parts: Vec = conditions.iter().map(|cond| { let col = cond["col"].as_str().unwrap_or(""); let op = cond["op"].as_str().unwrap_or(""); let val = &cond["val"]; if op == "eq" { format!("{} = '{}'", col, val.as_str().unwrap_or("")) } else if op == "neq" { format!("{} != '{}'", col, val.as_str().unwrap_or("")) } else if op == "lt" { format!("{} < {}", col, format_value_for_sql(val)) } else if op == "lte" { format!("{} <= {}", col, format_value_for_sql(val)) } else if op == "gt" { format!("{} > {}", col, format_value_for_sql(val)) } else if op == "gte" { format!("{} >= {}", col, format_value_for_sql(val)) } else if op == "between" { if let Some(vals) = val.as_array() { let low = vals[0].as_str().unwrap_or(""); let high = vals[1].as_str().unwrap_or(""); format!("{} BETWEEN '{}' AND '{}'", col, low, high) } else { String::new() } } else if op == "in" { if let Some(vals) = val.as_array() { let vals_str = vals.iter() .map(|v| format!("'{}'", v.as_str().unwrap_or(""))) .collect::>() .join(", "); format!("{} IN ({})", col, vals_str) } else { String::new() } } else { String::new() } }).collect(); if parts.is_empty() { String::new() } else { format!("WHERE {}", parts.join(" AND ")) } } fn format_value_for_sql(val: &serde_json::Value) -> String { if let Some(num) = val.as_f64() { if num.fract() == 0.0 { format!("{}", num as i64) } else { format!("{}", num) } } else if let Some(str_val) = val.as_str() { // Try to parse as number if let Ok(num) = str_val.parse::() { if num.fract() == 0.0 { format!("{}", num as i64) } else { format!("{}", num) } } else { format!("'{}'", str_val) } } else if let Some(num) = val.as_u64() { format!("{}", num) } else if let Some(num) = val.as_i64() { format!("{}", num) } else { val.to_string() } } fn select_to_sql(select: &Value) -> String { let Some(select_array) = select.as_array() else { return "*".to_string(); }; let parts: Vec = select_array.iter().map(|item| { if let Some(s) = item.as_str() { s.to_string() } else if let Some(obj) = item.as_object() { let mut result = String::new(); for (func, col) in obj { result = format!("{}({})", func.to_uppercase(), col.as_str().unwrap_or("")); } result } else { String::new() } }).collect(); if parts.is_empty() { "*".to_string() } else { parts.join(", ") } } fn group_by_to_sql(group_by: Option<&Value>) -> String { if let Some(gb) = group_by { if let Some(gb_array) = gb.as_array() { if !gb_array.is_empty() { let parts: Vec = gb_array.iter() .filter_map(|v| v.as_str()) .map(|s| s.to_string()) .collect(); if !parts.is_empty() { return format!("GROUP BY {}", parts.join(", ")); } } } } String::new() } fn order_by_to_sql(order_by: Option<&Value>) -> String { if let Some(ob) = order_by { if let Some(ob_array) = ob.as_array() { if !ob_array.is_empty() { let parts: Vec = ob_array.iter().map(|o| { let col = o["col"].as_str().unwrap_or(""); let dir = o.get("dir").and_then(|d| d.as_str()).unwrap_or("asc").to_uppercase(); format!("{} {}", col, dir) }).collect(); if !parts.is_empty() { return format!("ORDER BY {}", parts.join(", ")); } } } } String::new() } ``` ### src/data_loader.rs ```rust use duckdb::Connection; use std::path::PathBuf; use anyhow::Result; pub fn load_data(con: &Connection, data_dir: &PathBuf) -> Result> { // Determine parquet file/directory location (in data directory parent) let parquet_dir = data_dir.parent() .unwrap_or(data_dir) .join("events.parquet"); // Check if parquet already exists let parquet_exists = parquet_dir.is_dir() || parquet_dir.exists(); let parquet_path = if parquet_exists { // Parquet exists - use it directly, skip CSV entirely let parquet_pattern = if parquet_dir.is_dir() { // Directory with multiple parquet files - use glob pattern format!("{}/data_*.parquet", parquet_dir.to_string_lossy()) } else { // Single parquet file parquet_dir.to_string_lossy().to_string() }; // Create events view directly from Parquet con.execute( &format!( r#" CREATE OR REPLACE VIEW events AS SELECT ts, DATE_TRUNC('week', ts) AS week, DATE(ts) AS day, DATE_TRUNC('hour', ts) AS hour, STRFTIME(ts, '%Y-%m-%d %H:%M') AS minute, type, auction_id, advertiser_id, publisher_id, bid_price, user_id, total_price, country FROM read_parquet('{}') "#, parquet_pattern ), [], )?; Some(parquet_dir) } else { // Parquet doesn't exist - need to generate it from CSV let csv_pattern = format!("{}/events_part_*.csv", data_dir.to_string_lossy()); // Create the events view from CSV first con.execute( &format!( r#" CREATE OR REPLACE VIEW events AS WITH raw AS ( SELECT * FROM read_csv( '{}', AUTO_DETECT = FALSE, HEADER = TRUE, union_by_name = TRUE, COLUMNS = {{ 'ts': 'VARCHAR', 'type': 'VARCHAR', 'auction_id': 'VARCHAR', 'advertiser_id': 'VARCHAR', 'publisher_id': 'VARCHAR', 'bid_price': 'VARCHAR', 'user_id': 'VARCHAR', 'total_price': 'VARCHAR', 'country': 'VARCHAR' }} ) ), casted AS ( SELECT to_timestamp(TRY_CAST(ts AS DOUBLE) / 1000.0) AS ts, TRY_CAST(type AS ENUM('impression','serve','click','purchase')) AS type, auction_id, TRY_CAST(advertiser_id AS INTEGER) AS advertiser_id, TRY_CAST(publisher_id AS INTEGER) AS publisher_id, NULLIF(bid_price, '')::DOUBLE AS bid_price, TRY_CAST(user_id AS BIGINT) AS user_id, NULLIF(total_price, '')::DOUBLE AS total_price, country FROM raw ) SELECT ts, DATE_TRUNC('week', ts) AS week, DATE(ts) AS day, DATE_TRUNC('hour', ts) AS hour, STRFTIME(ts, '%Y-%m-%d %H:%M') AS minute, type, auction_id, advertiser_id, publisher_id, bid_price, user_id, total_price, country FROM casted "#, csv_pattern ), [], )?; // Generate parquet from CSV view // Use hardware-aware Parquet generation use crate::hardware::get_hardware_info; let hw = get_hardware_info(); // Estimate total rows (rough estimate: 245M for full dataset) let estimated_rows = 245_000_000; // Full dataset estimate let optimal_row_group_size = hw.optimal_row_group_size(estimated_rows); // Create directory if it doesn't exist std::fs::create_dir_all(&parquet_dir)?; let parquet_file = parquet_dir.join("data.parquet"); con.execute( &format!( "COPY (SELECT * FROM events) TO '{}' (FORMAT PARQUET, COMPRESSION ZSTD, PER_THREAD_OUTPUT, ROW_GROUP_SIZE {});", parquet_file.to_string_lossy(), optimal_row_group_size ), [], )?; // Replace events view to read from Parquet let parquet_pattern = format!("{}/data_*.parquet", parquet_dir.to_string_lossy()); con.execute( &format!( r#" CREATE OR REPLACE VIEW events AS SELECT ts, DATE_TRUNC('week', ts) AS week, DATE(ts) AS day, DATE_TRUNC('hour', ts) AS hour, STRFTIME(ts, '%Y-%m-%d %H:%M') AS minute, type, auction_id, advertiser_id, publisher_id, bid_price, user_id, total_price, country FROM read_parquet('{}') "#, parquet_pattern ), [], )?; Some(parq [truncated — 43 more characters] ``` [3 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]