use std::io::{IsTerminal, Read};
use std::path::PathBuf;
use ctx::error::{CtxError, Result};
use ctx::exit::Outcome;
pub const SQL_SCHEMA_REFERENCE: &str = include_str!("sql_schema.md");
#[cfg_attr(not(feature = "duckdb"), allow(dead_code))]
pub struct SqlArgs {
pub query: Option<String>,
pub file: Option<PathBuf>,
pub format: String,
pub json: bool,
pub max_rows: usize,
pub timeout: u64,
pub fail_on_rows: bool,
pub schema: bool,
pub snapshots: Option<PathBuf>,
}
pub fn run_sql(args: SqlArgs) -> Result<Outcome> {
if args.schema {
print!("{}", SQL_SCHEMA_REFERENCE);
return Ok(Outcome::Clean);
}
let format = if args.json {
"json"
} else {
args.format.as_str()
};
if !matches!(format, "table" | "csv" | "json") {
return Err(CtxError::Other(format!(
"unknown --format '{}'; expected table, csv, or json",
format
)));
}
let sql_text = resolve_query_text(&args)?;
if sql_text.trim().is_empty() {
return Err(CtxError::Other(
"no query provided; pass SQL as an argument, via --file, or on stdin".into(),
));
}
#[cfg(not(feature = "duckdb"))]
{
let _ = format;
Err(CtxError::Other(
"ctx sql requires the duckdb feature; reinstall with default features".into(),
))
}
#[cfg(feature = "duckdb")]
{
run_sql_duckdb(&args, format, &sql_text)
}
}
fn resolve_query_text(args: &SqlArgs) -> Result<String> {
match (&args.query, &args.file) {
(Some(q), Some(_)) if q != "-" => Err(CtxError::Other(
"provide the query as an argument OR via --file, not both".into(),
)),
(_, Some(path)) => Ok(std::fs::read_to_string(path)?),
(Some(q), None) if q != "-" => Ok(q.clone()),
_ => {
if std::io::stdin().is_terminal() {
return Err(CtxError::Other(
"no query provided; pass SQL as an argument, via --file, or on stdin".into(),
));
}
let mut buf = String::new();
std::io::stdin().read_to_string(&mut buf)?;
Ok(buf)
}
}
}
#[cfg_attr(not(feature = "duckdb"), allow(dead_code))]
fn split_sql_statements(sql: &str) -> Vec<String> {
#[derive(PartialEq)]
enum State {
Normal,
Single,
Double,
Line,
Block,
}
let mut statements = Vec::new();
let mut current = String::new();
let mut state = State::Normal;
let mut chars = sql.chars().peekable();
while let Some(c) = chars.next() {
match state {
State::Normal => match c {
'\'' => {
current.push(c);
state = State::Single;
}
'"' => {
current.push(c);
state = State::Double;
}
'-' if chars.peek() == Some(&'-') => {
current.push(c);
current.push(chars.next().unwrap());
state = State::Line;
}
'/' if chars.peek() == Some(&'*') => {
current.push(c);
current.push(chars.next().unwrap());
state = State::Block;
}
';' => {
let trimmed = current.trim();
if !trimmed.is_empty() {
statements.push(trimmed.to_string());
}
current.clear();
}
_ => current.push(c),
},
State::Single => {
current.push(c);
if c == '\'' {
if chars.peek() == Some(&'\'') {
current.push(chars.next().unwrap());
} else {
state = State::Normal;
}
}
}
State::Double => {
current.push(c);
if c == '"' {
if chars.peek() == Some(&'"') {
current.push(chars.next().unwrap());
} else {
state = State::Normal;
}
}
}
State::Line => {
current.push(c);
if c == '\n' {
state = State::Normal;
}
}
State::Block => {
current.push(c);
if c == '*' && chars.peek() == Some(&'/') {
current.push(chars.next().unwrap());
state = State::Normal;
}
}
}
}
let trimmed = current.trim();
if !trimmed.is_empty() {
statements.push(trimmed.to_string());
}
statements
}
#[cfg(feature = "duckdb")]
mod engine {
use super::*;
use ctx::analytics::{Analytics, SqlResult};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
enum EngineError {
MultipleResultSets,
Duck(duckdb::Error),
}
pub fn run_sql_duckdb(args: &SqlArgs, format: &str, sql_text: &str) -> Result<Outcome> {
use ctx::index::{CTX_DIR, DB_FILE};
let root = std::env::current_dir()?;
let db_path = root.join(CTX_DIR).join(DB_FILE);
if !db_path.exists() {
return Err(CtxError::Other(format!(
"no index found at {}; run `ctx index` first",
db_path.display()
)));
}
let statements = split_sql_statements(sql_text);
if statements.is_empty() {
return Err(CtxError::Other("no SQL statement found".into()));
}
let snapshots_dir = args.snapshots.as_ref().map(|dir| {
if dir.is_absolute() {
dir.clone()
} else {
root.join(dir)
}
});
let analytics = Analytics::open_sql_sandbox(&root, snapshots_dir.as_deref())?;
let timed_out = Arc::new(AtomicBool::new(false));
let done = Arc::new(AtomicBool::new(false));
let watchdog = if args.timeout > 0 {
let handle = analytics.interrupt_handle();
let timed_out = timed_out.clone();
let done = done.clone();
let timeout = args.timeout;
Some(std::thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(timeout);
while Instant::now() < deadline {
if done.load(Ordering::SeqCst) {
return;
}
std::thread::sleep(Duration::from_millis(50));
}
if !done.load(Ordering::SeqCst) {
timed_out.store(true, Ordering::SeqCst);
handle.interrupt();
}
}))
} else {
None
};
let start = Instant::now();
let result = execute_statements(&analytics, &statements, args.max_rows);
let elapsed_ms = start.elapsed().as_millis();
done.store(true, Ordering::SeqCst);
if let Some(w) = watchdog {
let _ = w.join();
}
let result = match result {
Ok(r) => r,
Err(EngineError::MultipleResultSets) => {
return Err(CtxError::Other(
"only the final statement may return rows; earlier statements must not produce a result set"
.into(),
));
}
Err(EngineError::Duck(e)) => {
if timed_out.load(Ordering::SeqCst) {
return Err(CtxError::Other(format!(
"query exceeded the {}s timeout; raise it with --timeout",
args.timeout
)));
}
return Err(CtxError::Other(format!("sql error: {}", e)));
}
};
render(&result, format, elapsed_ms);
if args.fail_on_rows && !result.rows.is_empty() {
Ok(Outcome::Findings)
} else {
Ok(Outcome::Clean)
}
}
fn execute_statements(
analytics: &Analytics,
statements: &[String],
max_rows: usize,
) -> std::result::Result<SqlResult, EngineError> {
let (last, leading) = statements
.split_last()
.expect("statements is non-empty (checked by caller)");
for stmt in leading {
let produced_result = analytics
.exec_non_final_statement(stmt)
.map_err(EngineError::Duck)?;
if produced_result {
return Err(EngineError::MultipleResultSets);
}
}
analytics
.run_final_statement(last, max_rows)
.map_err(EngineError::Duck)
}
fn render(result: &SqlResult, format: &str, elapsed_ms: u128) {
match format {
"json" => render_json(result, elapsed_ms),
"csv" => render_csv(result),
_ => render_table(result),
}
}
fn render_json(result: &SqlResult, elapsed_ms: u128) {
let columns: Vec<serde_json::Value> = result
.columns
.iter()
.map(|c| serde_json::json!({ "name": c.name, "type": c.type_name }))
.collect();
let data = serde_json::json!({
"columns": columns,
"rows": result.rows,
"row_count": result.rows.len(),
"truncated": result.truncated,
"elapsed_ms": elapsed_ms,
});
if let Err(e) = ctx::json::emit("sql", data) {
eprintln!("failed to serialize result: {}", e);
}
}
const MAX_CELL: usize = 200;
fn cell_string(value: &serde_json::Value) -> (String, bool) {
let raw = match value {
serde_json::Value::Null => return ("∅".to_string(), false),
serde_json::Value::String(s) => s.clone(),
serde_json::Value::Bool(b) => b.to_string(),
serde_json::Value::Number(n) => n.to_string(),
other => other.to_string(),
};
if raw.chars().count() > MAX_CELL {
let truncated: String = raw.chars().take(MAX_CELL).collect();
(format!("{}…", truncated), true)
} else {
(raw, false)
}
}
fn render_table(result: &SqlResult) {
let headers: Vec<&str> = result.columns.iter().map(|c| c.name.as_str()).collect();
let mut any_cell_truncated = false;
let rendered: Vec<Vec<String>> = result
.rows
.iter()
.map(|row| {
row.iter()
.map(|v| {
let (s, truncated) = cell_string(v);
any_cell_truncated |= truncated;
s
})
.collect()
})
.collect();
let mut widths: Vec<usize> = headers.iter().map(|h| h.chars().count()).collect();
for row in &rendered {
for (i, cell) in row.iter().enumerate() {
if i < widths.len() {
widths[i] = widths[i].max(cell.chars().count());
}
}
}
let mut out = String::new();
push_row(
&mut out,
&headers.iter().map(|h| h.to_string()).collect::<Vec<_>>(),
&widths,
);
push_separator(&mut out, &widths);
for row in &rendered {
push_row(&mut out, row, &widths);
}
print!("{}", out);
if result.truncated {
eprintln!(
"note: output capped at {} rows; raise the limit with --max-rows",
result.rows.len()
);
}
if any_cell_truncated {
eprintln!(
"note: some cells were truncated at {} characters; use --format json or csv for full values",
MAX_CELL
);
}
}
fn push_row(out: &mut String, cells: &[String], widths: &[usize]) {
let mut parts = Vec::with_capacity(cells.len());
for (i, cell) in cells.iter().enumerate() {
let width = widths.get(i).copied().unwrap_or(0);
let pad = width.saturating_sub(cell.chars().count());
parts.push(format!("{}{}", cell, " ".repeat(pad)));
}
out.push_str(parts.join(" ").trim_end());
out.push('\n');
}
fn push_separator(out: &mut String, widths: &[usize]) {
let parts: Vec<String> = widths.iter().map(|w| "-".repeat(*w)).collect();
out.push_str(parts.join(" ").trim_end());
out.push('\n');
}
fn render_csv(result: &SqlResult) {
let mut out = String::new();
let headers: Vec<String> = result.columns.iter().map(|c| csv_field(&c.name)).collect();
out.push_str(&headers.join(","));
out.push_str("\r\n");
for row in &result.rows {
let fields: Vec<String> = row
.iter()
.map(|v| match v {
serde_json::Value::Null => String::new(),
serde_json::Value::String(s) => csv_field(s),
serde_json::Value::Bool(b) => b.to_string(),
serde_json::Value::Number(n) => n.to_string(),
other => csv_field(&other.to_string()),
})
.collect();
out.push_str(&fields.join(","));
out.push_str("\r\n");
}
print!("{}", out);
if result.truncated {
eprintln!(
"note: output capped at {} rows; raise the limit with --max-rows",
result.rows.len()
);
}
}
fn csv_field(s: &str) -> String {
if s.contains([',', '"', '\n', '\r']) {
format!("\"{}\"", s.replace('"', "\"\""))
} else {
s.to_string()
}
}
}
#[cfg(feature = "duckdb")]
use engine::run_sql_duckdb;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn splits_simple_statements() {
let stmts = split_sql_statements("SELECT 1; SELECT 2");
assert_eq!(stmts, vec!["SELECT 1", "SELECT 2"]);
}
#[test]
fn ignores_semicolons_in_strings_and_comments() {
let stmts = split_sql_statements(
"SELECT ';' AS a; -- comment; not a split\nSELECT /* also; not */ 2",
);
assert_eq!(stmts.len(), 2);
assert!(stmts[0].contains("';'"));
assert!(stmts[1].contains("2"));
}
#[test]
fn single_statement_yields_one() {
assert_eq!(split_sql_statements("SELECT * FROM v1.symbols").len(), 1);
}
#[test]
fn trailing_semicolon_is_not_an_empty_statement() {
assert_eq!(split_sql_statements("SELECT 1;").len(), 1);
}
#[test]
fn schema_reference_matches_docs_page() {
let manifest = env!("CARGO_MANIFEST_DIR");
let doc_path = std::path::Path::new(manifest).join("docs/website/docs/sql-schema.md");
let doc = std::fs::read_to_string(&doc_path)
.unwrap_or_else(|e| panic!("read {}: {}", doc_path.display(), e));
let body = strip_front_matter(&doc);
assert_eq!(
body.trim(),
SQL_SCHEMA_REFERENCE.trim(),
"docs/website/docs/sql-schema.md content must match SQL_SCHEMA_REFERENCE"
);
}
fn strip_front_matter(doc: &str) -> &str {
if let Some(rest) = doc.strip_prefix("---") {
if let Some(end) = rest.find("\n---") {
let after = &rest[end + 4..];
return after.trim_start_matches(['\n', '\r']);
}
}
doc
}
}