1use crate::{detect_content, detect_extension, CanonicalValue, Format, SqlDialect};
30use serde_json::Number;
31use sqlparser::ast::{
32 BinaryOperator, Expr, FunctionArg, FunctionArgExpr, FunctionArguments, GroupByExpr,
33 JoinConstraint, JoinOperator, LimitClause, ObjectNamePart, OrderByExpr, OrderByKind, Query,
34 SelectItem, SetExpr, Statement, TableFactor, TableWithJoins, Value, ValueWithSpan,
35};
36use sqlparser::dialect::GenericDialect;
37use sqlparser::parser::Parser;
38use std::cmp::Ordering;
39use std::collections::{HashMap, HashSet};
40use std::error::Error;
41use std::path::{Path, PathBuf};
42
43type Row = Vec<(String, CanonicalValue)>;
44
45pub fn run(sql: &str, base_dir: &Path) -> Result<CanonicalValue, Box<dyn Error>> {
83 let stmts = Parser::parse_sql(&GenericDialect {}, sql)?;
84 let stmt = stmts.into_iter().next().ok_or("empty SQL")?;
85 match stmt {
86 Statement::Query(q) => execute_query(*q, base_dir),
87 _ => Err("only SELECT statements are supported".into()),
88 }
89}
90
91fn execute_query(query: Query, base_dir: &Path) -> Result<CanonicalValue, Box<dyn Error>> {
92 let body = (*query.body).clone();
93 let select = match body {
94 SetExpr::Select(s) => *s,
95 _ => return Err("only simple SELECT is supported".into()),
96 };
97
98 let mut rows = load_from(&select.from, base_dir)?;
99
100 if let Some(pred) = &select.selection {
101 let mut kept = Vec::new();
102 for row in rows {
103 if matches!(eval(pred.clone(), &row), Ok(CanonicalValue::Bool(true))) {
104 kept.push(row);
105 }
106 }
107 rows = kept;
108 }
109
110 let has_agg = select.projection.iter().any(|item| match item {
111 SelectItem::UnnamedExpr(e) | SelectItem::ExprWithAlias { expr: e, .. } => {
112 contains_aggregate(e)
113 }
114 _ => false,
115 });
116 let group_by = extract_group_by(&select.group_by);
117
118 let mut out_rows = if !group_by.is_empty() || has_agg {
119 aggregate(rows, &select.projection, &group_by, &select.having)?
120 } else {
121 project(rows, &select.projection)?
122 };
123
124 if select.distinct.is_some() {
125 let mut seen: HashSet<String> = HashSet::new();
126 out_rows.retain(|r| {
127 let key = r
128 .iter()
129 .map(|(_, v)| format!("{v:?}"))
130 .collect::<Vec<_>>()
131 .join("\x1f");
132 seen.insert(key)
133 });
134 }
135
136 if let Some(order) = &query.order_by {
137 if let OrderByKind::Expressions(exprs) = &order.kind {
138 sort_rows(&mut out_rows, exprs)?;
139 }
140 }
141
142 if let Some(lc) = &query.limit_clause {
143 if let Some(limit_expr) = extract_limit_expr(lc) {
144 if let Ok(CanonicalValue::Number(n)) = eval(limit_expr, &Vec::new()) {
145 if let Some(limit) = n.as_u64() {
146 out_rows.truncate(limit as usize);
147 }
148 }
149 }
150 }
151
152 Ok(CanonicalValue::Array(
153 out_rows.into_iter().map(CanonicalValue::Object).collect(),
154 ))
155}
156
157fn extract_group_by(g: &GroupByExpr) -> Vec<Expr> {
158 match g {
159 GroupByExpr::Expressions(exprs, _) => exprs.clone(),
160 _ => Vec::new(),
161 }
162}
163
164fn load_from(from: &[TableWithJoins], base_dir: &Path) -> Result<Vec<Row>, Box<dyn Error>> {
165 if from.is_empty() {
166 return Err("FROM is required".into());
167 }
168 let mut rows = load_table_with_joins(&from[0], base_dir)?;
169 for extra in &from[1..] {
170 let more = load_table_with_joins(extra, base_dir)?;
171 let mut crossed = Vec::new();
172 for a in &rows {
173 for b in &more {
174 let mut combined = a.clone();
175 combined.extend(b.clone());
176 crossed.push(combined);
177 }
178 }
179 rows = crossed;
180 }
181 Ok(rows)
182}
183
184fn load_table_with_joins(t: &TableWithJoins, base_dir: &Path) -> Result<Vec<Row>, Box<dyn Error>> {
185 let mut rows = load_factor(&t.relation, base_dir)?;
186 for join in &t.joins {
187 let right = load_factor(&join.relation, base_dir)?;
188 rows = apply_join(rows, right, &join.join_operator)?;
189 }
190 Ok(rows)
191}
192
193fn load_factor(factor: &TableFactor, base_dir: &Path) -> Result<Vec<Row>, Box<dyn Error>> {
194 let (name, alias) = match factor {
195 TableFactor::Table { name, alias, .. } => (name, alias),
196 _ => return Err("only table references are supported in FROM".into()),
197 };
198
199 let raw = name
200 .0
201 .iter()
202 .filter_map(|p| match p {
203 ObjectNamePart::Identifier(i) => Some(i.value.clone()),
204 _ => None,
205 })
206 .collect::<Vec<_>>()
207 .join(".");
208
209 let alias_name = alias
210 .as_ref()
211 .map(|a| a.name.value.clone())
212 .unwrap_or_else(|| {
213 Path::new(&raw)
214 .file_stem()
215 .and_then(|s| s.to_str())
216 .unwrap_or(&raw)
217 .to_string()
218 });
219
220 let mut all_rows = Vec::new();
221 if is_glob(&raw) {
222 let pattern = resolve_path(&raw, base_dir).to_string_lossy().to_string();
223 let entries = glob::glob(&pattern).map_err(|e| format!("invalid glob {pattern}: {e}"))?;
224 for entry in entries {
225 let p = entry?;
226 let mut r = load_file(&p)?;
227 all_rows.append(&mut r);
228 }
229 } else {
230 all_rows = load_file(&resolve_path(&raw, base_dir))?;
231 }
232
233 Ok(all_rows
234 .into_iter()
235 .map(|row| {
236 row.into_iter()
237 .map(|(k, v)| (format!("{alias_name}.{k}"), v))
238 .collect()
239 })
240 .collect())
241}
242
243fn is_glob(s: &str) -> bool {
244 s.contains('*') || s.contains('?') || s.contains('[')
245}
246
247fn resolve_path(raw: &str, base_dir: &Path) -> PathBuf {
248 let candidate = base_dir.join(raw);
249 if candidate.exists() || is_glob(raw) || Path::new(raw).is_absolute() {
250 return candidate;
251 }
252 let parts: Vec<&str> = raw.split('.').collect();
253 if parts.len() > 2 {
254 let prefix = parts[..parts.len() - 2].join("/");
255 let suffix = parts[parts.len() - 2..].join(".");
256 let slashed = base_dir.join(format!("{prefix}/{suffix}"));
257 if slashed.exists() {
258 return slashed;
259 }
260 }
261 candidate
262}
263
264fn load_file(path: &Path) -> Result<Vec<Row>, Box<dyn Error>> {
265 let bytes = std::fs::read(path)?;
266 let fmt = match detect_extension(path) {
267 Some(f) => f,
268 None => detect_content(&bytes)?,
269 };
270 let value = decode(fmt, &bytes)?;
271 match value {
272 CanonicalValue::Array(rows) => Ok(rows
273 .into_iter()
274 .filter_map(|r| match r {
275 CanonicalValue::Object(o) => Some(o),
276 _ => None,
277 })
278 .collect()),
279 _ => Err(format!("{} did not produce a table", path.display()).into()),
280 }
281}
282
283fn decode(fmt: Format, bytes: &[u8]) -> Result<CanonicalValue, Box<dyn Error>> {
284 let text = || -> Result<&str, Box<dyn Error>> { Ok(std::str::from_utf8(bytes)?) };
285 match fmt {
286 Format::Csv => CanonicalValue::from_csv(text()?),
287 Format::Tsv => CanonicalValue::from_tsv(text()?),
288 Format::Json => CanonicalValue::from_json(text()?),
289 Format::Ndjson => CanonicalValue::from_ndjson(text()?),
290 Format::Yaml => CanonicalValue::from_yaml(text()?),
291 Format::Xml => CanonicalValue::from_xml(text()?),
292 Format::Toml => CanonicalValue::from_toml(text()?),
293 Format::Xlsx => {
294 let mut sheets = CanonicalValue::from_xlsx_workbook(bytes)?;
295 if sheets.len() != 1 {
296 return Err("multi-sheet XLSX requires a single sheet".into());
297 }
298 Ok(sheets.remove(0).1)
299 }
300 Format::Ods => CanonicalValue::from_ods(bytes),
301 Format::Sql => {
302 let mut sheets = CanonicalValue::from_sql_workbook(text()?, SqlDialect::Generic)?;
303 if sheets.len() != 1 {
304 return Err("multi-table SQL requires a single table".into());
305 }
306 Ok(sheets.remove(0).1)
307 }
308 }
309}
310
311fn apply_join(
312 left: Vec<Row>,
313 right: Vec<Row>,
314 op: &JoinOperator,
315) -> Result<Vec<Row>, Box<dyn Error>> {
316 let (constraint, is_left) = match op {
317 JoinOperator::Inner(c) => (c, false),
318 JoinOperator::Join(c) => (c, false),
319 JoinOperator::LeftOuter(c) => (c, true),
320 other => return Err(format!("unsupported join variant: {other:?}").into()),
321 };
322 let on = match constraint {
323 JoinConstraint::On(e) => e.clone(),
324 _ => return Err("only JOIN ... ON is supported".into()),
325 };
326
327 let mut result = Vec::new();
328 for l in &left {
329 let mut matched = false;
330 for r in &right {
331 let mut combined = l.clone();
332 combined.extend(r.clone());
333 if matches!(eval(on.clone(), &combined), Ok(CanonicalValue::Bool(true))) {
334 result.push(combined);
335 matched = true;
336 }
337 }
338 if is_left && !matched {
339 result.push(l.clone());
340 }
341 }
342 Ok(result)
343}
344
345fn eval(expr: Expr, row: &Row) -> Result<CanonicalValue, Box<dyn Error>> {
346 match expr {
347 Expr::Identifier(i) => resolve_column(&i.value, row),
348 Expr::CompoundIdentifier(parts) => {
349 let name = parts
350 .iter()
351 .map(|p| p.value.clone())
352 .collect::<Vec<_>>()
353 .join(".");
354 resolve_column(&name, row)
355 }
356 Expr::Value(ValueWithSpan { value, .. }) => Ok(literal(value)),
357 Expr::BinaryOp { left, op, right } => {
358 let l = eval(*left, row)?;
359 let r = eval(*right, row)?;
360 binary_op(l, &op, r)
361 }
362 Expr::Nested(inner) => eval(*inner, row),
363 other => Err(format!("unsupported expression: {other:?}").into()),
364 }
365}
366
367fn resolve_column(name: &str, row: &Row) -> Result<CanonicalValue, Box<dyn Error>> {
368 if let Some((_, v)) = row.iter().find(|(k, _)| k == name) {
369 return Ok(v.clone());
370 }
371 let suffix = format!(".{name}");
372 if let Some((_, v)) = row.iter().find(|(k, _)| k.ends_with(&suffix)) {
373 return Ok(v.clone());
374 }
375 Err(format!("column {name} not found").into())
376}
377
378fn literal(v: Value) -> CanonicalValue {
379 match v {
380 Value::Number(n, _) => {
381 if let Ok(i) = n.parse::<i64>() {
382 CanonicalValue::Number(Number::from(i))
383 } else if let Ok(f) = n.parse::<f64>() {
384 Number::from_f64(f)
385 .map(CanonicalValue::Number)
386 .unwrap_or(CanonicalValue::Null)
387 } else {
388 CanonicalValue::String(n)
389 }
390 }
391 Value::SingleQuotedString(s) | Value::DoubleQuotedString(s) => CanonicalValue::String(s),
392 Value::Boolean(b) => CanonicalValue::Bool(b),
393 Value::Null => CanonicalValue::Null,
394 _ => CanonicalValue::Null,
395 }
396}
397
398fn binary_op(
399 l: CanonicalValue,
400 op: &BinaryOperator,
401 r: CanonicalValue,
402) -> Result<CanonicalValue, Box<dyn Error>> {
403 if matches!(l, CanonicalValue::Null) || matches!(r, CanonicalValue::Null) {
404 match op {
405 BinaryOperator::Eq
406 | BinaryOperator::NotEq
407 | BinaryOperator::Lt
408 | BinaryOperator::LtEq
409 | BinaryOperator::Gt
410 | BinaryOperator::GtEq => return Ok(CanonicalValue::Bool(false)),
411 _ => {}
412 }
413 }
414 let ord = || compare_values(&l, &r);
415 match op {
416 BinaryOperator::Eq => return Ok(CanonicalValue::Bool(ord() == Ordering::Equal)),
417 BinaryOperator::NotEq => return Ok(CanonicalValue::Bool(ord() != Ordering::Equal)),
418 BinaryOperator::Lt => return Ok(CanonicalValue::Bool(ord() == Ordering::Less)),
419 BinaryOperator::LtEq => return Ok(CanonicalValue::Bool(ord() != Ordering::Greater)),
420 BinaryOperator::Gt => return Ok(CanonicalValue::Bool(ord() == Ordering::Greater)),
421 BinaryOperator::GtEq => return Ok(CanonicalValue::Bool(ord() != Ordering::Less)),
422 BinaryOperator::And => {
423 let a = matches!(l, CanonicalValue::Bool(true));
424 let b = matches!(r, CanonicalValue::Bool(true));
425 return Ok(CanonicalValue::Bool(a && b));
426 }
427 BinaryOperator::Or => {
428 let a = matches!(l, CanonicalValue::Bool(true));
429 let b = matches!(r, CanonicalValue::Bool(true));
430 return Ok(CanonicalValue::Bool(a || b));
431 }
432 _ => {}
433 }
434 let a = to_f64(&l);
435 let b = to_f64(&r);
436 let result = match (a, b) {
437 (Some(a), Some(b)) => match op {
438 BinaryOperator::Plus => Some(a + b),
439 BinaryOperator::Minus => Some(a - b),
440 BinaryOperator::Multiply => Some(a * b),
441 BinaryOperator::Divide => Some(a / b),
442 _ => None,
443 },
444 _ => None,
445 };
446 match result {
447 Some(f) => Ok(num(f)),
448 None => Err(format!("cannot apply {op:?} to {l:?} and {r:?}").into()),
449 }
450}
451
452fn to_f64(v: &CanonicalValue) -> Option<f64> {
453 match v {
454 CanonicalValue::Number(n) => n.as_f64(),
455 CanonicalValue::String(s) => s.parse().ok(),
456 CanonicalValue::Bool(b) => Some(if *b { 1.0 } else { 0.0 }),
457 _ => None,
458 }
459}
460
461fn num(f: f64) -> CanonicalValue {
462 if f.is_finite() && f.fract() == 0.0 && f.abs() < i64::MAX as f64 {
463 CanonicalValue::Number(Number::from(f as i64))
464 } else {
465 Number::from_f64(f)
466 .map(CanonicalValue::Number)
467 .unwrap_or(CanonicalValue::Null)
468 }
469}
470
471fn project(rows: Vec<Row>, projection: &[SelectItem]) -> Result<Vec<Row>, Box<dyn Error>> {
472 let mut out = Vec::with_capacity(rows.len());
473 for row in rows {
474 let mut new_row = Vec::new();
475 for item in projection {
476 match item {
477 SelectItem::UnnamedExpr(expr) => {
478 let name = expr_name(expr);
479 new_row.push((name, eval(expr.clone(), &row)?));
480 }
481 SelectItem::ExprWithAlias { expr, alias } => {
482 new_row.push((alias.value.clone(), eval(expr.clone(), &row)?));
483 }
484 SelectItem::Wildcard(_) => {
485 for (k, v) in &row {
486 let name = k.rsplit('.').next().unwrap_or(k).to_string();
487 new_row.push((name, v.clone()));
488 }
489 }
490 SelectItem::QualifiedWildcard(name, _) => {
491 let prefix = format!("{name}.");
492 for (k, v) in &row {
493 if let Some(short) = k.strip_prefix(&prefix) {
494 new_row.push((short.to_string(), v.clone()));
495 }
496 }
497 }
498 SelectItem::ExprWithAliases { .. } => {
499 return Err("multi-alias SELECT not supported".into());
500 }
501 }
502 }
503 out.push(new_row);
504 }
505 Ok(out)
506}
507
508fn expr_name(expr: &Expr) -> String {
509 match expr {
510 Expr::Identifier(i) => i.value.clone(),
511 Expr::CompoundIdentifier(parts) => {
512 parts.last().map(|p| p.value.clone()).unwrap_or_default()
513 }
514 Expr::Function(f) => {
515 let fname = f.name.to_string().to_lowercase();
516 let arg = match &f.args {
517 FunctionArguments::List(l) => l
518 .args
519 .first()
520 .and_then(|a| match a {
521 FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => Some(expr_name(e)),
522 _ => None,
523 })
524 .unwrap_or_default(),
525 _ => String::new(),
526 };
527 format!("{fname}({arg})")
528 }
529 Expr::BinaryOp { left, op, right } => {
530 format!("{} {} {}", expr_name(left), op, expr_name(right))
531 }
532 Expr::Nested(inner) => expr_name(inner),
533 _ => "expr".to_string(),
534 }
535}
536
537fn aggregate(
538 rows: Vec<Row>,
539 projection: &[SelectItem],
540 group_by: &[Expr],
541 having: &Option<Expr>,
542) -> Result<Vec<Row>, Box<dyn Error>> {
543 let mut groups: HashMap<Vec<String>, Vec<Row>> = HashMap::new();
544 if group_by.is_empty() {
545 groups.insert(Vec::new(), rows);
546 } else {
547 for row in rows {
548 let mut key = Vec::new();
549 for g in group_by {
550 let v = eval(g.clone(), &row)?;
551 key.push(format!("{v:?}"));
552 }
553 groups.entry(key).or_default().push(row);
554 }
555 }
556
557 let mut out = Vec::new();
558 for group in groups.into_values() {
559 let mut new_row = Vec::new();
560 for item in projection {
561 match item {
562 SelectItem::UnnamedExpr(expr) => {
563 let name = expr_name(expr);
564 new_row.push((name, eval_aggregated(expr, &group)?));
565 }
566 SelectItem::ExprWithAlias { expr, alias } => {
567 new_row.push((alias.value.clone(), eval_aggregated(expr, &group)?));
568 }
569 _ => return Err("unsupported aggregate SELECT item".into()),
570 }
571 }
572 if let Some(h) = having {
573 if !matches!(eval(h.clone(), &new_row), Ok(CanonicalValue::Bool(true))) {
574 continue;
575 }
576 }
577 out.push(new_row);
578 }
579 Ok(out)
580}
581
582fn eval_aggregated(expr: &Expr, group: &[Row]) -> Result<CanonicalValue, Box<dyn Error>> {
583 if let Expr::Function(f) = expr {
584 let name = f.name.to_string().to_uppercase();
585 if matches!(name.as_str(), "SUM" | "AVG" | "MIN" | "MAX" | "COUNT") {
586 let arg = match &f.args {
587 FunctionArguments::List(l) => l.args.first().and_then(|a| match a {
588 FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => Some(e.clone()),
589 _ => None,
590 }),
591 _ => None,
592 };
593 return compute_aggregate(&name, arg, group);
594 }
595 }
596 if let Some(first) = group.first() {
597 return eval(expr.clone(), first);
598 }
599 Ok(CanonicalValue::Null)
600}
601
602fn compute_aggregate(
603 name: &str,
604 arg: Option<Expr>,
605 group: &[Row],
606) -> Result<CanonicalValue, Box<dyn Error>> {
607 let arg = match arg {
608 Some(a) => a,
609 None => {
610 return match name {
611 "COUNT" => Ok(CanonicalValue::Number(Number::from(group.len() as i64))),
612 _ => Err(format!("{name} requires an argument").into()),
613 };
614 }
615 };
616
617 let mut count = 0usize;
618 let mut sum = 0.0f64;
619 let mut best: Option<f64> = None;
620
621 for row in group {
622 if let Ok(v) = eval(arg.clone(), row) {
623 if let CanonicalValue::Null = v {
624 continue;
625 }
626 if let Some(f) = to_f64(&v) {
627 sum += f;
628 count += 1;
629 best = Some(match best {
630 None => f,
631 Some(b) => match name {
632 "MIN" => b.min(f),
633 "MAX" => b.max(f),
634 _ => b,
635 },
636 });
637 }
638 }
639 }
640
641 Ok(match name {
642 "SUM" => num(sum),
643 "AVG" => {
644 if count > 0 {
645 num(sum / count as f64)
646 } else {
647 CanonicalValue::Null
648 }
649 }
650 "COUNT" => CanonicalValue::Number(Number::from(count as i64)),
651 "MIN" | "MAX" => best.map(num).unwrap_or(CanonicalValue::Null),
652 _ => CanonicalValue::Null,
653 })
654}
655
656fn extract_limit_expr(lc: &LimitClause) -> Option<Expr> {
657 match lc {
658 LimitClause::LimitOffset { limit, .. } => limit.clone(),
659 LimitClause::OffsetCommaLimit { limit, .. } => Some(limit.clone()),
660 }
661}
662
663fn sort_rows(rows: &mut [Row], order_by: &[OrderByExpr]) -> Result<(), Box<dyn Error>> {
664 let keys: Vec<(Expr, bool)> = order_by
665 .iter()
666 .map(|o| (o.expr.clone(), o.options.asc.unwrap_or(true)))
667 .collect();
668 rows.sort_by(|a, b| {
669 for (expr, asc) in &keys {
670 let av = eval(expr.clone(), a).unwrap_or(CanonicalValue::Null);
671 let bv = eval(expr.clone(), b).unwrap_or(CanonicalValue::Null);
672 let ord = compare_values(&av, &bv);
673 let ord = if *asc { ord } else { ord.reverse() };
674 if ord != Ordering::Equal {
675 return ord;
676 }
677 }
678 Ordering::Equal
679 });
680 Ok(())
681}
682
683fn compare_values(a: &CanonicalValue, b: &CanonicalValue) -> Ordering {
684 match (a, b) {
685 (CanonicalValue::Null, CanonicalValue::Null) => Ordering::Equal,
686 (CanonicalValue::Null, _) => Ordering::Less,
687 (_, CanonicalValue::Null) => Ordering::Greater,
688 _ => {
689 if let (Some(x), Some(y)) = (to_f64(a), to_f64(b)) {
690 return x.partial_cmp(&y).unwrap_or(Ordering::Equal);
691 }
692 scalar_string(a).cmp(&scalar_string(b))
693 }
694 }
695}
696
697fn scalar_string(v: &CanonicalValue) -> String {
698 match v {
699 CanonicalValue::Null => String::new(),
700 CanonicalValue::Bool(b) => b.to_string(),
701 CanonicalValue::Number(n) => n.to_string(),
702 CanonicalValue::String(s) => s.clone(),
703 CanonicalValue::Bytes(b) => b.iter().map(|x| format!("{x:02x}")).collect(),
704 _ => String::new(),
705 }
706}
707
708fn contains_aggregate(expr: &Expr) -> bool {
709 match expr {
710 Expr::Function(f) => {
711 let name = f.name.to_string().to_uppercase();
712 matches!(name.as_str(), "SUM" | "AVG" | "MIN" | "MAX" | "COUNT")
713 }
714 Expr::BinaryOp { left, right, .. } => contains_aggregate(left) || contains_aggregate(right),
715 Expr::Nested(inner) => contains_aggregate(inner),
716 _ => false,
717 }
718}