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