1use rushdb_types::value::ID_FIELD;
18use serde_json::Value as Json;
19
20use crate::{AggregateRequest, Direction, Expr, FieldPred, Pagination, RelatedExpr, SearchRequest};
21
22#[derive(Debug, thiserror::Error)]
24pub enum QueryDocumentError {
25 #[error("invalid where clause: {0}")]
27 Parse(String),
28 #[error("unsupported where construct: {0}")]
30 Unsupported(String),
31}
32
33const OPS: &[&str] = &[
34 "$eq",
35 "$ne",
36 "$gt",
37 "$gte",
38 "$lt",
39 "$lte",
40 "$in",
41 "$nin",
42 "$contains",
43 "$startsWith",
44 "$endsWith",
45 "$exists",
46 "$type",
47 "$and",
48 "$or",
49 "$nor",
50 "$xor",
51 "$not",
52 "$id",
53 "$relation",
54 "$alias",
55 "$cycle",
56 "$year",
57 "$month",
58 "$day",
59 "$hour",
60 "$minute",
61 "$second",
62 "$millisecond",
63 "$microsecond",
64 "$nanosecond",
65];
66
67fn is_op_key(k: &str) -> bool {
68 OPS.contains(&k)
69}
70
71fn reject_unknown_operators(key: &str, value: &Json) -> Result<(), QueryDocumentError> {
74 let unknown = std::iter::once(key)
75 .chain(
76 value
77 .as_object()
78 .into_iter()
79 .flat_map(|m| m.keys().map(String::as_str)),
80 )
81 .find(|k| k.starts_with('$') && !is_op_key(k));
82 match unknown {
83 Some(op) => Err(QueryDocumentError::Parse(format!(
84 "unknown operator `{op}`"
85 ))),
86 None => Ok(()),
87 }
88}
89
90fn lit_json(v: &Json) -> Json {
91 v.clone()
92}
93
94pub fn parse_where(obj: &Json) -> Result<Expr, QueryDocumentError> {
102 let map = obj
103 .as_object()
104 .ok_or_else(|| QueryDocumentError::Parse("where must be an object".into()))?;
105 let mut parts: Vec<Expr> = Vec::new();
106 for (k, v) in map {
107 reject_unknown_operators(k, v)?;
108 if k == "$and" {
109 parts.push(logical(k, v, "and")?);
110 } else if k == "$or" {
111 parts.push(logical(k, v, "or")?);
112 } else if k == "$nor" {
113 parts.push(logical(k, v, "nor")?);
114 } else if k == "$xor" {
115 parts.push(logical(k, v, "xor")?);
116 } else if k == "$not" {
117 parts.push(Expr::Not(Box::new(parse_where(v)?)));
118 } else if k == "$id" {
119 parts.push(parse_id(v)?);
120 } else if k == "$cycle" {
121 parts.push(Expr::Related(parse_cycle(v)?));
122 } else if is_op_key(k) {
123 return Err(QueryDocumentError::Unsupported(format!(
124 "root-level operator key `{k}` (operators apply to fields)"
125 )));
126 } else if v.is_object() && looks_like_related_block(v) {
127 parts.push(Expr::Related(parse_related_block(k, v)?));
129 } else if v.is_object() && looks_like_related_logical_block(v) {
130 parts.push(Expr::Related(parse_related_block(k, v)?));
133 } else if v.is_object() && looks_like_operator_block(v) {
134 parts.push(parse_field_ops(k, v)?);
135 } else if v.is_array() && is_primitive_array(v) {
136 parts.push(Expr::Field(FieldPred::In(
140 norm(k),
141 v.as_array().cloned().unwrap_or_default(),
142 )));
143 } else if v.is_object() || v.is_array() {
144 parts.push(Expr::Related(parse_related_block(k, v)?));
146 } else {
147 parts.push(Expr::Field(FieldPred::Eq(norm(k), lit_json(v))));
149 }
150 }
151 if parts.is_empty() {
152 return Ok(Expr::And(vec![]));
153 }
154 Ok(Expr::And(parts))
155}
156
157fn logical(_k: &str, v: &Json, op: &str) -> Result<Expr, QueryDocumentError> {
158 let arr = v
159 .as_array()
160 .ok_or_else(|| QueryDocumentError::Parse(format!("${op} requires an array")))?;
161 let items: Result<Vec<Expr>, _> = arr.iter().map(parse_where).collect();
162 let items = items?;
163 Ok(match op {
164 "or" => Expr::Or(items),
165 "nor" => Expr::Nor(items),
166 "xor" => {
167 if items.len() != 2 {
168 return Err(QueryDocumentError::Parse(
169 "$xor requires exactly two conditions".into(),
170 ));
171 }
172 Expr::Xor(Box::new(items[0].clone()), Box::new(items[1].clone()))
173 }
174 _ => Expr::And(items),
175 })
176}
177
178fn parse_field_ops(field: &str, v: &Json) -> Result<Expr, QueryDocumentError> {
181 let map = v
182 .as_object()
183 .ok_or_else(|| QueryDocumentError::Parse("operator block must be an object".into()))?;
184 let mut parts = Vec::new();
185 for (op, value) in map {
186 if op == "$or" {
187 parts.push(Expr::Or(parse_field_conditions(field, value)?));
188 } else if op == "$and" {
189 parts.push(Expr::And(parse_field_conditions(field, value)?));
190 } else if op == "$nor" {
191 parts.push(Expr::Nor(parse_field_conditions(field, value)?));
192 } else if op == "$xor" {
193 let items = parse_field_conditions(field, value)?;
194 if items.len() != 2 {
195 return Err(QueryDocumentError::Parse(
196 "$xor requires exactly two conditions".into(),
197 ));
198 }
199 parts.push(Expr::Xor(
200 Box::new(items[0].clone()),
201 Box::new(items[1].clone()),
202 ));
203 } else if op == "$not" {
204 parts.push(Expr::Not(Box::new(parse_field_or_logical(field, value)?)));
205 } else {
206 parts.push(parse_one_op(field, op, value)?);
207 }
208 }
209 Ok(if parts.is_empty() {
210 Expr::And(vec![])
211 } else if parts.len() == 1 {
212 parts.remove(0)
213 } else {
214 Expr::And(parts)
215 })
216}
217
218fn parse_field_conditions(field: &str, v: &Json) -> Result<Vec<Expr>, QueryDocumentError> {
221 let arr = v
222 .as_array()
223 .ok_or_else(|| QueryDocumentError::Parse("logical operator requires an array".into()))?;
224 let mut out = Vec::new();
225 for item in arr {
226 if let Some(m) = item.as_object() {
227 if m.keys().all(|k| is_op_key(k)) {
228 out.push(parse_field_ops(field, item)?);
229 } else {
230 out.push(parse_where(item)?);
231 }
232 } else {
233 return Err(QueryDocumentError::Parse(
234 "condition must be an object".into(),
235 ));
236 }
237 }
238 Ok(out)
239}
240
241fn parse_field_or_logical(field: &str, v: &Json) -> Result<Expr, QueryDocumentError> {
242 if let Some(m) = v.as_object() {
243 if m.keys().all(|k| is_op_key(k)) {
244 return parse_field_ops(field, v);
245 }
246 }
247 parse_where(v)
248}
249
250fn looks_like_related_logical_block(v: &Json) -> bool {
251 let Some(m) = v.as_object() else { return false };
252 if !m
253 .keys()
254 .all(|k| matches!(k.as_str(), "$and" | "$or" | "$nor" | "$xor" | "$not"))
255 {
256 return false;
257 }
258 for val in m.values() {
261 if let Some(arr) = val.as_array() {
262 if arr.iter().any(|e| {
263 e.as_object()
264 .is_some_and(|o| o.keys().any(|k| !is_op_key(k)))
265 }) {
266 return true;
267 }
268 } else if let Some(o) = val.as_object() {
269 if o.keys().any(|k| !is_op_key(k)) {
270 return true;
271 }
272 }
273 }
274 false
275}
276
277fn looks_like_operator_block(v: &Json) -> bool {
278 v.as_object()
279 .is_some_and(|m| m.keys().any(|k| is_op_key(k) && !is_relation_key(k)))
280}
281
282fn is_relation_key(k: &str) -> bool {
283 matches!(k, "$relation" | "$alias" | "$cycle" | "$id")
284}
285
286fn is_primitive_array(v: &Json) -> bool {
290 v.as_array().is_some_and(|items| {
291 items
292 .iter()
293 .all(|item| !item.is_object() && !item.is_array())
294 })
295}
296
297fn looks_like_related_block(v: &Json) -> bool {
298 v.as_object()
299 .is_some_and(|m| m.keys().any(|k| is_relation_key(k)))
300}
301
302fn parse_one_op(field: &str, op: &str, value: &Json) -> Result<Expr, QueryDocumentError> {
303 let f = norm(field);
304 match op {
305 "$eq" => Ok(Expr::Field(FieldPred::Eq(f, dt_lit(value)))),
306 "$ne" => Ok(Expr::Field(FieldPred::Ne(f, dt_lit(value)))),
307 "$gt" => Ok(Expr::Field(FieldPred::Gt(f, dt_num(value)?))),
308 "$gte" => Ok(Expr::Field(FieldPred::Gte(f, dt_num(value)?))),
309 "$lt" => Ok(Expr::Field(FieldPred::Lt(f, dt_num(value)?))),
310 "$lte" => Ok(Expr::Field(FieldPred::Lte(f, dt_num(value)?))),
311 "$in" => Ok(Expr::Field(FieldPred::In(f, array_of(value)?))),
312 "$nin" => Ok(Expr::Field(FieldPred::Nin(f, array_of(value)?))),
313 "$contains" => Ok(Expr::Field(FieldPred::Contains(f, str_of(value)?))),
314 "$startsWith" => Ok(Expr::Field(FieldPred::StartsWith(f, str_of(value)?))),
315 "$endsWith" => Ok(Expr::Field(FieldPred::EndsWith(f, str_of(value)?))),
316 "$exists" => {
317 let want = value
318 .as_bool()
319 .ok_or_else(|| QueryDocumentError::Parse("$exists requires a boolean".into()))?;
320 Ok(Expr::Field(FieldPred::Exists(f, want)))
321 }
322 "$type" => Ok(Expr::Field(FieldPred::Type(f, str_of(value)?))),
323 "$id" => parse_id(value),
324 "$year" => Ok(Expr::Field(FieldPred::DatetimeComponent {
325 field: f,
326 which: 0,
327 value: comp_int(value)?,
328 })),
329 "$month" => Ok(Expr::Field(FieldPred::DatetimeComponent {
330 field: f,
331 which: 1,
332 value: comp_int(value)?,
333 })),
334 "$day" => Ok(Expr::Field(FieldPred::DatetimeComponent {
335 field: f,
336 which: 2,
337 value: comp_int(value)?,
338 })),
339 "$hour" => Ok(Expr::Field(FieldPred::DatetimeComponent {
340 field: f,
341 which: 3,
342 value: comp_int(value)?,
343 })),
344 "$minute" => Ok(Expr::Field(FieldPred::DatetimeComponent {
345 field: f,
346 which: 4,
347 value: comp_int(value)?,
348 })),
349 "$second" => Ok(Expr::Field(FieldPred::DatetimeComponent {
350 field: f,
351 which: 5,
352 value: comp_int(value)?,
353 })),
354 "$millisecond" => Ok(Expr::Field(FieldPred::DatetimeComponent {
355 field: f,
356 which: 6,
357 value: comp_int(value)?,
358 })),
359 "$microsecond" => Ok(Expr::Field(FieldPred::DatetimeComponent {
360 field: f,
361 which: 7,
362 value: comp_int(value)?,
363 })),
364 "$nanosecond" => Ok(Expr::Field(FieldPred::DatetimeComponent {
365 field: f,
366 which: 8,
367 value: comp_int(value)?,
368 })),
369 other => Err(QueryDocumentError::Unsupported(other.to_string())),
370 }
371}
372
373fn parse_id(v: &Json) -> Result<Expr, QueryDocumentError> {
374 if let Some(s) = v.as_str() {
375 return Ok(Expr::Field(FieldPred::Id(s.to_string())));
376 }
377 if let Some(map) = v.as_object() {
378 if let Some(arr) = map.get("$in").and_then(|x| x.as_array()) {
380 let ids = arr
381 .iter()
382 .map(|x| {
383 x.as_str()
384 .map(|id| Expr::Field(FieldPred::Id(id.to_string())))
385 .ok_or_else(|| {
386 QueryDocumentError::Parse("$id.$in requires string ids".into())
387 })
388 })
389 .collect::<Result<Vec<_>, _>>()?;
390 return Ok(Expr::Or(ids));
391 }
392 }
393 Err(QueryDocumentError::Parse(
394 "$id requires a string or $in list".into(),
395 ))
396}
397
398fn parse_related_logical(op: &str, v: &Json) -> Result<Expr, QueryDocumentError> {
403 let subs = |arr: &Json| -> Result<Vec<Expr>, QueryDocumentError> {
404 arr.as_array()
405 .ok_or_else(|| QueryDocumentError::Parse(format!("{op} requires an array")))?
406 .iter()
407 .map(parse_where)
408 .collect()
409 };
410 match op {
411 "$and" => Ok(Expr::And(subs(v)?)),
412 "$or" => Ok(Expr::Or(subs(v)?)),
413 "$nor" => Ok(Expr::Nor(subs(v)?)),
414 "$xor" => {
415 let arr = subs(v)?;
416 if arr.len() != 2 {
417 return Err(QueryDocumentError::Parse(
418 "$xor requires exactly two operands".into(),
419 ));
420 }
421 Ok(Expr::Xor(
422 Box::new(arr[0].clone()),
423 Box::new(arr[1].clone()),
424 ))
425 }
426 "$not" => Ok(Expr::Not(Box::new(parse_where(v)?))),
427 _ => unreachable!(),
428 }
429}
430
431fn parse_related_block(label: &str, v: &Json) -> Result<RelatedExpr, QueryDocumentError> {
432 let mut rel_type: Option<String> = None;
433 let mut direction = Direction::Both;
434 let mut hops: Option<(u64, u64)> = None;
435 let mut cycle = false;
436 let mut alias = None;
437
438 let mut field_preds: Vec<Expr> = Vec::new();
439 let mut nested_related: Vec<Expr> = Vec::new();
440
441 let items: Vec<(String, Json)> = if v.is_array() {
442 let arr = v.as_array().unwrap();
445 let mut out = Vec::new();
446 for item in arr {
447 if let Some(m) = item.as_object() {
448 out.extend(m.iter().map(|(k, val)| (k.clone(), val.clone())));
449 }
450 }
451 out
452 } else if let Some(m) = v.as_object() {
453 m.iter().map(|(k, val)| (k.clone(), val.clone())).collect()
454 } else {
455 return Err(QueryDocumentError::Parse(
456 "related block must be an object".into(),
457 ));
458 };
459
460 for (k, value) in items {
461 reject_unknown_operators(&k, &value)?;
462 if k == "$alias" {
463 alias = Some(str_of(&value)?);
464 } else if k == "$id" {
465 let id = value
467 .as_str()
468 .ok_or_else(|| QueryDocumentError::Parse("$id requires a string".into()))?;
469 field_preds.push(Expr::Field(FieldPred::Id(id.to_string())));
470 } else if k == "$relation" {
471 if let Some(s) = value.as_str() {
472 rel_type = Some(s.to_string());
473 } else if let Some(m) = value.as_object() {
474 if let Some(t) = m.get("type").and_then(|x| x.as_str()) {
475 rel_type = Some(t.to_string());
476 }
477 if let Some(d) = m.get("direction").and_then(|x| x.as_str()) {
478 direction = parse_direction(d)?;
479 }
480 if let Some(h) = m.get("hops") {
481 hops = Some(parse_hops(h)?);
482 }
483 }
484 } else if k == "$cycle" {
485 cycle = true;
486 let c = parse_cycle(&value)?;
487 rel_type = c.rel_type;
488 direction = c.direction;
489 hops = c.hops;
490 } else if matches!(k.as_str(), "$and" | "$or" | "$nor" | "$xor" | "$not") {
491 field_preds.push(parse_related_logical(&k, &value)?);
493 } else if is_op_key(&k) {
494 return Err(QueryDocumentError::Parse(format!(
495 "operator `{k}` cannot apply to the related block `{label}`; put it on a field"
496 )));
497 } else if value.is_object()
498 && (looks_like_related_block(&value) || looks_like_related_logical_block(&value))
499 {
500 nested_related.push(Expr::Related(parse_related_block(&k, &value)?));
504 } else if value.is_object() && looks_like_operator_block(&value) {
505 field_preds.push(parse_field_ops(&k, &value)?);
506 } else if value.is_object() || value.is_array() {
507 let child = parse_related_block(&k, &value)?;
509 nested_related.push(Expr::Related(child));
510 } else {
511 field_preds.push(Expr::Field(FieldPred::Eq(norm(&k), lit_json(&value))));
512 }
513 }
514
515 let mut filter_parts = field_preds;
516 filter_parts.extend(nested_related);
517 let filter = if filter_parts.is_empty() {
518 None
519 } else if filter_parts.len() == 1 {
520 Some(Box::new(filter_parts.remove(0)))
521 } else {
522 Some(Box::new(Expr::And(filter_parts)))
523 };
524
525 Ok(RelatedExpr {
526 target_labels: vec![label.to_string()],
527 alias,
528 rel_type,
529 direction,
530 hops,
531 cycle,
532 filter,
533 })
534}
535
536fn parse_direction(direction: &str) -> Result<Direction, QueryDocumentError> {
537 match direction {
538 "in" | "incoming" => Ok(Direction::Incoming),
539 "out" | "outgoing" => Ok(Direction::Outgoing),
540 "both" | "any" => Ok(Direction::Both),
541 other => Err(QueryDocumentError::Parse(format!(
542 "unknown relationship direction `{other}` (expected in, out or both)"
543 ))),
544 }
545}
546
547fn parse_cycle(v: &Json) -> Result<RelatedExpr, QueryDocumentError> {
548 let m = v
549 .as_object()
550 .ok_or_else(|| QueryDocumentError::Parse("$cycle requires an object".into()))?;
551 let mut rel_type = None;
552 let mut direction = Direction::Both;
553 if let Some(t) = m.get("type").and_then(|x| x.as_str()) {
554 rel_type = Some(t.to_string());
555 }
556 if let Some(d) = m.get("direction").and_then(|x| x.as_str()) {
557 direction = parse_direction(d)?;
558 }
559 let hops = match m.get("hops") {
560 Some(h) => parse_hops(h)?,
561 None => {
562 return Err(QueryDocumentError::Parse("$cycle requires hops".into()));
563 }
564 };
565 if hops.0 < 2 {
566 return Err(QueryDocumentError::Parse(
567 "$cycle hops.min must be at least 2".into(),
568 ));
569 }
570 Ok(RelatedExpr {
571 target_labels: Vec::new(),
572 alias: None,
573 rel_type,
574 direction,
575 hops: Some(hops),
576 cycle: true,
577 filter: None,
578 })
579}
580
581fn parse_hops(v: &Json) -> Result<(u64, u64), QueryDocumentError> {
583 if let Some(n) = v.as_u64() {
584 if n == 0 {
585 return Err(QueryDocumentError::Parse("hops must be at least 1".into()));
586 }
587 return Ok((n, n));
588 }
589 let m = v
590 .as_object()
591 .ok_or_else(|| QueryDocumentError::Parse("hops must be int or object".into()))?;
592 let min = m
593 .get("min")
594 .and_then(serde_json::Value::as_u64)
595 .unwrap_or(1);
596 let max = m
597 .get("max")
598 .and_then(serde_json::Value::as_u64)
599 .unwrap_or(10);
600 if max < min {
601 return Err(QueryDocumentError::Parse("hops.max < hops.min".into()));
602 }
603 if min == 0 {
604 return Err(QueryDocumentError::Parse(
605 "hops.min must be at least 1".into(),
606 ));
607 }
608 Ok((min, max))
609}
610
611fn comp_int(v: &Json) -> Result<i64, QueryDocumentError> {
612 v.as_i64()
613 .or_else(|| v.as_u64().and_then(|u| i64::try_from(u).ok()))
614 .ok_or_else(|| QueryDocumentError::Parse("expected integer".into()))
615}
616
617fn dt_lit(v: &Json) -> Json {
623 if let Some(m) = v.as_object() {
624 if m.contains_key("$year") {
625 if let Some(ms) = crate::datetime::parse_components(m) {
626 return serde_json::json!(ms);
627 }
628 }
629 }
630 lit_json(v)
631}
632
633#[expect(
634 clippy::cast_precision_loss,
635 reason = "epoch-millisecond bounds are compared as f64 throughout the evaluator, matching its numeric comparison semantics"
636)]
637fn dt_num(v: &Json) -> Result<f64, QueryDocumentError> {
638 if let Some(s) = v.as_str() {
640 if let Some(ms) = crate::datetime::parse_iso_8601(s) {
641 return Ok(ms as f64);
642 }
643 }
644 if let Some(m) = v.as_object() {
645 if let Some(ms) = crate::datetime::parse_components(m) {
646 return Ok(ms as f64);
647 }
648 }
649 num(v)
650}
651
652#[expect(
653 clippy::cast_precision_loss,
654 clippy::cast_possible_truncation,
655 reason = "the casts are paired with an i128 round-trip check; imprecise integer bounds are rejected"
656)]
657fn num(v: &Json) -> Result<f64, QueryDocumentError> {
658 let Some(number) = v.as_number() else {
659 return Err(QueryDocumentError::Parse("expected number".into()));
660 };
661 if let Some(i) = number.as_i64() {
662 let f = i as f64;
663 if (f as i128) == i128::from(i) {
664 return Ok(f);
665 }
666 return Err(QueryDocumentError::Parse(
667 "integer range bound is not exactly representable".into(),
668 ));
669 }
670 if let Some(u) = number.as_u64() {
671 let f = u as f64;
672 if (f as i128) == i128::from(u) {
673 return Ok(f);
674 }
675 return Err(QueryDocumentError::Parse(
676 "unsigned integer range bound is not exactly representable".into(),
677 ));
678 }
679 number
680 .as_f64()
681 .ok_or_else(|| QueryDocumentError::Parse("expected number".into()))
682}
683
684fn str_of(v: &Json) -> Result<String, QueryDocumentError> {
685 v.as_str()
686 .map(std::string::ToString::to_string)
687 .ok_or_else(|| QueryDocumentError::Parse("expected string".into()))
688}
689
690fn array_of(v: &Json) -> Result<Vec<Json>, QueryDocumentError> {
691 v.as_array()
692 .cloned()
693 .ok_or_else(|| QueryDocumentError::Parse("expected array".into()))
694}
695
696fn norm(s: &str) -> String {
697 crate::normalize(s)
698}
699
700pub fn parse_select(
710 select: &Json,
711 aliases: &CollectAliasMap,
712) -> Result<AggregateRequest, QueryDocumentError> {
713 let map = select
714 .as_object()
715 .ok_or_else(|| QueryDocumentError::Parse("select must be an object".into()))?;
716 let mut req = AggregateRequest::default();
717 let mut internal_metric = 0usize;
718 for (name, spec) in map {
719 let out = name.clone();
720 if let Some(ops) = spec.as_object() {
721 for (op, target) in ops {
722 match op.as_str() {
723 "$count" => {
724 let field = select_field(target)?;
725 req.counts.push((out.clone(), field));
726 }
727 "$sum" => {
728 if let Some(expr) = select_math_expr(target)? {
729 req.expr_sums.push((out.clone(), expr));
730 } else {
731 req.sums.push((out.clone(), select_field(target)?));
732 }
733 }
734 "$avg" => {
735 if let Some(expr) = select_math_expr(target)? {
736 req.expr_avgs.push((out.clone(), expr));
737 } else {
738 req.avg_terms.push((out.clone(), select_field(target)?));
739 }
740 if let Some(prec) =
741 ops.get("$precision").and_then(serde_json::Value::as_u64)
742 {
743 let prec = u32::try_from(prec)
744 .unwrap_or(u32::MAX)
745 .min(crate::MAX_AVG_PRECISION);
746 req.avg_precision.insert(out.clone(), prec);
747 }
748 }
749 "$min" => req.mins.push((out.clone(), select_field(target)?)),
750 "$max" => req.maxs.push((out.clone(), select_field(target)?)),
751 "$ref" => req
752 .outputs
753 .push((out.clone(), crate::OutputExpr::Ref(select_field(target)?))),
754 "$subtract" | "$add" | "$multiply" | "$divide" => {
755 let expr =
756 build_derived_math(&mut req, &out, &mut internal_metric, op, target)?;
757 req.outputs.push((out.clone(), expr));
758 }
759 "$timeBucket" => {
760 if let Some(bucket) = target.as_object() {
761 let f = bucket
762 .get("field")
763 .map(select_field)
764 .transpose()?
765 .unwrap_or_default();
766 let unit = crate::TimeUnit::from_name(
767 bucket
768 .get("unit")
769 .and_then(|x| x.as_str())
770 .unwrap_or("month"),
771 )?;
772 let size = bucket
773 .get("size")
774 .and_then(serde_json::Value::as_u64)
775 .unwrap_or(1);
776 if size == 0 {
777 return Err(QueryDocumentError::Parse(
778 "$timeBucket `size` must be at least 1".into(),
779 ));
780 }
781 req.time_bucket = Some(crate::TimeBucketSpec {
782 field: f,
783 unit,
784 size,
785 });
786 req.time_bucket_alias = Some(out.clone());
787 } else {
788 return Err(QueryDocumentError::Parse(
789 "$timeBucket requires an object".into(),
790 ));
791 }
792 }
793 "$collect" => {
794 req.collects.push(parse_collect(&out, target, aliases)?);
795 }
796 "$precision" => {}
797 other => {
798 return Err(QueryDocumentError::Unsupported(format!(
799 "select operator `{other}`"
800 )));
801 }
802 }
803 }
804 } else if let Some(f) = spec.as_str() {
805 let field = select_field(&Json::String(f.to_string()))?;
806 req.plain_fields.push((out, field));
807 }
808 }
809 Ok(req)
810}
811
812fn build_derived_math(
815 req: &mut AggregateRequest,
816 output: &str,
817 counter: &mut usize,
818 op: &str,
819 value: &Json,
820) -> Result<crate::OutputExpr, QueryDocumentError> {
821 let args = value
822 .as_array()
823 .ok_or_else(|| QueryDocumentError::Parse(format!("{op} requires an array")))?;
824 if args.len() != 2 {
825 return Err(QueryDocumentError::Parse(format!(
826 "{op} expects exactly two operands"
827 )));
828 }
829 let mut operands = Vec::with_capacity(2);
830 for arg in args {
831 operands.push(lower_derived_operand(req, output, counter, arg)?);
832 }
833 Ok(match op {
834 "$multiply" => crate::OutputExpr::Multiply(
835 Box::new(operands[0].clone()),
836 Box::new(operands[1].clone()),
837 ),
838 "$divide" => crate::OutputExpr::Divide {
839 dividend: Box::new(operands[0].clone()),
840 divisor: Box::new(operands[1].clone()),
841 precision: 8,
842 },
843 "$add" => {
844 crate::OutputExpr::Add(Box::new(operands[0].clone()), Box::new(operands[1].clone()))
845 }
846 "$subtract" => crate::OutputExpr::Subtract(
847 Box::new(operands[0].clone()),
848 Box::new(operands[1].clone()),
849 ),
850 other => {
851 return Err(QueryDocumentError::Unsupported(format!(
852 "select math `{other}`"
853 )))
854 }
855 })
856}
857
858fn lower_derived_operand(
859 req: &mut AggregateRequest,
860 output: &str,
861 counter: &mut usize,
862 value: &Json,
863) -> Result<crate::OutputExpr, QueryDocumentError> {
864 if value.is_number() {
865 return Ok(crate::OutputExpr::Literal(num(value)?));
868 }
869 if let Some(field) = value.as_str() {
870 return Ok(crate::OutputExpr::Ref(select_field(&Json::String(
871 field.to_string(),
872 ))?));
873 }
874 let ops = value
875 .as_object()
876 .ok_or_else(|| QueryDocumentError::Parse("unsupported derived operand".into()))?;
877 if let Some(reference) = ops.get("$ref") {
878 return Ok(crate::OutputExpr::Ref(select_field(reference)?));
879 }
880 for math in ["$add", "$subtract", "$multiply", "$divide"] {
881 if let Some(args) = ops.get(math) {
882 return build_derived_math(req, output, counter, math, args);
883 }
884 }
885
886 let alias = format!("__{output}_{}", *counter);
887 *counter += 1;
888 if let Some(target) = ops.get("$sum") {
889 if let Some(expr) = select_math_expr(target)? {
890 req.expr_sums.push((alias.clone(), expr));
891 } else {
892 req.sums.push((alias.clone(), select_field(target)?));
893 }
894 } else if let Some(target) = ops.get("$avg") {
895 if let Some(expr) = select_math_expr(target)? {
896 req.expr_avgs.push((alias.clone(), expr));
897 } else {
898 req.avg_terms.push((alias.clone(), select_field(target)?));
899 }
900 if let Some(precision) = ops.get("$precision").and_then(Json::as_u64) {
901 let precision = u32::try_from(precision)
902 .unwrap_or(u32::MAX)
903 .min(crate::MAX_AVG_PRECISION);
904 req.avg_precision.insert(alias.clone(), precision);
905 }
906 } else if let Some(target) = ops.get("$count") {
907 req.counts.push((alias.clone(), select_field(target)?));
908 } else if let Some(target) = ops.get("$min") {
909 req.mins.push((alias.clone(), select_field(target)?));
910 } else if let Some(target) = ops.get("$max") {
911 req.maxs.push((alias.clone(), select_field(target)?));
912 } else {
913 return Err(QueryDocumentError::Parse(
914 "expected aggregate or math operand".into(),
915 ));
916 }
917 req.hidden_outputs.insert(alias.clone());
918 Ok(crate::OutputExpr::Ref(alias))
919}
920
921fn build_math_from_array(op: &str, v: &Json) -> Result<crate::OutputExpr, QueryDocumentError> {
925 let argv = v
926 .as_array()
927 .ok_or_else(|| QueryDocumentError::Parse(format!("{op} requires an array")))?;
928 let operands: Result<Vec<crate::OutputExpr>, QueryDocumentError> = argv
929 .iter()
930 .map(|a| {
931 if let Some(s) = a.as_str() {
932 Ok(crate::OutputExpr::Field(select_field(&Json::String(
933 s.to_string(),
934 ))?))
935 } else if a.is_number() {
936 Ok(crate::OutputExpr::Literal(num(a)?))
940 } else if let Some(m) = a.as_object() {
941 if m.len() == 1 {
943 if let Some(name) = m.get("$ref").and_then(|x| x.as_str()) {
944 return Ok(crate::OutputExpr::Ref(select_field(&Json::String(
945 name.to_string(),
946 ))?));
947 }
948 }
949 select_math_expr(&Json::Object(m.clone()))?
950 .ok_or_else(|| QueryDocumentError::Parse("expected math expression".into()))
951 } else {
952 Err(QueryDocumentError::Parse("unsupported operand".into()))
953 }
954 })
955 .collect();
956 let operands = operands?;
957 if operands.len() != 2 {
958 return Err(QueryDocumentError::Parse(format!(
959 "{op} expects exactly two operands"
960 )));
961 }
962 Ok(match op {
963 "$multiply" => crate::OutputExpr::Multiply(
964 Box::new(operands[0].clone()),
965 Box::new(operands[1].clone()),
966 ),
967 "$divide" => crate::OutputExpr::Divide {
968 dividend: Box::new(operands[0].clone()),
969 divisor: Box::new(operands[1].clone()),
970 precision: 8,
971 },
972 "$add" => {
973 crate::OutputExpr::Add(Box::new(operands[0].clone()), Box::new(operands[1].clone()))
974 }
975 "$subtract" => crate::OutputExpr::Subtract(
976 Box::new(operands[0].clone()),
977 Box::new(operands[1].clone()),
978 ),
979 other => {
980 return Err(QueryDocumentError::Unsupported(format!(
981 "select math `{other}`"
982 )))
983 }
984 })
985}
986
987fn select_math_expr(v: &Json) -> Result<Option<crate::OutputExpr>, QueryDocumentError> {
988 let Some(ops) = v.as_object() else {
989 return Ok(None);
990 };
991 if ops.len() != 1 {
992 return Err(QueryDocumentError::Parse(
993 "math expression must have a single operator".into(),
994 ));
995 }
996 let (op, args) = ops.iter().next().unwrap();
997 args.as_array()
998 .ok_or_else(|| QueryDocumentError::Parse(format!("{op} requires an array")))?;
999 Ok(Some(build_math_from_array(op.as_str(), args)?))
1000}
1001
1002fn select_field(v: &Json) -> Result<String, QueryDocumentError> {
1003 let s = v
1004 .as_str()
1005 .ok_or_else(|| QueryDocumentError::Parse("select target must be a string".into()))?;
1006 let s = s
1010 .trim_start_matches("$record.")
1011 .trim_start_matches("$self.")
1012 .trim_start_matches('$');
1013 if s.is_empty() {
1014 return Err(QueryDocumentError::Parse(
1015 "select target must reference a field (e.g. `$record.name` or `$self.name`)".into(),
1016 ));
1017 }
1018 Ok(s.to_string())
1019}
1020
1021fn projected_order(
1024 order_by: &[(String, bool)],
1025 projection: &[(String, String)],
1026) -> Vec<(String, bool)> {
1027 order_by
1028 .iter()
1029 .map(|(name, desc)| {
1030 let field = projection
1031 .iter()
1032 .find(|(output, _)| norm(output) == norm(name))
1033 .map_or(name, |(_, field)| field);
1034 (field.clone(), *desc)
1035 })
1036 .collect()
1037}
1038
1039fn parse_order_by(v: Option<&Json>) -> Vec<(String, bool)> {
1042 match v {
1043 Some(v) if v.is_string() => vec![(ID_FIELD.into(), v.as_str().unwrap() == "desc")],
1044 Some(v) => v
1045 .as_object()
1046 .map(|m| {
1047 m.iter()
1048 .map(|(k, d)| {
1049 let field = if k == "id" {
1050 ID_FIELD.to_string()
1051 } else {
1052 norm(k)
1053 };
1054 (field, d.as_str().unwrap_or("asc") == "desc")
1055 })
1056 .collect()
1057 })
1058 .unwrap_or_default(),
1059 None => Vec::new(),
1060 }
1061}
1062
1063pub type CollectAliasMap = std::collections::HashMap<String, Vec<crate::TraversalStep>>;
1068
1069pub fn collect_alias_map(expr: &crate::Expr, out: &mut CollectAliasMap) {
1072 fn split_level(
1075 filter: Option<&crate::Expr>,
1076 children: &mut Vec<crate::Expr>,
1077 ) -> Option<crate::Expr> {
1078 let f = filter?;
1079 match f {
1080 crate::Expr::And(items) => {
1081 let mut preds: Vec<crate::Expr> = Vec::new();
1082 for item in items {
1083 match item {
1084 crate::Expr::Related(_) => children.push(item.clone()),
1085 other => preds.push(other.clone()),
1086 }
1087 }
1088 match preds.len() {
1089 0 => None,
1090 1 => Some(preds.remove(0)),
1091 _ => Some(crate::Expr::And(preds)),
1092 }
1093 }
1094 crate::Expr::Related(_) => {
1095 children.push(f.clone());
1096 None
1097 }
1098 other => Some(other.clone()),
1099 }
1100 }
1101
1102 fn walk(expr: &crate::Expr, path: &mut Vec<crate::TraversalStep>, out: &mut CollectAliasMap) {
1103 match expr {
1104 crate::Expr::And(items) | crate::Expr::Or(items) | crate::Expr::Nor(items) => {
1105 for item in items {
1106 walk(item, path, out);
1107 }
1108 }
1109 crate::Expr::Xor(a, b) => {
1110 walk(a, path, out);
1111 walk(b, path, out);
1112 }
1113 crate::Expr::Not(inner) => walk(inner, path, out),
1114 crate::Expr::Related(rel) => {
1115 let mut children: Vec<crate::Expr> = Vec::new();
1116 let filter = split_level(rel.filter.as_deref(), &mut children);
1117 let step = crate::TraversalStep {
1118 target_labels: rel.target_labels.clone(),
1119 rel_type: rel.rel_type.clone(),
1120 direction: rel.direction,
1121 filter,
1122 };
1123 path.push(step);
1124 if let Some(alias) = &rel.alias {
1125 out.insert(alias.trim_start_matches('$').to_string(), path.clone());
1126 }
1127 for child in &children {
1128 walk(child, path, out);
1129 }
1130 path.pop();
1131 }
1132 crate::Expr::Field(_) => {}
1133 }
1134 }
1135 let mut path: Vec<crate::TraversalStep> = Vec::new();
1136 walk(expr, &mut path, out);
1137}
1138
1139fn parse_collect(
1144 out_key: &str,
1145 spec: &Json,
1146 aliases: &CollectAliasMap,
1147) -> Result<crate::CollectSpec, QueryDocumentError> {
1148 let obj = spec
1149 .as_object()
1150 .ok_or_else(|| QueryDocumentError::Parse("$collect requires an object".into()))?;
1151 let from = obj.get("from").and_then(|v| v.as_str());
1152 let label = obj.get("label").and_then(|v| v.as_str());
1153 if from.is_some() && label.is_some() {
1154 return Err(QueryDocumentError::Parse(
1155 "$collect `from` and `label` are mutually exclusive".into(),
1156 ));
1157 }
1158 let mut target_labels: Vec<String> = Vec::new();
1159 let mut rel_type: Option<String> = None;
1160 let mut direction = crate::Direction::Outgoing;
1161 let mut steps: Vec<crate::TraversalStep> = Vec::new();
1162 if let Some(from_alias) = from {
1163 let key = from_alias.trim_start_matches('$');
1164 match aliases.get(key) {
1165 Some(steps_path) => {
1166 steps.clone_from(steps_path);
1167 if let Some(last) = steps.last() {
1168 target_labels.clone_from(&last.target_labels);
1169 rel_type.clone_from(&last.rel_type);
1170 direction = last.direction;
1171 }
1172 }
1173 None => {
1174 return Err(QueryDocumentError::Parse(format!(
1175 "$collect `from` references unknown traversal alias `{from_alias}`"
1176 )))
1177 }
1178 }
1179 } else if let Some(l) = label {
1180 target_labels = vec![l.to_string()];
1181 }
1182
1183 let mut fields: Vec<(String, String)> = Vec::new();
1184 let mut nested: Vec<crate::CollectSpec> = Vec::new();
1185 if let Some(sel) = obj.get("select") {
1186 let smap = sel.as_object().ok_or_else(|| {
1187 QueryDocumentError::Parse("$collect `select` must be an object".into())
1188 })?;
1189 for (key, val) in smap {
1190 if let Some(nested_spec) = val.as_object().and_then(|o| o.get("$collect")) {
1191 nested.push(parse_collect(key, nested_spec, aliases)?);
1192 } else if let Some(field_ref) = val.as_str() {
1193 let field = select_field(&Json::String(field_ref.to_string()))?;
1196 let field = field.split('.').next_back().unwrap_or(&field).to_string();
1197 fields.push((key.clone(), field));
1198 } else {
1199 return Err(QueryDocumentError::Parse(format!(
1200 "$collect `select.{key}` must be a field reference or a nested `$collect`"
1201 )));
1202 }
1203 }
1204 }
1205 let filter = match obj.get("where") {
1206 Some(w) => Some(parse_where(w)?),
1207 None => None,
1208 };
1209 let order_by = parse_order_by(obj.get("orderBy"));
1210 let unique = obj
1211 .get("unique")
1212 .and_then(serde_json::Value::as_bool)
1213 .unwrap_or(true);
1214 let skip = obj
1215 .get("skip")
1216 .and_then(serde_json::Value::as_u64)
1217 .unwrap_or(0);
1218 let limit = obj.get("limit").and_then(serde_json::Value::as_u64);
1219 Ok(crate::CollectSpec {
1220 alias: out_key.to_string(),
1221 rel_type,
1222 direction,
1223 target_labels,
1224 steps,
1225 fields,
1226 nested,
1227 filter,
1228 order_by,
1229 unique,
1230 skip,
1231 limit,
1232 })
1233}
1234
1235pub struct ParsedQuery {
1239 pub search: SearchRequest,
1243 pub aggregate: Option<AggregateRequest>,
1246 pub order_by: Vec<(String, bool)>,
1248}
1249
1250#[expect(
1258 clippy::too_many_lines,
1259 reason = "the top-level lowering keeps listing/aggregate/collect decisions together so the classification rules stay auditable"
1260)]
1261pub fn parse_document(document: &Json) -> Result<ParsedQuery, QueryDocumentError> {
1262 let obj = document
1263 .as_object()
1264 .ok_or_else(|| QueryDocumentError::Parse("query document must be an object".into()))?;
1265
1266 let labels: Vec<String> = match obj.get("labels") {
1267 None => Vec::new(),
1268 Some(value) => value
1269 .as_array()
1270 .ok_or_else(|| QueryDocumentError::Parse("labels must be an array".into()))?
1271 .iter()
1272 .map(|value| {
1273 value.as_str().map(String::from).ok_or_else(|| {
1274 QueryDocumentError::Parse("labels must contain only strings".into())
1275 })
1276 })
1277 .collect::<Result<_, _>>()?,
1278 };
1279
1280 let where_expr = match obj.get("where") {
1285 None | Some(serde_json::Value::Null) => None,
1286 Some(serde_json::Value::Object(map)) if map.is_empty() => None,
1287 Some(w) => Some(parse_where(w)?),
1288 };
1289
1290 let limit = match obj.get("limit") {
1291 None => None,
1292 Some(value) if value.is_null() => None,
1293 Some(value) => Some(value.as_u64().ok_or_else(|| {
1294 QueryDocumentError::Parse("limit must be a nonnegative integer".into())
1295 })?),
1296 };
1297 let skip = match obj.get("skip") {
1298 None => 0,
1299 Some(value) if value.is_null() => 0,
1300 Some(value) => value.as_u64().ok_or_else(|| {
1301 QueryDocumentError::Parse("skip must be a nonnegative integer".into())
1302 })?,
1303 };
1304
1305 let order_by = parse_order_by(obj.get("orderBy"));
1307
1308 let select = obj.get("select").cloned();
1309 let group_by: Vec<String> = obj
1310 .get("groupBy")
1311 .and_then(|x| x.as_array())
1312 .map(|a| {
1313 a.iter()
1314 .filter_map(|v| v.as_str())
1315 .map(|s| {
1316 s.trim_start_matches("$record.")
1317 .trim_start_matches('$')
1318 .to_string()
1319 })
1320 .collect()
1321 })
1322 .unwrap_or_default();
1323
1324 let mut aliases = CollectAliasMap::new();
1331 if let Some(where_expr) = &where_expr {
1332 collect_alias_map(where_expr, &mut aliases);
1333 }
1334 let mut collects: Vec<crate::CollectSpec> = Vec::new();
1335 let mut projection: Vec<(String, String)> = Vec::new();
1336 let mut listing = group_by.is_empty();
1337 match select.as_ref().map(|sel| sel.as_object()) {
1338 Some(Some(map)) => {
1339 listing &= !map.is_empty();
1340 for (key, val) in map {
1341 if let Some(collect) = val.as_object().and_then(|o| o.get("$collect")) {
1342 collects.push(parse_collect(key, collect, &aliases)?);
1343 } else if val.is_string() {
1344 projection.push((key.clone(), select_field(val)?));
1345 } else {
1346 listing = false;
1347 }
1348 }
1349 }
1350 Some(None) => listing = false,
1351 None => {}
1352 }
1353 let is_aggregate = (select.is_some() && !listing) || !group_by.is_empty();
1354
1355 let search = SearchRequest {
1356 labels: labels.clone(),
1357 where_: where_expr,
1358 related: Vec::new(),
1359 collects: collects.clone(),
1360 pagination: Pagination { skip, limit },
1361 order_by: if is_aggregate {
1363 order_by.clone()
1364 } else {
1365 projected_order(&order_by, &projection)
1366 },
1367 projection: if is_aggregate { Vec::new() } else { projection },
1368 force_scan: false,
1369 };
1370
1371 if is_aggregate {
1372 let mut agg = match &select {
1373 Some(sel) => parse_select(sel, &aliases)?,
1374 None => AggregateRequest::default(),
1375 };
1376 agg.labels = labels;
1377 agg.where_.clone_from(&search.where_);
1378 agg.group_by.clone_from(&group_by);
1379 agg.order_by.clone_from(&order_by);
1380 agg.skip = skip;
1381 agg.limit = limit;
1382 agg.collects = collects;
1384 if !group_by.is_empty() {
1387 let select_names: Vec<String> = select
1388 .as_ref()
1389 .and_then(|s| s.as_object())
1390 .map(|m| m.keys().map(|k| norm(k)).collect())
1391 .unwrap_or_default();
1392 let all_self = group_by.iter().all(|g| select_names.contains(&norm(g)));
1393 if all_self {
1394 agg.group_by.clear(); }
1396 }
1397 return Ok(ParsedQuery {
1398 search,
1399 aggregate: Some(agg),
1400 order_by,
1401 });
1402 }
1403
1404 Ok(ParsedQuery {
1405 search,
1406 aggregate: None,
1407 order_by,
1408 })
1409}