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