1use crate::ast::*;
2use crate::parser::{parse, ParseError};
3use crate::plan::*;
4use powdb_storage::stored_json_path::StoredJsonPathV1;
5
6#[derive(Debug, Clone, PartialEq, Eq)]
7pub(crate) enum RangeTarget {
8 Column(String),
9 JsonPath(StoredJsonPathV1),
10}
11
12pub(crate) type RangeBound = (RangeTarget, Option<(Expr, bool)>, Option<(Expr, bool)>);
14
15#[derive(Debug)]
17pub enum PlanError {
18 Parse(ParseError),
20 Semantic(String),
22}
23
24impl PlanError {
25 pub fn message(&self) -> String {
27 self.to_string()
28 }
29}
30
31impl std::fmt::Display for PlanError {
32 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33 match self {
34 Self::Parse(e) => write!(f, "{e}"),
35 Self::Semantic(message) => write!(f, "{message}"),
36 }
37 }
38}
39
40impl std::error::Error for PlanError {
41 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
42 match self {
43 Self::Parse(e) => Some(e),
44 Self::Semantic(_) => None,
45 }
46 }
47}
48
49impl From<ParseError> for PlanError {
50 fn from(e: ParseError) -> Self {
51 PlanError::Parse(e)
52 }
53}
54
55pub fn plan(input: &str) -> Result<PlanNode, PlanError> {
56 let stmt = parse(input)?;
57 plan_statement(stmt)
58}
59
60pub fn plan_statement(stmt: Statement) -> Result<PlanNode, PlanError> {
61 match stmt {
62 Statement::Query(q) => plan_query(q),
63 Statement::Insert(ins) => plan_insert(ins),
64 Statement::UpdateQuery(upd) => plan_update(upd),
65 Statement::DeleteQuery(del) => plan_delete(del),
66 Statement::CreateType(ct) => plan_create_type(ct),
67 Statement::CreateLink(cl) => Ok(PlanNode::CreateLink {
68 owner: cl.owner,
69 name: cl.name,
70 target: cl.target,
71 local_key: cl.local_key,
72 target_key: cl.target_key,
73 }),
74 Statement::DropLink(dl) => Ok(PlanNode::DropLink {
75 owner: dl.owner,
76 name: dl.name,
77 if_exists: dl.if_exists,
78 }),
79 Statement::AlterTable(at) => Ok(PlanNode::AlterTable {
80 table: at.table,
81 action: at.action,
82 }),
83 Statement::DropTable(dt) => Ok(PlanNode::DropTable {
84 name: dt.table,
85 if_exists: dt.if_exists,
86 }),
87 Statement::CreateView(cv) => Ok(PlanNode::CreateView {
88 name: cv.name,
89 query_text: cv.query_text,
90 }),
91 Statement::RefreshView(rv) => Ok(PlanNode::RefreshView { name: rv.name }),
92 Statement::DropView(dv) => Ok(PlanNode::DropView {
93 name: dv.name,
94 if_exists: dv.if_exists,
95 }),
96 Statement::ListTypes => Ok(PlanNode::ListTypes),
97 Statement::Describe(table) => Ok(PlanNode::Describe { table }),
98 Statement::ListLinks => Ok(PlanNode::ListLinks),
99 Statement::Union(u) => {
100 let left = plan_statement(*u.left)?;
101 let right = plan_statement(*u.right)?;
102 Ok(PlanNode::Union {
103 left: Box::new(left),
104 right: Box::new(right),
105 all: u.all,
106 })
107 }
108 Statement::Upsert(ups) => plan_upsert(ups),
109 Statement::Begin => Ok(PlanNode::Begin),
110 Statement::Commit => Ok(PlanNode::Commit),
111 Statement::Rollback => Ok(PlanNode::Rollback),
112 Statement::Explain(inner) => {
113 let inner_plan = plan_statement(*inner)?;
114 Ok(PlanNode::Explain {
115 input: Box::new(inner_plan),
116 })
117 }
118 }
119}
120
121fn plan_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
122 if let Some(agg) = q.aggregation.as_ref() {
127 if matches!(
128 agg.argument,
129 Some(Expr::LinkPath { .. }) | Some(Expr::NestedQuery(_))
130 ) {
131 return Err(PlanError::Semantic(
132 "aggregates over a nested or link projection are not supported: \
133 aggregate a plain column instead (e.g. `sum(Order { .total })`)"
134 .into(),
135 ));
136 }
137 }
138 if q.projection.as_ref().is_some_and(|proj| {
141 proj.iter()
142 .any(|pf| matches!(pf.expr, Expr::NestedQuery(_) | Expr::LinkPath { .. }))
143 }) {
144 return plan_nested_query(q);
145 }
146 if !q.joins.is_empty() {
152 return plan_joined_query(q);
153 }
154 let visible = q.alias.clone().unwrap_or_else(|| q.source.clone());
159 if let Some(filter) = q.filter.as_mut() {
160 resolve_scan_qualifiers(filter, &visible)?;
161 }
162 if let Some(proj) = q.projection.as_mut() {
163 for pf in proj.iter_mut() {
164 resolve_scan_qualifiers(&mut pf.expr, &visible)?;
165 }
166 }
167 if let Some(order) = q.order.as_mut() {
168 for key in order.keys.iter_mut() {
169 resolve_scan_qualifiers(&mut key.expr, &visible)?;
170 }
171 }
172 if let Some(group) = q.group_by.as_mut() {
173 for key in group.keys.iter_mut() {
174 resolve_scan_qualifiers(&mut key.expr, &visible)?;
175 }
176 if let Some(having) = group.having.as_mut() {
177 resolve_scan_qualifiers(having, &visible)?;
178 }
179 }
180 if let Some(agg) = q.aggregation.as_mut() {
181 if let Some(arg) = agg.argument.as_mut() {
182 resolve_scan_qualifiers(arg, &visible)?;
183 }
184 }
185 let source_aliases = std::collections::HashSet::from([q.source.clone()]);
186 let ordered_expr_scan = try_extract_ordered_expr_index_scan(&q);
198 let (source, filter) = if let Some(scan) = ordered_expr_scan {
199 q.order = None;
203 q.limit = None;
204 q.offset = None;
205 (scan, None)
206 } else {
207 match q.filter {
208 Some(pred) => match try_extract_eq_index_key(&q.source, &pred) {
209 Some(index_scan) => (index_scan, None),
210 None => match try_extract_range_index_keys(&q.source, &pred) {
211 Some(range_scan) => (range_scan, None),
212 None => (
213 PlanNode::SeqScan {
214 table: q.source.clone(),
215 },
216 Some(pred),
217 ),
218 },
219 },
220 None => (
221 PlanNode::SeqScan {
222 table: q.source.clone(),
223 },
224 None,
225 ),
226 }
227 };
228 let mut node = source;
229
230 if let Some(pred) = filter {
231 node = PlanNode::Filter {
232 input: Box::new(node),
233 predicate: pred,
234 };
235 }
236
237 if let Some(group) = q.group_by {
240 let mut grouped_order = q.order;
241 let mut proj_fields: Vec<ProjectField> = q
242 .projection
243 .map(|proj| {
244 proj.into_iter()
245 .map(|pf| ProjectField {
246 alias: pf.alias,
247 expr: pf.expr,
248 })
249 .collect()
250 })
251 .unwrap_or_default();
252 let mut having = group.having;
253 let aggregates = extract_aggregates(&mut proj_fields, &mut having, &source_aliases)?;
254 rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
255 rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
256
257 node = PlanNode::GroupBy {
258 input: Box::new(node),
259 keys: group.keys,
260 aggregates,
261 having,
262 };
263
264 if !proj_fields.is_empty() {
265 node = PlanNode::Project {
266 input: Box::new(node),
267 fields: proj_fields,
268 };
269 }
270
271 if q.distinct {
274 node = PlanNode::Distinct {
275 input: Box::new(node),
276 };
277 }
278
279 if let Some(order) = grouped_order {
280 node = PlanNode::Sort {
281 input: Box::new(node),
282 keys: order
283 .keys
284 .into_iter()
285 .map(|k| SortKey {
286 expr: k.expr,
287 descending: k.descending,
288 })
289 .collect(),
290 };
291 }
292 return Ok(slice_layer(node, q.offset, q.limit));
293 }
294
295 if let Some(order) = q.order {
296 node = PlanNode::Sort {
297 input: Box::new(node),
298 keys: order
299 .keys
300 .into_iter()
301 .map(|k| SortKey {
302 expr: k.expr,
303 descending: k.descending,
304 })
305 .collect(),
306 };
307 }
308
309 node = projected_tail(node, q.projection, q.distinct, q.offset, q.limit);
310
311 if let Some(agg) = q.aggregation {
312 let provenance_alias = symmetric_provenance_alias(
313 agg.function,
314 agg.argument.as_ref(),
315 agg.mode,
316 &source_aliases,
317 )?;
318 node = PlanNode::Aggregate {
319 input: Box::new(node),
320 function: agg.function,
321 argument: agg.argument,
322 mode: agg.mode,
323 provenance_alias,
324 };
325 }
326
327 Ok(node)
328}
329
330fn project_layer(input: PlanNode, projection: Vec<ProjectionField>) -> PlanNode {
334 let mut fields: Vec<ProjectField> = projection
335 .into_iter()
336 .map(|pf| ProjectField {
337 alias: pf.alias,
338 expr: pf.expr,
339 })
340 .collect();
341 let windows = extract_windows(&mut fields);
342 let input = if windows.is_empty() {
343 input
344 } else {
345 PlanNode::Window {
346 input: Box::new(input),
347 windows,
348 }
349 };
350 PlanNode::Project {
351 input: Box::new(input),
352 fields,
353 }
354}
355
356fn slice_layer(mut input: PlanNode, offset: Option<Expr>, limit: Option<Expr>) -> PlanNode {
361 if let Some(count) = offset {
362 input = PlanNode::Offset {
363 input: Box::new(input),
364 count,
365 };
366 }
367 if let Some(count) = limit {
368 input = PlanNode::Limit {
369 input: Box::new(input),
370 count,
371 };
372 }
373 input
374}
375
376fn projected_tail(
390 input: PlanNode,
391 projection: Option<Vec<ProjectionField>>,
392 distinct: bool,
393 offset: Option<Expr>,
394 limit: Option<Expr>,
395) -> PlanNode {
396 if distinct {
397 let mut node = match projection {
398 Some(projection) => project_layer(input, projection),
399 None => input,
400 };
401 node = PlanNode::Distinct {
402 input: Box::new(node),
403 };
404 slice_layer(node, offset, limit)
405 } else {
406 let node = slice_layer(input, offset, limit);
407 match projection {
408 Some(projection) => project_layer(node, projection),
409 None => node,
410 }
411 }
412}
413
414fn resolve_scan_qualifiers(expr: &mut Expr, visible: &str) -> Result<(), PlanError> {
437 match expr {
438 Expr::QualifiedField { qualifier, field } => {
439 if qualifier == visible {
440 *expr = Expr::Field(std::mem::take(field));
441 Ok(())
442 } else {
443 Err(PlanError::Semantic(format!(
444 "no such column: `{qualifier}.{field}` (the only table in this \
445 query is `{visible}`)"
446 )))
447 }
448 }
449 Expr::Field(_) | Expr::Literal(_) | Expr::Param(_) | Expr::ValueLit(_) | Expr::Null => {
450 Ok(())
451 }
452 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
453 resolve_scan_qualifiers(left, visible)?;
454 resolve_scan_qualifiers(right, visible)
455 }
456 Expr::UnaryOp(_, inner)
457 | Expr::Cast(inner, _)
458 | Expr::FunctionCall(_, inner, _)
459 | Expr::JsonPath { base: inner, .. } => resolve_scan_qualifiers(inner, visible),
460 Expr::ScalarFunc(_, args) => {
461 for arg in args.iter_mut() {
462 resolve_scan_qualifiers(arg, visible)?;
463 }
464 Ok(())
465 }
466 Expr::InList { expr, list, .. } => {
467 resolve_scan_qualifiers(expr, visible)?;
468 for item in list.iter_mut() {
469 resolve_scan_qualifiers(item, visible)?;
470 }
471 Ok(())
472 }
473 Expr::Case { whens, else_expr } => {
474 for (cond, res) in whens.iter_mut() {
475 resolve_scan_qualifiers(cond, visible)?;
476 resolve_scan_qualifiers(res, visible)?;
477 }
478 if let Some(e) = else_expr {
479 resolve_scan_qualifiers(e, visible)?;
480 }
481 Ok(())
482 }
483 Expr::Window {
484 args,
485 partition_by,
486 order_by,
487 ..
488 } => {
489 for a in args.iter_mut() {
490 resolve_scan_qualifiers(a, visible)?;
491 }
492 for p in partition_by.iter_mut() {
493 resolve_scan_qualifiers(p, visible)?;
494 }
495 for k in order_by.iter_mut() {
496 resolve_scan_qualifiers(&mut k.expr, visible)?;
497 }
498 Ok(())
499 }
500 Expr::InSubquery { expr, subquery, .. } => {
501 resolve_scan_qualifiers(expr, visible)?;
502 mark_outer_refs_in_query(subquery, visible);
503 Ok(())
504 }
505 Expr::ExistsSubquery { subquery, .. } => {
506 mark_outer_refs_in_query(subquery, visible);
507 Ok(())
508 }
509 Expr::NestedQuery(_) | Expr::LinkPath { .. } => Ok(()),
512 }
513}
514
515fn mark_outer_refs_in_query(query: &mut QueryExpr, outer_visible: &str) {
516 let query_visible = query.alias.as_deref().unwrap_or(&query.source);
517 let outer_is_shadowed = query_visible == outer_visible;
518
519 if let Some(filter) = query.filter.as_mut() {
520 mark_outer_refs(filter, outer_visible, outer_is_shadowed);
521 }
522 if let Some(order) = query.order.as_mut() {
523 for key in order.keys.iter_mut() {
524 mark_outer_refs(&mut key.expr, outer_visible, outer_is_shadowed);
525 }
526 }
527 if let Some(projection) = query.projection.as_mut() {
528 for field in projection.iter_mut() {
529 mark_outer_refs(&mut field.expr, outer_visible, outer_is_shadowed);
530 }
531 }
532 if let Some(group) = query.group_by.as_mut() {
533 for key in group.keys.iter_mut() {
534 mark_outer_refs(&mut key.expr, outer_visible, outer_is_shadowed);
535 }
536 if let Some(having) = group.having.as_mut() {
537 mark_outer_refs(having, outer_visible, outer_is_shadowed);
538 }
539 }
540 if let Some(aggregation) = query.aggregation.as_mut() {
541 if let Some(argument) = aggregation.argument.as_mut() {
542 mark_outer_refs(argument, outer_visible, outer_is_shadowed);
543 }
544 }
545}
546
547fn mark_outer_refs(expr: &mut Expr, outer_visible: &str, outer_is_shadowed: bool) {
548 match expr {
549 Expr::QualifiedField { qualifier, .. }
550 if !outer_is_shadowed && qualifier == outer_visible =>
551 {
552 *qualifier = outer_correlation_qualifier(outer_visible);
553 }
554 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
555 mark_outer_refs(left, outer_visible, outer_is_shadowed);
556 mark_outer_refs(right, outer_visible, outer_is_shadowed);
557 }
558 Expr::UnaryOp(_, inner)
559 | Expr::Cast(inner, _)
560 | Expr::FunctionCall(_, inner, _)
561 | Expr::JsonPath { base: inner, .. } => {
562 mark_outer_refs(inner, outer_visible, outer_is_shadowed);
563 }
564 Expr::ScalarFunc(_, args) => {
565 for arg in args {
566 mark_outer_refs(arg, outer_visible, outer_is_shadowed);
567 }
568 }
569 Expr::InList { expr, list, .. } => {
570 mark_outer_refs(expr, outer_visible, outer_is_shadowed);
571 for item in list {
572 mark_outer_refs(item, outer_visible, outer_is_shadowed);
573 }
574 }
575 Expr::Case { whens, else_expr } => {
576 for (cond, result) in whens {
577 mark_outer_refs(cond, outer_visible, outer_is_shadowed);
578 mark_outer_refs(result, outer_visible, outer_is_shadowed);
579 }
580 if let Some(else_expr) = else_expr {
581 mark_outer_refs(else_expr, outer_visible, outer_is_shadowed);
582 }
583 }
584 Expr::Window {
585 args,
586 partition_by,
587 order_by,
588 ..
589 } => {
590 for expr in args.iter_mut().chain(partition_by) {
591 mark_outer_refs(expr, outer_visible, outer_is_shadowed);
592 }
593 for key in order_by {
594 mark_outer_refs(&mut key.expr, outer_visible, outer_is_shadowed);
595 }
596 }
597 Expr::InSubquery { expr, subquery, .. } => {
598 mark_outer_refs(expr, outer_visible, outer_is_shadowed);
599 if !outer_is_shadowed {
600 mark_outer_refs_in_query(subquery, outer_visible);
601 }
602 }
603 Expr::ExistsSubquery { subquery, .. } if !outer_is_shadowed => {
604 mark_outer_refs_in_query(subquery, outer_visible);
605 }
606 _ => {}
607 }
608}
609
610fn plan_nested_query(q: QueryExpr) -> Result<PlanNode, PlanError> {
617 if q.aggregation.is_some() {
618 return Err(PlanError::Semantic(
622 "aggregates over a nested or link projection are not supported: the \
623 aggregate would ignore the projection and count parent rows; \
624 aggregate the parent table directly (e.g. `count(Order)`) or run \
625 the projection without an aggregate"
626 .into(),
627 ));
628 }
629 if !q.joins.is_empty() || q.group_by.is_some() || q.distinct {
630 return Err(PlanError::Semantic(
631 "nested projections require a plain aliased table scan (no joins, \
632 group, distinct, or aggregation)"
633 .into(),
634 ));
635 }
636 let parent_alias = q.alias.unwrap_or_else(|| q.source.clone());
637 let mut node = PlanNode::AliasScan {
638 table: q.source,
639 alias: parent_alias.clone(),
640 };
641 if let Some(pred) = q.filter {
642 node = PlanNode::Filter {
643 input: Box::new(node),
644 predicate: pred,
645 };
646 }
647 if let Some(order) = q.order {
648 node = PlanNode::Sort {
649 input: Box::new(node),
650 keys: order
651 .keys
652 .into_iter()
653 .map(|k| SortKey {
654 expr: k.expr,
655 descending: k.descending,
656 })
657 .collect(),
658 };
659 }
660 if let Some(off) = q.offset {
661 node = PlanNode::Offset {
662 input: Box::new(node),
663 count: off,
664 };
665 }
666 if let Some(lim) = q.limit {
667 node = PlanNode::Limit {
668 input: Box::new(node),
669 count: lim,
670 };
671 }
672 let fields = q
673 .projection
674 .expect("plan_nested_query is only called with a projection")
675 .into_iter()
676 .map(|pf| match pf.expr {
677 Expr::NestedQuery(nested) => {
678 let name = pf.alias.ok_or_else(|| {
679 PlanError::Semantic("nested projection field requires a name".into())
680 })?;
681 resolve_nested_projection(name, *nested, &parent_alias)
682 .map(|nested| NestedProjectField::Nested(Box::new(nested)))
683 }
684 Expr::LinkPath {
685 outer_alias,
686 links,
687 column,
688 } => {
689 if outer_alias != parent_alias {
690 return Err(PlanError::Semantic(format!(
691 "link path starts at unknown alias `{outer_alias}`; \
692 the outer scan is aliased `{parent_alias}`"
693 )));
694 }
695 let name = pf.alias.unwrap_or_else(|| {
696 let mut n = outer_alias.clone();
697 for link in &links {
698 n.push('.');
699 n.push_str(link);
700 }
701 n.push('.');
702 n.push_str(&column);
703 n
704 });
705 Ok(NestedProjectField::Link(Box::new(ScalarLinkField {
706 name,
707 outer_alias,
708 links,
709 column,
710 resolved: None,
711 })))
712 }
713 expr => Ok(NestedProjectField::Plain(ProjectField {
714 alias: pf.alias,
715 expr,
716 })),
717 })
718 .collect::<Result<Vec<_>, PlanError>>()?;
719 Ok(PlanNode::NestedProject {
720 input: Box::new(node),
721 fields,
722 })
723}
724
725fn resolve_nested_projection(
731 name: String,
732 nested: NestedQuery,
733 parent_alias: &str,
734) -> Result<NestedProjection, PlanError> {
735 resolve_nested_projection_inner(name, nested, parent_alias, true)
736}
737
738fn resolve_nested_projection_inner(
742 name: String,
743 nested: NestedQuery,
744 parent_alias: &str,
745 qualify_parent_key: bool,
746) -> Result<NestedProjection, PlanError> {
747 let via_link = nested.via_link.clone();
753 let mut conjuncts = Vec::new();
754 split_and_chain(nested.filter, &mut conjuncts);
755 let mut residual: Option<Expr> = None;
756
757 let (child_key, parent_key) = if via_link.is_some() {
758 for conjunct in conjuncts {
759 if matches!(conjunct, Expr::Literal(Literal::Bool(true))) {
761 continue;
762 }
763 let rewritten =
764 rewrite_residual_condition(conjunct, &name, &nested.alias, parent_alias)?;
765 residual = Some(match residual {
766 Some(existing) => {
767 Expr::BinaryOp(Box::new(existing), BinOp::And, Box::new(rewritten))
768 }
769 None => rewritten,
770 });
771 }
772 (String::new(), String::new())
774 } else {
775 let correlation_of = |expr: &Expr| -> Option<(String, String)> {
777 let Expr::BinaryOp(left, BinOp::Eq, right) = expr else {
778 return None;
779 };
780 let side = |expr: &Expr| match expr {
781 Expr::QualifiedField { qualifier, field } => {
782 Some((qualifier.clone(), field.clone()))
783 }
784 _ => None,
785 };
786 let ((lq, lf), (rq, rf)) = (side(left)?, side(right)?);
787 if lq == nested.alias && rq == parent_alias {
788 Some((lf, rf))
789 } else if rq == nested.alias && lq == parent_alias {
790 Some((rf, lf))
791 } else {
792 None
793 }
794 };
795 let mut correlation: Option<(String, String)> = None;
796 for conjunct in conjuncts {
797 match correlation_of(&conjunct) {
798 Some(keys) if correlation.is_none() => correlation = Some(keys),
799 Some(_) => {
800 return Err(PlanError::Semantic(format!(
801 "nested projection `{name}` links `{child}` to `{parent}` more than \
802 once; exactly one correlation predicate \
803 ({child}.<col> = {parent}.<col>) is supported",
804 child = nested.alias,
805 parent = parent_alias,
806 )))
807 }
808 None => {
809 let rewritten =
810 rewrite_residual_condition(conjunct, &name, &nested.alias, parent_alias)?;
811 residual = Some(match residual {
812 Some(existing) => {
813 Expr::BinaryOp(Box::new(existing), BinOp::And, Box::new(rewritten))
814 }
815 None => rewritten,
816 });
817 }
818 }
819 }
820 let Some(correlation) = correlation else {
821 return Err(PlanError::Semantic(format!(
822 "nested projection `{name}` requires an equi-correlation predicate linking \
823 `{child}` to the outer query ({child}.<col> = {parent}.<col>) somewhere in \
824 its filter",
825 child = nested.alias,
826 parent = parent_alias,
827 )));
828 };
829 correlation
830 };
831 let order = nested
832 .order
833 .map(|clause| {
834 clause
835 .keys
836 .into_iter()
837 .map(|key| {
838 let column = match &key.expr {
839 Expr::Field(field) => field.clone(),
840 Expr::QualifiedField { qualifier, field } if *qualifier == nested.alias => {
841 field.clone()
842 }
843 _ => {
844 return Err(PlanError::Semantic(format!(
845 "nested projection `{name}` order keys must be plain \
846 columns of `{child}` (`{child}.<col>` or `.<col>`)",
847 child = nested.alias,
848 )))
849 }
850 };
851 Ok((column, key.descending))
852 })
853 .collect::<Result<Vec<_>, PlanError>>()
854 })
855 .transpose()?
856 .unwrap_or_default();
857 let fields = nested
858 .fields
859 .into_iter()
860 .map(|pf| {
861 if let Expr::NestedQuery(inner) = pf.expr {
862 let inner_name = pf.alias.ok_or_else(|| {
863 PlanError::Semantic(
864 "nested projection field requires a name \
865 (`<name>: <Table> as <alias> ...`)"
866 .into(),
867 )
868 })?;
869 return resolve_nested_projection_inner(inner_name, *inner, &nested.alias, false)
872 .map(|inner| NestedField::Nested(Box::new(inner)));
873 }
874 let column = match &pf.expr {
875 Expr::Field(field) => field.clone(),
876 Expr::QualifiedField { qualifier, field } if *qualifier == nested.alias => {
877 field.clone()
878 }
879 _ => {
880 return Err(PlanError::Semantic(format!(
881 "nested projection `{name}` fields must be plain columns of `{}` \
882 (`{}.<col>` or `.<col>`) or a deeper nested projection",
883 nested.alias, nested.alias
884 )))
885 }
886 };
887 let key = pf.alias.unwrap_or_else(|| column.clone());
888 Ok(NestedField::Scalar { key, column })
889 })
890 .collect::<Result<Vec<_>, PlanError>>()?;
891 let is_via_link = via_link.is_some();
892 Ok(NestedProjection {
893 name,
894 table: nested.source,
895 via_link,
896 alias: nested.alias,
897 parent_alias: parent_alias.to_string(),
898 child_key,
899 parent_key: if is_via_link {
903 parent_key
904 } else if qualify_parent_key {
905 format!("{parent_alias}.{parent_key}")
906 } else {
907 parent_key
908 },
909 residual,
910 order,
911 limit: nested.limit,
912 offset: nested.offset,
913 offset_before_limit: nested.offset_before_limit,
914 fields,
915 })
916}
917
918fn split_and_chain(expr: Expr, out: &mut Vec<Expr>) {
920 match expr {
921 Expr::BinaryOp(left, BinOp::And, right) => {
922 split_and_chain(*left, out);
923 split_and_chain(*right, out);
924 }
925 other => out.push(other),
926 }
927}
928
929fn rewrite_residual_condition(
935 expr: Expr,
936 name: &str,
937 child_alias: &str,
938 parent_alias: &str,
939) -> Result<Expr, PlanError> {
940 let rewrite = |inner: Box<Expr>| -> Result<Box<Expr>, PlanError> {
941 Ok(Box::new(rewrite_residual_condition(
942 *inner,
943 name,
944 child_alias,
945 parent_alias,
946 )?))
947 };
948 match expr {
949 Expr::QualifiedField { qualifier, field } => {
950 if qualifier == child_alias {
951 Ok(Expr::Field(field))
952 } else if qualifier == parent_alias {
953 Err(PlanError::Semantic(format!(
954 "nested projection `{name}` filter references outer alias \
955 `{parent_alias}` (`{parent_alias}.{field}`) outside the correlation \
956 predicate; move that condition to the outer query's filter"
957 )))
958 } else {
959 Err(PlanError::Semantic(format!(
960 "nested projection `{name}` filter references unknown alias \
961 `{qualifier}`; only columns of `{child_alias}` may be used"
962 )))
963 }
964 }
965 Expr::Field(_) | Expr::Literal(_) | Expr::Param(_) | Expr::ValueLit(_) | Expr::Null => {
966 Ok(expr)
967 }
968 Expr::BinaryOp(left, op, right) => Ok(Expr::BinaryOp(rewrite(left)?, op, rewrite(right)?)),
969 Expr::UnaryOp(op, inner) => Ok(Expr::UnaryOp(op, rewrite(inner)?)),
970 Expr::Coalesce(left, right) => Ok(Expr::Coalesce(rewrite(left)?, rewrite(right)?)),
971 Expr::Cast(inner, ty) => Ok(Expr::Cast(rewrite(inner)?, ty)),
972 Expr::ScalarFunc(func, args) => Ok(Expr::ScalarFunc(
973 func,
974 args.into_iter()
975 .map(|arg| rewrite_residual_condition(arg, name, child_alias, parent_alias))
976 .collect::<Result<Vec<_>, _>>()?,
977 )),
978 Expr::InList {
979 expr,
980 list,
981 negated,
982 } => Ok(Expr::InList {
983 expr: rewrite(expr)?,
984 list: list
985 .into_iter()
986 .map(|item| rewrite_residual_condition(item, name, child_alias, parent_alias))
987 .collect::<Result<Vec<_>, _>>()?,
988 negated,
989 }),
990 Expr::Case { whens, else_expr } => Ok(Expr::Case {
991 whens: whens
992 .into_iter()
993 .map(|(cond, result)| Ok((rewrite(cond)?, rewrite(result)?)))
994 .collect::<Result<Vec<_>, PlanError>>()?,
995 else_expr: else_expr.map(rewrite).transpose()?,
996 }),
997 Expr::JsonPath { base, segments } => Ok(Expr::JsonPath {
998 base: rewrite(base)?,
999 segments,
1000 }),
1001 Expr::InSubquery { .. } | Expr::ExistsSubquery { .. } => Err(PlanError::Semantic(format!(
1002 "nested projection `{name}` filter cannot contain a subquery; \
1003 filter the outer query or the child columns directly"
1004 ))),
1005 Expr::FunctionCall(..) => Err(PlanError::Semantic(format!(
1006 "nested projection `{name}` filter cannot contain an aggregate function"
1007 ))),
1008 Expr::Window { .. } => Err(PlanError::Semantic(format!(
1009 "nested projection `{name}` filter cannot contain a window function"
1010 ))),
1011 Expr::NestedQuery(_) => Err(PlanError::Semantic(format!(
1012 "nested projection `{name}` filter cannot contain another nested projection"
1013 ))),
1014 Expr::LinkPath { .. } => Err(PlanError::Semantic(format!(
1015 "nested projection `{name}` filter cannot contain a link traversal; \
1016 link paths are only valid as projection fields"
1017 ))),
1018 }
1019}
1020
1021fn plan_joined_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
1043 let primary_alias = q.alias.clone().unwrap_or_else(|| q.source.clone());
1044 let mut aliases = std::collections::HashSet::new();
1045 aliases.insert(primary_alias.clone());
1046 let mut node = PlanNode::AliasScan {
1047 table: q.source.clone(),
1048 alias: primary_alias,
1049 };
1050
1051 for join in q.joins {
1052 let right_alias = join.alias.unwrap_or_else(|| join.source.clone());
1053 if !aliases.insert(right_alias.clone()) {
1054 return Err(ParseError::Syntax {
1055 message: format!(
1056 "duplicate source alias `{right_alias}` in join; every joined source needs a unique alias"
1057 ),
1058 position: None,
1059 }
1060 .into());
1061 }
1062 let right = PlanNode::AliasScan {
1063 table: join.source,
1064 alias: right_alias,
1065 };
1066 match join.kind {
1067 JoinKind::Inner | JoinKind::LeftOuter | JoinKind::Cross => {
1068 node = PlanNode::NestedLoopJoin {
1069 left: Box::new(node),
1070 right: Box::new(right),
1071 on: join.on,
1072 kind: join.kind,
1073 };
1074 }
1075 JoinKind::RightOuter => {
1076 node = PlanNode::NestedLoopJoin {
1078 left: Box::new(right),
1079 right: Box::new(node),
1080 on: join.on,
1081 kind: JoinKind::LeftOuter,
1082 };
1083 }
1084 }
1085 }
1086
1087 if let Some(pred) = q.filter {
1088 node = PlanNode::Filter {
1089 input: Box::new(node),
1090 predicate: pred,
1091 };
1092 }
1093
1094 if q.group_by.is_none() {
1095 if let Some(order) = q.order.take() {
1096 node = PlanNode::Sort {
1097 input: Box::new(node),
1098 keys: order
1099 .keys
1100 .into_iter()
1101 .map(|k| SortKey {
1102 expr: k.expr,
1103 descending: k.descending,
1104 })
1105 .collect(),
1106 };
1107 }
1108 }
1109
1110 if let Some(group) = q.group_by {
1112 let mut grouped_order = q.order;
1113 let mut proj_fields: Vec<ProjectField> = q
1114 .projection
1115 .map(|proj| {
1116 proj.into_iter()
1117 .map(|pf| ProjectField {
1118 alias: pf.alias,
1119 expr: pf.expr,
1120 })
1121 .collect()
1122 })
1123 .unwrap_or_default();
1124 let mut having = group.having;
1125 let aggregates = extract_aggregates(&mut proj_fields, &mut having, &aliases)?;
1126 rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
1127 rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
1128
1129 node = PlanNode::GroupBy {
1130 input: Box::new(node),
1131 keys: group.keys,
1132 aggregates,
1133 having,
1134 };
1135
1136 if !proj_fields.is_empty() {
1137 node = PlanNode::Project {
1138 input: Box::new(node),
1139 fields: proj_fields,
1140 };
1141 }
1142 if q.distinct {
1145 node = PlanNode::Distinct {
1146 input: Box::new(node),
1147 };
1148 }
1149 if let Some(order) = grouped_order {
1150 node = PlanNode::Sort {
1151 input: Box::new(node),
1152 keys: order
1153 .keys
1154 .into_iter()
1155 .map(|key| SortKey {
1156 expr: key.expr,
1157 descending: key.descending,
1158 })
1159 .collect(),
1160 };
1161 }
1162 return Ok(slice_layer(node, q.offset, q.limit));
1166 }
1167
1168 node = projected_tail(node, q.projection, q.distinct, q.offset, q.limit);
1169
1170 if let Some(agg) = q.aggregation {
1171 let provenance_alias =
1172 symmetric_provenance_alias(agg.function, agg.argument.as_ref(), agg.mode, &aliases)?;
1173 node = PlanNode::Aggregate {
1174 input: Box::new(node),
1175 function: agg.function,
1176 argument: agg.argument,
1177 mode: agg.mode,
1178 provenance_alias,
1179 };
1180 }
1181
1182 Ok(node)
1183}
1184
1185fn plan_insert(ins: InsertExpr) -> Result<PlanNode, PlanError> {
1186 Ok(PlanNode::Insert {
1187 table: ins.target,
1188 rows: ins.rows,
1189 returning: ins.returning,
1190 })
1191}
1192
1193fn plan_update(mut upd: UpdateExpr) -> Result<PlanNode, PlanError> {
1194 let visible = upd.alias.clone().unwrap_or_else(|| upd.source.clone());
1198 if let Some(filter) = upd.filter.as_mut() {
1199 resolve_scan_qualifiers(filter, &visible)?;
1200 }
1201 for assign in upd.assignments.iter_mut() {
1202 resolve_scan_qualifiers(&mut assign.value, &visible)?;
1203 }
1204 let source = match upd.filter {
1209 Some(pred) => match try_extract_eq_index_key(&upd.source, &pred) {
1210 Some(index_scan) => index_scan,
1211 None => match try_extract_range_index_keys(&upd.source, &pred) {
1212 Some(range_scan) => range_scan,
1213 None => PlanNode::Filter {
1214 input: Box::new(PlanNode::SeqScan {
1215 table: upd.source.clone(),
1216 }),
1217 predicate: pred,
1218 },
1219 },
1220 },
1221 None => PlanNode::SeqScan {
1222 table: upd.source.clone(),
1223 },
1224 };
1225 Ok(PlanNode::Update {
1226 input: Box::new(source),
1227 table: upd.source,
1228 assignments: upd.assignments,
1229 returning: upd.returning,
1230 })
1231}
1232
1233fn plan_delete(mut del: DeleteExpr) -> Result<PlanNode, PlanError> {
1234 let visible = del.alias.clone().unwrap_or_else(|| del.source.clone());
1238 if let Some(filter) = del.filter.as_mut() {
1239 resolve_scan_qualifiers(filter, &visible)?;
1240 }
1241 let source = match del.filter {
1242 Some(pred) => match try_extract_eq_index_key(&del.source, &pred) {
1243 Some(index_scan) => index_scan,
1244 None => match try_extract_range_index_keys(&del.source, &pred) {
1245 Some(range_scan) => range_scan,
1246 None => PlanNode::Filter {
1247 input: Box::new(PlanNode::SeqScan {
1248 table: del.source.clone(),
1249 }),
1250 predicate: pred,
1251 },
1252 },
1253 },
1254 None => PlanNode::SeqScan {
1255 table: del.source.clone(),
1256 },
1257 };
1258 Ok(PlanNode::Delete {
1259 input: Box::new(source),
1260 table: del.source,
1261 returning: del.returning,
1262 })
1263}
1264
1265fn plan_upsert(ups: UpsertExpr) -> Result<PlanNode, PlanError> {
1266 Ok(PlanNode::Upsert {
1267 table: ups.target,
1268 key_column: ups.key_column,
1269 assignments: ups.assignments,
1270 on_conflict: ups.on_conflict,
1271 })
1272}
1273
1274fn plan_create_type(ct: CreateTypeExpr) -> Result<PlanNode, PlanError> {
1275 let fields = ct
1276 .fields
1277 .into_iter()
1278 .map(|f| crate::plan::CreateField {
1279 name: f.name,
1280 type_name: f.type_name,
1281 required: f.required,
1282 unique: f.unique,
1283 default: f.default,
1284 auto: f.auto,
1285 })
1286 .collect();
1287 Ok(PlanNode::CreateTable {
1288 name: ct.name,
1289 fields,
1290 if_not_exists: ct.if_not_exists,
1291 })
1292}
1293
1294pub(crate) fn try_extract_eq_index_key(table: &str, pred: &Expr) -> Option<PlanNode> {
1304 let (lhs, op, rhs) = match pred {
1305 Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
1306 _ => return None,
1307 };
1308 if op != BinOp::Eq {
1309 return None;
1310 }
1311 match (lhs, rhs) {
1312 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some(PlanNode::ExprIndexScan {
1313 table: table.to_string(),
1314 path: stored_json_path(path)?,
1315 key: rhs.clone(),
1316 }),
1317 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some(PlanNode::ExprIndexScan {
1318 table: table.to_string(),
1319 path: stored_json_path(path)?,
1320 key: lhs.clone(),
1321 }),
1322 (Expr::Field(name), Expr::Literal(_)) => Some(PlanNode::IndexScan {
1323 table: table.to_string(),
1324 column: name.clone(),
1325 key: rhs.clone(),
1326 }),
1327 (Expr::Literal(_), Expr::Field(name)) => Some(PlanNode::IndexScan {
1328 table: table.to_string(),
1329 column: name.clone(),
1330 key: lhs.clone(),
1331 }),
1332 _ => None,
1333 }
1334}
1335
1336fn stored_json_path(expr: &Expr) -> Option<StoredJsonPathV1> {
1337 JsonPathIdentityV1::from_expr(expr)?.bind_table_local(None)
1338}
1339
1340pub(crate) fn extract_single_bound(pred: &Expr) -> Option<RangeBound> {
1343 let (lhs, op, rhs) = match pred {
1344 Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
1345 _ => return None,
1346 };
1347 match op {
1348 BinOp::Gt => match (lhs, rhs) {
1350 (Expr::Field(name), Expr::Literal(_)) => Some((
1351 RangeTarget::Column(name.clone()),
1352 Some((rhs.clone(), false)),
1353 None,
1354 )),
1355 (Expr::Literal(_), Expr::Field(name)) => {
1356 Some((
1358 RangeTarget::Column(name.clone()),
1359 None,
1360 Some((lhs.clone(), false)),
1361 ))
1362 }
1363 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1364 RangeTarget::JsonPath(stored_json_path(path)?),
1365 Some((rhs.clone(), false)),
1366 None,
1367 )),
1368 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1369 RangeTarget::JsonPath(stored_json_path(path)?),
1370 None,
1371 Some((lhs.clone(), false)),
1372 )),
1373 _ => None,
1374 },
1375 BinOp::Gte => match (lhs, rhs) {
1377 (Expr::Field(name), Expr::Literal(_)) => Some((
1378 RangeTarget::Column(name.clone()),
1379 Some((rhs.clone(), true)),
1380 None,
1381 )),
1382 (Expr::Literal(_), Expr::Field(name)) => Some((
1383 RangeTarget::Column(name.clone()),
1384 None,
1385 Some((lhs.clone(), true)),
1386 )),
1387 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1388 RangeTarget::JsonPath(stored_json_path(path)?),
1389 Some((rhs.clone(), true)),
1390 None,
1391 )),
1392 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1393 RangeTarget::JsonPath(stored_json_path(path)?),
1394 None,
1395 Some((lhs.clone(), true)),
1396 )),
1397 _ => None,
1398 },
1399 BinOp::Lt => match (lhs, rhs) {
1401 (Expr::Field(name), Expr::Literal(_)) => Some((
1402 RangeTarget::Column(name.clone()),
1403 None,
1404 Some((rhs.clone(), false)),
1405 )),
1406 (Expr::Literal(_), Expr::Field(name)) => Some((
1407 RangeTarget::Column(name.clone()),
1408 Some((lhs.clone(), false)),
1409 None,
1410 )),
1411 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1412 RangeTarget::JsonPath(stored_json_path(path)?),
1413 None,
1414 Some((rhs.clone(), false)),
1415 )),
1416 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1417 RangeTarget::JsonPath(stored_json_path(path)?),
1418 Some((lhs.clone(), false)),
1419 None,
1420 )),
1421 _ => None,
1422 },
1423 BinOp::Lte => match (lhs, rhs) {
1425 (Expr::Field(name), Expr::Literal(_)) => Some((
1426 RangeTarget::Column(name.clone()),
1427 None,
1428 Some((rhs.clone(), true)),
1429 )),
1430 (Expr::Literal(_), Expr::Field(name)) => Some((
1431 RangeTarget::Column(name.clone()),
1432 Some((lhs.clone(), true)),
1433 None,
1434 )),
1435 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1436 RangeTarget::JsonPath(stored_json_path(path)?),
1437 None,
1438 Some((rhs.clone(), true)),
1439 )),
1440 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1441 RangeTarget::JsonPath(stored_json_path(path)?),
1442 Some((lhs.clone(), true)),
1443 None,
1444 )),
1445 _ => None,
1446 },
1447 _ => None,
1448 }
1449}
1450
1451fn try_extract_range_index_keys(table: &str, pred: &Expr) -> Option<PlanNode> {
1474 if let Expr::BinaryOp(lhs, BinOp::And, rhs) = pred {
1477 if let (Some((col1, s1, e1)), Some((col2, s2, e2))) =
1478 (extract_single_bound(lhs), extract_single_bound(rhs))
1479 {
1480 if col1 == col2 {
1481 if let (Some(start), None, None, Some(end)) = (s1, e1, s2, e2) {
1482 return Some(range_scan_for_target(table, col1, Some(start), Some(end)));
1483 }
1484 }
1485 }
1486 }
1487
1488 if let Some((col, start, end)) = extract_single_bound(pred) {
1490 return Some(range_scan_for_target(table, col, start, end));
1491 }
1492
1493 None
1494}
1495
1496pub(crate) fn range_scan_for_target(
1497 table: &str,
1498 target: RangeTarget,
1499 start: Option<(Expr, bool)>,
1500 end: Option<(Expr, bool)>,
1501) -> PlanNode {
1502 match target {
1503 RangeTarget::Column(column) => PlanNode::RangeScan {
1504 table: table.to_string(),
1505 column,
1506 start,
1507 end,
1508 },
1509 RangeTarget::JsonPath(path) => PlanNode::ExprRangeScan {
1510 table: table.to_string(),
1511 path,
1512 start,
1513 end,
1514 },
1515 }
1516}
1517
1518fn try_extract_ordered_expr_index_scan(query: &QueryExpr) -> Option<PlanNode> {
1523 if query.alias.is_some()
1524 || !query.joins.is_empty()
1525 || query.filter.is_some()
1526 || query.group_by.is_some()
1527 || query.distinct
1528 || query.aggregation.is_some()
1529 || query.projection.as_ref().is_some_and(|fields| {
1530 fields
1531 .iter()
1532 .any(|field| matches!(field.expr, Expr::Window { .. }))
1533 })
1534 {
1535 return None;
1536 }
1537 let order = query.order.as_ref()?;
1538 let [key] = order.keys.as_slice() else {
1539 return None;
1540 };
1541 let path = stored_json_path(&key.expr)?;
1542 let limit = query.limit.as_ref()?;
1543 if !matches!(limit, Expr::Literal(Literal::Int(value)) if *value >= 0) {
1544 return None;
1545 }
1546 if !query
1547 .offset
1548 .as_ref()
1549 .is_none_or(|offset| matches!(offset, Expr::Literal(Literal::Int(value)) if *value >= 0))
1550 {
1551 return None;
1552 }
1553 Some(PlanNode::OrderedExprIndexScan {
1554 table: query.source.clone(),
1555 path,
1556 descending: key.descending,
1557 limit: limit.clone(),
1558 offset: query.offset.clone(),
1559 })
1560}
1561
1562fn extract_windows(proj_fields: &mut [ProjectField]) -> Vec<WindowDef> {
1567 let mut defs = Vec::new();
1568 let mut counter = 0usize;
1569 for f in proj_fields.iter_mut() {
1570 if let Expr::Window {
1571 function,
1572 args,
1573 mode,
1574 partition_by,
1575 order_by,
1576 } = &f.expr
1577 {
1578 let output_name = format!("__win_{counter}");
1579 defs.push(WindowDef {
1580 function: *function,
1581 args: args.clone(),
1582 mode: *mode,
1583 partition_by: partition_by.clone(),
1584 order_by: order_by
1585 .iter()
1586 .map(|k| SortKey {
1587 expr: k.expr.clone(),
1588 descending: k.descending,
1589 })
1590 .collect(),
1591 output_name: output_name.clone(),
1592 });
1593 f.expr = Expr::Field(output_name);
1594 counter += 1;
1595 }
1596 }
1597 defs
1598}
1599
1600fn extract_aggregates(
1606 proj_fields: &mut [ProjectField],
1607 having: &mut Option<Expr>,
1608 source_aliases: &std::collections::HashSet<String>,
1609) -> Result<Vec<GroupAgg>, PlanError> {
1610 let mut aggs: Vec<GroupAgg> = Vec::new();
1611 let mut counter = 0usize;
1612 for f in proj_fields.iter_mut() {
1613 if f.alias.is_none() && contains_aggregate(&f.expr) {
1620 f.alias = Some(projection_output_name(&f.expr));
1621 }
1622 rewrite_agg_expr(&mut f.expr, &mut aggs, &mut counter, source_aliases)?;
1623 }
1624 if let Some(h) = having {
1625 rewrite_agg_expr(h, &mut aggs, &mut counter, source_aliases)?;
1626 }
1627 Ok(aggs)
1628}
1629
1630fn rewrite_group_key_references(
1631 fields: &mut [ProjectField],
1632 having: &mut Option<Expr>,
1633 keys: &[GroupKey],
1634) {
1635 for field in fields {
1636 rewrite_group_key_expr(&mut field.expr, keys);
1637 }
1638 if let Some(having) = having {
1639 rewrite_group_key_expr(having, keys);
1640 }
1641}
1642
1643fn rewrite_group_order_keys(
1644 order: Option<&mut OrderClause>,
1645 projection: &[ProjectField],
1646 keys: &[GroupKey],
1647) {
1648 let Some(order) = order else {
1649 return;
1650 };
1651 for order_key in &mut order.keys {
1652 let Some(group_key) = keys.iter().find(|key| key.expr == order_key.expr) else {
1653 continue;
1654 };
1655 let projected_name = projection
1656 .iter()
1657 .find(|field| field.expr == group_key.expr)
1658 .and_then(|field| field.alias.clone())
1659 .unwrap_or_else(|| group_key.output_name());
1660 order_key.expr = Expr::Field(projected_name);
1661 }
1662}
1663
1664fn rewrite_group_key_expr(expr: &mut Expr, keys: &[GroupKey]) {
1665 if let Some(key) = keys.iter().find(|key| key.expr == *expr) {
1666 *expr = Expr::Field(key.output_name());
1667 return;
1668 }
1669 match expr {
1670 Expr::FunctionCall(..) => {}
1674 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1675 rewrite_group_key_expr(left, keys);
1676 rewrite_group_key_expr(right, keys);
1677 }
1678 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) => rewrite_group_key_expr(inner, keys),
1679 Expr::ScalarFunc(_, args) => {
1680 for arg in args {
1681 rewrite_group_key_expr(arg, keys);
1682 }
1683 }
1684 Expr::InList { expr, list, .. } => {
1685 rewrite_group_key_expr(expr, keys);
1686 for item in list {
1687 rewrite_group_key_expr(item, keys);
1688 }
1689 }
1690 Expr::Case { whens, else_expr } => {
1691 for (condition, result) in whens {
1692 rewrite_group_key_expr(condition, keys);
1693 rewrite_group_key_expr(result, keys);
1694 }
1695 if let Some(expr) = else_expr {
1696 rewrite_group_key_expr(expr, keys);
1697 }
1698 }
1699 _ => {}
1700 }
1701}
1702
1703fn contains_aggregate(expr: &Expr) -> bool {
1706 match expr {
1707 Expr::FunctionCall(..) => true,
1708 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1709 contains_aggregate(left) || contains_aggregate(right)
1710 }
1711 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) | Expr::JsonPath { base: inner, .. } => {
1712 contains_aggregate(inner)
1713 }
1714 Expr::ScalarFunc(_, args) => args.iter().any(contains_aggregate),
1715 Expr::InList { expr, list, .. } => {
1716 contains_aggregate(expr) || list.iter().any(contains_aggregate)
1717 }
1718 Expr::InSubquery { expr, .. } => contains_aggregate(expr),
1719 Expr::Case { whens, else_expr } => {
1720 whens
1721 .iter()
1722 .any(|(when, then)| contains_aggregate(when) || contains_aggregate(then))
1723 || else_expr.as_deref().is_some_and(contains_aggregate)
1724 }
1725 _ => false,
1726 }
1727}
1728
1729fn rewrite_agg_expr(
1730 expr: &mut Expr,
1731 aggs: &mut Vec<GroupAgg>,
1732 counter: &mut usize,
1733 source_aliases: &std::collections::HashSet<String>,
1734) -> Result<(), PlanError> {
1735 match expr {
1736 Expr::FunctionCall(func, inner, mode) => {
1737 let output = find_or_insert_agg(aggs, *func, inner, *mode, counter, source_aliases)?;
1738 *expr = Expr::Field(output);
1739 }
1740 Expr::BinaryOp(l, _, r) => {
1741 rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1742 rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1743 }
1744 Expr::UnaryOp(_, inner) => rewrite_agg_expr(inner, aggs, counter, source_aliases)?,
1745 Expr::Coalesce(l, r) => {
1746 rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1747 rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1748 }
1749 Expr::InList { expr: e, list, .. } => {
1750 rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1751 for item in list {
1752 rewrite_agg_expr(item, aggs, counter, source_aliases)?;
1753 }
1754 }
1755 Expr::InSubquery { expr: e, .. } => {
1756 rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1757 }
1758 _ => {}
1759 }
1760 Ok(())
1761}
1762
1763fn find_or_insert_agg(
1764 aggs: &mut Vec<GroupAgg>,
1765 func: AggFunc,
1766 argument: &Expr,
1767 mode: AggregateMode,
1768 counter: &mut usize,
1769 source_aliases: &std::collections::HashSet<String>,
1770) -> Result<String, PlanError> {
1771 for existing in aggs.iter() {
1772 if existing.function == func && existing.argument == *argument && existing.mode == mode {
1773 return Ok(existing.output_name.clone());
1774 }
1775 }
1776 let provenance_alias = symmetric_provenance_alias(func, Some(argument), mode, source_aliases)?;
1777 let output_name = format!("__agg_{counter}");
1778 aggs.push(GroupAgg {
1779 function: func,
1780 argument: argument.clone(),
1781 mode,
1782 provenance_alias,
1783 output_name: output_name.clone(),
1784 });
1785 *counter += 1;
1786 Ok(output_name)
1787}
1788
1789fn symmetric_provenance_alias(
1790 function: AggFunc,
1791 argument: Option<&Expr>,
1792 mode: AggregateMode,
1793 source_aliases: &std::collections::HashSet<String>,
1794) -> Result<Option<String>, PlanError> {
1795 if mode == AggregateMode::Raw
1796 || source_aliases.len() < 2
1797 || !matches!(function, AggFunc::Sum | AggFunc::Avg | AggFunc::Count)
1798 || (function == AggFunc::Count
1799 && argument.is_none_or(|argument| matches!(argument, Expr::Field(name) if name == "*")))
1800 {
1801 return Ok(None);
1802 }
1803 let Some(argument) = argument else {
1804 return Err(symmetric_aggregate_error(
1805 function,
1806 "does not reference a source row",
1807 ));
1808 };
1809
1810 let mut qualified = std::collections::HashSet::new();
1811 let mut has_unqualified = false;
1812 collect_expression_sources(argument, &mut qualified, &mut has_unqualified);
1813
1814 for alias in &qualified {
1815 if !source_aliases.contains(alias) {
1816 return Err(symmetric_aggregate_error(
1817 function,
1818 &format!("references unknown source alias '{alias}'"),
1819 ));
1820 }
1821 }
1822 if has_unqualified {
1823 if source_aliases.len() != 1 {
1824 return Err(symmetric_aggregate_error(
1825 function,
1826 "contains an ambiguous unqualified field",
1827 ));
1828 }
1829 qualified.extend(source_aliases.iter().cloned());
1830 }
1831 match qualified.len() {
1832 1 => Ok(qualified.into_iter().next()),
1833 0 => Err(symmetric_aggregate_error(
1834 function,
1835 "does not reference a source row",
1836 )),
1837 _ => Err(symmetric_aggregate_error(
1838 function,
1839 "references multiple source aliases",
1840 )),
1841 }
1842}
1843
1844fn symmetric_aggregate_error(function: AggFunc, reason: &str) -> PlanError {
1845 let name = format!("{function:?}").to_lowercase();
1846 PlanError::Semantic(format!(
1847 "symmetric {name} expression {reason}; reference exactly one source alias or use {name}(raw ...)"
1848 ))
1849}
1850
1851fn collect_expression_sources(
1852 expr: &Expr,
1853 qualified: &mut std::collections::HashSet<String>,
1854 has_unqualified: &mut bool,
1855) {
1856 match expr {
1857 Expr::Field(name) if name != "*" => *has_unqualified = true,
1858 Expr::QualifiedField { qualifier, .. } => {
1859 qualified.insert(qualifier.clone());
1860 }
1861 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1862 collect_expression_sources(left, qualified, has_unqualified);
1863 collect_expression_sources(right, qualified, has_unqualified);
1864 }
1865 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) | Expr::JsonPath { base: inner, .. } => {
1866 collect_expression_sources(inner, qualified, has_unqualified);
1867 }
1868 Expr::ScalarFunc(_, args) => {
1869 for argument in args {
1870 collect_expression_sources(argument, qualified, has_unqualified);
1871 }
1872 }
1873 Expr::InList { expr, list, .. } => {
1874 collect_expression_sources(expr, qualified, has_unqualified);
1875 for item in list {
1876 collect_expression_sources(item, qualified, has_unqualified);
1877 }
1878 }
1879 Expr::InSubquery { expr, .. } => {
1880 collect_expression_sources(expr, qualified, has_unqualified);
1881 }
1882 Expr::Case { whens, else_expr } => {
1883 for (condition, result) in whens {
1884 collect_expression_sources(condition, qualified, has_unqualified);
1885 collect_expression_sources(result, qualified, has_unqualified);
1886 }
1887 if let Some(expr) = else_expr {
1888 collect_expression_sources(expr, qualified, has_unqualified);
1889 }
1890 }
1891 Expr::Window {
1892 args,
1893 partition_by,
1894 order_by,
1895 ..
1896 } => {
1897 for expr in args.iter().chain(partition_by) {
1898 collect_expression_sources(expr, qualified, has_unqualified);
1899 }
1900 for key in order_by {
1901 collect_expression_sources(&key.expr, qualified, has_unqualified);
1902 }
1903 }
1904 Expr::FunctionCall(_, inner, _) => {
1905 collect_expression_sources(inner, qualified, has_unqualified);
1906 }
1907 Expr::LinkPath { outer_alias, .. } => {
1909 qualified.insert(outer_alias.clone());
1910 }
1911 Expr::ExistsSubquery { .. }
1912 | Expr::Field(_)
1913 | Expr::Literal(_)
1914 | Expr::Param(_)
1915 | Expr::ValueLit(_)
1916 | Expr::Null
1917 | Expr::NestedQuery(_) => {}
1918 }
1919}
1920
1921#[cfg(test)]
1922mod tests {
1923 use super::*;
1924 use crate::plan::PlanNode;
1925
1926 #[test]
1927 fn test_plan_simple_scan() {
1928 let plan = plan("User").unwrap();
1929 assert!(matches!(plan, PlanNode::SeqScan { table } if table == "User"));
1930 }
1931
1932 #[test]
1933 fn test_plan_filter() {
1934 let plan = plan("User filter .age > 30").unwrap();
1935 assert!(matches!(plan, PlanNode::RangeScan { .. }));
1936 }
1937
1938 #[test]
1939 fn test_plan_filter_with_projection() {
1940 let plan = plan("User filter .age > 30 { name, email }").unwrap();
1941 assert!(matches!(plan, PlanNode::Project { .. }));
1942 }
1943
1944 #[test]
1945 fn test_plan_insert() {
1946 let plan = plan(r#"insert User { name := "Alice", age := 30 }"#).unwrap();
1947 assert!(matches!(plan, PlanNode::Insert { .. }));
1948 }
1949
1950 #[test]
1951 fn test_plan_order_limit() {
1952 let plan = plan("User order .name limit 10").unwrap();
1953 match plan {
1954 PlanNode::Limit { input, .. } => {
1955 assert!(matches!(*input, PlanNode::Sort { .. }));
1956 }
1957 _ => panic!("expected Limit(Sort(SeqScan))"),
1958 }
1959 }
1960
1961 #[test]
1962 fn test_plan_count() {
1963 let plan = plan("count(User)").unwrap();
1964 assert!(matches!(plan, PlanNode::Aggregate { .. }));
1965 }
1966
1967 #[test]
1968 fn single_source_aggregates_do_not_request_provenance() {
1969 for query in [
1970 "sum(User { .amount })",
1971 "avg(User { .amount })",
1972 "count(User { .amount })",
1973 ] {
1974 match plan(query).unwrap() {
1975 PlanNode::Aggregate {
1976 provenance_alias, ..
1977 } => assert!(
1978 provenance_alias.is_none(),
1979 "unexpected provenance for {query}"
1980 ),
1981 other => panic!("expected Aggregate for {query}, got {other:?}"),
1982 }
1983 }
1984
1985 match plan("User group .dept { total: sum(.amount) }").unwrap() {
1986 PlanNode::Project { input, .. } => match *input {
1987 PlanNode::GroupBy { aggregates, .. } => {
1988 assert!(aggregates[0].provenance_alias.is_none());
1989 }
1990 other => panic!("expected GroupBy, got {other:?}"),
1991 },
1992 other => panic!("expected Project(GroupBy), got {other:?}"),
1993 }
1994 }
1995
1996 #[test]
1997 fn join_provenance_is_limited_to_fanout_sensitive_aggregates() {
1998 let base = "Account as a join Entry as e on a.id = e.account_id group a.dept";
1999 for (function, expects_provenance) in [
2000 ("sum(a.balance)", true),
2001 ("avg(a.balance)", true),
2002 ("count(a.balance)", true),
2003 ("min(a.balance)", false),
2004 ("max(a.balance)", false),
2005 ("count(distinct a.balance)", false),
2006 ("count(*)", false),
2007 ] {
2008 let query = format!("{base} {{ value: {function} }}");
2009 match plan(&query).unwrap() {
2010 PlanNode::Project { input, .. } => match *input {
2011 PlanNode::GroupBy { aggregates, .. } => assert_eq!(
2012 aggregates[0].provenance_alias.as_deref(),
2013 expects_provenance.then_some("a"),
2014 "unexpected provenance selection for {function}"
2015 ),
2016 other => panic!("expected GroupBy for {function}, got {other:?}"),
2017 },
2018 other => panic!("expected Project(GroupBy) for {function}, got {other:?}"),
2019 }
2020 }
2021 }
2022
2023 #[test]
2024 fn test_plan_eq_becomes_index_scan() {
2025 let plan = plan("User filter .id = 42").unwrap();
2028 match plan {
2029 PlanNode::IndexScan { table, column, key } => {
2030 assert_eq!(table, "User");
2031 assert_eq!(column, "id");
2032 assert!(matches!(key, Expr::Literal(Literal::Int(42))));
2033 }
2034 other => panic!("expected IndexScan, got {other:?}"),
2035 }
2036 }
2037
2038 #[test]
2039 fn test_plan_eq_reversed_becomes_index_scan() {
2040 let plan = plan(r#"User filter "NYC" = .city"#).unwrap();
2042 assert!(matches!(plan, PlanNode::IndexScan { .. }));
2043 }
2044
2045 #[test]
2046 fn json_path_equality_and_reversed_equality_are_speculative_expression_scans() {
2047 for query in ["Post filter .data->age = 21", "Post filter 21 = .data->age"] {
2048 match plan(query).unwrap() {
2049 PlanNode::ExprIndexScan { table, path, key } => {
2050 assert_eq!(table, "Post");
2051 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
2052 assert!(matches!(key, Expr::Literal(Literal::Int(21))));
2053 }
2054 other => panic!("expected ExprIndexScan for `{query}`, got {other:?}"),
2055 }
2056 }
2057 }
2058
2059 #[test]
2060 fn json_path_range_and_same_path_compound_bounds_are_speculative_scans() {
2061 for query in ["Post filter .data->age > 18", "Post filter 18 < .data->age"] {
2062 match plan(query).unwrap() {
2063 PlanNode::ExprRangeScan {
2064 path, start, end, ..
2065 } => {
2066 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
2067 assert!(start.is_some());
2068 assert!(end.is_none());
2069 }
2070 other => panic!("expected ExprRangeScan for `{query}`, got {other:?}"),
2071 }
2072 }
2073
2074 match plan("Post filter .data->age >= 18 and .data->age < 65").unwrap() {
2075 PlanNode::ExprRangeScan {
2076 path, start, end, ..
2077 } => {
2078 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
2079 assert_eq!(start, Some((Expr::Literal(Literal::Int(18)), true)));
2080 assert_eq!(end, Some((Expr::Literal(Literal::Int(65)), false)));
2081 }
2082 other => panic!("expected bounded ExprRangeScan, got {other:?}"),
2083 }
2084
2085 assert!(matches!(
2086 plan("Post filter .data->age >= 18 and .data->score < 65").unwrap(),
2087 PlanNode::Filter { .. }
2088 ));
2089 }
2090
2091 #[test]
2092 fn exact_single_path_order_limit_uses_ordered_expression_scan() {
2093 match plan("Post order .data->age desc limit 10 offset 2 { .id }").unwrap() {
2094 PlanNode::Project { input, .. } => match *input {
2095 PlanNode::OrderedExprIndexScan {
2096 table,
2097 path,
2098 descending,
2099 limit,
2100 offset,
2101 } => {
2102 assert_eq!(table, "Post");
2103 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
2104 assert!(descending);
2105 assert_eq!(limit, Expr::Literal(Literal::Int(10)));
2106 assert_eq!(offset, Some(Expr::Literal(Literal::Int(2))));
2107 }
2108 other => panic!("expected OrderedExprIndexScan, got {other:?}"),
2109 },
2110 other => panic!("expected Project(OrderedExprIndexScan), got {other:?}"),
2111 }
2112 }
2113
2114 #[test]
2115 fn incompatible_path_order_shapes_keep_generic_sort() {
2116 for query in [
2117 "Post order .data->age",
2118 "Post order .data->age, .id limit 10",
2119 "Post filter .data->active = true order .data->age limit 10",
2120 "Post order .data->age limit .id",
2121 ] {
2122 let planned = plan(query).unwrap();
2123 assert!(
2124 !plan_contains_ordered_expr_scan(&planned),
2125 "`{query}` must remain on the generic pipeline: {planned:?}"
2126 );
2127 }
2128 }
2129
2130 fn plan_contains_ordered_expr_scan(plan: &PlanNode) -> bool {
2131 match plan {
2132 PlanNode::OrderedExprIndexScan { .. } => true,
2133 PlanNode::Filter { input, .. }
2134 | PlanNode::Project { input, .. }
2135 | PlanNode::Sort { input, .. }
2136 | PlanNode::Limit { input, .. }
2137 | PlanNode::Offset { input, .. }
2138 | PlanNode::Aggregate { input, .. }
2139 | PlanNode::Distinct { input }
2140 | PlanNode::GroupBy { input, .. }
2141 | PlanNode::Update { input, .. }
2142 | PlanNode::Delete { input, .. }
2143 | PlanNode::Window { input, .. }
2144 | PlanNode::Explain { input } => plan_contains_ordered_expr_scan(input),
2145 PlanNode::NestedLoopJoin { left, right, .. } | PlanNode::Union { left, right, .. } => {
2146 plan_contains_ordered_expr_scan(left) || plan_contains_ordered_expr_scan(right)
2147 }
2148 _ => false,
2149 }
2150 }
2151
2152 #[test]
2153 fn test_plan_non_eq_stays_filter() {
2154 let plan = plan("User filter .age > 30").unwrap();
2156 match plan {
2157 PlanNode::RangeScan {
2158 column, start, end, ..
2159 } => {
2160 assert_eq!(column, "age");
2161 assert!(start.is_some(), "expected lower bound");
2162 assert!(end.is_none(), "expected no upper bound");
2163 let (_, inclusive) = start.unwrap();
2164 assert!(!inclusive, "expected exclusive lower bound for >");
2165 }
2166 other => panic!("expected RangeScan, got {other:?}"),
2167 }
2168 }
2169
2170 #[test]
2171 fn test_plan_index_scan_with_projection() {
2172 let plan = plan("User filter .id = 1 { .name }").unwrap();
2174 match plan {
2175 PlanNode::Project { input, .. } => {
2176 assert!(matches!(*input, PlanNode::IndexScan { .. }));
2177 }
2178 other => panic!("expected Project(IndexScan), got {other:?}"),
2179 }
2180 }
2181
2182 #[test]
2183 fn test_plan_update_by_pk_becomes_index_scan() {
2184 let plan = plan("User filter .id = 42 update { age := 31 }").unwrap();
2187 match plan {
2188 PlanNode::Update { input, .. } => {
2189 assert!(
2190 matches!(*input, PlanNode::IndexScan { .. }),
2191 "expected Update(IndexScan), got {input:?}"
2192 );
2193 }
2194 other => panic!("expected Update, got {other:?}"),
2195 }
2196 }
2197
2198 #[test]
2199 fn test_plan_update_range_stays_range_scan() {
2200 let plan = plan("User filter .age > 30 update { age := 31 }").unwrap();
2201 match plan {
2202 PlanNode::Update { input, .. } => {
2203 assert!(
2204 matches!(*input, PlanNode::RangeScan { .. }),
2205 "expected Update(RangeScan), got {input:?}"
2206 );
2207 }
2208 other => panic!("expected Update, got {other:?}"),
2209 }
2210 }
2211
2212 #[test]
2213 fn test_plan_delete_by_pk_becomes_index_scan() {
2214 let plan = plan("User filter .id = 7 delete").unwrap();
2215 match plan {
2216 PlanNode::Delete { input, .. } => {
2217 assert!(matches!(*input, PlanNode::IndexScan { .. }));
2218 }
2219 other => panic!("expected Delete, got {other:?}"),
2220 }
2221 }
2222
2223 #[test]
2224 fn test_plan_inner_join_builds_nested_loop() {
2225 let plan = plan("User as u join Order as o on u.id = o.user_id").unwrap();
2228 match plan {
2229 PlanNode::NestedLoopJoin {
2230 left,
2231 right,
2232 on,
2233 kind,
2234 } => {
2235 assert_eq!(kind, JoinKind::Inner);
2236 assert!(on.is_some());
2237 assert!(matches!(*left, PlanNode::AliasScan { .. }));
2238 assert!(matches!(*right, PlanNode::AliasScan { .. }));
2239 }
2240 other => panic!("expected NestedLoopJoin, got {other:?}"),
2241 }
2242 }
2243
2244 #[test]
2245 fn duplicate_join_aliases_are_rejected_before_execution() {
2246 let err = plan("A as x join A as x on x.id = x.id").unwrap_err();
2247 assert!(
2248 err.to_string().contains("duplicate source alias `x`"),
2249 "unexpected error: {err}"
2250 );
2251 }
2252
2253 #[test]
2254 fn test_plan_right_join_rewritten_as_left_with_swapped_inputs() {
2255 let plan = plan("User as u right join Order as o on u.id = o.user_id").unwrap();
2256 match plan {
2257 PlanNode::NestedLoopJoin {
2258 left, right, kind, ..
2259 } => {
2260 assert_eq!(kind, JoinKind::LeftOuter);
2261 match *left {
2263 PlanNode::AliasScan { table, .. } => assert_eq!(table, "Order"),
2264 other => panic!("expected AliasScan(Order), got {other:?}"),
2265 }
2266 match *right {
2267 PlanNode::AliasScan { table, .. } => assert_eq!(table, "User"),
2268 other => panic!("expected AliasScan(User), got {other:?}"),
2269 }
2270 }
2271 other => panic!("expected NestedLoopJoin, got {other:?}"),
2272 }
2273 }
2274
2275 #[test]
2276 fn test_plan_multi_join_is_left_deep() {
2277 let plan = plan(
2279 "User as u join Order as o on u.id = o.user_id \
2280 join Product as p on o.product_id = p.id",
2281 )
2282 .unwrap();
2283 match plan {
2284 PlanNode::NestedLoopJoin { left, right, .. } => {
2285 match *right {
2287 PlanNode::AliasScan { table, .. } => assert_eq!(table, "Product"),
2288 other => panic!("expected AliasScan(Product), got {other:?}"),
2289 }
2290 assert!(matches!(*left, PlanNode::NestedLoopJoin { .. }));
2292 }
2293 other => panic!("expected NestedLoopJoin, got {other:?}"),
2294 }
2295 }
2296
2297 #[test]
2298 fn test_plan_join_with_filter_tail_wraps_filter_on_top() {
2299 let plan =
2300 plan("User as u join Order as o on u.id = o.user_id filter o.total > 100").unwrap();
2301 match plan {
2302 PlanNode::Filter { input, .. } => {
2303 assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
2304 }
2305 other => panic!("expected Filter(NestedLoopJoin), got {other:?}"),
2306 }
2307 }
2308
2309 #[test]
2310 fn test_plan_group_by_builds_groupby_node() {
2311 let plan = plan("User group .status { .status, n: count(.name) }").unwrap();
2312 match plan {
2314 PlanNode::Project { input, fields } => {
2315 assert_eq!(fields.len(), 2);
2316 match *input {
2317 PlanNode::GroupBy {
2318 input: inner,
2319 keys,
2320 aggregates,
2321 having,
2322 } => {
2323 assert!(matches!(*inner, PlanNode::SeqScan { .. }));
2324 assert_eq!(
2325 keys,
2326 vec![GroupKey {
2327 expr: Expr::Field("status".into()),
2328 output_name: "status".into(),
2329 }]
2330 );
2331 assert_eq!(aggregates.len(), 1);
2332 assert_eq!(aggregates[0].function, AggFunc::Count);
2333 assert_eq!(aggregates[0].argument, Expr::Field("name".into()));
2334 assert!(having.is_none());
2335 }
2336 other => panic!("expected GroupBy, got {other:?}"),
2337 }
2338 }
2339 other => panic!("expected Project, got {other:?}"),
2340 }
2341 }
2342
2343 #[test]
2344 fn test_plan_joined_group_applies_order_offset_limit_after_grouping() {
2345 let plan = plan(
2346 "User as u join Order as o on u.id = o.user_id \
2347 group u.status { u.status, n: count(*) } order n desc offset 1 limit 2",
2348 )
2349 .unwrap();
2350
2351 let PlanNode::Limit { input, .. } = plan else {
2352 panic!("expected Limit at the grouped-result boundary");
2353 };
2354 let PlanNode::Offset { input, .. } = *input else {
2355 panic!("expected Offset below Limit");
2356 };
2357 let PlanNode::Sort { input, .. } = *input else {
2358 panic!("expected Sort below Offset");
2359 };
2360 let PlanNode::Project { input, .. } = *input else {
2361 panic!("expected Project below Sort");
2362 };
2363 let PlanNode::GroupBy { input, .. } = *input else {
2364 panic!("expected GroupBy below Project");
2365 };
2366 assert!(
2367 matches!(*input, PlanNode::NestedLoopJoin { .. }),
2368 "joined rows must flow into GroupBy before result limiting"
2369 );
2370 }
2371
2372 #[test]
2373 fn test_plan_group_by_having_rewrites_agg_in_having() {
2374 let plan = plan("User group .status having count(.name) > 1 { .status }").unwrap();
2375 match plan {
2376 PlanNode::Project { input, .. } => {
2377 match *input {
2378 PlanNode::GroupBy {
2379 having, aggregates, ..
2380 } => {
2381 assert_eq!(aggregates.len(), 1);
2384 assert_eq!(aggregates[0].output_name, "__agg_0");
2385 let h = having.expect("having should be Some");
2386 match h {
2387 Expr::BinaryOp(l, BinOp::Gt, _) => {
2388 assert!(
2389 matches!(*l, Expr::Field(ref name) if name == "__agg_0"),
2390 "expected Field(__agg_0), got {l:?}"
2391 );
2392 }
2393 other => panic!("expected BinaryOp, got {other:?}"),
2394 }
2395 }
2396 other => panic!("expected GroupBy, got {other:?}"),
2397 }
2398 }
2399 other => panic!("expected Project, got {other:?}"),
2400 }
2401 }
2402
2403 #[test]
2404 fn test_plan_window_inserts_window_node_before_project() {
2405 let plan = plan("User { .name, rn: row_number() over (order .age) }").unwrap();
2406 match plan {
2408 PlanNode::Project { input, fields } => {
2409 assert_eq!(fields.len(), 2);
2410 assert!(
2412 matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"),
2413 "expected Field(__win_0), got {:?}",
2414 fields[1].expr
2415 );
2416 match *input {
2417 PlanNode::Window {
2418 input: inner,
2419 windows,
2420 } => {
2421 assert_eq!(windows.len(), 1);
2422 assert_eq!(windows[0].output_name, "__win_0");
2423 assert!(matches!(*inner, PlanNode::SeqScan { .. }));
2424 }
2425 other => panic!("expected Window, got {other:?}"),
2426 }
2427 }
2428 other => panic!("expected Project, got {other:?}"),
2429 }
2430 }
2431
2432 #[test]
2433 fn test_plan_multiple_windows() {
2434 let plan = plan(
2435 "User { .name, rn: row_number() over (order .age), s: sum(.salary) over (partition .dept order .salary) }"
2436 ).unwrap();
2437 match plan {
2438 PlanNode::Project { input, fields } => {
2439 assert_eq!(fields.len(), 3);
2440 assert!(matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"));
2441 assert!(matches!(&fields[2].expr, Expr::Field(name) if name == "__win_1"));
2442 match *input {
2443 PlanNode::Window { windows, .. } => {
2444 assert_eq!(windows.len(), 2);
2445 assert_eq!(windows[0].output_name, "__win_0");
2446 assert_eq!(windows[1].output_name, "__win_1");
2447 }
2448 other => panic!("expected Window, got {other:?}"),
2449 }
2450 }
2451 other => panic!("expected Project, got {other:?}"),
2452 }
2453 }
2454
2455 #[test]
2456 fn test_plan_no_window_without_over() {
2457 let plan = plan("User group .dept { .dept, total: sum(.salary) }").unwrap();
2459 match plan {
2460 PlanNode::Project { input, .. } => {
2461 assert!(
2463 matches!(*input, PlanNode::GroupBy { .. }),
2464 "expected GroupBy under Project, got {:?}",
2465 input
2466 );
2467 }
2468 other => panic!("expected Project, got {other:?}"),
2469 }
2470 }
2471
2472 #[test]
2473 fn test_plan_explain_wraps_inner() {
2474 let plan = plan("explain User filter .age > 30").unwrap();
2475 match plan {
2476 PlanNode::Explain { input } => {
2477 assert!(
2478 matches!(*input, PlanNode::RangeScan { .. }),
2479 "expected Explain(RangeScan), got {:?}",
2480 input
2481 );
2482 }
2483 other => panic!("expected Explain, got {other:?}"),
2484 }
2485 }
2486
2487 #[test]
2488 fn test_plan_explain_simple_scan() {
2489 let plan = plan("explain User").unwrap();
2490 match plan {
2491 PlanNode::Explain { input } => {
2492 assert!(matches!(*input, PlanNode::SeqScan { .. }));
2493 }
2494 other => panic!("expected Explain(SeqScan), got {other:?}"),
2495 }
2496 }
2497
2498 #[test]
2499 fn test_plan_explain_join() {
2500 let plan = plan("explain User as u join Order as o on u.id = o.user_id").unwrap();
2501 match plan {
2502 PlanNode::Explain { input } => {
2503 assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
2504 }
2505 other => panic!("expected Explain(NestedLoopJoin), got {other:?}"),
2506 }
2507 }
2508}