From f9ad4393ce9107fadc87032dd94d6e65a1afd6ed Mon Sep 17 00:00:00 2001 From: Chris Agostino Date: Wed, 9 Sep 2026 19:19:35 -0500 Subject: [PATCH 1/2] feat(nql): log BigQuery bytes scanned per model and run total --- Cargo.toml | 2 +- src/runner.rs | 117 +++++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 113 insertions(+), 6 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 05b15c0..8fcc643 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nql" -version = "0.1.4" +version = "0.1.5" edition = "2021" authors = ["NPC Worldwide "] description = "NQL — SQL compiler with AI function calls targeting multiple databases" diff --git a/src/runner.rs b/src/runner.rs index e652c80..6572f56 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -161,24 +161,44 @@ impl Runner { } /// Compile all models and execute them against BigQuery via the `bq` CLI, - /// in dependency order. + /// in dependency order. Each query is tagged with a unique job id so we can + /// retrieve and log bytes scanned after it finishes. pub fn execute_bigquery(&self) -> Result<(), RunnerError> { + use std::time::{SystemTime, UNIX_EPOCH}; + let results = self.compile_all()?; - let mut bq_args: Vec = vec![ + let mut query_args: Vec = vec![ "query".to_string(), "--use_legacy_sql=false".to_string(), "--format=none".to_string(), ]; + let mut show_flags: Vec = vec!["--format=json".to_string()]; if let Ok(project) = std::env::var("NQL_BQ_PROJECT") { - bq_args.push(format!("--project_id={}", project)); + query_args.push(format!("--project_id={}", project)); + show_flags.push(format!("--project_id={}", project)); } if let Ok(location) = std::env::var("NQL_BQ_LOCATION") { - bq_args.push(format!("--location={}", location)); + query_args.push(format!("--location={}", location)); + show_flags.push(format!("--location={}", location)); } + let mut total_bytes: u64 = 0; + for (name, sql) in results { - eprintln!("[nql] Executing model on BigQuery: {}", name); + let job_id = format!( + "nql_{}_{}_{}", + name.replace(|c: char| !c.is_alphanumeric() && c != '-' && c != '_', "_"), + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + let mut bq_args = query_args.clone(); + bq_args.push(format!("--job_id={}", job_id)); + + eprintln!("[nql] Executing model on BigQuery: {} (job_id={})", name, job_id); let mut child = std::process::Command::new("bq") .args(&bq_args) .stdin(std::process::Stdio::piped()) @@ -193,6 +213,10 @@ impl Runner { let status = child .wait_with_output() .map_err(|e| RunnerError::Compilation(format!("Failed waiting for bq: {}", e)))?; + + // Log cost regardless of success/failure if the job was created. + let _ = Self::log_bq_job_cost(&show_flags, &job_id, &name); + if !status.status.success() { return Err(RunnerError::Compilation(format!( "bq query failed for model '{}': {}", @@ -200,11 +224,94 @@ impl Runner { String::from_utf8_lossy(&status.stderr) ))); } + + if let Some(bytes) = Self::bytes_from_bq_job(&show_flags, &job_id) { + total_bytes += bytes; + } } + let gb = total_bytes as f64 / (1024.0 * 1024.0 * 1024.0); + let tb = total_bytes as f64 / (1024.0 * 1024.0 * 1024.0 * 1024.0); + eprintln!( + "[nql][cost] TOTAL bytes processed this nql run: {} ({:.6} GB)", + total_bytes, gb + ); + eprintln!( + "[nql][cost] At $5/TB, estimated query scan cost: ${:.6}", + tb * 5.0 + ); + Ok(()) } + fn log_bq_job_cost(show_flags: &[String], job_id: &str, model: &str) -> Option<()> { + let output = std::process::Command::new("bq") + .args(show_flags) + .arg("show") + .arg("-j") + .arg(job_id) + .output() + .ok()?; + if !output.status.success() { + return None; + } + let json: serde_json::Value = serde_json::from_slice(&output.stdout).ok()?; + let query_stats = json + .get("statistics") + .and_then(|s| s.get("query")) + .cloned() + .unwrap_or_default(); + let job_id_out = json + .get("jobReference") + .and_then(|j| j.get("jobId")) + .and_then(|j| j.as_str()) + .unwrap_or(job_id); + let bytes = query_stats + .get("totalBytesProcessed") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()); + let cache_hit = query_stats + .get("cacheHit") + .and_then(|v| v.as_bool()) + .map(|b| b.to_string()) + .unwrap_or_else(|| "None".to_string()); + let billing_tier = query_stats + .get("billingTier") + .and_then(|v| v.as_i64()) + .map(|t| t.to_string()) + .unwrap_or_else(|| "None".to_string()); + + if let Some(b) = bytes { + let gb = b as f64 / (1024.0 * 1024.0 * 1024.0); + eprintln!( + "[nql][cost] {}: {} processed {:.6} GB (cache_hit={}, tier={})", + model, job_id_out, gb, cache_hit, billing_tier + ); + } else { + eprintln!("[nql][cost] {}: {} (no bytes metric)", model, job_id_out); + } + Some(()) + } + + fn bytes_from_bq_job(show_flags: &[String], job_id: &str) -> Option { + let output = std::process::Command::new("bq") + .args(show_flags) + .arg("show") + .arg("-j") + .arg(job_id) + .output() + .ok()?; + if !output.status.success() { + return None; + } + let json: serde_json::Value = serde_json::from_slice(&output.stdout).ok()?; + json.get("statistics") + .and_then(|s| s.get("query")) + .and_then(|q| q.get("totalBytesProcessed")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + } + /// Compile all models and return the SQL strings (for non-SQLite targets). pub fn compile_all(&self) -> Result, RunnerError> { let mut compiler = Compiler::new(self.target); From 3fb179614ceb586d97972b8f491aed908a20648a Mon Sep 17 00:00:00 2001 From: Chris Agostino Date: Mon, 28 Sep 2026 17:55:14 -0500 Subject: [PATCH 2/2] avoid text mentions comments in facts --- .gitignore | 1 + Cargo.toml | 2 +- src/compiler.rs | 2 +- src/sqlite_udf.rs | 1 + 4 files changed, 4 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index b9f381a..6121445 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ target/ example/ *.db +Cargo.lock diff --git a/Cargo.toml b/Cargo.toml index 8fcc643..412a148 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nql" -version = "0.1.5" +version = "0.1.6" edition = "2021" authors = ["NPC Worldwide "] description = "NQL — SQL compiler with AI function calls targeting multiple databases" diff --git a/src/compiler.rs b/src/compiler.rs index 9a7f357..67e3eb5 100644 --- a/src/compiler.rs +++ b/src/compiler.rs @@ -322,7 +322,7 @@ impl Compiler { format!("(AI.GENERATE(CONCAT('Analyze the sentiment of the following text. Respond with exactly one word: positive, negative, or neutral.\\n\\n', {}))).result", args) } "get_facts" => { - format!("(AI.GENERATE(CONCAT('Extract facts from this text. A fact is a specific statement that can be sourced from the text. Return as JSON array of objects with \"statement\", \"source_text\", and \"type\" (explicit or inferred) fields.{}\\n\\nText: ', {}))).result", BQ_JSON_FORMAT_SUFFIX, args) + format!("(AI.GENERATE(CONCAT('Extract facts from this text. A fact is a specific statement that can be sourced from the text. Return as JSON array of objects with \"statement\", \"source_text\", and \"type\" (explicit or inferred) fields. ABSOLUTE RULE: state every fact directly as a concrete domain claim; NEVER use meta-textual framing such as \"the text mentions\", \"the text refers to\", \"the content states\", \"according to the text\", or \"the passage says\".{}\\n\\nText: ', {}))).result", BQ_JSON_FORMAT_SUFFIX, args) } "identify_groups" => { format!("(AI.GENERATE(CONCAT('What are the main groups these items could be organized into? Return as JSON array of group names.{}\\n\\nItems: ', {}))).result", BQ_JSON_FORMAT_SUFFIX, args) diff --git a/src/sqlite_udf.rs b/src/sqlite_udf.rs index 89f2385..69d10a0 100644 --- a/src/sqlite_udf.rs +++ b/src/sqlite_udf.rs @@ -178,6 +178,7 @@ pub fn register_all_functions(conn: &Connection) -> rusqlite::Result<()> { register_udf(conn, "get_facts", &config, |text| { format!( "Extract facts from this text. A fact is a specific statement that can be sourced from the text.\n\n\ + ABSOLUTE RULE: state every fact directly as a concrete domain claim; NEVER use meta-textual framing such as \"the text mentions\", \"the text refers to\", \"the content states\", \"according to the text\", or \"the passage says\".\n\n\ Text: \"{}\"\n\n\ Return as JSON array of objects with \"statement\", \"source_text\", and \"type\" (explicit or inferred) fields.", text