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)]
7enum RangeTarget {
8 Column(String),
9 JsonPath(StoredJsonPathV1),
10}
11
12type 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::AlterTable(at) => Ok(PlanNode::AlterTable {
68 table: at.table,
69 action: at.action,
70 }),
71 Statement::DropTable(dt) => Ok(PlanNode::DropTable {
72 name: dt.table,
73 if_exists: dt.if_exists,
74 }),
75 Statement::CreateView(cv) => Ok(PlanNode::CreateView {
76 name: cv.name,
77 query_text: cv.query_text,
78 }),
79 Statement::RefreshView(rv) => Ok(PlanNode::RefreshView { name: rv.name }),
80 Statement::DropView(dv) => Ok(PlanNode::DropView {
81 name: dv.name,
82 if_exists: dv.if_exists,
83 }),
84 Statement::ListTypes => Ok(PlanNode::ListTypes),
85 Statement::Describe(table) => Ok(PlanNode::Describe { table }),
86 Statement::Union(u) => {
87 let left = plan_statement(*u.left)?;
88 let right = plan_statement(*u.right)?;
89 Ok(PlanNode::Union {
90 left: Box::new(left),
91 right: Box::new(right),
92 all: u.all,
93 })
94 }
95 Statement::Upsert(ups) => plan_upsert(ups),
96 Statement::Begin => Ok(PlanNode::Begin),
97 Statement::Commit => Ok(PlanNode::Commit),
98 Statement::Rollback => Ok(PlanNode::Rollback),
99 Statement::Explain(inner) => {
100 let inner_plan = plan_statement(*inner)?;
101 Ok(PlanNode::Explain {
102 input: Box::new(inner_plan),
103 })
104 }
105 }
106}
107
108fn plan_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
109 if !q.joins.is_empty() {
115 return plan_joined_query(q);
116 }
117 let source_aliases = std::collections::HashSet::from([q.source.clone()]);
118 let ordered_expr_scan = try_extract_ordered_expr_index_scan(&q);
127 let (source, filter) = if let Some(scan) = ordered_expr_scan {
128 q.order = None;
132 q.limit = None;
133 q.offset = None;
134 (scan, None)
135 } else {
136 match q.filter {
137 Some(pred) => match try_extract_eq_index_key(&q.source, &pred) {
138 Some(index_scan) => (index_scan, None),
139 None => match try_extract_range_index_keys(&q.source, &pred) {
140 Some(range_scan) => (range_scan, None),
141 None => (
142 PlanNode::SeqScan {
143 table: q.source.clone(),
144 },
145 Some(pred),
146 ),
147 },
148 },
149 None => (
150 PlanNode::SeqScan {
151 table: q.source.clone(),
152 },
153 None,
154 ),
155 }
156 };
157 let mut node = source;
158
159 if let Some(pred) = filter {
160 node = PlanNode::Filter {
161 input: Box::new(node),
162 predicate: pred,
163 };
164 }
165
166 if let Some(group) = q.group_by {
169 let mut grouped_order = q.order;
170 let mut proj_fields: Vec<ProjectField> = q
171 .projection
172 .map(|proj| {
173 proj.into_iter()
174 .map(|pf| ProjectField {
175 alias: pf.alias,
176 expr: pf.expr,
177 })
178 .collect()
179 })
180 .unwrap_or_default();
181 let mut having = group.having;
182 let aggregates = extract_aggregates(&mut proj_fields, &mut having, &source_aliases)?;
183 rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
184 rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
185
186 node = PlanNode::GroupBy {
187 input: Box::new(node),
188 keys: group.keys,
189 aggregates,
190 having,
191 };
192
193 if !proj_fields.is_empty() {
194 node = PlanNode::Project {
195 input: Box::new(node),
196 fields: proj_fields,
197 };
198 }
199
200 if let Some(order) = grouped_order {
201 node = PlanNode::Sort {
202 input: Box::new(node),
203 keys: order
204 .keys
205 .into_iter()
206 .map(|k| SortKey {
207 expr: k.expr,
208 descending: k.descending,
209 })
210 .collect(),
211 };
212 }
213 if let Some(off) = q.offset {
217 node = PlanNode::Offset {
218 input: Box::new(node),
219 count: off,
220 };
221 }
222 if let Some(lim) = q.limit {
223 node = PlanNode::Limit {
224 input: Box::new(node),
225 count: lim,
226 };
227 }
228 if q.distinct {
229 node = PlanNode::Distinct {
230 input: Box::new(node),
231 };
232 }
233 return Ok(node);
234 }
235
236 if let Some(order) = q.order {
237 node = PlanNode::Sort {
238 input: Box::new(node),
239 keys: order
240 .keys
241 .into_iter()
242 .map(|k| SortKey {
243 expr: k.expr,
244 descending: k.descending,
245 })
246 .collect(),
247 };
248 }
249
250 if let Some(off) = q.offset {
254 node = PlanNode::Offset {
255 input: Box::new(node),
256 count: off,
257 };
258 }
259
260 if let Some(lim) = q.limit {
261 node = PlanNode::Limit {
262 input: Box::new(node),
263 count: lim,
264 };
265 }
266
267 if let Some(proj) = q.projection {
268 let mut fields: Vec<ProjectField> = proj
269 .into_iter()
270 .map(|pf| ProjectField {
271 alias: pf.alias,
272 expr: pf.expr,
273 })
274 .collect();
275 let windows = extract_windows(&mut fields);
276 if !windows.is_empty() {
277 node = PlanNode::Window {
278 input: Box::new(node),
279 windows,
280 };
281 }
282 node = PlanNode::Project {
283 input: Box::new(node),
284 fields,
285 };
286 }
287
288 if q.distinct {
289 node = PlanNode::Distinct {
290 input: Box::new(node),
291 };
292 }
293
294 if let Some(agg) = q.aggregation {
295 let provenance_alias = symmetric_provenance_alias(
296 agg.function,
297 agg.argument.as_ref(),
298 agg.mode,
299 &source_aliases,
300 )?;
301 node = PlanNode::Aggregate {
302 input: Box::new(node),
303 function: agg.function,
304 argument: agg.argument,
305 mode: agg.mode,
306 provenance_alias,
307 };
308 }
309
310 Ok(node)
311}
312
313fn plan_joined_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
335 let primary_alias = q.alias.clone().unwrap_or_else(|| q.source.clone());
336 let mut aliases = std::collections::HashSet::new();
337 aliases.insert(primary_alias.clone());
338 let mut node = PlanNode::AliasScan {
339 table: q.source.clone(),
340 alias: primary_alias,
341 };
342
343 for join in q.joins {
344 let right_alias = join.alias.unwrap_or_else(|| join.source.clone());
345 if !aliases.insert(right_alias.clone()) {
346 return Err(ParseError::Syntax {
347 message: format!(
348 "duplicate source alias `{right_alias}` in join; every joined source needs a unique alias"
349 ),
350 }
351 .into());
352 }
353 let right = PlanNode::AliasScan {
354 table: join.source,
355 alias: right_alias,
356 };
357 match join.kind {
358 JoinKind::Inner | JoinKind::LeftOuter | JoinKind::Cross => {
359 node = PlanNode::NestedLoopJoin {
360 left: Box::new(node),
361 right: Box::new(right),
362 on: join.on,
363 kind: join.kind,
364 };
365 }
366 JoinKind::RightOuter => {
367 node = PlanNode::NestedLoopJoin {
369 left: Box::new(right),
370 right: Box::new(node),
371 on: join.on,
372 kind: JoinKind::LeftOuter,
373 };
374 }
375 }
376 }
377
378 if let Some(pred) = q.filter {
379 node = PlanNode::Filter {
380 input: Box::new(node),
381 predicate: pred,
382 };
383 }
384
385 if q.group_by.is_none() {
386 if let Some(order) = q.order.take() {
387 node = PlanNode::Sort {
388 input: Box::new(node),
389 keys: order
390 .keys
391 .into_iter()
392 .map(|k| SortKey {
393 expr: k.expr,
394 descending: k.descending,
395 })
396 .collect(),
397 };
398 }
399 }
400
401 if let Some(group) = q.group_by {
403 let mut grouped_order = q.order;
404 let mut proj_fields: Vec<ProjectField> = q
405 .projection
406 .map(|proj| {
407 proj.into_iter()
408 .map(|pf| ProjectField {
409 alias: pf.alias,
410 expr: pf.expr,
411 })
412 .collect()
413 })
414 .unwrap_or_default();
415 let mut having = group.having;
416 let aggregates = extract_aggregates(&mut proj_fields, &mut having, &aliases)?;
417 rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
418 rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
419
420 node = PlanNode::GroupBy {
421 input: Box::new(node),
422 keys: group.keys,
423 aggregates,
424 having,
425 };
426
427 if !proj_fields.is_empty() {
428 node = PlanNode::Project {
429 input: Box::new(node),
430 fields: proj_fields,
431 };
432 }
433 if let Some(order) = grouped_order {
434 node = PlanNode::Sort {
435 input: Box::new(node),
436 keys: order
437 .keys
438 .into_iter()
439 .map(|key| SortKey {
440 expr: key.expr,
441 descending: key.descending,
442 })
443 .collect(),
444 };
445 }
446 if let Some(off) = q.offset {
451 node = PlanNode::Offset {
452 input: Box::new(node),
453 count: off,
454 };
455 }
456 if let Some(lim) = q.limit {
457 node = PlanNode::Limit {
458 input: Box::new(node),
459 count: lim,
460 };
461 }
462 if q.distinct {
463 node = PlanNode::Distinct {
464 input: Box::new(node),
465 };
466 }
467 return Ok(node);
468 }
469
470 if let Some(off) = q.offset {
474 node = PlanNode::Offset {
475 input: Box::new(node),
476 count: off,
477 };
478 }
479
480 if let Some(lim) = q.limit {
481 node = PlanNode::Limit {
482 input: Box::new(node),
483 count: lim,
484 };
485 }
486
487 if let Some(proj) = q.projection {
488 let mut fields: Vec<ProjectField> = proj
489 .into_iter()
490 .map(|pf| ProjectField {
491 alias: pf.alias,
492 expr: pf.expr,
493 })
494 .collect();
495 let windows = extract_windows(&mut fields);
496 if !windows.is_empty() {
497 node = PlanNode::Window {
498 input: Box::new(node),
499 windows,
500 };
501 }
502 node = PlanNode::Project {
503 input: Box::new(node),
504 fields,
505 };
506 }
507
508 if q.distinct {
509 node = PlanNode::Distinct {
510 input: Box::new(node),
511 };
512 }
513
514 if let Some(agg) = q.aggregation {
515 let provenance_alias =
516 symmetric_provenance_alias(agg.function, agg.argument.as_ref(), agg.mode, &aliases)?;
517 node = PlanNode::Aggregate {
518 input: Box::new(node),
519 function: agg.function,
520 argument: agg.argument,
521 mode: agg.mode,
522 provenance_alias,
523 };
524 }
525
526 Ok(node)
527}
528
529fn plan_insert(ins: InsertExpr) -> Result<PlanNode, PlanError> {
530 Ok(PlanNode::Insert {
531 table: ins.target,
532 rows: ins.rows,
533 returning: ins.returning,
534 })
535}
536
537fn plan_update(upd: UpdateExpr) -> Result<PlanNode, PlanError> {
538 let source = match upd.filter {
543 Some(pred) => match try_extract_eq_index_key(&upd.source, &pred) {
544 Some(index_scan) => index_scan,
545 None => match try_extract_range_index_keys(&upd.source, &pred) {
546 Some(range_scan) => range_scan,
547 None => PlanNode::Filter {
548 input: Box::new(PlanNode::SeqScan {
549 table: upd.source.clone(),
550 }),
551 predicate: pred,
552 },
553 },
554 },
555 None => PlanNode::SeqScan {
556 table: upd.source.clone(),
557 },
558 };
559 Ok(PlanNode::Update {
560 input: Box::new(source),
561 table: upd.source,
562 assignments: upd.assignments,
563 returning: upd.returning,
564 })
565}
566
567fn plan_delete(del: DeleteExpr) -> Result<PlanNode, PlanError> {
568 let source = match del.filter {
569 Some(pred) => match try_extract_eq_index_key(&del.source, &pred) {
570 Some(index_scan) => index_scan,
571 None => match try_extract_range_index_keys(&del.source, &pred) {
572 Some(range_scan) => range_scan,
573 None => PlanNode::Filter {
574 input: Box::new(PlanNode::SeqScan {
575 table: del.source.clone(),
576 }),
577 predicate: pred,
578 },
579 },
580 },
581 None => PlanNode::SeqScan {
582 table: del.source.clone(),
583 },
584 };
585 Ok(PlanNode::Delete {
586 input: Box::new(source),
587 table: del.source,
588 returning: del.returning,
589 })
590}
591
592fn plan_upsert(ups: UpsertExpr) -> Result<PlanNode, PlanError> {
593 Ok(PlanNode::Upsert {
594 table: ups.target,
595 key_column: ups.key_column,
596 assignments: ups.assignments,
597 on_conflict: ups.on_conflict,
598 })
599}
600
601fn plan_create_type(ct: CreateTypeExpr) -> Result<PlanNode, PlanError> {
602 let fields = ct
603 .fields
604 .into_iter()
605 .map(|f| crate::plan::CreateField {
606 name: f.name,
607 type_name: f.type_name,
608 required: f.required,
609 unique: f.unique,
610 default: f.default,
611 auto: f.auto,
612 })
613 .collect();
614 Ok(PlanNode::CreateTable {
615 name: ct.name,
616 fields,
617 if_not_exists: ct.if_not_exists,
618 })
619}
620
621fn try_extract_eq_index_key(table: &str, pred: &Expr) -> Option<PlanNode> {
631 let (lhs, op, rhs) = match pred {
632 Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
633 _ => return None,
634 };
635 if op != BinOp::Eq {
636 return None;
637 }
638 match (lhs, rhs) {
639 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some(PlanNode::ExprIndexScan {
640 table: table.to_string(),
641 path: stored_json_path(path)?,
642 key: rhs.clone(),
643 }),
644 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some(PlanNode::ExprIndexScan {
645 table: table.to_string(),
646 path: stored_json_path(path)?,
647 key: lhs.clone(),
648 }),
649 (Expr::Field(name), Expr::Literal(_)) => Some(PlanNode::IndexScan {
650 table: table.to_string(),
651 column: name.clone(),
652 key: rhs.clone(),
653 }),
654 (Expr::Literal(_), Expr::Field(name)) => Some(PlanNode::IndexScan {
655 table: table.to_string(),
656 column: name.clone(),
657 key: lhs.clone(),
658 }),
659 _ => None,
660 }
661}
662
663fn stored_json_path(expr: &Expr) -> Option<StoredJsonPathV1> {
664 JsonPathIdentityV1::from_expr(expr)?.bind_table_local(None)
665}
666
667fn extract_single_bound(pred: &Expr) -> Option<RangeBound> {
670 let (lhs, op, rhs) = match pred {
671 Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
672 _ => return None,
673 };
674 match op {
675 BinOp::Gt => match (lhs, rhs) {
677 (Expr::Field(name), Expr::Literal(_)) => Some((
678 RangeTarget::Column(name.clone()),
679 Some((rhs.clone(), false)),
680 None,
681 )),
682 (Expr::Literal(_), Expr::Field(name)) => {
683 Some((
685 RangeTarget::Column(name.clone()),
686 None,
687 Some((lhs.clone(), false)),
688 ))
689 }
690 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
691 RangeTarget::JsonPath(stored_json_path(path)?),
692 Some((rhs.clone(), false)),
693 None,
694 )),
695 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
696 RangeTarget::JsonPath(stored_json_path(path)?),
697 None,
698 Some((lhs.clone(), false)),
699 )),
700 _ => None,
701 },
702 BinOp::Gte => match (lhs, rhs) {
704 (Expr::Field(name), Expr::Literal(_)) => Some((
705 RangeTarget::Column(name.clone()),
706 Some((rhs.clone(), true)),
707 None,
708 )),
709 (Expr::Literal(_), Expr::Field(name)) => Some((
710 RangeTarget::Column(name.clone()),
711 None,
712 Some((lhs.clone(), true)),
713 )),
714 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
715 RangeTarget::JsonPath(stored_json_path(path)?),
716 Some((rhs.clone(), true)),
717 None,
718 )),
719 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
720 RangeTarget::JsonPath(stored_json_path(path)?),
721 None,
722 Some((lhs.clone(), true)),
723 )),
724 _ => None,
725 },
726 BinOp::Lt => match (lhs, rhs) {
728 (Expr::Field(name), Expr::Literal(_)) => Some((
729 RangeTarget::Column(name.clone()),
730 None,
731 Some((rhs.clone(), false)),
732 )),
733 (Expr::Literal(_), Expr::Field(name)) => Some((
734 RangeTarget::Column(name.clone()),
735 Some((lhs.clone(), false)),
736 None,
737 )),
738 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
739 RangeTarget::JsonPath(stored_json_path(path)?),
740 None,
741 Some((rhs.clone(), false)),
742 )),
743 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
744 RangeTarget::JsonPath(stored_json_path(path)?),
745 Some((lhs.clone(), false)),
746 None,
747 )),
748 _ => None,
749 },
750 BinOp::Lte => match (lhs, rhs) {
752 (Expr::Field(name), Expr::Literal(_)) => Some((
753 RangeTarget::Column(name.clone()),
754 None,
755 Some((rhs.clone(), true)),
756 )),
757 (Expr::Literal(_), Expr::Field(name)) => Some((
758 RangeTarget::Column(name.clone()),
759 Some((lhs.clone(), true)),
760 None,
761 )),
762 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
763 RangeTarget::JsonPath(stored_json_path(path)?),
764 None,
765 Some((rhs.clone(), true)),
766 )),
767 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
768 RangeTarget::JsonPath(stored_json_path(path)?),
769 Some((lhs.clone(), true)),
770 None,
771 )),
772 _ => None,
773 },
774 _ => None,
775 }
776}
777
778fn try_extract_range_index_keys(table: &str, pred: &Expr) -> Option<PlanNode> {
783 if let Expr::BinaryOp(lhs, BinOp::And, rhs) = pred {
785 if let (Some((col1, s1, e1)), Some((col2, s2, e2))) =
786 (extract_single_bound(lhs), extract_single_bound(rhs))
787 {
788 if col1 == col2 {
789 let start = s1.or(s2);
790 let end = e1.or(e2);
791 if start.is_some() || end.is_some() {
792 return Some(range_scan_for_target(table, col1, start, end));
793 }
794 }
795 }
796 }
797
798 if let Some((col, start, end)) = extract_single_bound(pred) {
800 return Some(range_scan_for_target(table, col, start, end));
801 }
802
803 None
804}
805
806fn range_scan_for_target(
807 table: &str,
808 target: RangeTarget,
809 start: Option<(Expr, bool)>,
810 end: Option<(Expr, bool)>,
811) -> PlanNode {
812 match target {
813 RangeTarget::Column(column) => PlanNode::RangeScan {
814 table: table.to_string(),
815 column,
816 start,
817 end,
818 },
819 RangeTarget::JsonPath(path) => PlanNode::ExprRangeScan {
820 table: table.to_string(),
821 path,
822 start,
823 end,
824 },
825 }
826}
827
828fn try_extract_ordered_expr_index_scan(query: &QueryExpr) -> Option<PlanNode> {
833 if query.alias.is_some()
834 || !query.joins.is_empty()
835 || query.filter.is_some()
836 || query.group_by.is_some()
837 || query.distinct
838 || query.aggregation.is_some()
839 || query.projection.as_ref().is_some_and(|fields| {
840 fields
841 .iter()
842 .any(|field| matches!(field.expr, Expr::Window { .. }))
843 })
844 {
845 return None;
846 }
847 let order = query.order.as_ref()?;
848 let [key] = order.keys.as_slice() else {
849 return None;
850 };
851 let path = stored_json_path(&key.expr)?;
852 let limit = query.limit.as_ref()?;
853 if !matches!(limit, Expr::Literal(Literal::Int(value)) if *value >= 0) {
854 return None;
855 }
856 if !query
857 .offset
858 .as_ref()
859 .is_none_or(|offset| matches!(offset, Expr::Literal(Literal::Int(value)) if *value >= 0))
860 {
861 return None;
862 }
863 Some(PlanNode::OrderedExprIndexScan {
864 table: query.source.clone(),
865 path,
866 descending: key.descending,
867 limit: limit.clone(),
868 offset: query.offset.clone(),
869 })
870}
871
872fn extract_windows(proj_fields: &mut [ProjectField]) -> Vec<WindowDef> {
877 let mut defs = Vec::new();
878 let mut counter = 0usize;
879 for f in proj_fields.iter_mut() {
880 if let Expr::Window {
881 function,
882 args,
883 mode,
884 partition_by,
885 order_by,
886 } = &f.expr
887 {
888 let output_name = format!("__win_{counter}");
889 defs.push(WindowDef {
890 function: *function,
891 args: args.clone(),
892 mode: *mode,
893 partition_by: partition_by.clone(),
894 order_by: order_by
895 .iter()
896 .map(|k| SortKey {
897 expr: k.expr.clone(),
898 descending: k.descending,
899 })
900 .collect(),
901 output_name: output_name.clone(),
902 });
903 f.expr = Expr::Field(output_name);
904 counter += 1;
905 }
906 }
907 defs
908}
909
910fn extract_aggregates(
916 proj_fields: &mut [ProjectField],
917 having: &mut Option<Expr>,
918 source_aliases: &std::collections::HashSet<String>,
919) -> Result<Vec<GroupAgg>, PlanError> {
920 let mut aggs: Vec<GroupAgg> = Vec::new();
921 let mut counter = 0usize;
922 for f in proj_fields.iter_mut() {
923 rewrite_agg_expr(&mut f.expr, &mut aggs, &mut counter, source_aliases)?;
924 }
925 if let Some(h) = having {
926 rewrite_agg_expr(h, &mut aggs, &mut counter, source_aliases)?;
927 }
928 Ok(aggs)
929}
930
931fn rewrite_group_key_references(
932 fields: &mut [ProjectField],
933 having: &mut Option<Expr>,
934 keys: &[GroupKey],
935) {
936 for field in fields {
937 rewrite_group_key_expr(&mut field.expr, keys);
938 }
939 if let Some(having) = having {
940 rewrite_group_key_expr(having, keys);
941 }
942}
943
944fn rewrite_group_order_keys(
945 order: Option<&mut OrderClause>,
946 projection: &[ProjectField],
947 keys: &[GroupKey],
948) {
949 let Some(order) = order else {
950 return;
951 };
952 for order_key in &mut order.keys {
953 let Some(group_key) = keys.iter().find(|key| key.expr == order_key.expr) else {
954 continue;
955 };
956 let projected_name = projection
957 .iter()
958 .find(|field| field.expr == group_key.expr)
959 .and_then(|field| field.alias.clone())
960 .unwrap_or_else(|| group_key.output_name());
961 order_key.expr = Expr::Field(projected_name);
962 }
963}
964
965fn rewrite_group_key_expr(expr: &mut Expr, keys: &[GroupKey]) {
966 if let Some(key) = keys.iter().find(|key| key.expr == *expr) {
967 *expr = Expr::Field(key.output_name());
968 return;
969 }
970 match expr {
971 Expr::FunctionCall(..) => {}
975 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
976 rewrite_group_key_expr(left, keys);
977 rewrite_group_key_expr(right, keys);
978 }
979 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) => rewrite_group_key_expr(inner, keys),
980 Expr::ScalarFunc(_, args) => {
981 for arg in args {
982 rewrite_group_key_expr(arg, keys);
983 }
984 }
985 Expr::InList { expr, list, .. } => {
986 rewrite_group_key_expr(expr, keys);
987 for item in list {
988 rewrite_group_key_expr(item, keys);
989 }
990 }
991 Expr::Case { whens, else_expr } => {
992 for (condition, result) in whens {
993 rewrite_group_key_expr(condition, keys);
994 rewrite_group_key_expr(result, keys);
995 }
996 if let Some(expr) = else_expr {
997 rewrite_group_key_expr(expr, keys);
998 }
999 }
1000 _ => {}
1001 }
1002}
1003
1004fn rewrite_agg_expr(
1005 expr: &mut Expr,
1006 aggs: &mut Vec<GroupAgg>,
1007 counter: &mut usize,
1008 source_aliases: &std::collections::HashSet<String>,
1009) -> Result<(), PlanError> {
1010 match expr {
1011 Expr::FunctionCall(func, inner, mode) => {
1012 let output = find_or_insert_agg(aggs, *func, inner, *mode, counter, source_aliases)?;
1013 *expr = Expr::Field(output);
1014 }
1015 Expr::BinaryOp(l, _, r) => {
1016 rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1017 rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1018 }
1019 Expr::UnaryOp(_, inner) => rewrite_agg_expr(inner, aggs, counter, source_aliases)?,
1020 Expr::Coalesce(l, r) => {
1021 rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1022 rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1023 }
1024 Expr::InList { expr: e, list, .. } => {
1025 rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1026 for item in list {
1027 rewrite_agg_expr(item, aggs, counter, source_aliases)?;
1028 }
1029 }
1030 Expr::InSubquery { expr: e, .. } => {
1031 rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1032 }
1033 _ => {}
1034 }
1035 Ok(())
1036}
1037
1038fn find_or_insert_agg(
1039 aggs: &mut Vec<GroupAgg>,
1040 func: AggFunc,
1041 argument: &Expr,
1042 mode: AggregateMode,
1043 counter: &mut usize,
1044 source_aliases: &std::collections::HashSet<String>,
1045) -> Result<String, PlanError> {
1046 for existing in aggs.iter() {
1047 if existing.function == func && existing.argument == *argument && existing.mode == mode {
1048 return Ok(existing.output_name.clone());
1049 }
1050 }
1051 let provenance_alias = symmetric_provenance_alias(func, Some(argument), mode, source_aliases)?;
1052 let output_name = format!("__agg_{counter}");
1053 aggs.push(GroupAgg {
1054 function: func,
1055 argument: argument.clone(),
1056 mode,
1057 provenance_alias,
1058 output_name: output_name.clone(),
1059 });
1060 *counter += 1;
1061 Ok(output_name)
1062}
1063
1064fn symmetric_provenance_alias(
1065 function: AggFunc,
1066 argument: Option<&Expr>,
1067 mode: AggregateMode,
1068 source_aliases: &std::collections::HashSet<String>,
1069) -> Result<Option<String>, PlanError> {
1070 if mode == AggregateMode::Raw
1071 || source_aliases.len() < 2
1072 || !matches!(function, AggFunc::Sum | AggFunc::Avg | AggFunc::Count)
1073 || (function == AggFunc::Count
1074 && argument.is_none_or(|argument| matches!(argument, Expr::Field(name) if name == "*")))
1075 {
1076 return Ok(None);
1077 }
1078 let Some(argument) = argument else {
1079 return Err(symmetric_aggregate_error(
1080 function,
1081 "does not reference a source row",
1082 ));
1083 };
1084
1085 let mut qualified = std::collections::HashSet::new();
1086 let mut has_unqualified = false;
1087 collect_expression_sources(argument, &mut qualified, &mut has_unqualified);
1088
1089 for alias in &qualified {
1090 if !source_aliases.contains(alias) {
1091 return Err(symmetric_aggregate_error(
1092 function,
1093 &format!("references unknown source alias '{alias}'"),
1094 ));
1095 }
1096 }
1097 if has_unqualified {
1098 if source_aliases.len() != 1 {
1099 return Err(symmetric_aggregate_error(
1100 function,
1101 "contains an ambiguous unqualified field",
1102 ));
1103 }
1104 qualified.extend(source_aliases.iter().cloned());
1105 }
1106 match qualified.len() {
1107 1 => Ok(qualified.into_iter().next()),
1108 0 => Err(symmetric_aggregate_error(
1109 function,
1110 "does not reference a source row",
1111 )),
1112 _ => Err(symmetric_aggregate_error(
1113 function,
1114 "references multiple source aliases",
1115 )),
1116 }
1117}
1118
1119fn symmetric_aggregate_error(function: AggFunc, reason: &str) -> PlanError {
1120 let name = format!("{function:?}").to_lowercase();
1121 PlanError::Semantic(format!(
1122 "symmetric {name} expression {reason}; reference exactly one source alias or use {name}(raw ...)"
1123 ))
1124}
1125
1126fn collect_expression_sources(
1127 expr: &Expr,
1128 qualified: &mut std::collections::HashSet<String>,
1129 has_unqualified: &mut bool,
1130) {
1131 match expr {
1132 Expr::Field(name) if name != "*" => *has_unqualified = true,
1133 Expr::QualifiedField { qualifier, .. } => {
1134 qualified.insert(qualifier.clone());
1135 }
1136 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1137 collect_expression_sources(left, qualified, has_unqualified);
1138 collect_expression_sources(right, qualified, has_unqualified);
1139 }
1140 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) | Expr::JsonPath { base: inner, .. } => {
1141 collect_expression_sources(inner, qualified, has_unqualified);
1142 }
1143 Expr::ScalarFunc(_, args) => {
1144 for argument in args {
1145 collect_expression_sources(argument, qualified, has_unqualified);
1146 }
1147 }
1148 Expr::InList { expr, list, .. } => {
1149 collect_expression_sources(expr, qualified, has_unqualified);
1150 for item in list {
1151 collect_expression_sources(item, qualified, has_unqualified);
1152 }
1153 }
1154 Expr::InSubquery { expr, .. } => {
1155 collect_expression_sources(expr, qualified, has_unqualified);
1156 }
1157 Expr::Case { whens, else_expr } => {
1158 for (condition, result) in whens {
1159 collect_expression_sources(condition, qualified, has_unqualified);
1160 collect_expression_sources(result, qualified, has_unqualified);
1161 }
1162 if let Some(expr) = else_expr {
1163 collect_expression_sources(expr, qualified, has_unqualified);
1164 }
1165 }
1166 Expr::Window {
1167 args,
1168 partition_by,
1169 order_by,
1170 ..
1171 } => {
1172 for expr in args.iter().chain(partition_by) {
1173 collect_expression_sources(expr, qualified, has_unqualified);
1174 }
1175 for key in order_by {
1176 collect_expression_sources(&key.expr, qualified, has_unqualified);
1177 }
1178 }
1179 Expr::FunctionCall(_, inner, _) => {
1180 collect_expression_sources(inner, qualified, has_unqualified);
1181 }
1182 Expr::ExistsSubquery { .. }
1183 | Expr::Field(_)
1184 | Expr::Literal(_)
1185 | Expr::Param(_)
1186 | Expr::ValueLit(_)
1187 | Expr::Null => {}
1188 }
1189}
1190
1191#[cfg(test)]
1192mod tests {
1193 use super::*;
1194 use crate::plan::PlanNode;
1195
1196 #[test]
1197 fn test_plan_simple_scan() {
1198 let plan = plan("User").unwrap();
1199 assert!(matches!(plan, PlanNode::SeqScan { table } if table == "User"));
1200 }
1201
1202 #[test]
1203 fn test_plan_filter() {
1204 let plan = plan("User filter .age > 30").unwrap();
1205 assert!(matches!(plan, PlanNode::RangeScan { .. }));
1206 }
1207
1208 #[test]
1209 fn test_plan_filter_with_projection() {
1210 let plan = plan("User filter .age > 30 { name, email }").unwrap();
1211 assert!(matches!(plan, PlanNode::Project { .. }));
1212 }
1213
1214 #[test]
1215 fn test_plan_insert() {
1216 let plan = plan(r#"insert User { name := "Alice", age := 30 }"#).unwrap();
1217 assert!(matches!(plan, PlanNode::Insert { .. }));
1218 }
1219
1220 #[test]
1221 fn test_plan_order_limit() {
1222 let plan = plan("User order .name limit 10").unwrap();
1223 match plan {
1224 PlanNode::Limit { input, .. } => {
1225 assert!(matches!(*input, PlanNode::Sort { .. }));
1226 }
1227 _ => panic!("expected Limit(Sort(SeqScan))"),
1228 }
1229 }
1230
1231 #[test]
1232 fn test_plan_count() {
1233 let plan = plan("count(User)").unwrap();
1234 assert!(matches!(plan, PlanNode::Aggregate { .. }));
1235 }
1236
1237 #[test]
1238 fn single_source_aggregates_do_not_request_provenance() {
1239 for query in [
1240 "sum(User { .amount })",
1241 "avg(User { .amount })",
1242 "count(User { .amount })",
1243 ] {
1244 match plan(query).unwrap() {
1245 PlanNode::Aggregate {
1246 provenance_alias, ..
1247 } => assert!(
1248 provenance_alias.is_none(),
1249 "unexpected provenance for {query}"
1250 ),
1251 other => panic!("expected Aggregate for {query}, got {other:?}"),
1252 }
1253 }
1254
1255 match plan("User group .dept { total: sum(.amount) }").unwrap() {
1256 PlanNode::Project { input, .. } => match *input {
1257 PlanNode::GroupBy { aggregates, .. } => {
1258 assert!(aggregates[0].provenance_alias.is_none());
1259 }
1260 other => panic!("expected GroupBy, got {other:?}"),
1261 },
1262 other => panic!("expected Project(GroupBy), got {other:?}"),
1263 }
1264 }
1265
1266 #[test]
1267 fn join_provenance_is_limited_to_fanout_sensitive_aggregates() {
1268 let base = "Account as a join Entry as e on a.id = e.account_id group a.dept";
1269 for (function, expects_provenance) in [
1270 ("sum(a.balance)", true),
1271 ("avg(a.balance)", true),
1272 ("count(a.balance)", true),
1273 ("min(a.balance)", false),
1274 ("max(a.balance)", false),
1275 ("count(distinct a.balance)", false),
1276 ("count(*)", false),
1277 ] {
1278 let query = format!("{base} {{ value: {function} }}");
1279 match plan(&query).unwrap() {
1280 PlanNode::Project { input, .. } => match *input {
1281 PlanNode::GroupBy { aggregates, .. } => assert_eq!(
1282 aggregates[0].provenance_alias.as_deref(),
1283 expects_provenance.then_some("a"),
1284 "unexpected provenance selection for {function}"
1285 ),
1286 other => panic!("expected GroupBy for {function}, got {other:?}"),
1287 },
1288 other => panic!("expected Project(GroupBy) for {function}, got {other:?}"),
1289 }
1290 }
1291 }
1292
1293 #[test]
1294 fn test_plan_eq_becomes_index_scan() {
1295 let plan = plan("User filter .id = 42").unwrap();
1298 match plan {
1299 PlanNode::IndexScan { table, column, key } => {
1300 assert_eq!(table, "User");
1301 assert_eq!(column, "id");
1302 assert!(matches!(key, Expr::Literal(Literal::Int(42))));
1303 }
1304 other => panic!("expected IndexScan, got {other:?}"),
1305 }
1306 }
1307
1308 #[test]
1309 fn test_plan_eq_reversed_becomes_index_scan() {
1310 let plan = plan(r#"User filter "NYC" = .city"#).unwrap();
1312 assert!(matches!(plan, PlanNode::IndexScan { .. }));
1313 }
1314
1315 #[test]
1316 fn json_path_equality_and_reversed_equality_are_speculative_expression_scans() {
1317 for query in ["Post filter .data->age = 21", "Post filter 21 = .data->age"] {
1318 match plan(query).unwrap() {
1319 PlanNode::ExprIndexScan { table, path, key } => {
1320 assert_eq!(table, "Post");
1321 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1322 assert!(matches!(key, Expr::Literal(Literal::Int(21))));
1323 }
1324 other => panic!("expected ExprIndexScan for `{query}`, got {other:?}"),
1325 }
1326 }
1327 }
1328
1329 #[test]
1330 fn json_path_range_and_same_path_compound_bounds_are_speculative_scans() {
1331 for query in ["Post filter .data->age > 18", "Post filter 18 < .data->age"] {
1332 match plan(query).unwrap() {
1333 PlanNode::ExprRangeScan {
1334 path, start, end, ..
1335 } => {
1336 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1337 assert!(start.is_some());
1338 assert!(end.is_none());
1339 }
1340 other => panic!("expected ExprRangeScan for `{query}`, got {other:?}"),
1341 }
1342 }
1343
1344 match plan("Post filter .data->age >= 18 and .data->age < 65").unwrap() {
1345 PlanNode::ExprRangeScan {
1346 path, start, end, ..
1347 } => {
1348 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1349 assert_eq!(start, Some((Expr::Literal(Literal::Int(18)), true)));
1350 assert_eq!(end, Some((Expr::Literal(Literal::Int(65)), false)));
1351 }
1352 other => panic!("expected bounded ExprRangeScan, got {other:?}"),
1353 }
1354
1355 assert!(matches!(
1356 plan("Post filter .data->age >= 18 and .data->score < 65").unwrap(),
1357 PlanNode::Filter { .. }
1358 ));
1359 }
1360
1361 #[test]
1362 fn exact_single_path_order_limit_uses_ordered_expression_scan() {
1363 match plan("Post order .data->age desc limit 10 offset 2 { .id }").unwrap() {
1364 PlanNode::Project { input, .. } => match *input {
1365 PlanNode::OrderedExprIndexScan {
1366 table,
1367 path,
1368 descending,
1369 limit,
1370 offset,
1371 } => {
1372 assert_eq!(table, "Post");
1373 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1374 assert!(descending);
1375 assert_eq!(limit, Expr::Literal(Literal::Int(10)));
1376 assert_eq!(offset, Some(Expr::Literal(Literal::Int(2))));
1377 }
1378 other => panic!("expected OrderedExprIndexScan, got {other:?}"),
1379 },
1380 other => panic!("expected Project(OrderedExprIndexScan), got {other:?}"),
1381 }
1382 }
1383
1384 #[test]
1385 fn incompatible_path_order_shapes_keep_generic_sort() {
1386 for query in [
1387 "Post order .data->age",
1388 "Post order .data->age, .id limit 10",
1389 "Post filter .data->active = true order .data->age limit 10",
1390 "Post order .data->age limit .id",
1391 ] {
1392 let planned = plan(query).unwrap();
1393 assert!(
1394 !plan_contains_ordered_expr_scan(&planned),
1395 "`{query}` must remain on the generic pipeline: {planned:?}"
1396 );
1397 }
1398 }
1399
1400 fn plan_contains_ordered_expr_scan(plan: &PlanNode) -> bool {
1401 match plan {
1402 PlanNode::OrderedExprIndexScan { .. } => true,
1403 PlanNode::Filter { input, .. }
1404 | PlanNode::Project { input, .. }
1405 | PlanNode::Sort { input, .. }
1406 | PlanNode::Limit { input, .. }
1407 | PlanNode::Offset { input, .. }
1408 | PlanNode::Aggregate { input, .. }
1409 | PlanNode::Distinct { input }
1410 | PlanNode::GroupBy { input, .. }
1411 | PlanNode::Update { input, .. }
1412 | PlanNode::Delete { input, .. }
1413 | PlanNode::Window { input, .. }
1414 | PlanNode::Explain { input } => plan_contains_ordered_expr_scan(input),
1415 PlanNode::NestedLoopJoin { left, right, .. } | PlanNode::Union { left, right, .. } => {
1416 plan_contains_ordered_expr_scan(left) || plan_contains_ordered_expr_scan(right)
1417 }
1418 _ => false,
1419 }
1420 }
1421
1422 #[test]
1423 fn test_plan_non_eq_stays_filter() {
1424 let plan = plan("User filter .age > 30").unwrap();
1426 match plan {
1427 PlanNode::RangeScan {
1428 column, start, end, ..
1429 } => {
1430 assert_eq!(column, "age");
1431 assert!(start.is_some(), "expected lower bound");
1432 assert!(end.is_none(), "expected no upper bound");
1433 let (_, inclusive) = start.unwrap();
1434 assert!(!inclusive, "expected exclusive lower bound for >");
1435 }
1436 other => panic!("expected RangeScan, got {other:?}"),
1437 }
1438 }
1439
1440 #[test]
1441 fn test_plan_index_scan_with_projection() {
1442 let plan = plan("User filter .id = 1 { .name }").unwrap();
1444 match plan {
1445 PlanNode::Project { input, .. } => {
1446 assert!(matches!(*input, PlanNode::IndexScan { .. }));
1447 }
1448 other => panic!("expected Project(IndexScan), got {other:?}"),
1449 }
1450 }
1451
1452 #[test]
1453 fn test_plan_update_by_pk_becomes_index_scan() {
1454 let plan = plan("User filter .id = 42 update { age := 31 }").unwrap();
1457 match plan {
1458 PlanNode::Update { input, .. } => {
1459 assert!(
1460 matches!(*input, PlanNode::IndexScan { .. }),
1461 "expected Update(IndexScan), got {input:?}"
1462 );
1463 }
1464 other => panic!("expected Update, got {other:?}"),
1465 }
1466 }
1467
1468 #[test]
1469 fn test_plan_update_range_stays_range_scan() {
1470 let plan = plan("User filter .age > 30 update { age := 31 }").unwrap();
1471 match plan {
1472 PlanNode::Update { input, .. } => {
1473 assert!(
1474 matches!(*input, PlanNode::RangeScan { .. }),
1475 "expected Update(RangeScan), got {input:?}"
1476 );
1477 }
1478 other => panic!("expected Update, got {other:?}"),
1479 }
1480 }
1481
1482 #[test]
1483 fn test_plan_delete_by_pk_becomes_index_scan() {
1484 let plan = plan("User filter .id = 7 delete").unwrap();
1485 match plan {
1486 PlanNode::Delete { input, .. } => {
1487 assert!(matches!(*input, PlanNode::IndexScan { .. }));
1488 }
1489 other => panic!("expected Delete, got {other:?}"),
1490 }
1491 }
1492
1493 #[test]
1494 fn test_plan_inner_join_builds_nested_loop() {
1495 let plan = plan("User as u join Order as o on u.id = o.user_id").unwrap();
1498 match plan {
1499 PlanNode::NestedLoopJoin {
1500 left,
1501 right,
1502 on,
1503 kind,
1504 } => {
1505 assert_eq!(kind, JoinKind::Inner);
1506 assert!(on.is_some());
1507 assert!(matches!(*left, PlanNode::AliasScan { .. }));
1508 assert!(matches!(*right, PlanNode::AliasScan { .. }));
1509 }
1510 other => panic!("expected NestedLoopJoin, got {other:?}"),
1511 }
1512 }
1513
1514 #[test]
1515 fn duplicate_join_aliases_are_rejected_before_execution() {
1516 let err = plan("A as x join A as x on x.id = x.id").unwrap_err();
1517 assert!(
1518 err.to_string().contains("duplicate source alias `x`"),
1519 "unexpected error: {err}"
1520 );
1521 }
1522
1523 #[test]
1524 fn test_plan_right_join_rewritten_as_left_with_swapped_inputs() {
1525 let plan = plan("User as u right join Order as o on u.id = o.user_id").unwrap();
1526 match plan {
1527 PlanNode::NestedLoopJoin {
1528 left, right, kind, ..
1529 } => {
1530 assert_eq!(kind, JoinKind::LeftOuter);
1531 match *left {
1533 PlanNode::AliasScan { table, .. } => assert_eq!(table, "Order"),
1534 other => panic!("expected AliasScan(Order), got {other:?}"),
1535 }
1536 match *right {
1537 PlanNode::AliasScan { table, .. } => assert_eq!(table, "User"),
1538 other => panic!("expected AliasScan(User), got {other:?}"),
1539 }
1540 }
1541 other => panic!("expected NestedLoopJoin, got {other:?}"),
1542 }
1543 }
1544
1545 #[test]
1546 fn test_plan_multi_join_is_left_deep() {
1547 let plan = plan(
1549 "User as u join Order as o on u.id = o.user_id \
1550 join Product as p on o.product_id = p.id",
1551 )
1552 .unwrap();
1553 match plan {
1554 PlanNode::NestedLoopJoin { left, right, .. } => {
1555 match *right {
1557 PlanNode::AliasScan { table, .. } => assert_eq!(table, "Product"),
1558 other => panic!("expected AliasScan(Product), got {other:?}"),
1559 }
1560 assert!(matches!(*left, PlanNode::NestedLoopJoin { .. }));
1562 }
1563 other => panic!("expected NestedLoopJoin, got {other:?}"),
1564 }
1565 }
1566
1567 #[test]
1568 fn test_plan_join_with_filter_tail_wraps_filter_on_top() {
1569 let plan =
1570 plan("User as u join Order as o on u.id = o.user_id filter o.total > 100").unwrap();
1571 match plan {
1572 PlanNode::Filter { input, .. } => {
1573 assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
1574 }
1575 other => panic!("expected Filter(NestedLoopJoin), got {other:?}"),
1576 }
1577 }
1578
1579 #[test]
1580 fn test_plan_group_by_builds_groupby_node() {
1581 let plan = plan("User group .status { .status, n: count(.name) }").unwrap();
1582 match plan {
1584 PlanNode::Project { input, fields } => {
1585 assert_eq!(fields.len(), 2);
1586 match *input {
1587 PlanNode::GroupBy {
1588 input: inner,
1589 keys,
1590 aggregates,
1591 having,
1592 } => {
1593 assert!(matches!(*inner, PlanNode::SeqScan { .. }));
1594 assert_eq!(
1595 keys,
1596 vec![GroupKey {
1597 expr: Expr::Field("status".into()),
1598 output_name: "status".into(),
1599 }]
1600 );
1601 assert_eq!(aggregates.len(), 1);
1602 assert_eq!(aggregates[0].function, AggFunc::Count);
1603 assert_eq!(aggregates[0].argument, Expr::Field("name".into()));
1604 assert!(having.is_none());
1605 }
1606 other => panic!("expected GroupBy, got {other:?}"),
1607 }
1608 }
1609 other => panic!("expected Project, got {other:?}"),
1610 }
1611 }
1612
1613 #[test]
1614 fn test_plan_joined_group_applies_order_offset_limit_after_grouping() {
1615 let plan = plan(
1616 "User as u join Order as o on u.id = o.user_id \
1617 group u.status { u.status, n: count(*) } order n desc offset 1 limit 2",
1618 )
1619 .unwrap();
1620
1621 let PlanNode::Limit { input, .. } = plan else {
1622 panic!("expected Limit at the grouped-result boundary");
1623 };
1624 let PlanNode::Offset { input, .. } = *input else {
1625 panic!("expected Offset below Limit");
1626 };
1627 let PlanNode::Sort { input, .. } = *input else {
1628 panic!("expected Sort below Offset");
1629 };
1630 let PlanNode::Project { input, .. } = *input else {
1631 panic!("expected Project below Sort");
1632 };
1633 let PlanNode::GroupBy { input, .. } = *input else {
1634 panic!("expected GroupBy below Project");
1635 };
1636 assert!(
1637 matches!(*input, PlanNode::NestedLoopJoin { .. }),
1638 "joined rows must flow into GroupBy before result limiting"
1639 );
1640 }
1641
1642 #[test]
1643 fn test_plan_group_by_having_rewrites_agg_in_having() {
1644 let plan = plan("User group .status having count(.name) > 1 { .status }").unwrap();
1645 match plan {
1646 PlanNode::Project { input, .. } => {
1647 match *input {
1648 PlanNode::GroupBy {
1649 having, aggregates, ..
1650 } => {
1651 assert_eq!(aggregates.len(), 1);
1654 assert_eq!(aggregates[0].output_name, "__agg_0");
1655 let h = having.expect("having should be Some");
1656 match h {
1657 Expr::BinaryOp(l, BinOp::Gt, _) => {
1658 assert!(
1659 matches!(*l, Expr::Field(ref name) if name == "__agg_0"),
1660 "expected Field(__agg_0), got {l:?}"
1661 );
1662 }
1663 other => panic!("expected BinaryOp, got {other:?}"),
1664 }
1665 }
1666 other => panic!("expected GroupBy, got {other:?}"),
1667 }
1668 }
1669 other => panic!("expected Project, got {other:?}"),
1670 }
1671 }
1672
1673 #[test]
1674 fn test_plan_window_inserts_window_node_before_project() {
1675 let plan = plan("User { .name, rn: row_number() over (order .age) }").unwrap();
1676 match plan {
1678 PlanNode::Project { input, fields } => {
1679 assert_eq!(fields.len(), 2);
1680 assert!(
1682 matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"),
1683 "expected Field(__win_0), got {:?}",
1684 fields[1].expr
1685 );
1686 match *input {
1687 PlanNode::Window {
1688 input: inner,
1689 windows,
1690 } => {
1691 assert_eq!(windows.len(), 1);
1692 assert_eq!(windows[0].output_name, "__win_0");
1693 assert!(matches!(*inner, PlanNode::SeqScan { .. }));
1694 }
1695 other => panic!("expected Window, got {other:?}"),
1696 }
1697 }
1698 other => panic!("expected Project, got {other:?}"),
1699 }
1700 }
1701
1702 #[test]
1703 fn test_plan_multiple_windows() {
1704 let plan = plan(
1705 "User { .name, rn: row_number() over (order .age), s: sum(.salary) over (partition .dept order .salary) }"
1706 ).unwrap();
1707 match plan {
1708 PlanNode::Project { input, fields } => {
1709 assert_eq!(fields.len(), 3);
1710 assert!(matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"));
1711 assert!(matches!(&fields[2].expr, Expr::Field(name) if name == "__win_1"));
1712 match *input {
1713 PlanNode::Window { windows, .. } => {
1714 assert_eq!(windows.len(), 2);
1715 assert_eq!(windows[0].output_name, "__win_0");
1716 assert_eq!(windows[1].output_name, "__win_1");
1717 }
1718 other => panic!("expected Window, got {other:?}"),
1719 }
1720 }
1721 other => panic!("expected Project, got {other:?}"),
1722 }
1723 }
1724
1725 #[test]
1726 fn test_plan_no_window_without_over() {
1727 let plan = plan("User group .dept { .dept, total: sum(.salary) }").unwrap();
1729 match plan {
1730 PlanNode::Project { input, .. } => {
1731 assert!(
1733 matches!(*input, PlanNode::GroupBy { .. }),
1734 "expected GroupBy under Project, got {:?}",
1735 input
1736 );
1737 }
1738 other => panic!("expected Project, got {other:?}"),
1739 }
1740 }
1741
1742 #[test]
1743 fn test_plan_explain_wraps_inner() {
1744 let plan = plan("explain User filter .age > 30").unwrap();
1745 match plan {
1746 PlanNode::Explain { input } => {
1747 assert!(
1748 matches!(*input, PlanNode::RangeScan { .. }),
1749 "expected Explain(RangeScan), got {:?}",
1750 input
1751 );
1752 }
1753 other => panic!("expected Explain, got {other:?}"),
1754 }
1755 }
1756
1757 #[test]
1758 fn test_plan_explain_simple_scan() {
1759 let plan = plan("explain User").unwrap();
1760 match plan {
1761 PlanNode::Explain { input } => {
1762 assert!(matches!(*input, PlanNode::SeqScan { .. }));
1763 }
1764 other => panic!("expected Explain(SeqScan), got {other:?}"),
1765 }
1766 }
1767
1768 #[test]
1769 fn test_plan_explain_join() {
1770 let plan = plan("explain User as u join Order as o on u.id = o.user_id").unwrap();
1771 match plan {
1772 PlanNode::Explain { input } => {
1773 assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
1774 }
1775 other => panic!("expected Explain(NestedLoopJoin), got {other:?}"),
1776 }
1777 }
1778}