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 parse_order_by(v: Option<&Json>) -> Vec<(String, bool)> {
1024 match v {
1025 Some(v) if v.is_string() => vec![(ID_FIELD.into(), v.as_str().unwrap() == "desc")],
1026 Some(v) => v
1027 .as_object()
1028 .map(|m| {
1029 m.iter()
1030 .map(|(k, d)| {
1031 let field = if k == "id" {
1032 ID_FIELD.to_string()
1033 } else {
1034 norm(k)
1035 };
1036 (field, d.as_str().unwrap_or("asc") == "desc")
1037 })
1038 .collect()
1039 })
1040 .unwrap_or_default(),
1041 None => Vec::new(),
1042 }
1043}
1044
1045pub type CollectAliasMap = std::collections::HashMap<String, Vec<crate::TraversalStep>>;
1050
1051pub fn collect_alias_map(expr: &crate::Expr, out: &mut CollectAliasMap) {
1054 fn split_level(
1057 filter: Option<&crate::Expr>,
1058 children: &mut Vec<crate::Expr>,
1059 ) -> Option<crate::Expr> {
1060 let f = filter?;
1061 match f {
1062 crate::Expr::And(items) => {
1063 let mut preds: Vec<crate::Expr> = Vec::new();
1064 for item in items {
1065 match item {
1066 crate::Expr::Related(_) => children.push(item.clone()),
1067 other => preds.push(other.clone()),
1068 }
1069 }
1070 match preds.len() {
1071 0 => None,
1072 1 => Some(preds.remove(0)),
1073 _ => Some(crate::Expr::And(preds)),
1074 }
1075 }
1076 crate::Expr::Related(_) => {
1077 children.push(f.clone());
1078 None
1079 }
1080 other => Some(other.clone()),
1081 }
1082 }
1083
1084 fn walk(expr: &crate::Expr, path: &mut Vec<crate::TraversalStep>, out: &mut CollectAliasMap) {
1085 match expr {
1086 crate::Expr::And(items) | crate::Expr::Or(items) | crate::Expr::Nor(items) => {
1087 for item in items {
1088 walk(item, path, out);
1089 }
1090 }
1091 crate::Expr::Xor(a, b) => {
1092 walk(a, path, out);
1093 walk(b, path, out);
1094 }
1095 crate::Expr::Not(inner) => walk(inner, path, out),
1096 crate::Expr::Related(rel) => {
1097 let mut children: Vec<crate::Expr> = Vec::new();
1098 let filter = split_level(rel.filter.as_deref(), &mut children);
1099 let step = crate::TraversalStep {
1100 target_labels: rel.target_labels.clone(),
1101 rel_type: rel.rel_type.clone(),
1102 direction: rel.direction,
1103 filter,
1104 };
1105 path.push(step);
1106 if let Some(alias) = &rel.alias {
1107 out.insert(alias.trim_start_matches('$').to_string(), path.clone());
1108 }
1109 for child in &children {
1110 walk(child, path, out);
1111 }
1112 path.pop();
1113 }
1114 crate::Expr::Field(_) => {}
1115 }
1116 }
1117 let mut path: Vec<crate::TraversalStep> = Vec::new();
1118 walk(expr, &mut path, out);
1119}
1120
1121fn parse_collect(
1126 out_key: &str,
1127 spec: &Json,
1128 aliases: &CollectAliasMap,
1129) -> Result<crate::CollectSpec, QueryDocumentError> {
1130 let obj = spec
1131 .as_object()
1132 .ok_or_else(|| QueryDocumentError::Parse("$collect requires an object".into()))?;
1133 let from = obj.get("from").and_then(|v| v.as_str());
1134 let label = obj.get("label").and_then(|v| v.as_str());
1135 if from.is_some() && label.is_some() {
1136 return Err(QueryDocumentError::Parse(
1137 "$collect `from` and `label` are mutually exclusive".into(),
1138 ));
1139 }
1140 let mut target_labels: Vec<String> = Vec::new();
1141 let mut rel_type: Option<String> = None;
1142 let mut direction = crate::Direction::Outgoing;
1143 let mut steps: Vec<crate::TraversalStep> = Vec::new();
1144 if let Some(from_alias) = from {
1145 let key = from_alias.trim_start_matches('$');
1146 match aliases.get(key) {
1147 Some(steps_path) => {
1148 steps.clone_from(steps_path);
1149 if let Some(last) = steps.last() {
1150 target_labels.clone_from(&last.target_labels);
1151 rel_type.clone_from(&last.rel_type);
1152 direction = last.direction;
1153 }
1154 }
1155 None => {
1156 return Err(QueryDocumentError::Parse(format!(
1157 "$collect `from` references unknown traversal alias `{from_alias}`"
1158 )))
1159 }
1160 }
1161 } else if let Some(l) = label {
1162 target_labels = vec![l.to_string()];
1163 }
1164
1165 let mut fields: Vec<(String, String)> = Vec::new();
1166 let mut nested: Vec<crate::CollectSpec> = Vec::new();
1167 if let Some(sel) = obj.get("select") {
1168 let smap = sel.as_object().ok_or_else(|| {
1169 QueryDocumentError::Parse("$collect `select` must be an object".into())
1170 })?;
1171 for (key, val) in smap {
1172 if let Some(nested_spec) = val.as_object().and_then(|o| o.get("$collect")) {
1173 nested.push(parse_collect(key, nested_spec, aliases)?);
1174 } else if let Some(field_ref) = val.as_str() {
1175 let field = select_field(&Json::String(field_ref.to_string()))?;
1178 let field = field.split('.').next_back().unwrap_or(&field).to_string();
1179 fields.push((key.clone(), field));
1180 } else {
1181 return Err(QueryDocumentError::Parse(format!(
1182 "$collect `select.{key}` must be a field reference or a nested `$collect`"
1183 )));
1184 }
1185 }
1186 }
1187 let filter = match obj.get("where") {
1188 Some(w) => Some(parse_where(w)?),
1189 None => None,
1190 };
1191 let order_by = parse_order_by(obj.get("orderBy"));
1192 let unique = obj
1193 .get("unique")
1194 .and_then(serde_json::Value::as_bool)
1195 .unwrap_or(true);
1196 let skip = obj
1197 .get("skip")
1198 .and_then(serde_json::Value::as_u64)
1199 .unwrap_or(0);
1200 let limit = obj.get("limit").and_then(serde_json::Value::as_u64);
1201 Ok(crate::CollectSpec {
1202 alias: out_key.to_string(),
1203 rel_type,
1204 direction,
1205 target_labels,
1206 steps,
1207 fields,
1208 nested,
1209 filter,
1210 order_by,
1211 unique,
1212 skip,
1213 limit,
1214 })
1215}
1216
1217pub struct ParsedQuery {
1221 pub search: SearchRequest,
1223 pub aggregate: Option<AggregateRequest>,
1225 pub order_by: Vec<(String, bool)>,
1227}
1228
1229#[expect(
1237 clippy::too_many_lines,
1238 reason = "the top-level lowering keeps listing/aggregate/collect decisions together so the classification rules stay auditable"
1239)]
1240pub fn parse_document(document: &Json) -> Result<ParsedQuery, QueryDocumentError> {
1241 let obj = document
1242 .as_object()
1243 .ok_or_else(|| QueryDocumentError::Parse("query document must be an object".into()))?;
1244
1245 let labels: Vec<String> = match obj.get("labels") {
1246 None => Vec::new(),
1247 Some(value) => value
1248 .as_array()
1249 .ok_or_else(|| QueryDocumentError::Parse("labels must be an array".into()))?
1250 .iter()
1251 .map(|value| {
1252 value.as_str().map(String::from).ok_or_else(|| {
1253 QueryDocumentError::Parse("labels must contain only strings".into())
1254 })
1255 })
1256 .collect::<Result<_, _>>()?,
1257 };
1258
1259 let where_expr = match obj.get("where") {
1264 None | Some(serde_json::Value::Null) => None,
1265 Some(serde_json::Value::Object(map)) if map.is_empty() => None,
1266 Some(w) => Some(parse_where(w)?),
1267 };
1268
1269 let limit = match obj.get("limit") {
1270 None => None,
1271 Some(value) if value.is_null() => None,
1272 Some(value) => Some(value.as_u64().ok_or_else(|| {
1273 QueryDocumentError::Parse("limit must be a nonnegative integer".into())
1274 })?),
1275 };
1276 let skip = match obj.get("skip") {
1277 None => 0,
1278 Some(value) if value.is_null() => 0,
1279 Some(value) => value.as_u64().ok_or_else(|| {
1280 QueryDocumentError::Parse("skip must be a nonnegative integer".into())
1281 })?,
1282 };
1283
1284 let order_by = parse_order_by(obj.get("orderBy"));
1286
1287 let select = obj.get("select").cloned();
1289 let group_by: Vec<String> = obj
1290 .get("groupBy")
1291 .and_then(|x| x.as_array())
1292 .map(|a| {
1293 a.iter()
1294 .filter_map(|v| v.as_str())
1295 .map(|s| {
1296 s.trim_start_matches("$record.")
1297 .trim_start_matches('$')
1298 .to_string()
1299 })
1300 .collect()
1301 })
1302 .unwrap_or_default();
1303
1304 let mut aliases = CollectAliasMap::new();
1308 if let Some(where_expr) = &where_expr {
1309 collect_alias_map(where_expr, &mut aliases);
1310 }
1311 let mut collects: Vec<crate::CollectSpec> = Vec::new();
1312 let collect_only = select.as_ref().is_some_and(|sel| {
1313 sel.as_object().is_some_and(|m| {
1314 !m.is_empty()
1315 && m.values()
1316 .all(|v| v.as_object().and_then(|o| o.get("$collect")).is_some())
1317 })
1318 });
1319 if let Some(sel) = &select {
1320 if let Some(map) = sel.as_object() {
1321 for (key, val) in map {
1322 if let Some(collect) = val.as_object().and_then(|o| o.get("$collect")) {
1323 collects.push(parse_collect(key, collect, &aliases)?);
1324 }
1325 }
1326 }
1327 }
1328
1329 let is_aggregate = (select.is_some() && !collect_only) || !group_by.is_empty();
1330
1331 let search = SearchRequest {
1332 labels: labels.clone(),
1333 where_: where_expr,
1334 related: Vec::new(),
1335 collects: collects.clone(),
1336 pagination: Pagination { skip, limit },
1337 order_by: order_by.clone(),
1338 force_scan: false,
1339 };
1340
1341 if is_aggregate {
1342 let mut agg = match &select {
1343 Some(sel) => parse_select(sel, &aliases)?,
1344 None => AggregateRequest::default(),
1345 };
1346 agg.labels = labels;
1347 agg.where_.clone_from(&search.where_);
1348 agg.group_by.clone_from(&group_by);
1349 agg.order_by.clone_from(&order_by);
1350 agg.skip = skip;
1351 agg.limit = limit;
1352 agg.collects = collects;
1354 if !group_by.is_empty() {
1357 let select_names: Vec<String> = select
1358 .as_ref()
1359 .and_then(|s| s.as_object())
1360 .map(|m| m.keys().map(|k| norm(k)).collect())
1361 .unwrap_or_default();
1362 let all_self = group_by.iter().all(|g| select_names.contains(&norm(g)));
1363 if all_self {
1364 agg.group_by.clear(); }
1366 }
1367 return Ok(ParsedQuery {
1368 search,
1369 aggregate: Some(agg),
1370 order_by,
1371 });
1372 }
1373
1374 Ok(ParsedQuery {
1375 search,
1376 aggregate: None,
1377 order_by,
1378 })
1379}