use std::collections::HashMap;
use anyhow::{bail, Result};
use serde_json::{json, Value};
use crate::db::Db;
use crate::index::OrderedValue;
use crate::store::Node;
#[derive(Debug, Clone, PartialEq)]
enum Tok {
Kw(String, String),
Ident(String), Str(String), Num(f64), Op(String), Punct(char), Eof,
}
struct Lexer<'a> {
src: &'a str,
pos: usize,
}
impl<'a> Lexer<'a> {
fn new(src: &'a str) -> Self { Self { src, pos: 0 } }
fn peek_char(&self) -> Option<char> { self.src[self.pos..].chars().next() }
fn skip_ws(&mut self) {
while let Some(c) = self.peek_char() {
if c.is_whitespace() { self.pos += c.len_utf8(); } else { break; }
}
}
fn next_tok(&mut self) -> Tok {
self.skip_ws();
if self.pos >= self.src.len() { return Tok::Eof; }
let c = self.peek_char().unwrap();
if c == '"' {
self.pos += 1;
let mut s = String::new();
while let Some(ch) = self.peek_char() {
if ch == '"' {
break;
}
if ch == '\\' {
let next = self.src[self.pos + 1..].chars().next();
if next == Some('"') {
s.push('"');
self.pos += 1 + 1; continue;
}
}
s.push(ch);
self.pos += ch.len_utf8();
}
if self.peek_char() == Some('"') {
self.pos += 1;
}
return Tok::Str(s);
}
if self.pos + 1 < self.src.len() {
let two = &self.src[self.pos..self.pos+2];
if matches!(two, "!=" | ">=" | "<=") {
self.pos += 2;
return Tok::Op(two.to_string());
}
}
if matches!(c, '=' | '>' | '<') {
self.pos += 1;
return Tok::Op(c.to_string());
}
if matches!(c, '(' | ')' | ',') {
self.pos += 1;
return Tok::Punct(c);
}
if c.is_ascii_digit() || (c == '-' && self.src[self.pos+1..].starts_with(|d: char| d.is_ascii_digit())) {
let start = self.pos;
if c == '-' { self.pos += 1; }
while let Some(d) = self.peek_char() {
if d.is_ascii_digit() || d == '.' { self.pos += 1; } else { break; }
}
let n: f64 = self.src[start..self.pos].parse().unwrap_or(0.0);
return Tok::Num(n);
}
if c.is_alphabetic() || c == '_' {
let start = self.pos;
while let Some(ch) = self.peek_char() {
if ch.is_alphanumeric() || ch == '_' || ch == '.' || ch == ':' {
self.pos += ch.len_utf8();
} else { break; }
}
let word = &self.src[start..self.pos];
let upper = word.to_uppercase();
let keywords = ["FROM","AS","OF","VALID","WHERE","AND","OR","ORDER","BY",
"ASC","DESC","LIMIT","OFFSET","GROUP","HAVING",
"COUNT","SUM","AVG","MIN","MAX",
"TRACE","TRAVERSE","REVERSE","SEARCH","NOT","NULL","TRUE","FALSE",
"IN","BETWEEN","LIKE","ILIKE","IS"];
if keywords.contains(&upper.as_str()) {
return Tok::Kw(upper, word.to_string());
}
return Tok::Ident(word.to_string());
}
self.pos += c.len_utf8();
self.next_tok()
}
fn tokenize(&mut self) -> Vec<Tok> {
let mut toks = vec![];
loop {
let t = self.next_tok();
if t == Tok::Eof { break; }
toks.push(t);
}
toks
}
}
#[derive(Debug, Clone)]
pub enum Pred {
Cmp { field: String, op: String, value: Value },
In { field: String, values: Vec<Value>, negated: bool },
Between { field: String, low: Value, high: Value, negated: bool },
Like { field: String, pattern: String, negated: bool, ci: bool },
IsNull { field: String, negated: bool },
And(Vec<Pred>),
Or(Vec<Pred>),
Not(Box<Pred>),
}
#[derive(Debug, Clone, PartialEq)]
pub enum GroupAgg { Count, Sum, Avg, Min, Max }
impl GroupAgg {
fn name(&self) -> &'static str {
match self {
GroupAgg::Count => "count", GroupAgg::Sum => "sum",
GroupAgg::Avg => "avg", GroupAgg::Min => "min",
GroupAgg::Max => "max",
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct OrderKey {
pub field: String,
pub desc: bool,
}
#[derive(Debug, Clone)]
pub struct Aggregate {
pub group_field: Option<String>,
pub agg: GroupAgg,
pub agg_field: Option<String>,
}
#[derive(Debug, Clone)]
pub struct Query {
pub coll: String,
pub as_of: Option<u64>,
pub valid_as_of: Option<String>,
pub where_: Option<Pred>,
pub search: Option<String>,
pub order_by: Vec<OrderKey>,
pub limit: Option<usize>,
pub offset: Option<usize>,
pub aggregate: Option<Aggregate>,
pub having: Option<Pred>,
pub trace: Option<String>, pub trace_rev: bool,
pub traverse: Option<String>, }
struct Parser { toks: Vec<Tok>, pos: usize }
impl Parser {
fn new(toks: Vec<Tok>) -> Self { Self { toks, pos: 0 } }
fn peek(&self) -> &Tok { self.toks.get(self.pos).unwrap_or(&Tok::Eof) }
fn advance(&mut self) -> Tok { let t = self.peek().clone(); self.pos += 1; t }
fn expect_kw(&mut self, kw: &str) -> Result<()> {
match self.advance() {
Tok::Kw(k, _) if k == kw => Ok(()),
other => bail!("expected keyword {}, got {:?}", kw, other),
}
}
fn parse_value(&mut self) -> Result<Value> {
Ok(match self.advance() {
Tok::Str(s) => Value::String(s),
Tok::Num(n) => json!(n),
Tok::Kw(k, _) if k == "NULL" => Value::Null,
Tok::Kw(k, _) if k == "TRUE" => Value::Bool(true),
Tok::Kw(k, _) if k == "FALSE" => Value::Bool(false),
Tok::Ident(s) => Value::String(s),
other => bail!("expected a value (string, number, TRUE, FALSE or NULL), got {:?}", other),
})
}
fn peek_kw(&self, kw: &str) -> bool {
matches!(self.peek(), Tok::Kw(k, _) if k == kw)
}
fn eat_kw(&mut self, kw: &str) -> bool {
if self.peek_kw(kw) { self.advance(); true } else { false }
}
fn expect_punct(&mut self, c: char) -> Result<()> {
match self.advance() {
Tok::Punct(p) if p == c => Ok(()),
other => bail!("expected '{}', got {:?}", c, other),
}
}
fn parse_agg_kw(&mut self) -> Result<GroupAgg> {
Ok(match self.advance() {
Tok::Kw(a, _) if a == "COUNT" => GroupAgg::Count,
Tok::Kw(a, _) if a == "SUM" => GroupAgg::Sum,
Tok::Kw(a, _) if a == "AVG" => GroupAgg::Avg,
Tok::Kw(a, _) if a == "MIN" => GroupAgg::Min,
Tok::Kw(a, _) if a == "MAX" => GroupAgg::Max,
other => bail!("expected an aggregate (COUNT/SUM/AVG/MIN/MAX), got {:?}", other),
})
}
fn parse_field(&mut self, ctx: &str) -> Result<String> {
match self.advance() {
Tok::Ident(s) | Tok::Kw(_, s) => Ok(s),
other => bail!("{}: expected field name, got {:?}", ctx, other),
}
}
fn parse_pred(&mut self) -> Result<Pred> { self.parse_or() }
fn parse_or(&mut self) -> Result<Pred> {
let mut terms = vec![self.parse_and()?];
while self.eat_kw("OR") {
terms.push(self.parse_and()?);
}
Ok(if terms.len() == 1 { terms.pop().unwrap() } else { Pred::Or(terms) })
}
fn parse_and(&mut self) -> Result<Pred> {
let mut terms = vec![self.parse_not()?];
while self.peek_kw("AND") {
self.advance();
terms.push(self.parse_not()?);
}
Ok(if terms.len() == 1 { terms.pop().unwrap() } else { Pred::And(terms) })
}
fn parse_not(&mut self) -> Result<Pred> {
if self.eat_kw("NOT") {
return Ok(Pred::Not(Box::new(self.parse_not()?)));
}
self.parse_primary()
}
fn parse_primary(&mut self) -> Result<Pred> {
if matches!(self.peek(), Tok::Punct('(')) {
self.advance();
let inner = self.parse_pred()?;
self.expect_punct(')')?;
return Ok(inner);
}
self.parse_comparison()
}
fn parse_comparison(&mut self) -> Result<Pred> {
let field = self.parse_field("WHERE")?;
if self.eat_kw("IS") {
let negated = self.eat_kw("NOT");
if !self.eat_kw("NULL") {
bail!("WHERE: expected NULL after IS{}", if negated { " NOT" } else { "" });
}
return Ok(Pred::IsNull { field, negated });
}
let negated = self.eat_kw("NOT");
if self.eat_kw("IN") {
self.expect_punct('(')?;
let mut values = vec![];
loop {
values.push(self.parse_value()?);
if matches!(self.peek(), Tok::Punct(',')) { self.advance(); continue; }
break;
}
self.expect_punct(')')?;
if values.is_empty() {
bail!("WHERE: IN () needs at least one value");
}
return Ok(Pred::In { field, values, negated });
}
if self.eat_kw("BETWEEN") {
let low = self.parse_value()?;
if !self.eat_kw("AND") {
bail!("WHERE: BETWEEN expects AND between its bounds");
}
let high = self.parse_value()?;
return Ok(Pred::Between { field, low, high, negated });
}
let ci = self.peek_kw("ILIKE");
if ci || self.peek_kw("LIKE") {
self.advance();
let pattern = match self.advance() {
Tok::Str(s) => s,
Tok::Ident(s) => s,
other => bail!("WHERE: LIKE expects a pattern string, got {:?}", other),
};
return Ok(Pred::Like { field, pattern, negated, ci });
}
if negated {
bail!("WHERE: NOT must be followed by IN, BETWEEN, LIKE or ILIKE \
(use `NOT (field = value)` or `field != value` to negate a comparison)");
}
let op = match self.advance() {
Tok::Op(s) => s,
other => bail!("WHERE: expected operator, got {:?}", other),
};
let value = self.parse_value()?;
Ok(Pred::Cmp { field, op, value })
}
fn parse(&mut self) -> Result<Query> {
self.expect_kw("FROM")?;
let coll = match self.advance() {
Tok::Ident(s) | Tok::Kw(_, s) => s,
other => bail!("expected collection name, got {:?}", other),
};
let mut q = Query {
coll, as_of: None, valid_as_of: None,
where_: None, search: None,
order_by: vec![],
limit: None, offset: None,
aggregate: None, having: None,
trace: None, trace_rev: false,
traverse: None,
};
loop {
match self.peek() {
Tok::Eof => break,
Tok::Kw(k, _) if k == "AS" => {
self.advance();
self.expect_kw("OF")?;
match self.advance() {
Tok::Num(n) => q.as_of = Some(n as u64),
other => bail!("AS OF expects sequence number, got {:?}", other),
}
}
Tok::Kw(k, _) if k == "VALID" => {
self.advance();
self.expect_kw("AS")?;
self.expect_kw("OF")?;
match self.advance() {
Tok::Str(s) => q.valid_as_of = Some(s),
other => bail!("VALID AS OF expects date string, got {:?}", other),
}
}
Tok::Kw(k, _) if k == "WHERE" => {
self.advance();
let pred = self.parse_pred()?;
q.where_ = Some(match q.where_.take() {
None => pred,
Some(prev) => Pred::And(vec![prev, pred]),
});
}
Tok::Kw(k, _) if k == "SEARCH" => {
self.advance();
match self.advance() {
Tok::Str(s) => q.search = Some(s),
other => bail!("SEARCH expects quoted string, got {:?}", other),
}
}
Tok::Kw(k, _) if k == "ORDER" => {
self.advance();
self.expect_kw("BY")?;
loop {
let field = self.parse_field("ORDER BY")?;
let desc = if self.eat_kw("DESC") {
true
} else {
self.eat_kw("ASC");
false
};
q.order_by.push(OrderKey { field, desc });
if matches!(self.peek(), Tok::Punct(',')) { self.advance(); continue; }
break;
}
}
Tok::Kw(k, _) if k == "LIMIT" => {
self.advance();
match self.advance() {
Tok::Num(n) if n >= 0.0 => q.limit = Some(n as usize),
other => bail!("LIMIT expects a non-negative number, got {:?}", other),
}
}
Tok::Kw(k, _) if k == "OFFSET" => {
self.advance();
match self.advance() {
Tok::Num(n) if n >= 0.0 => q.offset = Some(n as usize),
other => bail!("OFFSET expects a non-negative number, got {:?}", other),
}
}
Tok::Kw(k, _) if k == "HAVING" => {
self.advance();
let pred = self.parse_pred()?;
q.having = Some(match q.having.take() {
None => pred,
Some(prev) => Pred::And(vec![prev, pred]),
});
}
Tok::Kw(k, _) if matches!(k.as_str(), "COUNT" | "SUM" | "AVG" | "MIN" | "MAX") => {
let agg = self.parse_agg_kw()?;
let agg_field = match agg {
GroupAgg::Count => None,
_ => Some(self.parse_field("aggregate")?),
};
if q.aggregate.is_some() {
bail!("only one aggregate per query");
}
q.aggregate = Some(Aggregate { group_field: None, agg, agg_field });
}
Tok::Kw(k, _) if k == "GROUP" => {
self.advance();
self.expect_kw("BY")?;
let field = match self.advance() {
Tok::Ident(s) | Tok::Kw(_, s) => s,
other => bail!("GROUP BY: expected field, got {:?}", other),
};
let agg = if matches!(self.peek(),
Tok::Kw(a, _) if matches!(a.as_str(), "COUNT"|"SUM"|"AVG"|"MIN"|"MAX"))
{
self.parse_agg_kw()?
} else {
GroupAgg::Count
};
let agg_field = match agg {
GroupAgg::Count => None,
_ => Some(self.parse_field("GROUP BY aggregate")?),
};
if q.aggregate.is_some() {
bail!("only one aggregate per query");
}
q.aggregate = Some(Aggregate {
group_field: Some(field), agg, agg_field,
});
}
Tok::Kw(k, _) if k == "TRACE" => {
self.advance();
let edge = match self.advance() {
Tok::Ident(s) | Tok::Kw(_, s) => s,
other => bail!("TRACE: expected edge type, got {:?}", other),
};
q.trace = Some(edge);
if let Tok::Kw(k, _) = self.peek() {
if k == "REVERSE" { self.advance(); q.trace_rev = true; }
}
}
Tok::Kw(k, _) if k == "TRAVERSE" => {
self.advance();
let rel = match self.advance() {
Tok::Ident(s) | Tok::Kw(_, s) => s,
other => bail!("TRAVERSE: expected relation name, got {:?}", other),
};
q.traverse = Some(rel);
}
other => bail!(
"unexpected {:?} in query. Expected one of: AS OF, VALID AS OF, \
WHERE, SEARCH, ORDER BY, LIMIT, OFFSET, GROUP BY, HAVING, \
COUNT, SUM, AVG, MIN, MAX, TRACE, TRAVERSE",
other
),
}
}
Ok(q)
}
}
fn field_value(node: &Node, field: &str) -> Value {
match field {
"_id" => Value::String(node.id.clone()),
"_coll" => Value::String(node.coll.clone()),
"_hash" => Value::String(node.hash.clone()),
"_seq" => json!(node.seq),
_ => node.data.get(field).cloned().unwrap_or(Value::Null),
}
}
fn cmp_op(a: &Value, op: &str, b: &Value) -> bool {
if matches!(op, "<" | "<=" | ">" | ">=") && a.is_null() {
return false;
}
let a = OrderedValue::from(a);
let b = OrderedValue::from(b);
match op {
"=" => a == b,
"!=" => a != b,
">" => a > b,
"<" => a < b,
">=" => a >= b,
"<=" => a <= b,
_ => false,
}
}
fn as_text(v: &Value) -> String {
match v {
Value::String(s) => s.clone(),
Value::Null => String::new(),
other => other.to_string(),
}
}
fn like_match(value: &str, pattern: &str, ci: bool) -> bool {
let (v, p): (Vec<char>, Vec<char>) = if ci {
(value.to_lowercase().chars().collect(), pattern.to_lowercase().chars().collect())
} else {
(value.chars().collect(), pattern.chars().collect())
};
let mut vi = 0usize;
let mut pi = 0usize;
let mut star: Option<(usize, usize)> = None;
while vi < v.len() {
if pi < p.len() && (p[pi] == '_' || p[pi] == v[vi]) {
vi += 1;
pi += 1;
} else if pi < p.len() && p[pi] == '%' {
star = Some((pi, vi));
pi += 1;
} else if let Some((sp, sv)) = star {
pi = sp + 1;
vi = sv + 1;
star = Some((sp, vi));
} else {
return false;
}
}
while pi < p.len() && p[pi] == '%' { pi += 1; }
pi == p.len()
}
fn eval_pred_with(get: &dyn Fn(&str) -> Value, pred: &Pred) -> bool {
match pred {
Pred::Cmp { field, op, value } => cmp_op(&get(field), op, value),
Pred::In { field, values, negated } => {
let fv = get(field);
let hit = values.iter().any(|v| cmp_op(&fv, "=", v));
hit != *negated
}
Pred::Between { field, low, high, negated } => {
let fv = get(field);
let hit = cmp_op(&fv, ">=", low) && cmp_op(&fv, "<=", high);
hit != *negated
}
Pred::Like { field, pattern, negated, ci } => {
let fv = get(field);
if fv.is_null() { return false; }
let hit = like_match(&as_text(&fv), pattern, *ci);
hit != *negated
}
Pred::IsNull { field, negated } => {
get(field).is_null() != *negated
}
Pred::And(terms) => terms.iter().all(|t| eval_pred_with(get, t)),
Pred::Or(terms) => terms.iter().any(|t| eval_pred_with(get, t)),
Pred::Not(inner) => !eval_pred_with(get, inner),
}
}
fn eval_pred(node: &Node, pred: &Pred) -> bool {
eval_pred_with(&|f| field_value(node, f), pred)
}
fn eval_pred_json(obj: &Value, pred: &Pred) -> bool {
eval_pred_with(&|f| obj.get(f).cloned().unwrap_or(Value::Null), pred)
}
fn sort_by_keys<T>(rows: &mut [T], keys: &[OrderKey], get: impl Fn(&T, &str) -> Value) {
rows.sort_by(|a, b| {
for k in keys {
let av = OrderedValue::from(&get(a, &k.field));
let bv = OrderedValue::from(&get(b, &k.field));
let ord = if k.desc { bv.cmp(&av) } else { av.cmp(&bv) };
if ord != std::cmp::Ordering::Equal {
return ord;
}
}
std::cmp::Ordering::Equal
});
}
fn paginate<T>(rows: Vec<T>, offset: Option<usize>, limit: Option<usize>) -> Vec<T> {
let mut it = rows;
if let Some(off) = offset {
if off >= it.len() {
return vec![];
}
it.drain(..off);
}
if let Some(n) = limit {
it.truncate(n);
}
it
}
fn aggregate_rows(rows: &[Node], spec: &Aggregate) -> Vec<Value> {
struct Group { count: usize, nums: Vec<f64>, ints: Vec<i64>, all_int: bool }
let mut order: Vec<String> = vec![];
let mut groups: HashMap<String, Group> = HashMap::new();
const WHOLE: &str = "";
for node in rows {
let key = match spec.group_field {
None => WHOLE.to_string(),
Some(ref gf) => node.data.get(gf)
.map(as_text)
.unwrap_or_else(|| "null".to_string()),
};
let entry = groups.entry(key.clone()).or_insert_with(|| {
order.push(key.clone());
Group { count: 0, nums: vec![], ints: vec![], all_int: true }
});
entry.count += 1;
if let Some(ref af) = spec.agg_field {
match node.data.get(af) {
Some(Value::Number(n)) => {
if let Some(i) = n.as_i64() {
entry.ints.push(i);
entry.nums.push(i as f64);
} else if let Some(f) = n.as_f64() {
entry.all_int = false;
entry.nums.push(f);
}
}
_ => {}
}
}
}
if spec.group_field.is_none() && order.is_empty() {
order.push(WHOLE.to_string());
groups.insert(WHOLE.to_string(),
Group { count: 0, nums: vec![], ints: vec![], all_int: true });
}
order.into_iter().map(|k| {
let g = &groups[&k];
let mut obj = serde_json::Map::new();
if let Some(ref gf) = spec.group_field {
obj.insert(gf.clone(), Value::String(k.clone()));
}
obj.insert("count".to_string(), json!(g.count));
let int_path = g.all_int && !g.ints.is_empty();
let agg_val: Value = match spec.agg {
GroupAgg::Count => json!(g.count),
_ if g.nums.is_empty() => Value::Null,
GroupAgg::Sum if int_path => {
match g.ints.iter().try_fold(0i64, |a, &b| a.checked_add(b)) {
Some(t) => json!(t),
None => json!(g.nums.iter().sum::<f64>()),
}
}
GroupAgg::Min if int_path => json!(g.ints.iter().min().copied().unwrap()),
GroupAgg::Max if int_path => json!(g.ints.iter().max().copied().unwrap()),
GroupAgg::Sum => json!(g.nums.iter().sum::<f64>()),
GroupAgg::Avg => json!(g.nums.iter().sum::<f64>() / g.nums.len() as f64),
GroupAgg::Min => json!(g.nums.iter().cloned().fold(f64::INFINITY, f64::min)),
GroupAgg::Max => json!(g.nums.iter().cloned().fold(f64::NEG_INFINITY, f64::max)),
};
if let Some(ref af) = spec.agg_field {
obj.insert(format!("{}_{}", spec.agg.name(), af), agg_val.clone());
}
obj.insert("value".to_string(), agg_val);
Value::Object(obj)
}).collect()
}
#[derive(Debug, Clone)]
enum IndexPlan {
Range {
field: String,
low: Option<Value>,
high: Option<Value>,
low_incl: bool,
high_incl: bool,
},
Values { field: String, values: Vec<Value> },
}
impl IndexPlan {
fn field(&self) -> &str {
match self {
IndexPlan::Range { field, .. } => field,
IndexPlan::Values { field, .. } => field,
}
}
}
fn collect_index_constraints(pred: &Pred, out: &mut Vec<IndexPlan>) {
match pred {
Pred::And(terms) => {
for t in terms {
collect_index_constraints(t, out);
}
}
Pred::Cmp { field, op, value } => {
if field == "_id" {
return;
}
match op.as_str() {
"=" => out.push(IndexPlan::Values {
field: field.clone(),
values: vec![value.clone()],
}),
">" | ">=" => out.push(IndexPlan::Range {
field: field.clone(),
low: Some(value.clone()),
high: None,
low_incl: op == ">=",
high_incl: true,
}),
"<" | "<=" => out.push(IndexPlan::Range {
field: field.clone(),
low: None,
high: Some(value.clone()),
low_incl: true,
high_incl: op == "<=",
}),
_ => {}
}
}
Pred::Between { field, low, high, negated: false } => {
out.push(IndexPlan::Range {
field: field.clone(),
low: Some(low.clone()),
high: Some(high.clone()),
low_incl: true, high_incl: true,
});
}
Pred::In { field, values, negated: false } => {
out.push(IndexPlan::Values {
field: field.clone(),
values: values.clone(),
});
}
_ => {}
}
}
fn choose_index_plan(db: &Db, coll: &str, pred: &Pred) -> Option<IndexPlan> {
let mut raw = vec![];
collect_index_constraints(pred, &mut raw);
raw.retain(|p| db.has_sorted_index(coll, p.field()));
if raw.is_empty() {
return None;
}
let mut merged: Vec<IndexPlan> = vec![];
for plan in raw {
let field = plan.field().to_string();
let existing = merged.iter().position(|m| m.field() == field);
match (existing, plan) {
(None, p) => merged.push(p),
(Some(i), IndexPlan::Range { low, high, low_incl, high_incl, .. }) => {
if let IndexPlan::Range {
low: ref mut elow, high: ref mut ehigh,
low_incl: ref mut eli, high_incl: ref mut ehi, ..
} = merged[i] {
if let Some(l) = low {
let tighter = match elow {
None => true,
Some(cur) => OrderedValue::from(&l) > OrderedValue::from(&*cur),
};
if tighter { *elow = Some(l); *eli = low_incl; }
}
if let Some(h) = high {
let tighter = match ehigh {
None => true,
Some(cur) => OrderedValue::from(&h) < OrderedValue::from(&*cur),
};
if tighter { *ehigh = Some(h); *ehi = high_incl; }
}
}
}
(Some(i), p @ IndexPlan::Values { .. }) => {
if matches!(merged[i], IndexPlan::Range { .. }) {
merged[i] = p;
}
}
}
}
let mut best: Option<(usize, IndexPlan)> = None;
for plan in merged {
let card = match &plan {
IndexPlan::Range { field, low, high, low_incl, high_incl } => db
.range_cardinality(coll, field, low.as_ref(), high.as_ref(),
*low_incl, *high_incl)
.unwrap_or(usize::MAX),
IndexPlan::Values { field, values } => values
.iter()
.map(|v| db.range_cardinality(coll, field, Some(v), Some(v), true, true)
.unwrap_or(usize::MAX))
.fold(0usize, |a, b| a.saturating_add(b)),
};
if best.as_ref().map(|(c, _)| card < *c).unwrap_or(true) {
best = Some((card, plan));
}
}
best.map(|(_, p)| p)
}
fn id_point_lookup(pred: &Pred) -> Option<String> {
match pred {
Pred::Cmp { field, op, value } if field == "_id" && op == "=" => {
if let Value::String(s) = value { Some(s.clone()) } else { None }
}
Pred::And(terms) => terms.iter().find_map(id_point_lookup),
_ => None,
}
}
fn matches_valid_as_of(node: &Node, date: &str) -> bool {
let from_ok = node.valid_from.as_deref().map(|f| f <= date).unwrap_or(true);
let to_ok = node.valid_to.as_deref().map(|t| t > date).unwrap_or(true);
from_ok && to_ok
}
fn node_contains_text(node: &Node, text: &str) -> bool {
let s = node.data.to_string().to_lowercase();
s.contains(&text.to_lowercase())
}
pub fn node_to_json(node: &Node) -> Value {
let mut obj = if let Value::Object(m) = &node.data {
m.clone()
} else {
serde_json::Map::new()
};
obj.insert("_id".to_string(), Value::String(node.id.clone()));
obj.insert("_hash".to_string(), Value::String(node.hash.clone()));
obj.insert("_seq".to_string(), json!(node.seq));
obj.insert("_coll".to_string(), Value::String(node.coll.clone()));
if let Some(ref vf) = node.valid_from {
obj.insert("_valid_from".to_string(), Value::String(vf.clone()));
}
if let Some(ref vt) = node.valid_to {
obj.insert("_valid_to".to_string(), Value::String(vt.clone()));
}
if !node.caused_by.is_empty() {
obj.insert("_caused_by".to_string(), Value::Array(
node.caused_by.iter().map(|h| Value::String(h.clone())).collect()
));
}
Value::Object(obj)
}
pub fn parse(nql: &str) -> Result<Query> {
let mut lexer = Lexer::new(nql);
let toks = lexer.tokenize();
let mut parser = Parser::new(toks);
parser.parse()
}
pub fn execute(db: &Db, nql: &str) -> Result<Vec<Value>> {
let q = parse(nql)?;
let id_eq_fast_path: Option<String> = if q.as_of.is_none() && q.trace.is_none() {
q.where_.as_ref().and_then(id_point_lookup)
} else { None };
let candidates: Vec<Node> = if let Some(ref target_id) = id_eq_fast_path {
db.get(&q.coll, target_id).into_iter().collect()
} else if let Some(seq_target) = q.as_of {
db.id_index.list_ids(&q.coll).into_iter()
.filter_map(|id| db.get_as_of(&q.coll, &id, seq_target))
.collect()
} else if let Some(plan) = q.where_.as_ref()
.filter(|_| q.as_of.is_none())
.and_then(|p| choose_index_plan(db, &q.coll, p))
{
let got = match &plan {
IndexPlan::Range { field, low, high, low_incl, high_incl } => db.range_scan(
&q.coll, field, low.as_ref(), high.as_ref(), *low_incl, *high_incl),
IndexPlan::Values { field, values } => db.index_lookup(&q.coll, field, values),
};
match got {
Some(nodes) => nodes,
None => db.list(&q.coll),
}
} else if q.order_by.len() == 1 && q.aggregate.is_none() {
let key = &q.order_by[0];
let has_post_filters = q.where_.is_some()
|| q.search.is_some()
|| q.valid_as_of.is_some();
let limit = if has_post_filters {
9_999_999
} else {
match q.limit {
Some(n) => n.saturating_add(q.offset.unwrap_or(0)),
None => 9_999_999,
}
};
if key.desc {
db.order_by_desc(&q.coll, &key.field, limit)
} else {
db.order_by_asc(&q.coll, &key.field, limit)
}
} else if let (Some(n), true) = (q.limit, q.where_.is_none()
&& q.search.is_none() && q.trace.is_none()
&& q.traverse.is_none() && q.aggregate.is_none()
&& q.order_by.is_empty() && q.offset.is_none()
&& q.valid_as_of.is_none()) {
db.id_index
.list_ids(&q.coll)
.into_iter()
.take(n)
.filter_map(|id| db.get(&q.coll, &id))
.collect()
} else {
db.list(&q.coll)
};
let mut rows: Vec<Node> = candidates.into_iter()
.filter(|n| q.where_.as_ref().map(|p| eval_pred(n, p)).unwrap_or(true))
.filter(|n| q.valid_as_of.as_deref()
.map(|d| matches_valid_as_of(n, d))
.unwrap_or(true))
.filter(|n| q.search.as_deref()
.map(|t| node_contains_text(n, t))
.unwrap_or(true))
.collect();
if let Some(ref _edge_type) = q.trace {
let limit = q.limit.unwrap_or(1000);
let mut traced: Vec<Node> = vec![];
for root in &rows {
let chain = db.trace(&root.hash, q.trace_rev, limit);
traced.extend(chain);
}
rows = traced;
}
if let Some(ref rel) = q.traverse {
let mut traversed: Vec<Node> = vec![];
for root in &rows {
let frm = format!("{}:{}", root.coll, root.id);
let neighbors = db.neighbors(&frm, rel);
traversed.extend(neighbors);
}
rows = traversed;
}
if let Some(ref spec) = q.aggregate {
let mut out = aggregate_rows(&rows, spec);
if let Some(ref pred) = q.having {
out.retain(|row| eval_pred_json(row, pred));
}
if !q.order_by.is_empty() {
sort_by_keys(&mut out, &q.order_by,
|row, f| row.get(f).cloned().unwrap_or(Value::Null));
} else if let Some(ref gf) = spec.group_field {
let gf = gf.clone();
out.sort_by(|a, b| {
as_text(&a.get(&gf).cloned().unwrap_or(Value::Null))
.cmp(&as_text(&b.get(&gf).cloned().unwrap_or(Value::Null)))
});
}
return Ok(paginate(out, q.offset, q.limit));
}
if q.having.is_some() {
bail!("HAVING requires an aggregate — add GROUP BY <field>, or use WHERE \
to filter individual rows");
}
if !q.order_by.is_empty() {
let index_path_held = q.order_by.len() == 1
&& q.as_of.is_none()
&& q.where_.is_none()
&& q.search.is_none()
&& q.valid_as_of.is_none()
&& q.trace.is_none()
&& q.traverse.is_none();
if !index_path_held {
sort_by_keys(&mut rows, &q.order_by,
|n, f| field_value(n, f));
}
}
let rows = paginate(rows, q.offset, q.limit);
Ok(rows.into_iter().map(|n| node_to_json(&n)).collect())
}
pub fn query(db: &Db, nql: &str) -> Result<(Vec<Value>, usize)> {
let rows = execute(db, nql)?;
let count = rows.len();
Ok((rows, count))
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
use crate::db::Db;
fn setup() -> (tempfile::TempDir, Db) {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("blocks", "height");
for h in 1u64..=5 {
db.put("blocks", &h.to_string(),
serde_json::json!({"height": h, "hash": format!("000{}", h), "n_tx": h * 2}),
vec![], None, None).unwrap();
}
(dir, db)
}
#[test]
fn from_all() {
let (_tmp, db) = setup();
let (rows, count) = query(&db, "FROM blocks").unwrap();
assert_eq!(count, 5);
let _ = rows;
}
#[test]
fn where_eq() {
let (_tmp, db) = setup();
let (rows, count) = query(&db, r#"FROM blocks WHERE _id = "3""#).unwrap();
assert_eq!(count, 1);
assert_eq!(rows[0]["_id"], "3");
}
#[test]
fn order_by_limit() {
let (_tmp, db) = setup();
let (rows, count) = query(&db, "FROM blocks ORDER BY height ASC LIMIT 3").unwrap();
assert_eq!(count, 3);
assert_eq!(rows[0]["height"], 1);
assert_eq!(rows[2]["height"], 3);
}
#[test]
fn order_by_desc() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks ORDER BY height DESC LIMIT 2").unwrap();
assert_eq!(rows[0]["height"], 5);
}
#[test]
fn where_gt() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height > 3").unwrap();
assert_eq!(rows.len(), 2);
}
#[test]
fn where_order_limit_does_not_truncate_before_filter() {
let (_tmp, db) = setup();
let (rows, count) =
query(&db, "FROM blocks WHERE n_tx >= 8 ORDER BY height LIMIT 2").unwrap();
assert_eq!(count, 2, "both matching rows must survive the limit");
let heights: Vec<u64> = rows.iter()
.filter_map(|r| r["height"].as_u64())
.collect();
assert_eq!(heights, vec![4, 5]);
let (rows_d, _) =
query(&db, "FROM blocks WHERE n_tx >= 8 ORDER BY height DESC LIMIT 1").unwrap();
assert_eq!(rows_d.len(), 1);
assert_eq!(rows_d[0]["height"], 5);
}
fn setup_text() -> (tempfile::TempDir, Db) {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
let rows = [
("1", serde_json::json!({"status": "open", "miner": "Acme Pool", "fee": 10})),
("2", serde_json::json!({"status": "pending", "miner": "acme solo", "fee": 20})),
("3", serde_json::json!({"status": "closed", "miner": "Zenith", "fee": 30})),
("4", serde_json::json!({"status": "open", "fee": 40})),
("5", serde_json::json!({"status": "voided", "miner": Value::Null, "fee": 50})),
];
for (id, data) in rows {
db.put("jobs", id, data, vec![], None, None).unwrap();
}
(dir, db)
}
fn ids(rows: &[Value]) -> Vec<String> {
let mut v: Vec<String> = rows.iter()
.filter_map(|r| r["_id"].as_str().map(String::from))
.collect();
v.sort();
v
}
#[test]
fn where_in_list() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height IN (2, 4)").unwrap();
assert_eq!(ids(&rows), vec!["2", "4"]);
}
#[test]
fn where_in_strings() {
let (_tmp, db) = setup_text();
let (rows, _) = query(&db, r#"FROM jobs WHERE status IN ("open", "closed")"#).unwrap();
assert_eq!(ids(&rows), vec!["1", "3", "4"]);
}
#[test]
fn where_not_in() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height NOT IN (1, 2, 3)").unwrap();
assert_eq!(ids(&rows), vec!["4", "5"]);
}
#[test]
fn where_in_single_value_equals_eq() {
let (_tmp, db) = setup();
let (a, _) = query(&db, "FROM blocks WHERE height IN (3)").unwrap();
let (b, _) = query(&db, "FROM blocks WHERE height = 3").unwrap();
assert_eq!(ids(&a), ids(&b));
}
#[test]
fn where_between_is_inclusive() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height BETWEEN 2 AND 4").unwrap();
assert_eq!(ids(&rows), vec!["2", "3", "4"]);
}
#[test]
fn where_not_between() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height NOT BETWEEN 2 AND 4").unwrap();
assert_eq!(ids(&rows), vec!["1", "5"]);
}
#[test]
fn between_and_does_not_swallow_the_conjunction() {
let (_tmp, db) = setup();
let (rows, _) = query(
&db, "FROM blocks WHERE height BETWEEN 2 AND 4 AND n_tx > 4").unwrap();
assert_eq!(ids(&rows), vec!["3", "4"]);
}
#[test]
fn where_like_prefix_suffix_and_infix() {
let (_tmp, db) = setup_text();
let (pre, _) = query(&db, r#"FROM jobs WHERE miner LIKE "Acme%""#).unwrap();
assert_eq!(ids(&pre), vec!["1"]);
let (suf, _) = query(&db, r#"FROM jobs WHERE miner LIKE "%Pool""#).unwrap();
assert_eq!(ids(&suf), vec!["1"]);
let (inf, _) = query(&db, r#"FROM jobs WHERE status LIKE "%pen%""#).unwrap();
assert_eq!(ids(&inf), vec!["1", "2", "4"]); }
#[test]
fn where_like_underscore_matches_exactly_one_char() {
let (_tmp, db) = setup_text();
let (rows, _) = query(&db, r#"FROM jobs WHERE status LIKE "open_""#).unwrap();
assert!(rows.is_empty(), "`open_` must not match the 4-char value `open`");
let (rows2, _) = query(&db, r#"FROM jobs WHERE status LIKE "ope_""#).unwrap();
assert_eq!(ids(&rows2), vec!["1", "4"]);
}
#[test]
fn where_ilike_is_case_insensitive_and_like_is_not() {
let (_tmp, db) = setup_text();
let (ci, _) = query(&db, r#"FROM jobs WHERE miner ILIKE "acme%""#).unwrap();
assert_eq!(ci.len(), 2, "ILIKE matches both `Acme Pool` and `acme solo`");
let (cs, _) = query(&db, r#"FROM jobs WHERE miner LIKE "acme%""#).unwrap();
assert_eq!(ids(&cs), vec!["2"], "LIKE stays case-sensitive");
}
#[test]
fn like_backtracks_across_multiple_wildcards() {
assert!(like_match("abcabcabd", "%abc%abd", false));
assert!(like_match("aaa", "%a", false));
assert!(like_match("", "%", false));
assert!(like_match("x", "%%%", false));
assert!(!like_match("abc", "%abd", false));
assert!(!like_match("ab", "ab_", false));
assert!(like_match("héllo wörld", "h_llo w%d", false));
}
#[test]
fn where_not_like() {
let (_tmp, db) = setup_text();
let (rows, _) = query(&db, r#"FROM jobs WHERE status NOT LIKE "open""#).unwrap();
assert_eq!(ids(&rows), vec!["2", "3", "5"]);
}
#[test]
fn like_over_null_is_false_in_both_polarities() {
let (_tmp, db) = setup_text();
let (pos, _) = query(&db, r#"FROM jobs WHERE miner LIKE "%""#).unwrap();
let (neg, _) = query(&db, r#"FROM jobs WHERE miner NOT LIKE "%""#).unwrap();
assert!(!ids(&pos).contains(&"4".to_string()));
assert!(!ids(&neg).contains(&"4".to_string()));
assert!(!ids(&pos).contains(&"5".to_string()));
assert!(!ids(&neg).contains(&"5".to_string()));
}
#[test]
fn where_is_null_catches_absent_and_explicit_null() {
let (_tmp, db) = setup_text();
let (rows, _) = query(&db, "FROM jobs WHERE miner IS NULL").unwrap();
assert_eq!(ids(&rows), vec!["4", "5"]);
}
#[test]
fn where_is_not_null() {
let (_tmp, db) = setup_text();
let (rows, _) = query(&db, "FROM jobs WHERE miner IS NOT NULL").unwrap();
assert_eq!(ids(&rows), vec!["1", "2", "3"]);
}
#[test]
fn where_or() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height = 1 OR height = 5").unwrap();
assert_eq!(ids(&rows), vec!["1", "5"]);
}
#[test]
fn and_binds_tighter_than_or() {
let (_tmp, db) = setup();
let (rows, _) = query(
&db, "FROM blocks WHERE height = 1 OR height = 5 AND n_tx = 10").unwrap();
assert_eq!(ids(&rows), vec!["1", "5"]);
let (rows2, _) = query(
&db, "FROM blocks WHERE height = 1 OR height = 5 AND n_tx = 99").unwrap();
assert_eq!(ids(&rows2), vec!["1"], "the AND arm must not match");
}
#[test]
fn parens_override_precedence() {
let (_tmp, db) = setup();
let (rows, _) = query(
&db, "FROM blocks WHERE (height = 1 OR height = 5) AND n_tx = 10").unwrap();
assert_eq!(ids(&rows), vec!["5"]);
}
#[test]
fn nested_parens() {
let (_tmp, db) = setup();
let (rows, _) = query(
&db,
"FROM blocks WHERE ((height >= 2 AND height <= 4) OR height = 1) AND n_tx != 6",
).unwrap();
assert_eq!(ids(&rows), vec!["1", "2", "4"]);
}
#[test]
fn not_negates_a_group() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE NOT (height > 2)").unwrap();
assert_eq!(ids(&rows), vec!["1", "2"]);
}
#[test]
fn prefix_not_before_a_comparison() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE NOT height = 1").unwrap();
assert_eq!(ids(&rows), vec!["2", "3", "4", "5"]);
let (double, _) = query(&db, "FROM blocks WHERE NOT NOT height = 1").unwrap();
assert_eq!(ids(&double), vec!["1"]);
let (mixed, _) = query(&db, "FROM blocks WHERE NOT height = 1 AND height < 4").unwrap();
assert_eq!(ids(&mixed), vec!["2", "3"]);
}
#[test]
fn id_equality_under_or_does_not_become_a_point_lookup() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, r#"FROM blocks WHERE _id = "1" OR height > 3"#).unwrap();
assert_eq!(ids(&rows), vec!["1", "4", "5"],
"the OR arm must survive the id fast path");
}
#[test]
fn id_equality_under_and_still_point_looks_up() {
let (_tmp, db) = setup();
let (hit, _) = query(&db, r#"FROM blocks WHERE _id = "3" AND n_tx = 6"#).unwrap();
assert_eq!(ids(&hit), vec!["3"]);
let (miss, _) = query(&db, r#"FROM blocks WHERE _id = "3" AND n_tx = 999"#).unwrap();
assert!(miss.is_empty(), "the second conjunct must still be applied");
}
#[test]
fn metadata_fields_are_filterable() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE _seq >= 0 AND _coll = blocks").unwrap();
assert_eq!(rows.len(), 5);
let (tail, _) = query(&db, "FROM blocks WHERE _seq > 0").unwrap();
assert_eq!(tail.len(), 4);
}
#[test]
fn combined_with_order_and_limit() {
let (_tmp, db) = setup();
let (rows, _) = query(
&db,
"FROM blocks WHERE height IN (1, 3, 5) ORDER BY height DESC LIMIT 2",
).unwrap();
let heights: Vec<u64> = rows.iter().filter_map(|r| r["height"].as_u64()).collect();
assert_eq!(heights, vec![5, 3]);
}
#[test]
fn unknown_clauses_are_errors_not_silent_skips() {
let (_tmp, db) = setup();
for bad in [
"FROM blocks ORDRE BY height", "FROM blocks WHERE height > 3 JUNK", "FROM blocks SELECT height", "FROM blocks LIMIT", "FROM blocks OFFSET", "FROM blocks ORDER BY", "FROM blocks ORDER BY height,", ] {
assert!(query(&db, bad).is_err(), "`{}` must be rejected, not silently reinterpreted", bad);
}
}
#[test]
fn malformed_predicates_are_errors() {
let (_tmp, db) = setup();
for bad in [
"FROM blocks WHERE height >", "FROM blocks WHERE height IN (", "FROM blocks WHERE height IN ()", "FROM blocks WHERE height BETWEEN 1", "FROM blocks WHERE height BETWEEN 1 3", "FROM blocks WHERE (height = 1", "FROM blocks WHERE height IS 3", "FROM blocks WHERE height NOT = 1", "FROM blocks WHERE height LIKE", ] {
assert!(query(&db, bad).is_err(), "`{}` must be a parse error", bad);
}
}
#[test]
fn asc_is_accepted_explicitly() {
let (_tmp, db) = setup();
let (asc, _) = query(&db, "FROM blocks ORDER BY height ASC").unwrap();
let (plain, _) = query(&db, "FROM blocks ORDER BY height").unwrap();
assert_eq!(asc[0]["height"], 1);
assert_eq!(plain[0]["height"], 1);
}
#[test]
fn new_keywords_are_case_insensitive() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "from blocks where height between 2 and 3").unwrap();
assert_eq!(ids(&rows), vec!["2", "3"]);
let (rows2, _) = query(&db, "FROM blocks Where height In (1) Or height In (2)").unwrap();
assert_eq!(ids(&rows2), vec!["1", "2"]);
}
#[test]
fn group_by_count() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks GROUP BY n_tx COUNT").unwrap();
assert_eq!(rows.len(), 5); }
fn twin(fields: &[&str]) -> (tempfile::TempDir, tempfile::TempDir, Db, Db) {
let d1 = tempdir().unwrap();
let d2 = tempdir().unwrap();
let indexed = Db::open(d1.path(), None).unwrap();
let plain = Db::open(d2.path(), None).unwrap();
for f in fields {
indexed.create_sorted_index("t", f);
}
let rows: Vec<(String, Value)> = (0..40u64).map(|i| {
let mut o = serde_json::Map::new();
if i % 7 != 0 {
o.insert("fee".into(), json!(i % 13));
}
if i % 11 == 0 {
o.insert("note".into(), Value::Null);
} else {
o.insert("note".into(), json!(format!("n{}", i % 5)));
}
o.insert("rank".into(), json!(40 - i));
(i.to_string(), Value::Object(o))
}).collect();
for (id, doc) in &rows {
indexed.put("t", id, doc.clone(), vec![], None, None).unwrap();
plain.put("t", id, doc.clone(), vec![], None, None).unwrap();
}
(d1, d2, indexed, plain)
}
fn same(a: &Db, b: &Db, nql: &str) -> (Vec<String>, Vec<String>) {
let ga = {
let (rows, _) = query(a, nql).unwrap();
let mut v: Vec<String> = rows.iter()
.filter_map(|r| r["_id"].as_str().map(String::from)).collect();
v.sort(); v
};
let gb = {
let (rows, _) = query(b, nql).unwrap();
let mut v: Vec<String> = rows.iter()
.filter_map(|r| r["_id"].as_str().map(String::from)).collect();
v.sort(); v
};
(ga, gb)
}
#[test]
fn indexed_and_unindexed_agree_on_every_predicate_shape() {
let (_t1, _t2, idx, plain) = twin(&["fee", "note", "rank"]);
for nql in [
"FROM t WHERE fee > 5",
"FROM t WHERE fee >= 5",
"FROM t WHERE fee < 5",
"FROM t WHERE fee <= 5",
"FROM t WHERE fee = 5",
"FROM t WHERE fee BETWEEN 3 AND 8",
"FROM t WHERE fee NOT BETWEEN 3 AND 8",
"FROM t WHERE fee IN (1, 5, 9)",
"FROM t WHERE fee NOT IN (1, 5, 9)",
"FROM t WHERE fee != 5",
"FROM t WHERE fee > 3 AND fee < 9",
"FROM t WHERE fee >= 3 AND fee <= 9",
"FROM t WHERE fee > 3 AND fee < 9 AND fee != 5",
"FROM t WHERE fee BETWEEN 2 AND 10 AND fee > 6",
"FROM t WHERE fee > 5 AND rank < 20",
"FROM t WHERE fee IN (2, 3) AND rank > 10",
"FROM t WHERE fee = 4 AND rank = 8",
"FROM t WHERE fee IS NULL",
"FROM t WHERE fee IS NOT NULL",
"FROM t WHERE note IS NULL",
"FROM t WHERE note IS NOT NULL",
"FROM t WHERE fee IS NULL AND rank > 20",
r#"FROM t WHERE note LIKE "n_""#,
r#"FROM t WHERE fee > 5 AND note LIKE "n1""#,
r#"FROM t WHERE note NOT LIKE "n1" AND fee < 4"#,
"FROM t WHERE fee > 11 OR rank > 38",
"FROM t WHERE fee = 1 OR note IS NULL",
"FROM t WHERE (fee > 11 OR rank > 38) AND rank < 39",
"FROM t WHERE fee IN (1) OR fee IN (2)",
"FROM t WHERE NOT (fee > 5)",
"FROM t WHERE NOT (fee IN (1, 2))",
"FROM t WHERE NOT (fee > 5) AND rank < 30",
"FROM t WHERE fee > 5 ORDER BY rank DESC LIMIT 5",
"FROM t WHERE fee BETWEEN 2 AND 8 ORDER BY fee, rank DESC",
"FROM t WHERE fee > 5 GROUP BY note COUNT",
"FROM t WHERE fee > 5 COUNT",
"FROM t WHERE fee > 5 ORDER BY rank LIMIT 3 OFFSET 2",
"FROM t WHERE fee > 9999",
"FROM t WHERE fee IN (9999)",
"FROM t WHERE fee BETWEEN 100 AND 200",
] {
let (a, b) = same(&idx, &plain, nql);
assert_eq!(a, b, "indexed and unindexed disagree on `{}`", nql);
}
}
#[test]
fn index_path_still_honours_order_by() {
let (_t1, _t2, idx, plain) = twin(&["fee", "rank"]);
for nql in [
"FROM t WHERE fee > 4 ORDER BY rank",
"FROM t WHERE fee > 4 ORDER BY rank DESC",
"FROM t WHERE fee > 4 ORDER BY note, rank DESC",
"FROM t WHERE fee BETWEEN 2 AND 9 ORDER BY rank LIMIT 4",
"FROM t WHERE fee IN (3, 6) ORDER BY rank DESC LIMIT 2",
] {
let ra = query(&idx, nql).unwrap().0;
let rb = query(&plain, nql).unwrap().0;
let ia: Vec<&str> = ra.iter().filter_map(|r| r["_id"].as_str()).collect();
let ib: Vec<&str> = rb.iter().filter_map(|r| r["_id"].as_str()).collect();
assert_eq!(ia, ib, "row ORDER differs on `{}`", nql);
}
}
#[test]
fn an_ordering_comparison_against_a_missing_field_is_false() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "has", json!({"fee": 1}), vec![], None, None).unwrap();
db.put("t", "none", json!({"other": 1}), vec![], None, None).unwrap();
db.put("t", "null", json!({"fee": Value::Null}), vec![], None, None).unwrap();
for nql in ["FROM t WHERE fee < 5", "FROM t WHERE fee <= 5"] {
let (r, _) = query(&db, nql).unwrap();
let ids: Vec<&str> = r.iter().filter_map(|x| x["_id"].as_str()).collect();
assert_eq!(ids, vec!["has"],
"`{}` must not match a row whose fee is absent or null", nql);
}
for nql in ["FROM t WHERE fee > 0", "FROM t WHERE fee >= 0"] {
let (r, _) = query(&db, nql).unwrap();
let ids: Vec<&str> = r.iter().filter_map(|x| x["_id"].as_str()).collect();
assert_eq!(ids, vec!["has"], "`{}`", nql);
}
let (b, _) = query(&db, "FROM t WHERE fee BETWEEN 0 AND 9").unwrap();
assert_eq!(b.len(), 1);
let (ne, _) = query(&db, "FROM t WHERE fee != 5").unwrap();
assert_eq!(ne.len(), 3, "!= still matches absent and null fields");
let (isnull, _) = query(&db, "FROM t WHERE fee = NULL").unwrap();
assert_eq!(isnull.len(), 2, "absent and explicit-null both equal NULL");
let d2 = tempdir().unwrap();
let idx = Db::open(d2.path(), None).unwrap();
idx.create_sorted_index("t", "fee");
idx.put("t", "has", json!({"fee": 1}), vec![], None, None).unwrap();
idx.put("t", "none", json!({"other": 1}), vec![], None, None).unwrap();
idx.put("t", "null", json!({"fee": Value::Null}), vec![], None, None).unwrap();
for nql in ["FROM t WHERE fee < 5", "FROM t WHERE fee <= 5",
"FROM t WHERE fee > 0", "FROM t WHERE fee BETWEEN 0 AND 9"] {
let (a, _) = query(&db, nql).unwrap();
let (b, _) = query(&idx, nql).unwrap();
let ia: Vec<&str> = a.iter().filter_map(|x| x["_id"].as_str()).collect();
let ib: Vec<&str> = b.iter().filter_map(|x| x["_id"].as_str()).collect();
assert_eq!(ia, ib, "indexed and unindexed disagree on `{}`", nql);
}
}
#[test]
fn is_null_never_uses_the_index() {
let (_t1, _t2, idx, plain) = twin(&["fee"]);
let (a, b) = same(&idx, &plain, "FROM t WHERE fee IS NULL");
assert_eq!(a, b);
assert_eq!(a, vec!["0", "14", "21", "28", "35", "7"]);
assert!(!a.is_empty(), "the fixture must actually contain absent fields");
}
#[test]
fn a_disjunct_is_never_used_to_narrow() {
let (_t1, _t2, idx, plain) = twin(&["fee", "rank"]);
let nql = "FROM t WHERE fee = 1 OR rank = 40";
let (a, b) = same(&idx, &plain, nql);
assert_eq!(a, b);
assert!(a.contains(&"0".to_string()),
"the OR arm matching a doc with no indexed field must survive: {:?}", a);
assert!(a.len() > 1, "both arms must contribute: {:?}", a);
}
#[test]
fn as_of_does_not_use_the_current_version_index() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("t", "fee");
db.put("t", "a", json!({"fee": 5}), vec![], None, None).unwrap();
let snap = db.put("t", "b", json!({"fee": 5}), vec![], None, None).unwrap().seq;
db.put("t", "a", json!({"fee": 999}), vec![], None, None).unwrap();
db.put("t", "b", json!({"fee": 999}), vec![], None, None).unwrap();
let (now, _) = query(&db, "FROM t WHERE fee = 5").unwrap();
assert!(now.is_empty(), "current versions have fee 999: {:?}", now);
let (then, _) = query(&db, &format!("FROM t AS OF {} WHERE fee = 5", snap)).unwrap();
let mut ids: Vec<&str> = then.iter().filter_map(|r| r["_id"].as_str()).collect();
ids.sort();
assert_eq!(ids, vec!["a", "b"], "AS OF must see the historical values");
let (r, _) = query(&db, &format!("FROM t AS OF {} WHERE fee BETWEEN 1 AND 9", snap)).unwrap();
assert_eq!(r.len(), 2);
let (i, _) = query(&db, &format!("FROM t AS OF {} WHERE fee IN (5)", snap)).unwrap();
assert_eq!(i.len(), 2);
}
#[test]
fn the_index_path_returns_current_versions_only() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("t", "fee");
for i in 0..5u64 {
db.put("t", &i.to_string(), json!({"fee": i}), vec![], None, None).unwrap();
}
db.put("t", "0", json!({"fee": 100}), vec![], None, None).unwrap();
let (low, _) = query(&db, "FROM t WHERE fee BETWEEN 0 AND 4").unwrap();
let mut ids: Vec<&str> = low.iter().filter_map(|r| r["_id"].as_str()).collect();
ids.sort();
assert_eq!(ids, vec!["1", "2", "3", "4"],
"doc 0 moved to fee 100 and must not appear in 0..4");
let (high, _) = query(&db, "FROM t WHERE fee = 100").unwrap();
assert_eq!(high.len(), 1);
assert_eq!(high[0]["_id"], "0");
assert_eq!(high[0]["fee"], json!(100), "the CURRENT value, not the old one");
}
#[test]
fn index_scans_do_not_duplicate_rows() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("t", "fee");
for i in 0..6u64 {
db.put("t", &i.to_string(), json!({"fee": i % 2}), vec![], None, None).unwrap();
}
let (dup, _) = query(&db, "FROM t WHERE fee IN (0, 0, 1, 1)").unwrap();
assert_eq!(dup.len(), 6, "each row once despite repeated IN arms");
let (r, _) = query(&db, "FROM t WHERE fee BETWEEN 0 AND 1").unwrap();
assert_eq!(r.len(), 6);
let mut ids: Vec<&str> = dup.iter().filter_map(|r| r["_id"].as_str()).collect();
ids.sort();
ids.dedup();
assert_eq!(ids.len(), 6, "no duplicate _ids");
}
#[test]
fn index_ranges_work_on_strings() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("t", "name");
for (i, n) in ["alpha", "bravo", "charlie", "delta", "echo"].iter().enumerate() {
db.put("t", &i.to_string(), json!({"name": n}), vec![], None, None).unwrap();
}
let (r, _) = query(&db, r#"FROM t WHERE name BETWEEN "bravo" AND "delta""#).unwrap();
let mut got: Vec<&str> = r.iter().filter_map(|x| x["name"].as_str()).collect();
got.sort();
assert_eq!(got, vec!["bravo", "charlie", "delta"]);
let (gt, _) = query(&db, r#"FROM t WHERE name > "charlie""#).unwrap();
assert_eq!(gt.len(), 2);
}
#[test]
fn same_field_bounds_are_merged_tightest_wins() {
let (_t1, _t2, idx, plain) = twin(&["fee"]);
for (a_nql, b_nql) in [
("FROM t WHERE fee > 2 AND fee > 6", "FROM t WHERE fee > 6"),
("FROM t WHERE fee > 6 AND fee > 2", "FROM t WHERE fee > 6"),
("FROM t WHERE fee < 9 AND fee < 4", "FROM t WHERE fee < 4"),
("FROM t WHERE fee BETWEEN 0 AND 12 AND fee >= 5 AND fee <= 7",
"FROM t WHERE fee >= 5 AND fee <= 7"),
] {
let (ia, _) = same(&idx, &plain, a_nql);
let (ib, _) = same(&idx, &plain, b_nql);
assert_eq!(ia, ib, "`{}` should equal `{}`", a_nql, b_nql);
}
}
#[test]
fn a_predicate_on_an_unindexed_field_still_answers() {
let (_t1, _t2, idx, plain) = twin(&["fee"]); for nql in [
"FROM t WHERE rank > 30",
"FROM t WHERE rank BETWEEN 10 AND 20",
"FROM t WHERE rank IN (40, 39)",
"FROM t WHERE rank > 30 AND fee > 2",
] {
let (a, b) = same(&idx, &plain, nql);
assert_eq!(a, b, "`{}`", nql);
}
}
#[test]
fn range_cardinality_counts_without_reading() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("t", "fee");
for i in 0..20u64 {
db.put("t", &i.to_string(), json!({"fee": i}), vec![], None, None).unwrap();
}
assert_eq!(db.range_cardinality("t", "fee", None, None, true, true), Some(20));
assert_eq!(
db.range_cardinality("t", "fee", Some(&json!(5)), Some(&json!(9)), true, true),
Some(5), "5..=9 inclusive is five values");
assert_eq!(
db.range_cardinality("t", "fee", Some(&json!(5)), Some(&json!(9)), false, false),
Some(3), "exclusive bounds drop both ends");
assert_eq!(
db.range_cardinality("t", "fee", Some(&json!(18)), None, true, true),
Some(2));
assert_eq!(
db.range_cardinality("t", "fee", Some(&json!(999)), None, true, true),
Some(0), "an empty range is 0, not an error");
assert_eq!(db.range_cardinality("t", "nope", None, None, true, true), None);
}
#[test]
fn the_narrower_index_is_preferred() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.create_sorted_index("t", "wide");
db.create_sorted_index("t", "narrow");
for i in 0..100u64 {
db.put("t", &i.to_string(),
json!({"wide": i % 2, "narrow": i}), vec![], None, None).unwrap();
}
let wide = db.range_cardinality("t", "wide", Some(&json!(0)), Some(&json!(0)), true, true);
let narrow = db.range_cardinality("t", "narrow", Some(&json!(7)), Some(&json!(7)), true, true);
assert_eq!(wide, Some(50));
assert_eq!(narrow, Some(1));
let (r, _) = query(&db, "FROM t WHERE wide = 0 AND narrow = 7").unwrap();
assert!(r.is_empty(), "narrow 7 has wide 1, so nothing matches");
let (r2, _) = query(&db, "FROM t WHERE wide = 0 AND narrow = 8").unwrap();
assert_eq!(r2.len(), 1);
assert_eq!(r2[0]["_id"], "8");
}
fn heights(rows: &[Value]) -> Vec<u64> {
rows.iter().filter_map(|r| r["height"].as_u64()).collect()
}
#[test]
fn offset_skips_result_rows() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks ORDER BY height OFFSET 2").unwrap();
assert_eq!(heights(&rows), vec![3, 4, 5]);
}
#[test]
fn offset_with_limit_pages() {
let (_tmp, db) = setup();
let mut seen = vec![];
for page in 0..3 {
let (rows, _) = query(
&db,
&format!("FROM blocks ORDER BY height LIMIT 2 OFFSET {}", page * 2),
).unwrap();
seen.extend(heights(&rows));
}
assert_eq!(seen, vec![1, 2, 3, 4, 5]);
}
#[test]
fn offset_past_the_end_is_an_empty_page() {
let (_tmp, db) = setup();
let (rows, count) = query(&db, "FROM blocks OFFSET 99").unwrap();
assert!(rows.is_empty());
assert_eq!(count, 0);
let (zero, _) = query(&db, "FROM blocks OFFSET 0").unwrap();
assert_eq!(zero.len(), 5, "OFFSET 0 skips nothing");
}
#[test]
fn offset_applies_after_the_filter() {
let (_tmp, db) = setup();
let (rows, _) = query(
&db, "FROM blocks WHERE n_tx >= 6 ORDER BY height OFFSET 1").unwrap();
assert_eq!(heights(&rows), vec![4, 5]);
}
#[test]
fn order_by_multiple_keys() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for (i, (s, f)) in [("open", 30), ("open", 10), ("closed", 20),
("open", 20), ("closed", 5)].iter().enumerate() {
db.put("t", &i.to_string(),
serde_json::json!({"status": s, "fee": f}), vec![], None, None).unwrap();
}
let (rows, _) = query(&db, "FROM t ORDER BY status, fee DESC").unwrap();
let got: Vec<(String, u64)> = rows.iter()
.map(|r| (r["status"].as_str().unwrap().to_string(), r["fee"].as_u64().unwrap()))
.collect();
assert_eq!(got, vec![
("closed".into(), 20), ("closed".into(), 5),
("open".into(), 30), ("open".into(), 20), ("open".into(), 10),
]);
}
#[test]
fn order_by_mixed_directions() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for (i, (a, b)) in [(1, 1), (1, 2), (2, 1), (2, 2)].iter().enumerate() {
db.put("t", &i.to_string(),
serde_json::json!({"a": a, "b": b}), vec![], None, None).unwrap();
}
let (rows, _) = query(&db, "FROM t ORDER BY a DESC, b ASC").unwrap();
let got: Vec<(u64, u64)> = rows.iter()
.map(|r| (r["a"].as_u64().unwrap(), r["b"].as_u64().unwrap()))
.collect();
assert_eq!(got, vec![(2, 1), (2, 2), (1, 1), (1, 2)]);
}
#[test]
fn limit_applies_to_grouped_rows_not_to_the_input() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for i in 0..12 {
let status = ["open", "closed", "void"][i % 3];
db.put("t", &i.to_string(),
serde_json::json!({"status": status, "fee": i}),
vec![], None, None).unwrap();
}
let (all, _) = query(&db, "FROM t GROUP BY status COUNT").unwrap();
assert_eq!(all.len(), 3);
let total: u64 = all.iter().filter_map(|r| r["count"].as_u64()).sum();
assert_eq!(total, 12, "every input row must be counted");
let (limited, _) = query(&db, "FROM t GROUP BY status COUNT LIMIT 2").unwrap();
assert_eq!(limited.len(), 2, "LIMIT caps the number of groups");
for r in &limited {
assert_eq!(r["count"], json!(4),
"each group keeps its true count, got {:?}", r);
}
}
#[test]
fn order_by_sorts_the_grouped_rows() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for (i, s) in ["a", "b", "b", "b", "c", "c"].iter().enumerate() {
db.put("t", &i.to_string(),
serde_json::json!({"g": s, "n": i}), vec![], None, None).unwrap();
}
let (rows, _) = query(&db, "FROM t GROUP BY g COUNT ORDER BY count DESC").unwrap();
let got: Vec<(String, u64)> = rows.iter()
.map(|r| (r["g"].as_str().unwrap().to_string(), r["count"].as_u64().unwrap()))
.collect();
assert_eq!(got, vec![("b".into(), 3), ("c".into(), 2), ("a".into(), 1)]);
let (by_key, _) = query(&db, "FROM t GROUP BY g COUNT ORDER BY g DESC").unwrap();
let keys: Vec<&str> = by_key.iter().map(|r| r["g"].as_str().unwrap()).collect();
assert_eq!(keys, vec!["c", "b", "a"]);
}
#[test]
fn order_by_an_aggregate_key() {
let (_tmp, db) = setup_items();
let (rows, _) = query(
&db, "FROM items GROUP BY cat SUM price ORDER BY sum_price DESC").unwrap();
let cats: Vec<&str> = rows.iter().map(|r| r["cat"].as_str().unwrap()).collect();
assert_eq!(cats, vec!["y", "x"], "y sums to 60, x to 15");
}
#[test]
fn offset_and_limit_page_grouped_rows() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for i in 0..9 {
db.put("t", &i.to_string(),
serde_json::json!({"g": format!("g{}", i % 3)}), vec![], None, None).unwrap();
}
let (page, _) = query(
&db, "FROM t GROUP BY g COUNT ORDER BY g LIMIT 1 OFFSET 1").unwrap();
assert_eq!(page.len(), 1);
assert_eq!(page[0]["g"], "g1");
}
#[test]
fn having_filters_groups_by_count() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for (i, s) in ["a", "b", "b", "b", "c", "c"].iter().enumerate() {
db.put("t", &i.to_string(),
serde_json::json!({"g": s, "n": i}), vec![], None, None).unwrap();
}
let (rows, _) = query(&db, "FROM t GROUP BY g COUNT HAVING count > 1").unwrap();
let mut keys: Vec<&str> = rows.iter().map(|r| r["g"].as_str().unwrap()).collect();
keys.sort();
assert_eq!(keys, vec!["b", "c"], "the single-row group `a` is filtered out");
}
#[test]
fn having_filters_on_the_aggregate_value() {
let (_tmp, db) = setup_items();
let (rows, _) = query(
&db, "FROM items GROUP BY cat SUM price HAVING sum_price > 20").unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0]["cat"], "y");
}
#[test]
fn having_supports_the_whole_predicate_surface() {
let (_tmp, db) = setup_items();
let (in_, _) = query(
&db, r#"FROM items GROUP BY cat COUNT HAVING cat IN ("x")"#).unwrap();
assert_eq!(in_.len(), 1);
assert_eq!(in_[0]["cat"], "x");
let (btw, _) = query(
&db, "FROM items GROUP BY cat SUM price HAVING sum_price BETWEEN 10 AND 20").unwrap();
assert_eq!(btw.len(), 1);
assert_eq!(btw[0]["cat"], "x");
let (like, _) = query(
&db, r#"FROM items GROUP BY cat COUNT HAVING cat LIKE "y""#).unwrap();
assert_eq!(like.len(), 1);
let (or_, _) = query(
&db, "FROM items GROUP BY cat SUM price HAVING sum_price < 20 OR count = 3").unwrap();
assert_eq!(or_.len(), 2);
}
#[test]
fn where_and_having_are_different_stages() {
let (_tmp, db) = setup_items();
let (w, _) = query(
&db, "FROM items WHERE price > 10 GROUP BY cat SUM price").unwrap();
let x = w.iter().find(|r| r["cat"] == "x");
assert!(x.is_none(), "x's rows (0,5,10) are all filtered out by WHERE");
let (h, _) = query(
&db, "FROM items GROUP BY cat SUM price HAVING sum_price > 10").unwrap();
assert_eq!(h.len(), 2, "both groups sum above 10 when nothing is pre-filtered");
}
#[test]
fn having_without_an_aggregate_is_an_error() {
let (_tmp, db) = setup();
assert!(query(&db, "FROM blocks HAVING height > 3").is_err());
}
#[test]
fn bare_count_returns_one_row() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks COUNT").unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0]["count"], json!(5));
assert_eq!(rows[0]["value"], json!(5));
}
#[test]
fn bare_count_respects_the_filter() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks WHERE height > 3 COUNT").unwrap();
assert_eq!(rows[0]["count"], json!(2));
}
#[test]
fn bare_sum_avg_min_max() {
let (_tmp, db) = setup();
let (s, _) = query(&db, "FROM blocks SUM n_tx").unwrap();
assert_eq!(s[0]["sum_n_tx"], json!(30), "integer inputs give an integer sum");
let (a, _) = query(&db, "FROM blocks AVG n_tx").unwrap();
assert_eq!(a[0]["avg_n_tx"], json!(6.0));
let (mn, _) = query(&db, "FROM blocks MIN n_tx").unwrap();
assert_eq!(mn[0]["min_n_tx"], json!(2));
let (mx, _) = query(&db, "FROM blocks MAX n_tx").unwrap();
assert_eq!(mx[0]["max_n_tx"], json!(10));
}
#[test]
fn integer_aggregates_stay_integers_and_keep_full_precision() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
let big: [i64; 3] = [9_007_199_254_740_993, 9_007_199_254_740_995, 1];
for (i, v) in big.iter().enumerate() {
db.put("t", &i.to_string(), serde_json::json!({"v": v}),
vec![], None, None).unwrap();
}
let (s, _) = query(&db, "FROM t SUM v").unwrap();
assert_eq!(s[0]["sum_v"], json!(18_014_398_509_481_989i64),
"exact i64 sum, not a rounded f64");
assert!(s[0]["sum_v"].is_i64(), "must serialise as an integer");
let (mx, _) = query(&db, "FROM t MAX v").unwrap();
assert_eq!(mx[0]["max_v"], json!(9_007_199_254_740_995i64));
let (mn, _) = query(&db, "FROM t MIN v").unwrap();
assert_eq!(mn[0]["min_v"], json!(1));
}
#[test]
fn a_single_float_makes_the_aggregate_fractional() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "1", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
db.put("t", "2", serde_json::json!({"v": 2.5}), vec![], None, None).unwrap();
let (s, _) = query(&db, "FROM t SUM v").unwrap();
assert_eq!(s[0]["sum_v"], json!(3.5));
let (a, _) = query(&db, "FROM t AVG v").unwrap();
assert_eq!(a[0]["avg_v"], json!(1.75));
}
#[test]
fn booleans_are_not_aggregated_as_numbers() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "1", serde_json::json!({"v": true}), vec![], None, None).unwrap();
db.put("t", "2", serde_json::json!({"v": 5}), vec![], None, None).unwrap();
let (s, _) = query(&db, "FROM t SUM v").unwrap();
assert_eq!(s[0]["sum_v"], json!(5), "the bool contributes nothing");
assert_eq!(s[0]["count"], json!(2), "but it still counts toward the group");
}
#[test]
fn bare_count_of_nothing_is_zero_not_empty() {
let (_tmp, db) = setup();
let (rows, count) = query(&db, "FROM blocks WHERE height > 999 COUNT").unwrap();
assert_eq!(count, 1, "still exactly one row");
assert_eq!(rows[0]["count"], json!(0));
let (g, _) = query(&db, "FROM blocks WHERE height > 999 GROUP BY height COUNT").unwrap();
assert!(g.is_empty());
}
#[test]
fn bare_aggregate_over_an_empty_collection() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM nonexistent COUNT").unwrap();
assert_eq!(rows[0]["count"], json!(0));
let (s, _) = query(&db, "FROM nonexistent SUM n_tx").unwrap();
assert_eq!(s[0]["sum_n_tx"], Value::Null, "sum of nothing is null, not 0");
}
#[test]
fn bare_aggregate_carries_no_group_key() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, "FROM blocks COUNT").unwrap();
if let Value::Object(m) = &rows[0] {
let mut keys: Vec<&String> = m.keys().collect();
keys.sort();
assert_eq!(keys, vec!["count", "value"]);
} else {
panic!("expected an object");
}
}
#[test]
fn a_field_named_like_a_keyword_is_still_addressable() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "1", serde_json::json!({
"count": 5, "min": 1, "max": 9, "sum": 3, "avg": 2,
"value": "keep", "limit": 7, "offset": 8, "group": "g1", "search": "s",
}), vec![], None, None).unwrap();
db.put("t", "2", serde_json::json!({
"count": 1, "min": 0, "max": 2, "sum": 0, "avg": 0,
"value": "drop", "limit": 0, "offset": 0, "group": "g2", "search": "t",
}), vec![], None, None).unwrap();
for (nql, want) in [
("FROM t WHERE count > 3", "1"),
("FROM t WHERE min = 1", "1"),
("FROM t WHERE max >= 9", "1"),
("FROM t WHERE sum = 3", "1"),
("FROM t WHERE avg = 2", "1"),
(r#"FROM t WHERE value = "keep""#, "1"),
("FROM t WHERE limit = 7", "1"),
("FROM t WHERE offset = 8", "1"),
(r#"FROM t WHERE group = "g1""#, "1"),
] {
let (rows, _) = query(&db, nql).unwrap();
assert_eq!(rows.len(), 1, "`{}` matched {} rows", nql, rows.len());
assert_eq!(rows[0]["_id"], want, "`{}`", nql);
}
let (ord, _) = query(&db, "FROM t ORDER BY count DESC").unwrap();
assert_eq!(ord[0]["_id"], "1");
let (grp, _) = query(&db, "FROM t GROUP BY group COUNT").unwrap();
assert_eq!(grp.len(), 2);
let keys: Vec<&str> = grp.iter().filter_map(|r| r["group"].as_str()).collect();
assert!(keys.contains(&"g1") && keys.contains(&"g2"), "{:?}", grp);
}
#[test]
fn keyword_matching_stays_case_insensitive() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "1", serde_json::json!({"Count": 5, "n": 1}),
vec![], None, None).unwrap();
let (rows, _) = query(&db, "from t where Count = 5 order by n").unwrap();
assert_eq!(rows.len(), 1);
let (miss, _) = query(&db, "FROM t WHERE count = 5").unwrap();
assert!(miss.is_empty(), "`count` and `Count` are distinct field names");
}
#[test]
fn two_aggregates_is_an_error() {
let (_tmp, db) = setup();
assert!(query(&db, "FROM blocks COUNT SUM n_tx").is_err());
assert!(query(&db, "FROM blocks GROUP BY height COUNT SUM n_tx").is_err());
}
#[test]
fn bare_aggregate_with_having() {
let (_tmp, db) = setup();
let (keep, _) = query(&db, "FROM blocks COUNT HAVING count > 3").unwrap();
assert_eq!(keep.len(), 1);
let (drop, _) = query(&db, "FROM blocks COUNT HAVING count > 99").unwrap();
assert!(drop.is_empty());
}
fn setup_items() -> (tempfile::TempDir, Db) {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
for i in 0..6 {
db.put("items", &i.to_string(),
serde_json::json!({"cat": if i < 3 {"x"} else {"y"}, "price": i * 5}),
vec![], None, None).unwrap();
}
(dir, db)
}
fn group(rows: &[Value], field: &str, key: &str) -> Value {
rows.iter()
.find(|r| r[field] == Value::String(key.to_string()))
.unwrap_or_else(|| panic!("no group {:?} in {:?}", key, rows))
.clone()
}
#[test]
fn group_by_aggregates_the_target_field_not_the_group_field() {
let (_tmp, db) = setup_items();
let (mins, _) = query(&db, "FROM items GROUP BY cat MIN price").unwrap();
assert_eq!(group(&mins, "cat", "x")["min_price"], json!(0));
assert_eq!(group(&mins, "cat", "y")["min_price"], json!(15));
let (maxs, _) = query(&db, "FROM items GROUP BY cat MAX price").unwrap();
assert_eq!(group(&maxs, "cat", "y")["max_price"], json!(25));
assert_eq!(group(&maxs, "cat", "x")["max_price"], json!(10));
let (sums, _) = query(&db, "FROM items GROUP BY cat SUM price").unwrap();
assert_eq!(group(&sums, "cat", "x")["sum_price"], json!(15)); assert_eq!(group(&sums, "cat", "y")["sum_price"], json!(60));
let (avgs, _) = query(&db, "FROM items GROUP BY cat AVG price").unwrap();
assert_eq!(group(&avgs, "cat", "x")["avg_price"], json!(5.0));
assert_eq!(group(&avgs, "cat", "y")["avg_price"], json!(20.0));
}
#[test]
fn group_by_emits_python_parity_keys_and_the_value_alias() {
let (_tmp, db) = setup_items();
let (rows, _) = query(&db, "FROM items GROUP BY cat SUM price").unwrap();
let x = group(&rows, "cat", "x");
assert_eq!(x["sum_price"], json!(15), "python-parity key");
assert_eq!(x["value"], json!(15), "back-compat alias must agree");
assert_eq!(x["count"], json!(3), "count is the group size");
}
#[test]
fn count_is_group_size_while_aggregate_skips_non_numeric() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "1", serde_json::json!({"g": "a", "n": 10}), vec![], None, None).unwrap();
db.put("t", "2", serde_json::json!({"g": "a", "n": 20}), vec![], None, None).unwrap();
db.put("t", "3", serde_json::json!({"g": "a", "n": "N/A"}), vec![], None, None).unwrap();
let (rows, _) = query(&db, "FROM t GROUP BY g AVG n").unwrap();
let a = group(&rows, "g", "a");
assert_eq!(a["count"], json!(3), "every row counts toward the group");
assert_eq!(a["avg_n"], json!(15.0), "only the two numeric rows average");
}
#[test]
fn empty_aggregate_input_is_null() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("t", "1", serde_json::json!({"g": "a", "n": "x"}), vec![], None, None).unwrap();
let (rows, _) = query(&db, "FROM t GROUP BY g MIN n").unwrap();
assert_eq!(rows[0]["min_n"], Value::Null);
assert_eq!(rows[0]["count"], json!(1));
}
#[test]
fn bare_group_by_without_an_aggregate_counts() {
let (_tmp, db) = setup_items();
let (rows, _) = query(&db, "FROM items GROUP BY cat").unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(group(&rows, "cat", "x")["count"], json!(3));
assert_eq!(group(&rows, "cat", "y")["count"], json!(3));
}
#[test]
fn aggregate_without_a_target_field_is_an_error() {
let (_tmp, db) = setup_items();
for bad in [
"FROM items GROUP BY cat SUM",
"FROM items GROUP BY cat AVG",
"FROM items GROUP BY cat MIN",
] {
assert!(query(&db, bad).is_err(), "`{}` must be rejected", bad);
}
}
#[test]
fn group_order_is_stable_across_runs() {
let (_tmp, db) = setup_items();
let first = query(&db, "FROM items GROUP BY cat SUM price").unwrap().0;
for _ in 0..8 {
let again = query(&db, "FROM items GROUP BY cat SUM price").unwrap().0;
assert_eq!(first, again);
}
}
#[test]
fn group_by_after_an_in_predicate() {
let (_tmp, db) = setup_items();
let (rows, _) = query(
&db, "FROM items WHERE price IN (0, 5, 25) GROUP BY cat SUM price").unwrap();
assert_eq!(group(&rows, "cat", "x")["sum_price"], json!(5));
assert_eq!(group(&rows, "cat", "y")["sum_price"], json!(25));
}
#[test]
fn search() {
let (_tmp, db) = setup();
let (rows, _) = query(&db, r#"FROM blocks SEARCH "0003""#).unwrap();
assert_eq!(rows.len(), 1);
}
#[test]
fn as_of() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
let v1 = db.put("docs", "x", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
db.put("docs", "x", serde_json::json!({"v": 2}), vec![], None, None).unwrap();
let (rows, _) = query(&db, &format!("FROM docs AS OF {}", v1.seq)).unwrap();
assert_eq!(rows[0]["v"], 1);
}
#[test]
fn valid_as_of() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("events", "e1", serde_json::json!({"type": "a"}), vec![],
Some("2025-01-01".to_string()), Some("2025-06-01".to_string())).unwrap();
db.put("events", "e2", serde_json::json!({"type": "b"}), vec![],
Some("2026-01-01".to_string()), None).unwrap();
let (rows, _) = query(&db, r#"FROM events VALID AS OF "2025-03-01""#).unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0]["type"], "a");
}
#[test]
fn escaped_quote_matches_a_value_containing_a_quote() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("m", "q", serde_json::json!({ "name": "say \"hi\"" }), vec![], None, None)
.unwrap();
db.put("m", "p", serde_json::json!({ "name": "plain" }), vec![], None, None).unwrap();
let (rows, count) = query(&db, r#"FROM m WHERE name = "say \"hi\"""#).unwrap();
assert_eq!(count, 1, "the escaped-quote literal matches exactly one row");
assert_eq!(rows[0]["_id"], "q");
}
#[test]
fn raw_backslash_still_matches_literally() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("m", "b", serde_json::json!({ "p": "back\\slash" }), vec![], None, None).unwrap();
let (rows, count) = query(&db, r#"FROM m WHERE p = "back\slash""#).unwrap();
assert_eq!(count, 1, "a raw backslash literal matches as before");
assert_eq!(rows[0]["_id"], "b");
}
#[test]
fn a_quote_can_no_longer_inject_trailing_clauses() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("m", "x", serde_json::json!({ "v": "a\"b" }), vec![], None, None).unwrap();
let (rows, count) = query(&db, r#"FROM m WHERE v = "a\"b""#).unwrap();
assert_eq!(count, 1);
assert_eq!(rows[0]["_id"], "x");
}
}
#[cfg(test)]
mod tests_traverse {
use super::*;
use tempfile::tempdir;
use crate::db::Db;
#[test]
fn traverse_one_hop() {
let db = Db::in_memory();
db.put("driver", "d1", serde_json::json!({"name": "Bob"}), vec![], None, None).unwrap();
db.put("driver", "d2", serde_json::json!({"name": "Carol"}), vec![], None, None).unwrap();
db.put("trip", "t1", serde_json::json!({"status": "req"}), vec![], None, None).unwrap();
db.put("trip", "t2", serde_json::json!({"status": "ok"}), vec![], None, None).unwrap();
db.link("driver:d1", "handles", "trip:t1").unwrap();
db.link("driver:d1", "handles", "trip:t2").unwrap();
let (rows, count) = query(&db, r#"FROM driver WHERE _id = "d1" TRAVERSE handles"#).unwrap();
assert_eq!(count, 2);
let ids: std::collections::HashSet<&str> = rows.iter()
.filter_map(|r| r["_id"].as_str())
.collect();
assert!(ids.contains("t1") && ids.contains("t2"));
}
#[test]
fn traverse_returns_empty_when_no_links() {
let db = Db::in_memory();
db.put("driver", "d1", serde_json::json!({"name": "Bob"}), vec![], None, None).unwrap();
let (rows, count) = query(&db, r#"FROM driver WHERE _id = "d1" TRAVERSE handles"#).unwrap();
assert_eq!(count, 0);
assert!(rows.is_empty());
}
#[test]
fn traverse_multi_source() {
let db = Db::in_memory();
db.put("driver", "d1", serde_json::json!({"status": "active"}), vec![], None, None).unwrap();
db.put("driver", "d2", serde_json::json!({"status": "active"}), vec![], None, None).unwrap();
db.put("trip", "t1", serde_json::json!({"n": 1}), vec![], None, None).unwrap();
db.put("trip", "t2", serde_json::json!({"n": 2}), vec![], None, None).unwrap();
db.put("trip", "t3", serde_json::json!({"n": 3}), vec![], None, None).unwrap();
db.link("driver:d1", "handles", "trip:t1").unwrap();
db.link("driver:d1", "handles", "trip:t2").unwrap();
db.link("driver:d2", "handles", "trip:t3").unwrap();
let (_rows, count) = query(&db, r#"FROM driver WHERE status = "active" TRAVERSE handles"#).unwrap();
assert_eq!(count, 3);
}
#[test]
fn traverse_nql_keyword_case_insensitive() {
let db = Db::in_memory();
db.put("driver", "d1", serde_json::json!({}), vec![], None, None).unwrap();
db.put("trip", "t1", serde_json::json!({}), vec![], None, None).unwrap();
db.link("driver:d1", "handles", "trip:t1").unwrap();
let (r1, c1) = query(&db, r#"FROM driver WHERE _id = "d1" TRAVERSE handles"#).unwrap();
assert_eq!(c1, 1);
let (r2, c2) = query(&db, r#"FROM driver WHERE _id = "d1" traverse handles"#).unwrap();
assert_eq!(c2, 1);
assert_eq!(r1[0]["_id"], r2[0]["_id"]);
}
#[test]
fn traverse_durable() {
let dir = tempdir().unwrap();
{
let db = Db::open(dir.path(), None).unwrap();
db.put("driver", "d1", serde_json::json!({"name": "Bob"}), vec![], None, None).unwrap();
db.put("trip", "t1", serde_json::json!({"status": "req"}), vec![], None, None).unwrap();
db.link("driver:d1", "handles", "trip:t1").unwrap();
}
let db2 = Db::open(dir.path(), None).unwrap();
db2.startup_ready.store(true, std::sync::atomic::Ordering::SeqCst);
let (rows, count) = query(&db2, r#"FROM driver WHERE _id = "d1" TRAVERSE handles"#).unwrap();
assert_eq!(count, 1);
assert_eq!(rows[0]["_id"], "t1");
}
}