Skip to main content

rushdb_query/
document.rs

1//! Native `RushDB` query-document parser.
2//!
3//! Accepts the database query-document shape and lowers it onto the
4//! shared `Expr`/`RelatedExpr` execution model:
5//!
6//! ```json
7//! { "labels": ["..."], "where": { "status": {"$in": [...]} },
8//!   "select": {...}, "groupBy": [...], "orderBy": {...}, "limit": 10, "skip": 0 }
9//! ```
10//!
11//! `where` supports flat-key implicit `$and`, `$eq/$ne/$gt/$gte/$lt/$lte/
12//! $in/$nin/$contains/$startsWith/$endsWith/$exists/$type`, logical
13//! `$and/$or/$nor/$xor/$not`, `$id`, label-key relationship blocks with
14//! `$alias`, `$relation` (type + direction + `hops`), `$cycle`, and nested
15//! multi-level blocks.
16
17use rushdb_types::value::ID_FIELD;
18use serde_json::Value as Json;
19
20use crate::{AggregateRequest, Direction, Expr, FieldPred, Pagination, RelatedExpr, SearchRequest};
21
22/// Errors produced while lowering a query document.
23#[derive(Debug, thiserror::Error)]
24pub enum QueryDocumentError {
25    /// The document is malformed for a recognized construct.
26    #[error("invalid where clause: {0}")]
27    Parse(String),
28    /// The document uses a construct this parser does not support.
29    #[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
71/// A `$`-prefixed key that is not an operator is a typo (`$regx`), never a
72/// label or field name: refuse it instead of matching nothing.
73fn 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
94/// Parse a documented `where` object into a root `Expr` (with `Related`
95/// nodes for traversal blocks). Flat keys become implicit `$and` conjuncts.
96///
97/// # Errors
98///
99/// Returns [`QueryDocumentError::Parse`] when the clause shape is invalid and
100/// [`QueryDocumentError::Unsupported`] for unsupported operator placements.
101pub 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            // label-key relationship block (with $relation/$alias/$cycle)
128            parts.push(Expr::Related(parse_related_block(k, v)?));
129        } else if v.is_object() && looks_like_related_logical_block(v) {
130            // `{ Label: { $or: [ {field:...}, ... ] } }` — logical combo over a
131            // related block (operands are field-predicate objects)
132            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            // array of primitives under a field key: element-match ($in)
137            // semantics — `{ tags: ["a", "b"] }` matches any element, the same
138            // way the baseline solution wraps single values into IN predicates.
139            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            // label-key relationship block (or array of blocks under a label)
145            parts.push(Expr::Related(parse_related_block(k, v)?));
146        } else {
147            // flat equality: { field: value }
148            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
178/// A field operator block: `{ field: { $gte: 1, $lt: 5 } }` is an implicit
179/// $and over each operator; multiple operators compose.
180fn 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
218/// Logical conditions scoped to a field: each array element is either an
219/// operator-only object (applied to `field`) or a full condition object.
220fn 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    // related-block only when an operand is a field-predicate object (has a
259    // non-operator key); a `$or` of pure operator blocks is field-value logic
260    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
286/// True when `v` is a JSON array whose elements are all scalars (no nested
287/// objects/arrays). Used to disambiguate `{ field: [v1, v2] }` element-match
288/// semantics from a label-key array of relationship blocks.
289fn 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        // { $id: { $in: [...] } } form
379        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
398/// `{ LABEL: { $alias, $relation, ...props, NESTED_LABEL: {...} } }`
399/// `$and`/`$or`/`$nor`/`$not` over predicates on a related record. Each
400/// element lowers through `parse_where` so both field predicates and nested
401/// related blocks are supported.
402fn 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        // single label can map to an array of blocks; treat each as the
443        // same label block (union semantics approximated by And-of-first)
444        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            // filter the related record by its identity
466            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            // logical combination of predicates on the related record
492            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 label block with `$relation`/`$alias`, or a logical
501            // combination over one (`{ CONCEPT: { $or: [{ name: … }] } }`),
502            // decided as `parse_where` decides it at the root (Q-1)
503            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            // nested label block → child RelatedExpr
508            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
581/// `hops: 3` → (3,3); `hops: {min, max}` → bounds (min defaults 1).
582fn 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
617/// Datetime-aware literal for equality: ISO strings / component objects
618/// lower to epoch-ms numbers (matching the engine `DateTime` type).
619/// An equality literal. ISO-8601 text stays text: it equals a string holding
620/// that text and a datetime at that instant, which the index probes and the
621/// evaluator both resolve. A component object names one instant.
622fn 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    // numeric compares accept ISO strings / component objects as epoch ms
639    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
700/// Parse a `select` object into aggregate request metrics.
701/// select value forms: `"$record.field"`, `"$alias"`, `"$alias.field"`,
702/// `{"$count": "*" | "$alias" | "$alias.field"}`, `{"$sum": "$record.field"}`,
703/// `{"$avg": ...}`, `{"$min": ...}`, `{"$max": ...}`.
704///
705/// # Errors
706///
707/// Returns [`QueryDocumentError::Parse`] for malformed select/collect shapes
708/// and [`QueryDocumentError::Unsupported`] for unknown select operators.
709pub 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
812/// Lower post-aggregation arithmetic. Inline aggregate operands are assigned
813/// private aliases and materialized before the derived expression is evaluated.
814fn 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        // Preserve the query language's exact integer-bound rule here too;
866        // `Json::as_f64` would silently round integers above 2^53.
867        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
921/// Parse a nested math expression used as an aggregate target, e.g.
922/// `{ $sum: { $multiply: ['$record.price', '$record.qty'] } }`.
923/// Build a math expression from an operand array for a given operator.
924fn 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                // Keep nested aggregate math on the same exact numeric gate as
937                // top-level derived expressions; direct `as_f64` would round
938                // integers above 2^53 before planning.
939                Ok(crate::OutputExpr::Literal(num(a)?))
940            } else if let Some(m) = a.as_object() {
941                // { $ref: 'name' } operand → derived-output reference
942                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    // `$record.field` and `$self.field` resolve to `field` (the current /
1007    // root record); `$alias.field` keeps the dotted alias path (used by
1008    // count/sum over related aliases).
1009    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
1021/// Parse an `orderBy` value (`"asc"|"desc"` or `{field:"asc"}`) into
1022/// `(normalized_field, descending)` pairs.
1023fn 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
1045/// Alias → full traversal path (each level's labels/rel-type/direction plus
1046/// that level's `where` predicates), resolved from the `where` clause so
1047/// alias-based `$collect.from` reuses a declared traversal alias. Multi-hop
1048/// aliases keep every step; alias `$count`/`$sum` share the same shape.
1049pub type CollectAliasMap = std::collections::HashMap<String, Vec<crate::TraversalStep>>;
1050
1051/// Collect every traversal alias (and nested aliases) reachable in a `where`
1052/// expression, recording the full path from the root to each alias level.
1053pub fn collect_alias_map(expr: &crate::Expr, out: &mut CollectAliasMap) {
1054    /// Split a level's filter into endpoint predicates and nested label-block
1055    /// children (each child becomes a further path step).
1056    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
1121/// Parse a `$collect` expression into a `CollectSpec`. `from` (alias-based)
1122/// resolves the traversal from `aliases`; `label` (label-based) traverses
1123/// inline with an optional `where` filter, output-key→field projection map,
1124/// and nested `$collect`s for hierarchical output.
1125fn 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                // A collect field reference reads the collected record's own
1176                // field: strip `$alias.` / `$self.` / `$record.` prefixes.
1177                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
1217/// Full query document lowered onto the execution model. Also exposes whether
1218/// the query is a listing (no groupBy) or aggregate (groupBy or select with
1219/// metrics).
1220pub struct ParsedQuery {
1221    /// Records-perspective request built from `labels`/`where`/pagination.
1222    pub search: SearchRequest,
1223    /// Aggregate request when the document carries a `select` or `groupBy`.
1224    pub aggregate: Option<AggregateRequest>,
1225    /// Parsed `orderBy` pairs shared by the listing and aggregate paths.
1226    pub order_by: Vec<(String, bool)>,
1227}
1228
1229/// Parse a complete query document into its execution model.
1230///
1231/// # Errors
1232///
1233/// Returns [`QueryDocumentError::Parse`] when the document shape is invalid
1234/// and [`QueryDocumentError::Unsupported`] when it uses an unsupported
1235/// construct.
1236#[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    // An absent, `null` or empty `where` is "no filter". It must parse to
1260    // `None`, not to an empty expression: the O(1) catalog paths (`$count`,
1261    // listing `total`) only apply to an unfiltered request (C8,
1262    // cloud-capacity F9).
1263    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    // orderBy: "asc"|"desc" | {field:"asc"}
1285    let order_by = parse_order_by(obj.get("orderBy"));
1286
1287    // select + groupBy → aggregate; else listing
1288    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    // `$collect` is a per-record output shape: extract every `$collect` from
1305    // `select` onto the search request so the listing path attaches them, and
1306    // treat a select made only of `$collect`s as a listing (per-record rows).
1307    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        // per-root `$collect` shapes attach to each aggregate row.
1353        agg.collects = collects;
1354        // self-group detection: groupBy entries equal select output names (compared
1355        // normalized — groupBy uses the select key names, which may be camelCase)
1356        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(); // single global group
1365            }
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}