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.projection.as_ref().is_some_and(|proj| {
112 proj.iter()
113 .any(|pf| matches!(pf.expr, Expr::NestedQuery(_)))
114 }) {
115 return plan_nested_query(q);
116 }
117 if !q.joins.is_empty() {
123 return plan_joined_query(q);
124 }
125 let visible = q.alias.clone().unwrap_or_else(|| q.source.clone());
130 if let Some(filter) = q.filter.as_mut() {
131 resolve_scan_qualifiers(filter, &visible)?;
132 }
133 if let Some(proj) = q.projection.as_mut() {
134 for pf in proj.iter_mut() {
135 resolve_scan_qualifiers(&mut pf.expr, &visible)?;
136 }
137 }
138 if let Some(order) = q.order.as_mut() {
139 for key in order.keys.iter_mut() {
140 resolve_scan_qualifiers(&mut key.expr, &visible)?;
141 }
142 }
143 if let Some(group) = q.group_by.as_mut() {
144 for key in group.keys.iter_mut() {
145 resolve_scan_qualifiers(&mut key.expr, &visible)?;
146 }
147 if let Some(having) = group.having.as_mut() {
148 resolve_scan_qualifiers(having, &visible)?;
149 }
150 }
151 if let Some(agg) = q.aggregation.as_mut() {
152 if let Some(arg) = agg.argument.as_mut() {
153 resolve_scan_qualifiers(arg, &visible)?;
154 }
155 }
156 let source_aliases = std::collections::HashSet::from([q.source.clone()]);
157 let ordered_expr_scan = try_extract_ordered_expr_index_scan(&q);
169 let (source, filter) = if let Some(scan) = ordered_expr_scan {
170 q.order = None;
174 q.limit = None;
175 q.offset = None;
176 (scan, None)
177 } else {
178 match q.filter {
179 Some(pred) => match try_extract_eq_index_key(&q.source, &pred) {
180 Some(index_scan) => (index_scan, None),
181 None => match try_extract_range_index_keys(&q.source, &pred) {
182 Some(range_scan) => (range_scan, None),
183 None => (
184 PlanNode::SeqScan {
185 table: q.source.clone(),
186 },
187 Some(pred),
188 ),
189 },
190 },
191 None => (
192 PlanNode::SeqScan {
193 table: q.source.clone(),
194 },
195 None,
196 ),
197 }
198 };
199 let mut node = source;
200
201 if let Some(pred) = filter {
202 node = PlanNode::Filter {
203 input: Box::new(node),
204 predicate: pred,
205 };
206 }
207
208 if let Some(group) = q.group_by {
211 let mut grouped_order = q.order;
212 let mut proj_fields: Vec<ProjectField> = q
213 .projection
214 .map(|proj| {
215 proj.into_iter()
216 .map(|pf| ProjectField {
217 alias: pf.alias,
218 expr: pf.expr,
219 })
220 .collect()
221 })
222 .unwrap_or_default();
223 let mut having = group.having;
224 let aggregates = extract_aggregates(&mut proj_fields, &mut having, &source_aliases)?;
225 rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
226 rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
227
228 node = PlanNode::GroupBy {
229 input: Box::new(node),
230 keys: group.keys,
231 aggregates,
232 having,
233 };
234
235 if !proj_fields.is_empty() {
236 node = PlanNode::Project {
237 input: Box::new(node),
238 fields: proj_fields,
239 };
240 }
241
242 if let Some(order) = grouped_order {
243 node = PlanNode::Sort {
244 input: Box::new(node),
245 keys: order
246 .keys
247 .into_iter()
248 .map(|k| SortKey {
249 expr: k.expr,
250 descending: k.descending,
251 })
252 .collect(),
253 };
254 }
255 if let Some(off) = q.offset {
259 node = PlanNode::Offset {
260 input: Box::new(node),
261 count: off,
262 };
263 }
264 if let Some(lim) = q.limit {
265 node = PlanNode::Limit {
266 input: Box::new(node),
267 count: lim,
268 };
269 }
270 if q.distinct {
271 node = PlanNode::Distinct {
272 input: Box::new(node),
273 };
274 }
275 return Ok(node);
276 }
277
278 if let Some(order) = q.order {
279 node = PlanNode::Sort {
280 input: Box::new(node),
281 keys: order
282 .keys
283 .into_iter()
284 .map(|k| SortKey {
285 expr: k.expr,
286 descending: k.descending,
287 })
288 .collect(),
289 };
290 }
291
292 if let Some(off) = q.offset {
296 node = PlanNode::Offset {
297 input: Box::new(node),
298 count: off,
299 };
300 }
301
302 if let Some(lim) = q.limit {
303 node = PlanNode::Limit {
304 input: Box::new(node),
305 count: lim,
306 };
307 }
308
309 if let Some(proj) = q.projection {
310 let mut fields: Vec<ProjectField> = proj
311 .into_iter()
312 .map(|pf| ProjectField {
313 alias: pf.alias,
314 expr: pf.expr,
315 })
316 .collect();
317 let windows = extract_windows(&mut fields);
318 if !windows.is_empty() {
319 node = PlanNode::Window {
320 input: Box::new(node),
321 windows,
322 };
323 }
324 node = PlanNode::Project {
325 input: Box::new(node),
326 fields,
327 };
328 }
329
330 if q.distinct {
331 node = PlanNode::Distinct {
332 input: Box::new(node),
333 };
334 }
335
336 if let Some(agg) = q.aggregation {
337 let provenance_alias = symmetric_provenance_alias(
338 agg.function,
339 agg.argument.as_ref(),
340 agg.mode,
341 &source_aliases,
342 )?;
343 node = PlanNode::Aggregate {
344 input: Box::new(node),
345 function: agg.function,
346 argument: agg.argument,
347 mode: agg.mode,
348 provenance_alias,
349 };
350 }
351
352 Ok(node)
353}
354
355fn resolve_scan_qualifiers(expr: &mut Expr, visible: &str) -> Result<(), PlanError> {
376 match expr {
377 Expr::QualifiedField { qualifier, field } => {
378 if qualifier == visible {
379 *expr = Expr::Field(std::mem::take(field));
380 Ok(())
381 } else {
382 Err(PlanError::Semantic(format!(
383 "no such column: `{qualifier}.{field}` (the only table in this \
384 query is `{visible}`)"
385 )))
386 }
387 }
388 Expr::Field(_) | Expr::Literal(_) | Expr::Param(_) | Expr::ValueLit(_) | Expr::Null => {
389 Ok(())
390 }
391 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
392 resolve_scan_qualifiers(left, visible)?;
393 resolve_scan_qualifiers(right, visible)
394 }
395 Expr::UnaryOp(_, inner)
396 | Expr::Cast(inner, _)
397 | Expr::FunctionCall(_, inner, _)
398 | Expr::JsonPath { base: inner, .. } => resolve_scan_qualifiers(inner, visible),
399 Expr::ScalarFunc(_, args) => {
400 for arg in args.iter_mut() {
401 resolve_scan_qualifiers(arg, visible)?;
402 }
403 Ok(())
404 }
405 Expr::InList { expr, list, .. } => {
406 resolve_scan_qualifiers(expr, visible)?;
407 for item in list.iter_mut() {
408 resolve_scan_qualifiers(item, visible)?;
409 }
410 Ok(())
411 }
412 Expr::Case { whens, else_expr } => {
413 for (cond, res) in whens.iter_mut() {
414 resolve_scan_qualifiers(cond, visible)?;
415 resolve_scan_qualifiers(res, visible)?;
416 }
417 if let Some(e) = else_expr {
418 resolve_scan_qualifiers(e, visible)?;
419 }
420 Ok(())
421 }
422 Expr::Window {
423 args,
424 partition_by,
425 order_by,
426 ..
427 } => {
428 for a in args.iter_mut() {
429 resolve_scan_qualifiers(a, visible)?;
430 }
431 for p in partition_by.iter_mut() {
432 resolve_scan_qualifiers(p, visible)?;
433 }
434 for k in order_by.iter_mut() {
435 resolve_scan_qualifiers(&mut k.expr, visible)?;
436 }
437 Ok(())
438 }
439 Expr::InSubquery { .. } | Expr::ExistsSubquery { .. } | Expr::NestedQuery(_) => Ok(()),
442 }
443}
444
445fn plan_nested_query(q: QueryExpr) -> Result<PlanNode, PlanError> {
452 if !q.joins.is_empty() || q.group_by.is_some() || q.aggregation.is_some() || q.distinct {
453 return Err(PlanError::Semantic(
454 "nested projections require a plain aliased table scan (no joins, \
455 group, distinct, or aggregation)"
456 .into(),
457 ));
458 }
459 let parent_alias = q.alias.unwrap_or_else(|| q.source.clone());
460 let mut node = PlanNode::AliasScan {
461 table: q.source,
462 alias: parent_alias.clone(),
463 };
464 if let Some(pred) = q.filter {
465 node = PlanNode::Filter {
466 input: Box::new(node),
467 predicate: pred,
468 };
469 }
470 if let Some(order) = q.order {
471 node = PlanNode::Sort {
472 input: Box::new(node),
473 keys: order
474 .keys
475 .into_iter()
476 .map(|k| SortKey {
477 expr: k.expr,
478 descending: k.descending,
479 })
480 .collect(),
481 };
482 }
483 if let Some(off) = q.offset {
484 node = PlanNode::Offset {
485 input: Box::new(node),
486 count: off,
487 };
488 }
489 if let Some(lim) = q.limit {
490 node = PlanNode::Limit {
491 input: Box::new(node),
492 count: lim,
493 };
494 }
495 let fields = q
496 .projection
497 .expect("plan_nested_query is only called with a projection")
498 .into_iter()
499 .map(|pf| match pf.expr {
500 Expr::NestedQuery(nested) => {
501 let name = pf.alias.ok_or_else(|| {
502 PlanError::Semantic("nested projection field requires a name".into())
503 })?;
504 resolve_nested_projection(name, *nested, &parent_alias)
505 .map(|nested| NestedProjectField::Nested(Box::new(nested)))
506 }
507 expr => Ok(NestedProjectField::Plain(ProjectField {
508 alias: pf.alias,
509 expr,
510 })),
511 })
512 .collect::<Result<Vec<_>, PlanError>>()?;
513 Ok(PlanNode::NestedProject {
514 input: Box::new(node),
515 fields,
516 })
517}
518
519fn resolve_nested_projection(
525 name: String,
526 nested: NestedQuery,
527 parent_alias: &str,
528) -> Result<NestedProjection, PlanError> {
529 resolve_nested_projection_inner(name, nested, parent_alias, true)
530}
531
532fn resolve_nested_projection_inner(
536 name: String,
537 nested: NestedQuery,
538 parent_alias: &str,
539 qualify_parent_key: bool,
540) -> Result<NestedProjection, PlanError> {
541 let mut conjuncts = Vec::new();
542 split_and_chain(nested.filter, &mut conjuncts);
543
544 let correlation_of = |expr: &Expr| -> Option<(String, String)> {
546 let Expr::BinaryOp(left, BinOp::Eq, right) = expr else {
547 return None;
548 };
549 let side = |expr: &Expr| match expr {
550 Expr::QualifiedField { qualifier, field } => Some((qualifier.clone(), field.clone())),
551 _ => None,
552 };
553 let ((lq, lf), (rq, rf)) = (side(left)?, side(right)?);
554 if lq == nested.alias && rq == parent_alias {
555 Some((lf, rf))
556 } else if rq == nested.alias && lq == parent_alias {
557 Some((rf, lf))
558 } else {
559 None
560 }
561 };
562 let mut correlation: Option<(String, String)> = None;
563 let mut residual: Option<Expr> = None;
564 for conjunct in conjuncts {
565 match correlation_of(&conjunct) {
566 Some(keys) if correlation.is_none() => correlation = Some(keys),
567 Some(_) => {
568 return Err(PlanError::Semantic(format!(
569 "nested projection `{name}` links `{child}` to `{parent}` more than \
570 once; exactly one correlation predicate \
571 ({child}.<col> = {parent}.<col>) is supported",
572 child = nested.alias,
573 parent = parent_alias,
574 )))
575 }
576 None => {
577 let rewritten =
578 rewrite_residual_condition(conjunct, &name, &nested.alias, parent_alias)?;
579 residual = Some(match residual {
580 Some(existing) => {
581 Expr::BinaryOp(Box::new(existing), BinOp::And, Box::new(rewritten))
582 }
583 None => rewritten,
584 });
585 }
586 }
587 }
588 let Some((child_key, parent_key)) = correlation else {
589 return Err(PlanError::Semantic(format!(
590 "nested projection `{name}` requires an equi-correlation predicate linking \
591 `{child}` to the outer query ({child}.<col> = {parent}.<col>) somewhere in \
592 its filter",
593 child = nested.alias,
594 parent = parent_alias,
595 )));
596 };
597 let order = nested
598 .order
599 .map(|clause| {
600 clause
601 .keys
602 .into_iter()
603 .map(|key| {
604 let column = match &key.expr {
605 Expr::Field(field) => field.clone(),
606 Expr::QualifiedField { qualifier, field } if *qualifier == nested.alias => {
607 field.clone()
608 }
609 _ => {
610 return Err(PlanError::Semantic(format!(
611 "nested projection `{name}` order keys must be plain \
612 columns of `{child}` (`{child}.<col>` or `.<col>`)",
613 child = nested.alias,
614 )))
615 }
616 };
617 Ok((column, key.descending))
618 })
619 .collect::<Result<Vec<_>, PlanError>>()
620 })
621 .transpose()?
622 .unwrap_or_default();
623 let fields = nested
624 .fields
625 .into_iter()
626 .map(|pf| {
627 if let Expr::NestedQuery(inner) = pf.expr {
628 let inner_name = pf.alias.ok_or_else(|| {
629 PlanError::Semantic(
630 "nested projection field requires a name \
631 (`<name>: <Table> as <alias> ...`)"
632 .into(),
633 )
634 })?;
635 return resolve_nested_projection_inner(inner_name, *inner, &nested.alias, false)
638 .map(|inner| NestedField::Nested(Box::new(inner)));
639 }
640 let column = match &pf.expr {
641 Expr::Field(field) => field.clone(),
642 Expr::QualifiedField { qualifier, field } if *qualifier == nested.alias => {
643 field.clone()
644 }
645 _ => {
646 return Err(PlanError::Semantic(format!(
647 "nested projection `{name}` fields must be plain columns of `{}` \
648 (`{}.<col>` or `.<col>`) or a deeper nested projection",
649 nested.alias, nested.alias
650 )))
651 }
652 };
653 let key = pf.alias.unwrap_or_else(|| column.clone());
654 Ok(NestedField::Scalar { key, column })
655 })
656 .collect::<Result<Vec<_>, PlanError>>()?;
657 Ok(NestedProjection {
658 name,
659 table: nested.source,
660 alias: nested.alias,
661 parent_alias: parent_alias.to_string(),
662 child_key,
663 parent_key: if qualify_parent_key {
664 format!("{parent_alias}.{parent_key}")
665 } else {
666 parent_key
667 },
668 residual,
669 order,
670 limit: nested.limit,
671 offset: nested.offset,
672 offset_before_limit: nested.offset_before_limit,
673 fields,
674 })
675}
676
677fn split_and_chain(expr: Expr, out: &mut Vec<Expr>) {
679 match expr {
680 Expr::BinaryOp(left, BinOp::And, right) => {
681 split_and_chain(*left, out);
682 split_and_chain(*right, out);
683 }
684 other => out.push(other),
685 }
686}
687
688fn rewrite_residual_condition(
694 expr: Expr,
695 name: &str,
696 child_alias: &str,
697 parent_alias: &str,
698) -> Result<Expr, PlanError> {
699 let rewrite = |inner: Box<Expr>| -> Result<Box<Expr>, PlanError> {
700 Ok(Box::new(rewrite_residual_condition(
701 *inner,
702 name,
703 child_alias,
704 parent_alias,
705 )?))
706 };
707 match expr {
708 Expr::QualifiedField { qualifier, field } => {
709 if qualifier == child_alias {
710 Ok(Expr::Field(field))
711 } else if qualifier == parent_alias {
712 Err(PlanError::Semantic(format!(
713 "nested projection `{name}` filter references outer alias \
714 `{parent_alias}` (`{parent_alias}.{field}`) outside the correlation \
715 predicate; move that condition to the outer query's filter"
716 )))
717 } else {
718 Err(PlanError::Semantic(format!(
719 "nested projection `{name}` filter references unknown alias \
720 `{qualifier}`; only columns of `{child_alias}` may be used"
721 )))
722 }
723 }
724 Expr::Field(_) | Expr::Literal(_) | Expr::Param(_) | Expr::ValueLit(_) | Expr::Null => {
725 Ok(expr)
726 }
727 Expr::BinaryOp(left, op, right) => Ok(Expr::BinaryOp(rewrite(left)?, op, rewrite(right)?)),
728 Expr::UnaryOp(op, inner) => Ok(Expr::UnaryOp(op, rewrite(inner)?)),
729 Expr::Coalesce(left, right) => Ok(Expr::Coalesce(rewrite(left)?, rewrite(right)?)),
730 Expr::Cast(inner, ty) => Ok(Expr::Cast(rewrite(inner)?, ty)),
731 Expr::ScalarFunc(func, args) => Ok(Expr::ScalarFunc(
732 func,
733 args.into_iter()
734 .map(|arg| rewrite_residual_condition(arg, name, child_alias, parent_alias))
735 .collect::<Result<Vec<_>, _>>()?,
736 )),
737 Expr::InList {
738 expr,
739 list,
740 negated,
741 } => Ok(Expr::InList {
742 expr: rewrite(expr)?,
743 list: list
744 .into_iter()
745 .map(|item| rewrite_residual_condition(item, name, child_alias, parent_alias))
746 .collect::<Result<Vec<_>, _>>()?,
747 negated,
748 }),
749 Expr::Case { whens, else_expr } => Ok(Expr::Case {
750 whens: whens
751 .into_iter()
752 .map(|(cond, result)| Ok((rewrite(cond)?, rewrite(result)?)))
753 .collect::<Result<Vec<_>, PlanError>>()?,
754 else_expr: else_expr.map(rewrite).transpose()?,
755 }),
756 Expr::JsonPath { base, segments } => Ok(Expr::JsonPath {
757 base: rewrite(base)?,
758 segments,
759 }),
760 Expr::InSubquery { .. } | Expr::ExistsSubquery { .. } => Err(PlanError::Semantic(format!(
761 "nested projection `{name}` filter cannot contain a subquery; \
762 filter the outer query or the child columns directly"
763 ))),
764 Expr::FunctionCall(..) => Err(PlanError::Semantic(format!(
765 "nested projection `{name}` filter cannot contain an aggregate function"
766 ))),
767 Expr::Window { .. } => Err(PlanError::Semantic(format!(
768 "nested projection `{name}` filter cannot contain a window function"
769 ))),
770 Expr::NestedQuery(_) => Err(PlanError::Semantic(format!(
771 "nested projection `{name}` filter cannot contain another nested projection"
772 ))),
773 }
774}
775
776fn plan_joined_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
798 let primary_alias = q.alias.clone().unwrap_or_else(|| q.source.clone());
799 let mut aliases = std::collections::HashSet::new();
800 aliases.insert(primary_alias.clone());
801 let mut node = PlanNode::AliasScan {
802 table: q.source.clone(),
803 alias: primary_alias,
804 };
805
806 for join in q.joins {
807 let right_alias = join.alias.unwrap_or_else(|| join.source.clone());
808 if !aliases.insert(right_alias.clone()) {
809 return Err(ParseError::Syntax {
810 message: format!(
811 "duplicate source alias `{right_alias}` in join; every joined source needs a unique alias"
812 ),
813 }
814 .into());
815 }
816 let right = PlanNode::AliasScan {
817 table: join.source,
818 alias: right_alias,
819 };
820 match join.kind {
821 JoinKind::Inner | JoinKind::LeftOuter | JoinKind::Cross => {
822 node = PlanNode::NestedLoopJoin {
823 left: Box::new(node),
824 right: Box::new(right),
825 on: join.on,
826 kind: join.kind,
827 };
828 }
829 JoinKind::RightOuter => {
830 node = PlanNode::NestedLoopJoin {
832 left: Box::new(right),
833 right: Box::new(node),
834 on: join.on,
835 kind: JoinKind::LeftOuter,
836 };
837 }
838 }
839 }
840
841 if let Some(pred) = q.filter {
842 node = PlanNode::Filter {
843 input: Box::new(node),
844 predicate: pred,
845 };
846 }
847
848 if q.group_by.is_none() {
849 if let Some(order) = q.order.take() {
850 node = PlanNode::Sort {
851 input: Box::new(node),
852 keys: order
853 .keys
854 .into_iter()
855 .map(|k| SortKey {
856 expr: k.expr,
857 descending: k.descending,
858 })
859 .collect(),
860 };
861 }
862 }
863
864 if let Some(group) = q.group_by {
866 let mut grouped_order = q.order;
867 let mut proj_fields: Vec<ProjectField> = q
868 .projection
869 .map(|proj| {
870 proj.into_iter()
871 .map(|pf| ProjectField {
872 alias: pf.alias,
873 expr: pf.expr,
874 })
875 .collect()
876 })
877 .unwrap_or_default();
878 let mut having = group.having;
879 let aggregates = extract_aggregates(&mut proj_fields, &mut having, &aliases)?;
880 rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
881 rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
882
883 node = PlanNode::GroupBy {
884 input: Box::new(node),
885 keys: group.keys,
886 aggregates,
887 having,
888 };
889
890 if !proj_fields.is_empty() {
891 node = PlanNode::Project {
892 input: Box::new(node),
893 fields: proj_fields,
894 };
895 }
896 if let Some(order) = grouped_order {
897 node = PlanNode::Sort {
898 input: Box::new(node),
899 keys: order
900 .keys
901 .into_iter()
902 .map(|key| SortKey {
903 expr: key.expr,
904 descending: key.descending,
905 })
906 .collect(),
907 };
908 }
909 if let Some(off) = q.offset {
914 node = PlanNode::Offset {
915 input: Box::new(node),
916 count: off,
917 };
918 }
919 if let Some(lim) = q.limit {
920 node = PlanNode::Limit {
921 input: Box::new(node),
922 count: lim,
923 };
924 }
925 if q.distinct {
926 node = PlanNode::Distinct {
927 input: Box::new(node),
928 };
929 }
930 return Ok(node);
931 }
932
933 if let Some(off) = q.offset {
937 node = PlanNode::Offset {
938 input: Box::new(node),
939 count: off,
940 };
941 }
942
943 if let Some(lim) = q.limit {
944 node = PlanNode::Limit {
945 input: Box::new(node),
946 count: lim,
947 };
948 }
949
950 if let Some(proj) = q.projection {
951 let mut fields: Vec<ProjectField> = proj
952 .into_iter()
953 .map(|pf| ProjectField {
954 alias: pf.alias,
955 expr: pf.expr,
956 })
957 .collect();
958 let windows = extract_windows(&mut fields);
959 if !windows.is_empty() {
960 node = PlanNode::Window {
961 input: Box::new(node),
962 windows,
963 };
964 }
965 node = PlanNode::Project {
966 input: Box::new(node),
967 fields,
968 };
969 }
970
971 if q.distinct {
972 node = PlanNode::Distinct {
973 input: Box::new(node),
974 };
975 }
976
977 if let Some(agg) = q.aggregation {
978 let provenance_alias =
979 symmetric_provenance_alias(agg.function, agg.argument.as_ref(), agg.mode, &aliases)?;
980 node = PlanNode::Aggregate {
981 input: Box::new(node),
982 function: agg.function,
983 argument: agg.argument,
984 mode: agg.mode,
985 provenance_alias,
986 };
987 }
988
989 Ok(node)
990}
991
992fn plan_insert(ins: InsertExpr) -> Result<PlanNode, PlanError> {
993 Ok(PlanNode::Insert {
994 table: ins.target,
995 rows: ins.rows,
996 returning: ins.returning,
997 })
998}
999
1000fn plan_update(mut upd: UpdateExpr) -> Result<PlanNode, PlanError> {
1001 let visible = upd.alias.clone().unwrap_or_else(|| upd.source.clone());
1005 if let Some(filter) = upd.filter.as_mut() {
1006 resolve_scan_qualifiers(filter, &visible)?;
1007 }
1008 for assign in upd.assignments.iter_mut() {
1009 resolve_scan_qualifiers(&mut assign.value, &visible)?;
1010 }
1011 let source = match upd.filter {
1016 Some(pred) => match try_extract_eq_index_key(&upd.source, &pred) {
1017 Some(index_scan) => index_scan,
1018 None => match try_extract_range_index_keys(&upd.source, &pred) {
1019 Some(range_scan) => range_scan,
1020 None => PlanNode::Filter {
1021 input: Box::new(PlanNode::SeqScan {
1022 table: upd.source.clone(),
1023 }),
1024 predicate: pred,
1025 },
1026 },
1027 },
1028 None => PlanNode::SeqScan {
1029 table: upd.source.clone(),
1030 },
1031 };
1032 Ok(PlanNode::Update {
1033 input: Box::new(source),
1034 table: upd.source,
1035 assignments: upd.assignments,
1036 returning: upd.returning,
1037 })
1038}
1039
1040fn plan_delete(mut del: DeleteExpr) -> Result<PlanNode, PlanError> {
1041 let visible = del.alias.clone().unwrap_or_else(|| del.source.clone());
1045 if let Some(filter) = del.filter.as_mut() {
1046 resolve_scan_qualifiers(filter, &visible)?;
1047 }
1048 let source = match del.filter {
1049 Some(pred) => match try_extract_eq_index_key(&del.source, &pred) {
1050 Some(index_scan) => index_scan,
1051 None => match try_extract_range_index_keys(&del.source, &pred) {
1052 Some(range_scan) => range_scan,
1053 None => PlanNode::Filter {
1054 input: Box::new(PlanNode::SeqScan {
1055 table: del.source.clone(),
1056 }),
1057 predicate: pred,
1058 },
1059 },
1060 },
1061 None => PlanNode::SeqScan {
1062 table: del.source.clone(),
1063 },
1064 };
1065 Ok(PlanNode::Delete {
1066 input: Box::new(source),
1067 table: del.source,
1068 returning: del.returning,
1069 })
1070}
1071
1072fn plan_upsert(ups: UpsertExpr) -> Result<PlanNode, PlanError> {
1073 Ok(PlanNode::Upsert {
1074 table: ups.target,
1075 key_column: ups.key_column,
1076 assignments: ups.assignments,
1077 on_conflict: ups.on_conflict,
1078 })
1079}
1080
1081fn plan_create_type(ct: CreateTypeExpr) -> Result<PlanNode, PlanError> {
1082 let fields = ct
1083 .fields
1084 .into_iter()
1085 .map(|f| crate::plan::CreateField {
1086 name: f.name,
1087 type_name: f.type_name,
1088 required: f.required,
1089 unique: f.unique,
1090 default: f.default,
1091 auto: f.auto,
1092 })
1093 .collect();
1094 Ok(PlanNode::CreateTable {
1095 name: ct.name,
1096 fields,
1097 if_not_exists: ct.if_not_exists,
1098 })
1099}
1100
1101pub(crate) fn try_extract_eq_index_key(table: &str, pred: &Expr) -> Option<PlanNode> {
1111 let (lhs, op, rhs) = match pred {
1112 Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
1113 _ => return None,
1114 };
1115 if op != BinOp::Eq {
1116 return None;
1117 }
1118 match (lhs, rhs) {
1119 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some(PlanNode::ExprIndexScan {
1120 table: table.to_string(),
1121 path: stored_json_path(path)?,
1122 key: rhs.clone(),
1123 }),
1124 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some(PlanNode::ExprIndexScan {
1125 table: table.to_string(),
1126 path: stored_json_path(path)?,
1127 key: lhs.clone(),
1128 }),
1129 (Expr::Field(name), Expr::Literal(_)) => Some(PlanNode::IndexScan {
1130 table: table.to_string(),
1131 column: name.clone(),
1132 key: rhs.clone(),
1133 }),
1134 (Expr::Literal(_), Expr::Field(name)) => Some(PlanNode::IndexScan {
1135 table: table.to_string(),
1136 column: name.clone(),
1137 key: lhs.clone(),
1138 }),
1139 _ => None,
1140 }
1141}
1142
1143fn stored_json_path(expr: &Expr) -> Option<StoredJsonPathV1> {
1144 JsonPathIdentityV1::from_expr(expr)?.bind_table_local(None)
1145}
1146
1147pub(crate) fn extract_single_bound(pred: &Expr) -> Option<RangeBound> {
1150 let (lhs, op, rhs) = match pred {
1151 Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
1152 _ => return None,
1153 };
1154 match op {
1155 BinOp::Gt => match (lhs, rhs) {
1157 (Expr::Field(name), Expr::Literal(_)) => Some((
1158 RangeTarget::Column(name.clone()),
1159 Some((rhs.clone(), false)),
1160 None,
1161 )),
1162 (Expr::Literal(_), Expr::Field(name)) => {
1163 Some((
1165 RangeTarget::Column(name.clone()),
1166 None,
1167 Some((lhs.clone(), false)),
1168 ))
1169 }
1170 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1171 RangeTarget::JsonPath(stored_json_path(path)?),
1172 Some((rhs.clone(), false)),
1173 None,
1174 )),
1175 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1176 RangeTarget::JsonPath(stored_json_path(path)?),
1177 None,
1178 Some((lhs.clone(), false)),
1179 )),
1180 _ => None,
1181 },
1182 BinOp::Gte => match (lhs, rhs) {
1184 (Expr::Field(name), Expr::Literal(_)) => Some((
1185 RangeTarget::Column(name.clone()),
1186 Some((rhs.clone(), true)),
1187 None,
1188 )),
1189 (Expr::Literal(_), Expr::Field(name)) => Some((
1190 RangeTarget::Column(name.clone()),
1191 None,
1192 Some((lhs.clone(), true)),
1193 )),
1194 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1195 RangeTarget::JsonPath(stored_json_path(path)?),
1196 Some((rhs.clone(), true)),
1197 None,
1198 )),
1199 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1200 RangeTarget::JsonPath(stored_json_path(path)?),
1201 None,
1202 Some((lhs.clone(), true)),
1203 )),
1204 _ => None,
1205 },
1206 BinOp::Lt => match (lhs, rhs) {
1208 (Expr::Field(name), Expr::Literal(_)) => Some((
1209 RangeTarget::Column(name.clone()),
1210 None,
1211 Some((rhs.clone(), false)),
1212 )),
1213 (Expr::Literal(_), Expr::Field(name)) => Some((
1214 RangeTarget::Column(name.clone()),
1215 Some((lhs.clone(), false)),
1216 None,
1217 )),
1218 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1219 RangeTarget::JsonPath(stored_json_path(path)?),
1220 None,
1221 Some((rhs.clone(), false)),
1222 )),
1223 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1224 RangeTarget::JsonPath(stored_json_path(path)?),
1225 Some((lhs.clone(), false)),
1226 None,
1227 )),
1228 _ => None,
1229 },
1230 BinOp::Lte => match (lhs, rhs) {
1232 (Expr::Field(name), Expr::Literal(_)) => Some((
1233 RangeTarget::Column(name.clone()),
1234 None,
1235 Some((rhs.clone(), true)),
1236 )),
1237 (Expr::Literal(_), Expr::Field(name)) => Some((
1238 RangeTarget::Column(name.clone()),
1239 Some((lhs.clone(), true)),
1240 None,
1241 )),
1242 (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
1243 RangeTarget::JsonPath(stored_json_path(path)?),
1244 None,
1245 Some((rhs.clone(), true)),
1246 )),
1247 (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
1248 RangeTarget::JsonPath(stored_json_path(path)?),
1249 Some((lhs.clone(), true)),
1250 None,
1251 )),
1252 _ => None,
1253 },
1254 _ => None,
1255 }
1256}
1257
1258fn try_extract_range_index_keys(table: &str, pred: &Expr) -> Option<PlanNode> {
1281 if let Expr::BinaryOp(lhs, BinOp::And, rhs) = pred {
1284 if let (Some((col1, s1, e1)), Some((col2, s2, e2))) =
1285 (extract_single_bound(lhs), extract_single_bound(rhs))
1286 {
1287 if col1 == col2 {
1288 if let (Some(start), None, None, Some(end)) = (s1, e1, s2, e2) {
1289 return Some(range_scan_for_target(table, col1, Some(start), Some(end)));
1290 }
1291 }
1292 }
1293 }
1294
1295 if let Some((col, start, end)) = extract_single_bound(pred) {
1297 return Some(range_scan_for_target(table, col, start, end));
1298 }
1299
1300 None
1301}
1302
1303pub(crate) fn range_scan_for_target(
1304 table: &str,
1305 target: RangeTarget,
1306 start: Option<(Expr, bool)>,
1307 end: Option<(Expr, bool)>,
1308) -> PlanNode {
1309 match target {
1310 RangeTarget::Column(column) => PlanNode::RangeScan {
1311 table: table.to_string(),
1312 column,
1313 start,
1314 end,
1315 },
1316 RangeTarget::JsonPath(path) => PlanNode::ExprRangeScan {
1317 table: table.to_string(),
1318 path,
1319 start,
1320 end,
1321 },
1322 }
1323}
1324
1325fn try_extract_ordered_expr_index_scan(query: &QueryExpr) -> Option<PlanNode> {
1330 if query.alias.is_some()
1331 || !query.joins.is_empty()
1332 || query.filter.is_some()
1333 || query.group_by.is_some()
1334 || query.distinct
1335 || query.aggregation.is_some()
1336 || query.projection.as_ref().is_some_and(|fields| {
1337 fields
1338 .iter()
1339 .any(|field| matches!(field.expr, Expr::Window { .. }))
1340 })
1341 {
1342 return None;
1343 }
1344 let order = query.order.as_ref()?;
1345 let [key] = order.keys.as_slice() else {
1346 return None;
1347 };
1348 let path = stored_json_path(&key.expr)?;
1349 let limit = query.limit.as_ref()?;
1350 if !matches!(limit, Expr::Literal(Literal::Int(value)) if *value >= 0) {
1351 return None;
1352 }
1353 if !query
1354 .offset
1355 .as_ref()
1356 .is_none_or(|offset| matches!(offset, Expr::Literal(Literal::Int(value)) if *value >= 0))
1357 {
1358 return None;
1359 }
1360 Some(PlanNode::OrderedExprIndexScan {
1361 table: query.source.clone(),
1362 path,
1363 descending: key.descending,
1364 limit: limit.clone(),
1365 offset: query.offset.clone(),
1366 })
1367}
1368
1369fn extract_windows(proj_fields: &mut [ProjectField]) -> Vec<WindowDef> {
1374 let mut defs = Vec::new();
1375 let mut counter = 0usize;
1376 for f in proj_fields.iter_mut() {
1377 if let Expr::Window {
1378 function,
1379 args,
1380 mode,
1381 partition_by,
1382 order_by,
1383 } = &f.expr
1384 {
1385 let output_name = format!("__win_{counter}");
1386 defs.push(WindowDef {
1387 function: *function,
1388 args: args.clone(),
1389 mode: *mode,
1390 partition_by: partition_by.clone(),
1391 order_by: order_by
1392 .iter()
1393 .map(|k| SortKey {
1394 expr: k.expr.clone(),
1395 descending: k.descending,
1396 })
1397 .collect(),
1398 output_name: output_name.clone(),
1399 });
1400 f.expr = Expr::Field(output_name);
1401 counter += 1;
1402 }
1403 }
1404 defs
1405}
1406
1407fn extract_aggregates(
1413 proj_fields: &mut [ProjectField],
1414 having: &mut Option<Expr>,
1415 source_aliases: &std::collections::HashSet<String>,
1416) -> Result<Vec<GroupAgg>, PlanError> {
1417 let mut aggs: Vec<GroupAgg> = Vec::new();
1418 let mut counter = 0usize;
1419 for f in proj_fields.iter_mut() {
1420 rewrite_agg_expr(&mut f.expr, &mut aggs, &mut counter, source_aliases)?;
1421 }
1422 if let Some(h) = having {
1423 rewrite_agg_expr(h, &mut aggs, &mut counter, source_aliases)?;
1424 }
1425 Ok(aggs)
1426}
1427
1428fn rewrite_group_key_references(
1429 fields: &mut [ProjectField],
1430 having: &mut Option<Expr>,
1431 keys: &[GroupKey],
1432) {
1433 for field in fields {
1434 rewrite_group_key_expr(&mut field.expr, keys);
1435 }
1436 if let Some(having) = having {
1437 rewrite_group_key_expr(having, keys);
1438 }
1439}
1440
1441fn rewrite_group_order_keys(
1442 order: Option<&mut OrderClause>,
1443 projection: &[ProjectField],
1444 keys: &[GroupKey],
1445) {
1446 let Some(order) = order else {
1447 return;
1448 };
1449 for order_key in &mut order.keys {
1450 let Some(group_key) = keys.iter().find(|key| key.expr == order_key.expr) else {
1451 continue;
1452 };
1453 let projected_name = projection
1454 .iter()
1455 .find(|field| field.expr == group_key.expr)
1456 .and_then(|field| field.alias.clone())
1457 .unwrap_or_else(|| group_key.output_name());
1458 order_key.expr = Expr::Field(projected_name);
1459 }
1460}
1461
1462fn rewrite_group_key_expr(expr: &mut Expr, keys: &[GroupKey]) {
1463 if let Some(key) = keys.iter().find(|key| key.expr == *expr) {
1464 *expr = Expr::Field(key.output_name());
1465 return;
1466 }
1467 match expr {
1468 Expr::FunctionCall(..) => {}
1472 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1473 rewrite_group_key_expr(left, keys);
1474 rewrite_group_key_expr(right, keys);
1475 }
1476 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) => rewrite_group_key_expr(inner, keys),
1477 Expr::ScalarFunc(_, args) => {
1478 for arg in args {
1479 rewrite_group_key_expr(arg, keys);
1480 }
1481 }
1482 Expr::InList { expr, list, .. } => {
1483 rewrite_group_key_expr(expr, keys);
1484 for item in list {
1485 rewrite_group_key_expr(item, keys);
1486 }
1487 }
1488 Expr::Case { whens, else_expr } => {
1489 for (condition, result) in whens {
1490 rewrite_group_key_expr(condition, keys);
1491 rewrite_group_key_expr(result, keys);
1492 }
1493 if let Some(expr) = else_expr {
1494 rewrite_group_key_expr(expr, keys);
1495 }
1496 }
1497 _ => {}
1498 }
1499}
1500
1501fn rewrite_agg_expr(
1502 expr: &mut Expr,
1503 aggs: &mut Vec<GroupAgg>,
1504 counter: &mut usize,
1505 source_aliases: &std::collections::HashSet<String>,
1506) -> Result<(), PlanError> {
1507 match expr {
1508 Expr::FunctionCall(func, inner, mode) => {
1509 let output = find_or_insert_agg(aggs, *func, inner, *mode, counter, source_aliases)?;
1510 *expr = Expr::Field(output);
1511 }
1512 Expr::BinaryOp(l, _, r) => {
1513 rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1514 rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1515 }
1516 Expr::UnaryOp(_, inner) => rewrite_agg_expr(inner, aggs, counter, source_aliases)?,
1517 Expr::Coalesce(l, r) => {
1518 rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1519 rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1520 }
1521 Expr::InList { expr: e, list, .. } => {
1522 rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1523 for item in list {
1524 rewrite_agg_expr(item, aggs, counter, source_aliases)?;
1525 }
1526 }
1527 Expr::InSubquery { expr: e, .. } => {
1528 rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1529 }
1530 _ => {}
1531 }
1532 Ok(())
1533}
1534
1535fn find_or_insert_agg(
1536 aggs: &mut Vec<GroupAgg>,
1537 func: AggFunc,
1538 argument: &Expr,
1539 mode: AggregateMode,
1540 counter: &mut usize,
1541 source_aliases: &std::collections::HashSet<String>,
1542) -> Result<String, PlanError> {
1543 for existing in aggs.iter() {
1544 if existing.function == func && existing.argument == *argument && existing.mode == mode {
1545 return Ok(existing.output_name.clone());
1546 }
1547 }
1548 let provenance_alias = symmetric_provenance_alias(func, Some(argument), mode, source_aliases)?;
1549 let output_name = format!("__agg_{counter}");
1550 aggs.push(GroupAgg {
1551 function: func,
1552 argument: argument.clone(),
1553 mode,
1554 provenance_alias,
1555 output_name: output_name.clone(),
1556 });
1557 *counter += 1;
1558 Ok(output_name)
1559}
1560
1561fn symmetric_provenance_alias(
1562 function: AggFunc,
1563 argument: Option<&Expr>,
1564 mode: AggregateMode,
1565 source_aliases: &std::collections::HashSet<String>,
1566) -> Result<Option<String>, PlanError> {
1567 if mode == AggregateMode::Raw
1568 || source_aliases.len() < 2
1569 || !matches!(function, AggFunc::Sum | AggFunc::Avg | AggFunc::Count)
1570 || (function == AggFunc::Count
1571 && argument.is_none_or(|argument| matches!(argument, Expr::Field(name) if name == "*")))
1572 {
1573 return Ok(None);
1574 }
1575 let Some(argument) = argument else {
1576 return Err(symmetric_aggregate_error(
1577 function,
1578 "does not reference a source row",
1579 ));
1580 };
1581
1582 let mut qualified = std::collections::HashSet::new();
1583 let mut has_unqualified = false;
1584 collect_expression_sources(argument, &mut qualified, &mut has_unqualified);
1585
1586 for alias in &qualified {
1587 if !source_aliases.contains(alias) {
1588 return Err(symmetric_aggregate_error(
1589 function,
1590 &format!("references unknown source alias '{alias}'"),
1591 ));
1592 }
1593 }
1594 if has_unqualified {
1595 if source_aliases.len() != 1 {
1596 return Err(symmetric_aggregate_error(
1597 function,
1598 "contains an ambiguous unqualified field",
1599 ));
1600 }
1601 qualified.extend(source_aliases.iter().cloned());
1602 }
1603 match qualified.len() {
1604 1 => Ok(qualified.into_iter().next()),
1605 0 => Err(symmetric_aggregate_error(
1606 function,
1607 "does not reference a source row",
1608 )),
1609 _ => Err(symmetric_aggregate_error(
1610 function,
1611 "references multiple source aliases",
1612 )),
1613 }
1614}
1615
1616fn symmetric_aggregate_error(function: AggFunc, reason: &str) -> PlanError {
1617 let name = format!("{function:?}").to_lowercase();
1618 PlanError::Semantic(format!(
1619 "symmetric {name} expression {reason}; reference exactly one source alias or use {name}(raw ...)"
1620 ))
1621}
1622
1623fn collect_expression_sources(
1624 expr: &Expr,
1625 qualified: &mut std::collections::HashSet<String>,
1626 has_unqualified: &mut bool,
1627) {
1628 match expr {
1629 Expr::Field(name) if name != "*" => *has_unqualified = true,
1630 Expr::QualifiedField { qualifier, .. } => {
1631 qualified.insert(qualifier.clone());
1632 }
1633 Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1634 collect_expression_sources(left, qualified, has_unqualified);
1635 collect_expression_sources(right, qualified, has_unqualified);
1636 }
1637 Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) | Expr::JsonPath { base: inner, .. } => {
1638 collect_expression_sources(inner, qualified, has_unqualified);
1639 }
1640 Expr::ScalarFunc(_, args) => {
1641 for argument in args {
1642 collect_expression_sources(argument, qualified, has_unqualified);
1643 }
1644 }
1645 Expr::InList { expr, list, .. } => {
1646 collect_expression_sources(expr, qualified, has_unqualified);
1647 for item in list {
1648 collect_expression_sources(item, qualified, has_unqualified);
1649 }
1650 }
1651 Expr::InSubquery { expr, .. } => {
1652 collect_expression_sources(expr, qualified, has_unqualified);
1653 }
1654 Expr::Case { whens, else_expr } => {
1655 for (condition, result) in whens {
1656 collect_expression_sources(condition, qualified, has_unqualified);
1657 collect_expression_sources(result, qualified, has_unqualified);
1658 }
1659 if let Some(expr) = else_expr {
1660 collect_expression_sources(expr, qualified, has_unqualified);
1661 }
1662 }
1663 Expr::Window {
1664 args,
1665 partition_by,
1666 order_by,
1667 ..
1668 } => {
1669 for expr in args.iter().chain(partition_by) {
1670 collect_expression_sources(expr, qualified, has_unqualified);
1671 }
1672 for key in order_by {
1673 collect_expression_sources(&key.expr, qualified, has_unqualified);
1674 }
1675 }
1676 Expr::FunctionCall(_, inner, _) => {
1677 collect_expression_sources(inner, qualified, has_unqualified);
1678 }
1679 Expr::ExistsSubquery { .. }
1680 | Expr::Field(_)
1681 | Expr::Literal(_)
1682 | Expr::Param(_)
1683 | Expr::ValueLit(_)
1684 | Expr::Null
1685 | Expr::NestedQuery(_) => {}
1686 }
1687}
1688
1689#[cfg(test)]
1690mod tests {
1691 use super::*;
1692 use crate::plan::PlanNode;
1693
1694 #[test]
1695 fn test_plan_simple_scan() {
1696 let plan = plan("User").unwrap();
1697 assert!(matches!(plan, PlanNode::SeqScan { table } if table == "User"));
1698 }
1699
1700 #[test]
1701 fn test_plan_filter() {
1702 let plan = plan("User filter .age > 30").unwrap();
1703 assert!(matches!(plan, PlanNode::RangeScan { .. }));
1704 }
1705
1706 #[test]
1707 fn test_plan_filter_with_projection() {
1708 let plan = plan("User filter .age > 30 { name, email }").unwrap();
1709 assert!(matches!(plan, PlanNode::Project { .. }));
1710 }
1711
1712 #[test]
1713 fn test_plan_insert() {
1714 let plan = plan(r#"insert User { name := "Alice", age := 30 }"#).unwrap();
1715 assert!(matches!(plan, PlanNode::Insert { .. }));
1716 }
1717
1718 #[test]
1719 fn test_plan_order_limit() {
1720 let plan = plan("User order .name limit 10").unwrap();
1721 match plan {
1722 PlanNode::Limit { input, .. } => {
1723 assert!(matches!(*input, PlanNode::Sort { .. }));
1724 }
1725 _ => panic!("expected Limit(Sort(SeqScan))"),
1726 }
1727 }
1728
1729 #[test]
1730 fn test_plan_count() {
1731 let plan = plan("count(User)").unwrap();
1732 assert!(matches!(plan, PlanNode::Aggregate { .. }));
1733 }
1734
1735 #[test]
1736 fn single_source_aggregates_do_not_request_provenance() {
1737 for query in [
1738 "sum(User { .amount })",
1739 "avg(User { .amount })",
1740 "count(User { .amount })",
1741 ] {
1742 match plan(query).unwrap() {
1743 PlanNode::Aggregate {
1744 provenance_alias, ..
1745 } => assert!(
1746 provenance_alias.is_none(),
1747 "unexpected provenance for {query}"
1748 ),
1749 other => panic!("expected Aggregate for {query}, got {other:?}"),
1750 }
1751 }
1752
1753 match plan("User group .dept { total: sum(.amount) }").unwrap() {
1754 PlanNode::Project { input, .. } => match *input {
1755 PlanNode::GroupBy { aggregates, .. } => {
1756 assert!(aggregates[0].provenance_alias.is_none());
1757 }
1758 other => panic!("expected GroupBy, got {other:?}"),
1759 },
1760 other => panic!("expected Project(GroupBy), got {other:?}"),
1761 }
1762 }
1763
1764 #[test]
1765 fn join_provenance_is_limited_to_fanout_sensitive_aggregates() {
1766 let base = "Account as a join Entry as e on a.id = e.account_id group a.dept";
1767 for (function, expects_provenance) in [
1768 ("sum(a.balance)", true),
1769 ("avg(a.balance)", true),
1770 ("count(a.balance)", true),
1771 ("min(a.balance)", false),
1772 ("max(a.balance)", false),
1773 ("count(distinct a.balance)", false),
1774 ("count(*)", false),
1775 ] {
1776 let query = format!("{base} {{ value: {function} }}");
1777 match plan(&query).unwrap() {
1778 PlanNode::Project { input, .. } => match *input {
1779 PlanNode::GroupBy { aggregates, .. } => assert_eq!(
1780 aggregates[0].provenance_alias.as_deref(),
1781 expects_provenance.then_some("a"),
1782 "unexpected provenance selection for {function}"
1783 ),
1784 other => panic!("expected GroupBy for {function}, got {other:?}"),
1785 },
1786 other => panic!("expected Project(GroupBy) for {function}, got {other:?}"),
1787 }
1788 }
1789 }
1790
1791 #[test]
1792 fn test_plan_eq_becomes_index_scan() {
1793 let plan = plan("User filter .id = 42").unwrap();
1796 match plan {
1797 PlanNode::IndexScan { table, column, key } => {
1798 assert_eq!(table, "User");
1799 assert_eq!(column, "id");
1800 assert!(matches!(key, Expr::Literal(Literal::Int(42))));
1801 }
1802 other => panic!("expected IndexScan, got {other:?}"),
1803 }
1804 }
1805
1806 #[test]
1807 fn test_plan_eq_reversed_becomes_index_scan() {
1808 let plan = plan(r#"User filter "NYC" = .city"#).unwrap();
1810 assert!(matches!(plan, PlanNode::IndexScan { .. }));
1811 }
1812
1813 #[test]
1814 fn json_path_equality_and_reversed_equality_are_speculative_expression_scans() {
1815 for query in ["Post filter .data->age = 21", "Post filter 21 = .data->age"] {
1816 match plan(query).unwrap() {
1817 PlanNode::ExprIndexScan { table, path, key } => {
1818 assert_eq!(table, "Post");
1819 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1820 assert!(matches!(key, Expr::Literal(Literal::Int(21))));
1821 }
1822 other => panic!("expected ExprIndexScan for `{query}`, got {other:?}"),
1823 }
1824 }
1825 }
1826
1827 #[test]
1828 fn json_path_range_and_same_path_compound_bounds_are_speculative_scans() {
1829 for query in ["Post filter .data->age > 18", "Post filter 18 < .data->age"] {
1830 match plan(query).unwrap() {
1831 PlanNode::ExprRangeScan {
1832 path, start, end, ..
1833 } => {
1834 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1835 assert!(start.is_some());
1836 assert!(end.is_none());
1837 }
1838 other => panic!("expected ExprRangeScan for `{query}`, got {other:?}"),
1839 }
1840 }
1841
1842 match plan("Post filter .data->age >= 18 and .data->age < 65").unwrap() {
1843 PlanNode::ExprRangeScan {
1844 path, start, end, ..
1845 } => {
1846 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1847 assert_eq!(start, Some((Expr::Literal(Literal::Int(18)), true)));
1848 assert_eq!(end, Some((Expr::Literal(Literal::Int(65)), false)));
1849 }
1850 other => panic!("expected bounded ExprRangeScan, got {other:?}"),
1851 }
1852
1853 assert!(matches!(
1854 plan("Post filter .data->age >= 18 and .data->score < 65").unwrap(),
1855 PlanNode::Filter { .. }
1856 ));
1857 }
1858
1859 #[test]
1860 fn exact_single_path_order_limit_uses_ordered_expression_scan() {
1861 match plan("Post order .data->age desc limit 10 offset 2 { .id }").unwrap() {
1862 PlanNode::Project { input, .. } => match *input {
1863 PlanNode::OrderedExprIndexScan {
1864 table,
1865 path,
1866 descending,
1867 limit,
1868 offset,
1869 } => {
1870 assert_eq!(table, "Post");
1871 assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1872 assert!(descending);
1873 assert_eq!(limit, Expr::Literal(Literal::Int(10)));
1874 assert_eq!(offset, Some(Expr::Literal(Literal::Int(2))));
1875 }
1876 other => panic!("expected OrderedExprIndexScan, got {other:?}"),
1877 },
1878 other => panic!("expected Project(OrderedExprIndexScan), got {other:?}"),
1879 }
1880 }
1881
1882 #[test]
1883 fn incompatible_path_order_shapes_keep_generic_sort() {
1884 for query in [
1885 "Post order .data->age",
1886 "Post order .data->age, .id limit 10",
1887 "Post filter .data->active = true order .data->age limit 10",
1888 "Post order .data->age limit .id",
1889 ] {
1890 let planned = plan(query).unwrap();
1891 assert!(
1892 !plan_contains_ordered_expr_scan(&planned),
1893 "`{query}` must remain on the generic pipeline: {planned:?}"
1894 );
1895 }
1896 }
1897
1898 fn plan_contains_ordered_expr_scan(plan: &PlanNode) -> bool {
1899 match plan {
1900 PlanNode::OrderedExprIndexScan { .. } => true,
1901 PlanNode::Filter { input, .. }
1902 | PlanNode::Project { input, .. }
1903 | PlanNode::Sort { input, .. }
1904 | PlanNode::Limit { input, .. }
1905 | PlanNode::Offset { input, .. }
1906 | PlanNode::Aggregate { input, .. }
1907 | PlanNode::Distinct { input }
1908 | PlanNode::GroupBy { input, .. }
1909 | PlanNode::Update { input, .. }
1910 | PlanNode::Delete { input, .. }
1911 | PlanNode::Window { input, .. }
1912 | PlanNode::Explain { input } => plan_contains_ordered_expr_scan(input),
1913 PlanNode::NestedLoopJoin { left, right, .. } | PlanNode::Union { left, right, .. } => {
1914 plan_contains_ordered_expr_scan(left) || plan_contains_ordered_expr_scan(right)
1915 }
1916 _ => false,
1917 }
1918 }
1919
1920 #[test]
1921 fn test_plan_non_eq_stays_filter() {
1922 let plan = plan("User filter .age > 30").unwrap();
1924 match plan {
1925 PlanNode::RangeScan {
1926 column, start, end, ..
1927 } => {
1928 assert_eq!(column, "age");
1929 assert!(start.is_some(), "expected lower bound");
1930 assert!(end.is_none(), "expected no upper bound");
1931 let (_, inclusive) = start.unwrap();
1932 assert!(!inclusive, "expected exclusive lower bound for >");
1933 }
1934 other => panic!("expected RangeScan, got {other:?}"),
1935 }
1936 }
1937
1938 #[test]
1939 fn test_plan_index_scan_with_projection() {
1940 let plan = plan("User filter .id = 1 { .name }").unwrap();
1942 match plan {
1943 PlanNode::Project { input, .. } => {
1944 assert!(matches!(*input, PlanNode::IndexScan { .. }));
1945 }
1946 other => panic!("expected Project(IndexScan), got {other:?}"),
1947 }
1948 }
1949
1950 #[test]
1951 fn test_plan_update_by_pk_becomes_index_scan() {
1952 let plan = plan("User filter .id = 42 update { age := 31 }").unwrap();
1955 match plan {
1956 PlanNode::Update { input, .. } => {
1957 assert!(
1958 matches!(*input, PlanNode::IndexScan { .. }),
1959 "expected Update(IndexScan), got {input:?}"
1960 );
1961 }
1962 other => panic!("expected Update, got {other:?}"),
1963 }
1964 }
1965
1966 #[test]
1967 fn test_plan_update_range_stays_range_scan() {
1968 let plan = plan("User filter .age > 30 update { age := 31 }").unwrap();
1969 match plan {
1970 PlanNode::Update { input, .. } => {
1971 assert!(
1972 matches!(*input, PlanNode::RangeScan { .. }),
1973 "expected Update(RangeScan), got {input:?}"
1974 );
1975 }
1976 other => panic!("expected Update, got {other:?}"),
1977 }
1978 }
1979
1980 #[test]
1981 fn test_plan_delete_by_pk_becomes_index_scan() {
1982 let plan = plan("User filter .id = 7 delete").unwrap();
1983 match plan {
1984 PlanNode::Delete { input, .. } => {
1985 assert!(matches!(*input, PlanNode::IndexScan { .. }));
1986 }
1987 other => panic!("expected Delete, got {other:?}"),
1988 }
1989 }
1990
1991 #[test]
1992 fn test_plan_inner_join_builds_nested_loop() {
1993 let plan = plan("User as u join Order as o on u.id = o.user_id").unwrap();
1996 match plan {
1997 PlanNode::NestedLoopJoin {
1998 left,
1999 right,
2000 on,
2001 kind,
2002 } => {
2003 assert_eq!(kind, JoinKind::Inner);
2004 assert!(on.is_some());
2005 assert!(matches!(*left, PlanNode::AliasScan { .. }));
2006 assert!(matches!(*right, PlanNode::AliasScan { .. }));
2007 }
2008 other => panic!("expected NestedLoopJoin, got {other:?}"),
2009 }
2010 }
2011
2012 #[test]
2013 fn duplicate_join_aliases_are_rejected_before_execution() {
2014 let err = plan("A as x join A as x on x.id = x.id").unwrap_err();
2015 assert!(
2016 err.to_string().contains("duplicate source alias `x`"),
2017 "unexpected error: {err}"
2018 );
2019 }
2020
2021 #[test]
2022 fn test_plan_right_join_rewritten_as_left_with_swapped_inputs() {
2023 let plan = plan("User as u right join Order as o on u.id = o.user_id").unwrap();
2024 match plan {
2025 PlanNode::NestedLoopJoin {
2026 left, right, kind, ..
2027 } => {
2028 assert_eq!(kind, JoinKind::LeftOuter);
2029 match *left {
2031 PlanNode::AliasScan { table, .. } => assert_eq!(table, "Order"),
2032 other => panic!("expected AliasScan(Order), got {other:?}"),
2033 }
2034 match *right {
2035 PlanNode::AliasScan { table, .. } => assert_eq!(table, "User"),
2036 other => panic!("expected AliasScan(User), got {other:?}"),
2037 }
2038 }
2039 other => panic!("expected NestedLoopJoin, got {other:?}"),
2040 }
2041 }
2042
2043 #[test]
2044 fn test_plan_multi_join_is_left_deep() {
2045 let plan = plan(
2047 "User as u join Order as o on u.id = o.user_id \
2048 join Product as p on o.product_id = p.id",
2049 )
2050 .unwrap();
2051 match plan {
2052 PlanNode::NestedLoopJoin { left, right, .. } => {
2053 match *right {
2055 PlanNode::AliasScan { table, .. } => assert_eq!(table, "Product"),
2056 other => panic!("expected AliasScan(Product), got {other:?}"),
2057 }
2058 assert!(matches!(*left, PlanNode::NestedLoopJoin { .. }));
2060 }
2061 other => panic!("expected NestedLoopJoin, got {other:?}"),
2062 }
2063 }
2064
2065 #[test]
2066 fn test_plan_join_with_filter_tail_wraps_filter_on_top() {
2067 let plan =
2068 plan("User as u join Order as o on u.id = o.user_id filter o.total > 100").unwrap();
2069 match plan {
2070 PlanNode::Filter { input, .. } => {
2071 assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
2072 }
2073 other => panic!("expected Filter(NestedLoopJoin), got {other:?}"),
2074 }
2075 }
2076
2077 #[test]
2078 fn test_plan_group_by_builds_groupby_node() {
2079 let plan = plan("User group .status { .status, n: count(.name) }").unwrap();
2080 match plan {
2082 PlanNode::Project { input, fields } => {
2083 assert_eq!(fields.len(), 2);
2084 match *input {
2085 PlanNode::GroupBy {
2086 input: inner,
2087 keys,
2088 aggregates,
2089 having,
2090 } => {
2091 assert!(matches!(*inner, PlanNode::SeqScan { .. }));
2092 assert_eq!(
2093 keys,
2094 vec![GroupKey {
2095 expr: Expr::Field("status".into()),
2096 output_name: "status".into(),
2097 }]
2098 );
2099 assert_eq!(aggregates.len(), 1);
2100 assert_eq!(aggregates[0].function, AggFunc::Count);
2101 assert_eq!(aggregates[0].argument, Expr::Field("name".into()));
2102 assert!(having.is_none());
2103 }
2104 other => panic!("expected GroupBy, got {other:?}"),
2105 }
2106 }
2107 other => panic!("expected Project, got {other:?}"),
2108 }
2109 }
2110
2111 #[test]
2112 fn test_plan_joined_group_applies_order_offset_limit_after_grouping() {
2113 let plan = plan(
2114 "User as u join Order as o on u.id = o.user_id \
2115 group u.status { u.status, n: count(*) } order n desc offset 1 limit 2",
2116 )
2117 .unwrap();
2118
2119 let PlanNode::Limit { input, .. } = plan else {
2120 panic!("expected Limit at the grouped-result boundary");
2121 };
2122 let PlanNode::Offset { input, .. } = *input else {
2123 panic!("expected Offset below Limit");
2124 };
2125 let PlanNode::Sort { input, .. } = *input else {
2126 panic!("expected Sort below Offset");
2127 };
2128 let PlanNode::Project { input, .. } = *input else {
2129 panic!("expected Project below Sort");
2130 };
2131 let PlanNode::GroupBy { input, .. } = *input else {
2132 panic!("expected GroupBy below Project");
2133 };
2134 assert!(
2135 matches!(*input, PlanNode::NestedLoopJoin { .. }),
2136 "joined rows must flow into GroupBy before result limiting"
2137 );
2138 }
2139
2140 #[test]
2141 fn test_plan_group_by_having_rewrites_agg_in_having() {
2142 let plan = plan("User group .status having count(.name) > 1 { .status }").unwrap();
2143 match plan {
2144 PlanNode::Project { input, .. } => {
2145 match *input {
2146 PlanNode::GroupBy {
2147 having, aggregates, ..
2148 } => {
2149 assert_eq!(aggregates.len(), 1);
2152 assert_eq!(aggregates[0].output_name, "__agg_0");
2153 let h = having.expect("having should be Some");
2154 match h {
2155 Expr::BinaryOp(l, BinOp::Gt, _) => {
2156 assert!(
2157 matches!(*l, Expr::Field(ref name) if name == "__agg_0"),
2158 "expected Field(__agg_0), got {l:?}"
2159 );
2160 }
2161 other => panic!("expected BinaryOp, got {other:?}"),
2162 }
2163 }
2164 other => panic!("expected GroupBy, got {other:?}"),
2165 }
2166 }
2167 other => panic!("expected Project, got {other:?}"),
2168 }
2169 }
2170
2171 #[test]
2172 fn test_plan_window_inserts_window_node_before_project() {
2173 let plan = plan("User { .name, rn: row_number() over (order .age) }").unwrap();
2174 match plan {
2176 PlanNode::Project { input, fields } => {
2177 assert_eq!(fields.len(), 2);
2178 assert!(
2180 matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"),
2181 "expected Field(__win_0), got {:?}",
2182 fields[1].expr
2183 );
2184 match *input {
2185 PlanNode::Window {
2186 input: inner,
2187 windows,
2188 } => {
2189 assert_eq!(windows.len(), 1);
2190 assert_eq!(windows[0].output_name, "__win_0");
2191 assert!(matches!(*inner, PlanNode::SeqScan { .. }));
2192 }
2193 other => panic!("expected Window, got {other:?}"),
2194 }
2195 }
2196 other => panic!("expected Project, got {other:?}"),
2197 }
2198 }
2199
2200 #[test]
2201 fn test_plan_multiple_windows() {
2202 let plan = plan(
2203 "User { .name, rn: row_number() over (order .age), s: sum(.salary) over (partition .dept order .salary) }"
2204 ).unwrap();
2205 match plan {
2206 PlanNode::Project { input, fields } => {
2207 assert_eq!(fields.len(), 3);
2208 assert!(matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"));
2209 assert!(matches!(&fields[2].expr, Expr::Field(name) if name == "__win_1"));
2210 match *input {
2211 PlanNode::Window { windows, .. } => {
2212 assert_eq!(windows.len(), 2);
2213 assert_eq!(windows[0].output_name, "__win_0");
2214 assert_eq!(windows[1].output_name, "__win_1");
2215 }
2216 other => panic!("expected Window, got {other:?}"),
2217 }
2218 }
2219 other => panic!("expected Project, got {other:?}"),
2220 }
2221 }
2222
2223 #[test]
2224 fn test_plan_no_window_without_over() {
2225 let plan = plan("User group .dept { .dept, total: sum(.salary) }").unwrap();
2227 match plan {
2228 PlanNode::Project { input, .. } => {
2229 assert!(
2231 matches!(*input, PlanNode::GroupBy { .. }),
2232 "expected GroupBy under Project, got {:?}",
2233 input
2234 );
2235 }
2236 other => panic!("expected Project, got {other:?}"),
2237 }
2238 }
2239
2240 #[test]
2241 fn test_plan_explain_wraps_inner() {
2242 let plan = plan("explain User filter .age > 30").unwrap();
2243 match plan {
2244 PlanNode::Explain { input } => {
2245 assert!(
2246 matches!(*input, PlanNode::RangeScan { .. }),
2247 "expected Explain(RangeScan), got {:?}",
2248 input
2249 );
2250 }
2251 other => panic!("expected Explain, got {other:?}"),
2252 }
2253 }
2254
2255 #[test]
2256 fn test_plan_explain_simple_scan() {
2257 let plan = plan("explain User").unwrap();
2258 match plan {
2259 PlanNode::Explain { input } => {
2260 assert!(matches!(*input, PlanNode::SeqScan { .. }));
2261 }
2262 other => panic!("expected Explain(SeqScan), got {other:?}"),
2263 }
2264 }
2265
2266 #[test]
2267 fn test_plan_explain_join() {
2268 let plan = plan("explain User as u join Order as o on u.id = o.user_id").unwrap();
2269 match plan {
2270 PlanNode::Explain { input } => {
2271 assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
2272 }
2273 other => panic!("expected Explain(NestedLoopJoin), got {other:?}"),
2274 }
2275 }
2276}