Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
target/
example/
*.db
Cargo.lock
2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "nql"
version = "0.1.4"
version = "0.1.6"
edition = "2021"
authors = ["NPC Worldwide <info@npcworldwi.de>"]
description = "NQL — SQL compiler with AI function calls targeting multiple databases"
Expand Down
2 changes: 1 addition & 1 deletion src/compiler.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
117 changes: 112 additions & 5 deletions src/runner.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<String> = vec![
let mut query_args: Vec<String> = vec![
"query".to_string(),
"--use_legacy_sql=false".to_string(),
"--format=none".to_string(),
];
let mut show_flags: Vec<String> = 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())
Expand All @@ -193,18 +213,105 @@ 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 '{}': {}",
name,
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::<u64>().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<u64> {
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::<u64>().ok())
}

/// Compile all models and return the SQL strings (for non-SQLite targets).
pub fn compile_all(&self) -> Result<Vec<(String, String)>, RunnerError> {
let mut compiler = Compiler::new(self.target);
Expand Down
1 change: 1 addition & 0 deletions src/sqlite_udf.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading