1pub mod aggregate_expr;
13mod error;
14pub mod knn_optimizer;
15pub mod logical_plan;
16pub mod name_resolver;
17pub mod type_checker;
18pub mod typed_expr;
19pub mod types;
20
21#[cfg(test)]
22mod planner_tests;
23
24pub use aggregate_expr::{AggregateExpr, AggregateFunction};
25pub use error::PlannerError;
26pub use knn_optimizer::{KnnPattern, SortDirection, detect_knn_pattern};
27pub use logical_plan::{
28 JoinType, LogicalPlan, OffsetWindowFunction, RecursiveCteLimits, SetOperator, WindowExpr,
29 WindowFunction,
30};
31pub use name_resolver::{NameResolver, ResolvedColumn};
32pub use type_checker::{ScopedTable, TypeChecker};
33pub use typed_expr::{
34 ProjectedColumn, Projection, SortExpr, TypedAssignment, TypedCaseWhen, TypedExpr, TypedExprKind,
35};
36pub use types::ResolvedType;
37
38use crate::ast::ddl::{
39 ColumnConstraint, ColumnDef, CreateIndex, CreateTable, DropIndex, DropTable,
40};
41use crate::ast::dml::{
42 Delete, FromItem, Insert, InsertSource, LITERAL_TABLE, OrderByExpr, Select, SelectItem,
43 SetOperator as AstSetOperator, Update,
44};
45use crate::ast::expr::{Expr, ExprKind, Literal};
46use crate::ast::{PragmaValue, Spanned, Statement, StatementKind};
47use crate::catalog::{Catalog, ColumnMetadata, IndexMetadata, TableMetadata};
48use crate::{AlopexDialect, DataSourceFormat, Parser, SqlError, TableType};
49use std::collections::{HashMap, HashSet};
50
51#[derive(Clone)]
52struct PlannedRelation {
53 plan: LogicalPlan,
54 schema: Vec<ColumnMetadata>,
55 scope: Vec<ScopedTable>,
56}
57
58type CtePlans = HashMap<String, PlannedRelation>;
59
60fn direct_from_reference_count(items: &[FromItem], name: &str) -> usize {
61 items
62 .iter()
63 .map(|item| match item {
64 FromItem::Table {
65 name: table_name, ..
66 } => usize::from(table_name == name),
67 FromItem::Join { left, right, .. } => {
68 direct_from_item_reference_count(left, name)
69 + direct_from_item_reference_count(right, name)
70 }
71 FromItem::Derived { .. } => 0,
72 })
73 .sum()
74}
75
76fn direct_from_item_reference_count(item: &FromItem, name: &str) -> usize {
77 match item {
78 FromItem::Table {
79 name: table_name, ..
80 } => usize::from(table_name == name),
81 FromItem::Join { left, right, .. } => {
82 direct_from_item_reference_count(left, name)
83 + direct_from_item_reference_count(right, name)
84 }
85 FromItem::Derived { .. } => 0,
86 }
87}
88
89fn select_table_reference_count(select: &Select, name: &str) -> usize {
90 fn from_item_count(item: &FromItem, name: &str) -> usize {
91 match item {
92 FromItem::Table {
93 name: table_name, ..
94 } => usize::from(table_name == name),
95 FromItem::Join { left, right, .. } => {
96 from_item_count(left, name) + from_item_count(right, name)
97 }
98 FromItem::Derived { subquery, .. } => match &subquery.kind {
99 StatementKind::Select(select) => select_table_reference_count(select, name),
100 _ => 0,
101 },
102 }
103 }
104
105 let from_count = select
106 .from
107 .iter()
108 .map(|item| from_item_count(item, name))
109 .sum::<usize>();
110 let set_count = select
111 .set_operations
112 .iter()
113 .map(|operation| select_table_reference_count(&operation.right, name))
114 .sum::<usize>();
115 let with_count = select.with.as_ref().map_or(0, |with| {
116 with.ctes
117 .iter()
118 .map(|cte| match &cte.query.kind {
119 StatementKind::Select(select) => select_table_reference_count(select, name),
120 _ => 0,
121 })
122 .sum()
123 });
124 from_count + set_count + with_count
125}
126
127fn cte_dependency_cycle(with: &crate::ast::WithClause) -> bool {
128 fn visit(node: usize, dependencies: &[Vec<usize>], state: &mut [u8]) -> bool {
129 state[node] = 1;
130 for &dependency in &dependencies[node] {
131 if state[dependency] == 1
132 || (state[dependency] == 0 && visit(dependency, dependencies, state))
133 {
134 return true;
135 }
136 }
137 state[node] = 2;
138 false
139 }
140
141 let dependencies = with
142 .ctes
143 .iter()
144 .map(|cte| {
145 with.ctes
146 .iter()
147 .enumerate()
148 .filter_map(|(dependency, candidate)| match &cte.query.kind {
149 StatementKind::Select(select)
150 if select_table_reference_count(select, &candidate.name) > 0 =>
151 {
152 Some(dependency)
153 }
154 _ => None,
155 })
156 .collect::<Vec<_>>()
157 })
158 .collect::<Vec<_>>();
159 let mut state = vec![0; dependencies.len()];
160 (0..dependencies.len()).any(|node| state[node] == 0 && visit(node, &dependencies, &mut state))
161}
162
163fn expr_contains_subquery(expr: &Expr) -> bool {
164 match &expr.kind {
165 ExprKind::Literal { .. } | ExprKind::ColumnRef { .. } | ExprKind::VectorLiteral { .. } => {
166 false
167 }
168 ExprKind::BinaryOp { left, right, .. } => {
169 expr_contains_subquery(left) || expr_contains_subquery(right)
170 }
171 ExprKind::UnaryOp { operand, .. } => expr_contains_subquery(operand),
172 ExprKind::Case {
173 operand,
174 branches,
175 else_expr,
176 } => {
177 operand.as_deref().is_some_and(expr_contains_subquery)
178 || branches.iter().any(|branch| {
179 expr_contains_subquery(&branch.when) || expr_contains_subquery(&branch.then)
180 })
181 || else_expr.as_deref().is_some_and(expr_contains_subquery)
182 }
183 ExprKind::FunctionCall { args, over, .. } => {
184 args.iter().any(expr_contains_subquery)
185 || over.as_ref().is_some_and(|window| {
186 window.partition_by.iter().any(expr_contains_subquery)
187 || window
188 .order_by
189 .iter()
190 .any(|order| expr_contains_subquery(&order.expr))
191 })
192 }
193 ExprKind::Cast { expr, .. } | ExprKind::IsNull { expr, .. } => expr_contains_subquery(expr),
194 ExprKind::Between {
195 expr, low, high, ..
196 } => {
197 expr_contains_subquery(expr)
198 || expr_contains_subquery(low)
199 || expr_contains_subquery(high)
200 }
201 ExprKind::Like {
202 expr,
203 pattern,
204 escape,
205 ..
206 } => {
207 expr_contains_subquery(expr)
208 || expr_contains_subquery(pattern)
209 || escape.as_deref().is_some_and(expr_contains_subquery)
210 }
211 ExprKind::InList { expr, list, .. } => {
212 expr_contains_subquery(expr) || list.iter().any(expr_contains_subquery)
213 }
214 ExprKind::ScalarSubquery { .. }
215 | ExprKind::InSubquery { .. }
216 | ExprKind::Exists { .. }
217 | ExprKind::Quantified { .. } => true,
218 }
219}
220
221fn from_item_contains_subquery(item: &FromItem) -> bool {
222 match item {
223 FromItem::Table { .. } => false,
224 FromItem::Derived { .. } => true,
225 FromItem::Join {
226 left,
227 right,
228 condition,
229 ..
230 } => {
231 from_item_contains_subquery(left)
232 || from_item_contains_subquery(right)
233 || condition.as_ref().is_some_and(expr_contains_subquery)
234 }
235 }
236}
237
238fn select_contains_subquery(select: &Select) -> bool {
239 select.projection.iter().any(|item| match item {
240 SelectItem::Expr { expr, .. } => expr_contains_subquery(expr),
241 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
242 }) || select.from.iter().any(from_item_contains_subquery)
243 || select
244 .selection
245 .as_ref()
246 .is_some_and(expr_contains_subquery)
247 || select
248 .group_by
249 .as_ref()
250 .is_some_and(|group| group.iter().any(expr_contains_subquery))
251 || select.having.as_ref().is_some_and(expr_contains_subquery)
252 || select
253 .set_operations
254 .iter()
255 .any(|operation| select_contains_subquery(&operation.right))
256 || select
257 .order_by
258 .iter()
259 .any(|order| expr_contains_subquery(&order.expr))
260 || select.limit.as_ref().is_some_and(expr_contains_subquery)
261 || select.offset.as_ref().is_some_and(expr_contains_subquery)
262 || select.with.as_ref().is_some_and(|with| {
263 with.ctes.iter().any(|cte| match &cte.query.kind {
264 StatementKind::Select(select) => select_contains_subquery(select),
265 _ => false,
266 })
267 })
268}
269
270#[derive(Debug, Clone)]
276pub struct PlannedStatement {
277 pub plan: LogicalPlan,
279 pub routing_input: RoutingInput,
281}
282
283impl PlannedStatement {
284 pub fn statement_kind(&self) -> &StatementKind {
286 &self.routing_input.statement_kind
287 }
288
289 pub fn table_references(&self) -> &[TableReference] {
291 &self.routing_input.table_references
292 }
293
294 pub fn diagnostics(&self) -> &[PlanningDiagnostic] {
297 &self.routing_input.diagnostics
298 }
299}
300
301#[derive(Debug, Clone)]
303pub struct RoutingInput {
304 pub statement_kind: StatementKind,
307 pub table_references: Vec<TableReference>,
309 pub diagnostics: Vec<PlanningDiagnostic>,
311}
312
313#[derive(Debug, Clone, PartialEq, Eq)]
315pub struct TableReference {
316 pub table_name: String,
318 pub access: TableReferenceAccess,
320 pub source: TableReferenceSource,
322}
323
324impl TableReference {
325 pub fn new(
326 table_name: impl Into<String>,
327 access: TableReferenceAccess,
328 source: TableReferenceSource,
329 ) -> Self {
330 Self {
331 table_name: table_name.into(),
332 access,
333 source,
334 }
335 }
336}
337
338#[derive(Debug, Clone, Copy, PartialEq, Eq)]
340pub enum TableReferenceAccess {
341 Read,
343 Write,
345 Create,
347 Drop,
349 Metadata,
351}
352
353#[derive(Debug, Clone, Copy, PartialEq, Eq)]
355pub enum TableReferenceSource {
356 TopLevelPlanTableName,
358 LogicalPlanScan,
360 LogicalPlanMutationTarget,
362 LogicalPlanDdlTarget,
364 LogicalPlanIndexTarget,
366 TypedExprSubquery,
368}
369
370#[derive(Debug, Clone, Copy, PartialEq, Eq)]
372pub enum PlanningDiagnosticSeverity {
373 Info,
374 Warning,
375}
376
377#[derive(Debug, Clone, PartialEq, Eq)]
379pub struct PlanningDiagnostic {
380 pub code: &'static str,
382 pub severity: PlanningDiagnosticSeverity,
384 pub message: String,
386}
387
388impl PlanningDiagnostic {
389 pub fn info(code: &'static str, message: impl Into<String>) -> Self {
390 Self {
391 code,
392 severity: PlanningDiagnosticSeverity::Info,
393 message: message.into(),
394 }
395 }
396
397 pub fn warning(code: &'static str, message: impl Into<String>) -> Self {
398 Self {
399 code,
400 severity: PlanningDiagnosticSeverity::Warning,
401 message: message.into(),
402 }
403 }
404}
405
406pub fn plan_sql_for_routing<C: Catalog + ?Sized>(
408 catalog: &C,
409 sql: &str,
410) -> Result<Vec<PlannedStatement>, SqlError> {
411 let statements = Parser::parse_sql(&AlopexDialect, sql).map_err(SqlError::from)?;
412 statements
413 .iter()
414 .map(|statement| plan_statement_for_routing(catalog, statement).map_err(SqlError::from))
415 .collect()
416}
417
418pub fn plan_statement_for_routing<C: Catalog + ?Sized>(
420 catalog: &C,
421 statement: &Statement,
422) -> Result<PlannedStatement, PlannerError> {
423 let planner = Planner::new(catalog);
424 let plan = planner.plan(statement)?;
425 let routing_input = routing_input_for_plan(statement, &plan)?;
426 Ok(PlannedStatement {
427 plan,
428 routing_input,
429 })
430}
431
432fn routing_input_for_plan(
433 statement: &Statement,
434 plan: &LogicalPlan,
435) -> Result<RoutingInput, PlannerError> {
436 let mut diagnostics = Vec::new();
437 let extractor = TableReferenceExtractor::new();
438 let table_references = extractor.extract_from_logical_plan(
439 plan,
440 table_reference_access(statement)?,
441 &mut diagnostics,
442 );
443
444 Ok(RoutingInput {
445 statement_kind: statement.kind.clone(),
446 table_references,
447 diagnostics,
448 })
449}
450
451#[derive(Debug, Default, Clone, Copy)]
453pub struct TableReferenceExtractor;
454
455impl TableReferenceExtractor {
456 pub fn new() -> Self {
457 Self
458 }
459
460 pub fn extract_from_logical_plan(
463 &self,
464 plan: &LogicalPlan,
465 root_access: TableReferenceAccess,
466 diagnostics: &mut Vec<PlanningDiagnostic>,
467 ) -> Vec<TableReference> {
468 let mut references = Vec::new();
469 self.extract_plan(
470 plan,
471 root_access,
472 TableReferenceSource::LogicalPlanScan,
473 diagnostics,
474 &mut references,
475 );
476 if references.is_empty() {
477 diagnostics.push(PlanningDiagnostic::info(
478 "ALOPEX-PLAN-ROUTE-001",
479 "statement has no physical table reference",
480 ));
481 }
482 references
483 }
484
485 pub fn extract_from_subquery_context(
487 &self,
488 plan: &LogicalPlan,
489 diagnostics: &mut Vec<PlanningDiagnostic>,
490 ) -> Vec<TableReference> {
491 let mut references = Vec::new();
492 self.extract_plan(
493 plan,
494 TableReferenceAccess::Read,
495 TableReferenceSource::TypedExprSubquery,
496 diagnostics,
497 &mut references,
498 );
499 references
500 }
501
502 fn extract_plan(
503 &self,
504 plan: &LogicalPlan,
505 root_access: TableReferenceAccess,
506 scan_source: TableReferenceSource,
507 diagnostics: &mut Vec<PlanningDiagnostic>,
508 references: &mut Vec<TableReference>,
509 ) {
510 match plan {
511 LogicalPlan::Scan { table, projection } => {
512 if table != LITERAL_TABLE {
513 push_table_reference(
514 references,
515 table,
516 TableReferenceAccess::Read,
517 scan_source,
518 );
519 }
520 self.extract_projection(projection, diagnostics, references);
521 }
522 LogicalPlan::Filter { input, predicate } => {
523 self.extract_plan(input, root_access, scan_source, diagnostics, references);
524 self.extract_typed_expr(predicate, diagnostics, references);
525 }
526 LogicalPlan::Project { input, projection } => {
527 self.extract_plan(input, root_access, scan_source, diagnostics, references);
528 self.extract_projection(projection, diagnostics, references);
529 }
530 LogicalPlan::Join {
531 left,
532 right,
533 condition,
534 ..
535 } => {
536 self.extract_plan(
537 left,
538 TableReferenceAccess::Read,
539 scan_source,
540 diagnostics,
541 references,
542 );
543 self.extract_plan(
544 right,
545 TableReferenceAccess::Read,
546 scan_source,
547 diagnostics,
548 references,
549 );
550 if let Some(condition) = condition {
551 self.extract_typed_expr(condition, diagnostics, references);
552 }
553 }
554 LogicalPlan::Aggregate {
555 input,
556 group_keys,
557 aggregates,
558 having,
559 projection,
560 } => {
561 self.extract_plan(input, root_access, scan_source, diagnostics, references);
562 for expr in group_keys {
563 self.extract_typed_expr(expr, diagnostics, references);
564 }
565 for aggregate in aggregates {
566 if let Some(arg) = &aggregate.arg {
567 self.extract_typed_expr(arg, diagnostics, references);
568 }
569 }
570 if let Some(having) = having {
571 self.extract_typed_expr(having, diagnostics, references);
572 }
573 self.extract_projection(projection, diagnostics, references);
574 }
575 LogicalPlan::Window { input, windows } => {
576 self.extract_plan(input, root_access, scan_source, diagnostics, references);
577 for window in windows {
578 for expr in &window.partition_by {
579 self.extract_typed_expr(expr, diagnostics, references);
580 }
581 for sort_expr in &window.order_by {
582 self.extract_typed_expr(&sort_expr.expr, diagnostics, references);
583 }
584 match &window.function {
585 WindowFunction::Aggregate(aggregate) => {
586 if let Some(arg) = &aggregate.arg {
587 self.extract_typed_expr(arg, diagnostics, references);
588 }
589 }
590 WindowFunction::Lag(function) | WindowFunction::Lead(function) => {
591 self.extract_typed_expr(&function.value, diagnostics, references);
592 if let Some(offset) = &function.offset {
593 self.extract_typed_expr(offset, diagnostics, references);
594 }
595 if let Some(default) = &function.default {
596 self.extract_typed_expr(default, diagnostics, references);
597 }
598 }
599 WindowFunction::RowNumber
600 | WindowFunction::Rank
601 | WindowFunction::DenseRank => {}
602 }
603 }
604 }
605 LogicalPlan::SetOperation { left, right, .. } => {
606 self.extract_plan(left, root_access, scan_source, diagnostics, references);
607 self.extract_plan(right, root_access, scan_source, diagnostics, references);
608 }
609 LogicalPlan::RecursiveCte {
610 anchor,
611 recursive_term,
612 ..
613 } => {
614 self.extract_plan(anchor, root_access, scan_source, diagnostics, references);
615 self.extract_plan(
616 recursive_term,
617 root_access,
618 scan_source,
619 diagnostics,
620 references,
621 );
622 }
623 LogicalPlan::RecursiveReference { .. } => {}
624 LogicalPlan::Sort { input, order_by } => {
625 self.extract_plan(input, root_access, scan_source, diagnostics, references);
626 for sort_expr in order_by {
627 self.extract_typed_expr(&sort_expr.expr, diagnostics, references);
628 }
629 }
630 LogicalPlan::Limit { input, .. } => {
631 self.extract_plan(input, root_access, scan_source, diagnostics, references);
632 }
633 LogicalPlan::Insert { table, values, .. } => {
634 push_table_reference(
635 references,
636 table,
637 root_access,
638 TableReferenceSource::LogicalPlanMutationTarget,
639 );
640 for row in values {
641 for value in row {
642 self.extract_typed_expr(value, diagnostics, references);
643 }
644 }
645 }
646 LogicalPlan::InsertSelect { table, source, .. } => {
647 push_table_reference(
648 references,
649 table,
650 root_access,
651 TableReferenceSource::LogicalPlanMutationTarget,
652 );
653 self.extract_plan(
654 source,
655 TableReferenceAccess::Read,
656 scan_source,
657 diagnostics,
658 references,
659 );
660 }
661 LogicalPlan::Update {
662 table,
663 assignments,
664 filter,
665 } => {
666 push_table_reference(
667 references,
668 table,
669 root_access,
670 TableReferenceSource::LogicalPlanMutationTarget,
671 );
672 for assignment in assignments {
673 self.extract_typed_expr(&assignment.value, diagnostics, references);
674 }
675 if let Some(filter) = filter {
676 self.extract_typed_expr(filter, diagnostics, references);
677 }
678 }
679 LogicalPlan::Delete { table, filter } => {
680 push_table_reference(
681 references,
682 table,
683 root_access,
684 TableReferenceSource::LogicalPlanMutationTarget,
685 );
686 if let Some(filter) = filter {
687 self.extract_typed_expr(filter, diagnostics, references);
688 }
689 }
690 LogicalPlan::CreateTable { table, .. } => push_table_reference(
691 references,
692 &table.name,
693 root_access,
694 TableReferenceSource::LogicalPlanDdlTarget,
695 ),
696 LogicalPlan::DropTable { name, .. } => push_table_reference(
697 references,
698 name,
699 root_access,
700 TableReferenceSource::LogicalPlanDdlTarget,
701 ),
702 LogicalPlan::CreateIndex { index, .. } => push_table_reference(
703 references,
704 &index.table,
705 root_access,
706 TableReferenceSource::LogicalPlanIndexTarget,
707 ),
708 LogicalPlan::DropIndex { name, .. } => diagnostics.push(PlanningDiagnostic::warning(
709 "ALOPEX-PLAN-ROUTE-003",
710 format!(
711 "DROP INDEX {name} does not expose a target table in the current logical plan"
712 ),
713 )),
714 LogicalPlan::Pragma { .. } => {}
715 }
716 }
717
718 fn extract_projection(
719 &self,
720 projection: &Projection,
721 diagnostics: &mut Vec<PlanningDiagnostic>,
722 references: &mut Vec<TableReference>,
723 ) {
724 if let Projection::Columns(columns) = projection {
725 for column in columns {
726 self.extract_typed_expr(&column.expr, diagnostics, references);
727 }
728 }
729 }
730
731 fn extract_typed_expr(
732 &self,
733 expr: &TypedExpr,
734 diagnostics: &mut Vec<PlanningDiagnostic>,
735 references: &mut Vec<TableReference>,
736 ) {
737 match &expr.kind {
738 TypedExprKind::Literal(_)
739 | TypedExprKind::ColumnRef { .. }
740 | TypedExprKind::VectorLiteral(_) => {}
741 TypedExprKind::BinaryOp { left, right, .. } => {
742 self.extract_typed_expr(left, diagnostics, references);
743 self.extract_typed_expr(right, diagnostics, references);
744 }
745 TypedExprKind::UnaryOp { operand, .. }
746 | TypedExprKind::Cast { expr: operand, .. }
747 | TypedExprKind::IsNull { expr: operand, .. } => {
748 self.extract_typed_expr(operand, diagnostics, references);
749 }
750 TypedExprKind::Case {
751 operand,
752 branches,
753 else_expr,
754 } => {
755 if let Some(operand) = operand {
756 self.extract_typed_expr(operand, diagnostics, references);
757 }
758 for branch in branches {
759 self.extract_typed_expr(&branch.when, diagnostics, references);
760 self.extract_typed_expr(&branch.then, diagnostics, references);
761 }
762 if let Some(else_expr) = else_expr {
763 self.extract_typed_expr(else_expr, diagnostics, references);
764 }
765 }
766 TypedExprKind::FunctionCall { args, .. } => {
767 for arg in args {
768 self.extract_typed_expr(arg, diagnostics, references);
769 }
770 }
771 TypedExprKind::Between {
772 expr, low, high, ..
773 } => {
774 self.extract_typed_expr(expr, diagnostics, references);
775 self.extract_typed_expr(low, diagnostics, references);
776 self.extract_typed_expr(high, diagnostics, references);
777 }
778 TypedExprKind::Like {
779 expr,
780 pattern,
781 escape,
782 ..
783 } => {
784 self.extract_typed_expr(expr, diagnostics, references);
785 self.extract_typed_expr(pattern, diagnostics, references);
786 if let Some(escape) = escape {
787 self.extract_typed_expr(escape, diagnostics, references);
788 }
789 }
790 TypedExprKind::InList { expr, list, .. } => {
791 self.extract_typed_expr(expr, diagnostics, references);
792 for item in list {
793 self.extract_typed_expr(item, diagnostics, references);
794 }
795 }
796 TypedExprKind::ScalarSubquery(subquery) => self.extract_plan(
797 subquery,
798 TableReferenceAccess::Read,
799 TableReferenceSource::TypedExprSubquery,
800 diagnostics,
801 references,
802 ),
803 TypedExprKind::InSubquery { expr, subquery, .. } => {
804 self.extract_typed_expr(expr, diagnostics, references);
805 self.extract_plan(
806 subquery,
807 TableReferenceAccess::Read,
808 TableReferenceSource::TypedExprSubquery,
809 diagnostics,
810 references,
811 );
812 }
813 TypedExprKind::Exists { subquery, .. } => self.extract_plan(
814 subquery,
815 TableReferenceAccess::Read,
816 TableReferenceSource::TypedExprSubquery,
817 diagnostics,
818 references,
819 ),
820 TypedExprKind::Quantified { expr, subquery, .. } => {
821 self.extract_typed_expr(expr, diagnostics, references);
822 self.extract_plan(
823 subquery,
824 TableReferenceAccess::Read,
825 TableReferenceSource::TypedExprSubquery,
826 diagnostics,
827 references,
828 );
829 }
830 }
831 }
832}
833
834fn push_table_reference(
835 references: &mut Vec<TableReference>,
836 table_name: &str,
837 access: TableReferenceAccess,
838 source: TableReferenceSource,
839) {
840 if !references.iter().any(|reference| {
841 reference.table_name == table_name
842 && reference.access == access
843 && reference.source == source
844 }) {
845 references.push(TableReference::new(table_name, access, source));
846 }
847}
848
849#[derive(Debug)]
850enum GenericHostStatement<'a> {
851 CreateTable(&'a CreateTable),
852 DropTable(&'a DropTable),
853 CreateIndex(&'a CreateIndex),
854 DropIndex(&'a DropIndex),
855 Pragma {
856 name: &'a str,
857 value: &'a Option<PragmaValue>,
858 },
859 Select(&'a Select),
860 Insert(&'a Insert),
861 Update(&'a Update),
862 Delete(&'a Delete),
863 Unsupported,
864}
865
866fn classify_generic_host_statement(statement_kind: &StatementKind) -> GenericHostStatement<'_> {
867 #[allow(unreachable_patterns)]
870 match statement_kind {
871 StatementKind::CreateTable(statement) => GenericHostStatement::CreateTable(statement),
872 StatementKind::DropTable(statement) => GenericHostStatement::DropTable(statement),
873 StatementKind::CreateIndex(statement) => GenericHostStatement::CreateIndex(statement),
874 StatementKind::DropIndex(statement) => GenericHostStatement::DropIndex(statement),
875 StatementKind::Pragma { name, value } => GenericHostStatement::Pragma { name, value },
876 StatementKind::Select(statement) => GenericHostStatement::Select(statement),
877 StatementKind::Insert(statement) => GenericHostStatement::Insert(statement),
878 StatementKind::Update(statement) => GenericHostStatement::Update(statement),
879 StatementKind::Delete(statement) => GenericHostStatement::Delete(statement),
880 _ => GenericHostStatement::Unsupported,
881 }
882}
883
884fn unsupported_generic_statement(statement: &Statement) -> PlannerError {
885 PlannerError::unsupported_feature(
886 "statement kind for the generic SQL planner",
887 "a statement-specific planner",
888 statement.span,
889 )
890}
891
892fn table_reference_access(statement: &Statement) -> Result<TableReferenceAccess, PlannerError> {
893 table_reference_access_for_classified(
894 statement,
895 classify_generic_host_statement(&statement.kind),
896 )
897}
898
899fn table_reference_access_for_classified(
900 statement: &Statement,
901 classified: GenericHostStatement<'_>,
902) -> Result<TableReferenceAccess, PlannerError> {
903 match classified {
904 GenericHostStatement::Select(_) => Ok(TableReferenceAccess::Read),
905 GenericHostStatement::Insert(_)
906 | GenericHostStatement::Update(_)
907 | GenericHostStatement::Delete(_) => Ok(TableReferenceAccess::Write),
908 GenericHostStatement::CreateTable(_) => Ok(TableReferenceAccess::Create),
909 GenericHostStatement::DropTable(_) => Ok(TableReferenceAccess::Drop),
910 GenericHostStatement::CreateIndex(_)
911 | GenericHostStatement::DropIndex(_)
912 | GenericHostStatement::Pragma { .. } => Ok(TableReferenceAccess::Metadata),
913 GenericHostStatement::Unsupported => Err(unsupported_generic_statement(statement)),
914 }
915}
916
917pub struct Planner<'a, C: Catalog + ?Sized> {
944 catalog: &'a C,
945 name_resolver: NameResolver<'a, C>,
946 type_checker: TypeChecker<'a, C>,
947}
948
949impl<'a, C: Catalog + ?Sized> Planner<'a, C> {
950 pub fn new(catalog: &'a C) -> Self {
952 Self {
953 catalog,
954 name_resolver: NameResolver::new(catalog),
955 type_checker: TypeChecker::new(catalog),
956 }
957 }
958
959 pub fn plan(&self, stmt: &Statement) -> Result<LogicalPlan, PlannerError> {
970 self.plan_classified_statement(stmt, classify_generic_host_statement(&stmt.kind))
971 }
972
973 fn plan_classified_statement(
974 &self,
975 stmt: &Statement,
976 classified: GenericHostStatement<'_>,
977 ) -> Result<LogicalPlan, PlannerError> {
978 match classified {
979 GenericHostStatement::CreateTable(statement) => self.plan_create_table(statement),
981 GenericHostStatement::DropTable(statement) => self.plan_drop_table(statement),
982 GenericHostStatement::CreateIndex(statement) => self.plan_create_index(statement),
983 GenericHostStatement::DropIndex(statement) => self.plan_drop_index(statement),
984 GenericHostStatement::Pragma { name, value } => self.plan_pragma(name, value),
985
986 GenericHostStatement::Select(statement) => self.plan_select(statement),
988 GenericHostStatement::Insert(statement) => self.plan_insert(statement),
989 GenericHostStatement::Update(statement) => self.plan_update(statement),
990 GenericHostStatement::Delete(statement) => self.plan_delete(statement),
991 GenericHostStatement::Unsupported => Err(unsupported_generic_statement(stmt)),
992 }
993 }
994
995 fn plan_pragma(
996 &self,
997 raw_name: &str,
998 value: &Option<PragmaValue>,
999 ) -> Result<LogicalPlan, PlannerError> {
1000 let name = raw_name.to_ascii_lowercase();
1001 if !matches!(name.as_str(), "cache_size" | "memory_limit" | "io_stats") {
1002 return Err(PlannerError::InvalidPragma {
1003 name,
1004 reason: "supported names are cache_size, memory_limit, and io_stats".to_string(),
1005 });
1006 }
1007 match name.as_str() {
1008 "cache_size" => match value {
1009 Some(PragmaValue::Int(v)) if *v > 0 => {}
1010 Some(PragmaValue::Int(_)) => {
1011 return Err(PlannerError::InvalidPragma {
1012 name,
1013 reason: "cache_size must be a positive page count".to_string(),
1014 });
1015 }
1016 Some(PragmaValue::Text(_)) => {
1017 return Err(PlannerError::InvalidPragma {
1018 name,
1019 reason: "cache_size requires an integer page count".to_string(),
1020 });
1021 }
1022 None => {}
1023 },
1024 "memory_limit" => {
1025 if let Some(PragmaValue::Int(v)) = value
1026 && *v < 0
1027 {
1028 return Err(PlannerError::InvalidPragma {
1029 name,
1030 reason: "memory_limit cannot be negative".to_string(),
1031 });
1032 }
1033 }
1034 "io_stats" => {
1035 if value.is_some() {
1036 return Err(PlannerError::InvalidPragma {
1037 name,
1038 reason: "io_stats does not accept a value".to_string(),
1039 });
1040 }
1041 }
1042 _ => unreachable!(),
1043 }
1044 Ok(LogicalPlan::Pragma {
1045 name,
1046 value: value.clone(),
1047 })
1048 }
1049
1050 fn plan_create_table(&self, stmt: &CreateTable) -> Result<LogicalPlan, PlannerError> {
1059 if !stmt.if_not_exists && self.catalog.table_exists(&stmt.name) {
1061 return Err(PlannerError::table_already_exists(&stmt.name));
1062 }
1063
1064 let columns: Vec<ColumnMetadata> = stmt
1066 .columns
1067 .iter()
1068 .map(|col| self.convert_column_def(col))
1069 .collect();
1070
1071 let primary_key = Self::extract_primary_key(stmt);
1073
1074 let mut table = TableMetadata::new(stmt.name.clone(), columns);
1077 if let Some(pk) = primary_key {
1078 table = table.with_primary_key(pk);
1079 }
1080 table.catalog_name = "default".to_string();
1081 table.namespace_name = "default".to_string();
1082 table.table_type = TableType::Managed;
1083 table.data_source_format = DataSourceFormat::Alopex;
1084 table.properties = HashMap::new();
1085
1086 Ok(LogicalPlan::CreateTable {
1087 table,
1088 if_not_exists: stmt.if_not_exists,
1089 with_options: stmt
1090 .with_options
1091 .iter()
1092 .map(|opt| (opt.key.clone(), opt.value.clone()))
1093 .collect(),
1094 })
1095 }
1096
1097 fn convert_column_def(&self, col: &ColumnDef) -> ColumnMetadata {
1099 let data_type = ResolvedType::from_ast(&col.data_type);
1100 let mut meta = ColumnMetadata::new(col.name.clone(), data_type);
1101
1102 for constraint in &col.constraints {
1104 meta = Self::apply_column_constraint(meta, constraint);
1105 }
1106
1107 meta
1108 }
1109
1110 fn apply_column_constraint(
1112 mut meta: ColumnMetadata,
1113 constraint: &ColumnConstraint,
1114 ) -> ColumnMetadata {
1115 match constraint {
1116 ColumnConstraint::NotNull { .. } => {
1117 meta.not_null = true;
1118 }
1119 ColumnConstraint::PrimaryKey { .. } => {
1120 meta.primary_key = true;
1121 meta.not_null = true; }
1123 ColumnConstraint::Unique { .. } => {
1124 meta.unique = true;
1125 }
1126 ColumnConstraint::Default { value: expr, .. } => {
1127 meta.default = Some(expr.clone());
1128 }
1129 }
1130 meta
1131 }
1132
1133 fn extract_primary_key(stmt: &CreateTable) -> Option<Vec<String>> {
1135 use crate::ast::ddl::TableConstraint;
1136
1137 if let Some(TableConstraint::PrimaryKey { columns, .. }) = stmt.constraints.first() {
1141 return Some(columns.clone());
1142 }
1143
1144 let pk_columns: Vec<String> = stmt
1146 .columns
1147 .iter()
1148 .filter(|col| col.constraints.iter().any(Self::is_primary_key_constraint))
1149 .map(|col| col.name.clone())
1150 .collect();
1151
1152 if pk_columns.is_empty() {
1153 None
1154 } else {
1155 Some(pk_columns)
1156 }
1157 }
1158
1159 fn is_primary_key_constraint(constraint: &ColumnConstraint) -> bool {
1161 matches!(constraint, ColumnConstraint::PrimaryKey { .. })
1162 }
1163
1164 fn plan_drop_table(&self, stmt: &DropTable) -> Result<LogicalPlan, PlannerError> {
1168 if !stmt.if_exists && !self.table_exists_in_default(&stmt.name) {
1170 return Err(PlannerError::TableNotFound {
1171 name: stmt.name.clone(),
1172 line: stmt.span.start.line,
1173 column: stmt.span.start.column,
1174 });
1175 }
1176
1177 Ok(LogicalPlan::DropTable {
1178 name: stmt.name.clone(),
1179 if_exists: stmt.if_exists,
1180 })
1181 }
1182
1183 fn table_exists_in_default(&self, name: &str) -> bool {
1184 match self.catalog.get_table(name) {
1185 Some(table) => table.catalog_name == "default" && table.namespace_name == "default",
1186 None => false,
1187 }
1188 }
1189
1190 fn plan_create_index(&self, stmt: &CreateIndex) -> Result<LogicalPlan, PlannerError> {
1197 if !stmt.if_not_exists && self.catalog.index_exists(&stmt.name) {
1199 return Err(PlannerError::index_already_exists(&stmt.name));
1200 }
1201
1202 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
1204
1205 self.name_resolver
1207 .resolve_column(table, &stmt.column, stmt.span)?;
1208
1209 let mut index = IndexMetadata::new(
1213 0,
1214 stmt.name.clone(),
1215 stmt.table.clone(),
1216 vec![stmt.column.clone()],
1217 );
1218
1219 if let Some(method) = stmt.method {
1220 index = index.with_method(method);
1221 }
1222
1223 let options: Vec<(String, String)> = stmt
1224 .options
1225 .iter()
1226 .map(|opt| (opt.key.clone(), opt.value.clone()))
1227 .collect();
1228 if !options.is_empty() {
1229 index = index.with_options(options);
1230 }
1231
1232 Ok(LogicalPlan::CreateIndex {
1233 index,
1234 if_not_exists: stmt.if_not_exists,
1235 })
1236 }
1237
1238 fn plan_drop_index(&self, stmt: &DropIndex) -> Result<LogicalPlan, PlannerError> {
1242 if !stmt.if_exists && !self.index_exists_in_default(&stmt.name) {
1244 return Err(PlannerError::index_not_found(&stmt.name));
1245 }
1246
1247 Ok(LogicalPlan::DropIndex {
1248 name: stmt.name.clone(),
1249 if_exists: stmt.if_exists,
1250 })
1251 }
1252
1253 fn index_exists_in_default(&self, name: &str) -> bool {
1254 match self.catalog.get_index(name) {
1255 Some(index) => index.catalog_name == "default" && index.namespace_name == "default",
1256 None => false,
1257 }
1258 }
1259
1260 fn plan_select(&self, stmt: &Select) -> Result<LogicalPlan, PlannerError> {
1269 self.plan_select_relation(stmt, &[], &CtePlans::new())
1270 .map(|relation| relation.plan)
1271 }
1272
1273 fn plan_ctes(
1274 &self,
1275 stmt: &Select,
1276 enclosing_ctes: &CtePlans,
1277 ) -> Result<CtePlans, PlannerError> {
1278 let Some(with) = &stmt.with else {
1279 return Ok(enclosing_ctes.clone());
1280 };
1281 let mut declared_names = HashSet::new();
1282 for cte in &with.ctes {
1283 if !declared_names.insert(&cte.name) {
1284 return Err(PlannerError::invalid_expression(format!(
1285 "common table expression '{}' is defined more than once",
1286 cte.name
1287 )));
1288 }
1289 }
1290 if with.recursive && cte_dependency_cycle(with) {
1291 return self.plan_recursive_cte(with, enclosing_ctes);
1292 }
1293
1294 let mut plans = enclosing_ctes.clone();
1295 let mut local_names = HashSet::new();
1296 for cte in &with.ctes {
1297 if !local_names.insert(cte.name.clone()) {
1298 return Err(PlannerError::invalid_expression(format!(
1299 "common table expression '{}' is defined more than once",
1300 cte.name
1301 )));
1302 }
1303 let StatementKind::Select(select) = &cte.query.kind else {
1304 return Err(PlannerError::unsupported_feature(
1305 "non-SELECT common table expression",
1306 "a future version",
1307 cte.span,
1308 ));
1309 };
1310 let mut relation = self.plan_select_relation(select, &[], &plans)?;
1311 if !cte.columns.is_empty() {
1312 if cte.columns.len() != relation.schema.len() {
1313 return Err(PlannerError::cte_column_count_mismatch(
1314 &cte.name,
1315 cte.columns.len(),
1316 relation.schema.len(),
1317 cte.span,
1318 ));
1319 }
1320
1321 let mut column_names = HashSet::new();
1322 for column_name in &cte.columns {
1323 if !column_names.insert(column_name) {
1324 return Err(PlannerError::duplicate_cte_column(
1325 &cte.name,
1326 column_name,
1327 cte.span,
1328 ));
1329 }
1330 }
1331
1332 for (column, column_name) in relation.schema.iter_mut().zip(&cte.columns) {
1333 column.name.clone_from(column_name);
1334 }
1335 }
1336 plans.insert(cte.name.clone(), relation);
1337 }
1338 Ok(plans)
1339 }
1340
1341 fn plan_recursive_cte(
1342 &self,
1343 with: &crate::ast::WithClause,
1344 enclosing_ctes: &CtePlans,
1345 ) -> Result<CtePlans, PlannerError> {
1346 if with.ctes.len() != 1 {
1347 return Err(PlannerError::unsupported_feature(
1348 "recursive WITH containing anything other than exactly one common table expression",
1349 "a future version",
1350 with.span,
1351 ));
1352 }
1353
1354 let cte = &with.ctes[0];
1355 let mut column_names = HashSet::new();
1356 for column_name in &cte.columns {
1357 if !column_names.insert(column_name) {
1358 return Err(PlannerError::duplicate_cte_column(
1359 &cte.name,
1360 column_name,
1361 cte.span,
1362 ));
1363 }
1364 }
1365
1366 let StatementKind::Select(body) = &cte.query.kind else {
1367 return Err(PlannerError::unsupported_feature(
1368 "non-SELECT recursive common table expression",
1369 "a future version",
1370 cte.span,
1371 ));
1372 };
1373 if body.set_operations.len() != 1 {
1374 return Err(PlannerError::unsupported_feature(
1375 "recursive common table expression without exactly one UNION or UNION ALL",
1376 "a future version",
1377 cte.span,
1378 ));
1379 }
1380 if !body.order_by.is_empty() || body.limit.is_some() || body.offset.is_some() {
1381 return Err(PlannerError::unsupported_feature(
1382 "ORDER BY, LIMIT, or OFFSET inside a recursive common table expression",
1383 "a future version",
1384 body.span,
1385 ));
1386 }
1387
1388 let operation = &body.set_operations[0];
1389 if operation.operator != AstSetOperator::Union {
1390 return Err(PlannerError::unsupported_feature(
1391 "recursive common table expression using an operator other than UNION or UNION ALL",
1392 "a future version",
1393 operation.span,
1394 ));
1395 }
1396
1397 let mut anchor = body.clone();
1398 anchor.with = None;
1399 anchor.set_operations.clear();
1400 if select_table_reference_count(&anchor, &cte.name) != 0 {
1401 return Err(PlannerError::unsupported_feature(
1402 "recursive common table expression without an anchor term that does not reference itself",
1403 "a future version",
1404 anchor.span,
1405 ));
1406 }
1407
1408 let recursive_term = operation.right.as_ref();
1409 if select_contains_subquery(recursive_term) {
1410 return Err(PlannerError::unsupported_feature(
1411 "subquery in a recursive term",
1412 "a future version",
1413 recursive_term.span,
1414 ));
1415 }
1416 let total_references = select_table_reference_count(recursive_term, &cte.name);
1417 let direct_references = direct_from_reference_count(&recursive_term.from, &cte.name);
1418 if total_references != 1 || direct_references != 1 {
1419 return Err(PlannerError::unsupported_feature(
1420 "recursive term without exactly one direct self-reference",
1421 "a future version",
1422 recursive_term.span,
1423 ));
1424 }
1425 if recursive_term.with.is_some() || !recursive_term.set_operations.is_empty() {
1426 return Err(PlannerError::unsupported_feature(
1427 "nested WITH or set operation in a recursive term",
1428 "a future version",
1429 recursive_term.span,
1430 ));
1431 }
1432
1433 let mut anchor_relation = self.plan_select_relation(&anchor, &[], enclosing_ctes)?;
1434 if cte.columns.is_empty() {
1435 let mut anchor_names = HashSet::new();
1436 for column in &anchor_relation.schema {
1437 if !anchor_names.insert(&column.name) {
1438 return Err(PlannerError::duplicate_cte_column(
1439 &cte.name,
1440 &column.name,
1441 cte.span,
1442 ));
1443 }
1444 }
1445 } else {
1446 if cte.columns.len() != anchor_relation.schema.len() {
1447 return Err(PlannerError::cte_column_count_mismatch(
1448 &cte.name,
1449 cte.columns.len(),
1450 anchor_relation.schema.len(),
1451 cte.span,
1452 ));
1453 }
1454 for (column, name) in anchor_relation.schema.iter_mut().zip(&cte.columns) {
1455 column.name.clone_from(name);
1456 }
1457 }
1458
1459 let mut recursive_scope = enclosing_ctes.clone();
1460 recursive_scope.insert(
1461 cte.name.clone(),
1462 PlannedRelation {
1463 plan: LogicalPlan::RecursiveReference {
1464 name: cte.name.clone(),
1465 schema: anchor_relation.schema.clone(),
1466 },
1467 schema: anchor_relation.schema.clone(),
1468 scope: vec![ScopedTable::new(
1469 TableMetadata::new(&cte.name, anchor_relation.schema.clone()),
1470 0,
1471 )],
1472 },
1473 );
1474 let recursive_relation =
1475 self.plan_select_relation(recursive_term, &[], &recursive_scope)?;
1476 if recursive_relation.schema.len() != anchor_relation.schema.len() {
1477 return Err(PlannerError::set_operation_column_count_mismatch(
1478 anchor_relation.schema.len(),
1479 recursive_relation.schema.len(),
1480 operation.span,
1481 ));
1482 }
1483 for (anchor_column, recursive_column) in anchor_relation
1484 .schema
1485 .iter()
1486 .zip(&recursive_relation.schema)
1487 {
1488 if anchor_column.data_type != recursive_column.data_type {
1489 return Err(PlannerError::type_mismatch(
1490 anchor_column.data_type.type_name(),
1491 recursive_column.data_type.type_name(),
1492 operation.span,
1493 ));
1494 }
1495 }
1496
1497 let schema = anchor_relation.schema.clone();
1498 let relation = PlannedRelation {
1499 plan: LogicalPlan::RecursiveCte {
1500 name: cte.name.clone(),
1501 anchor: Box::new(anchor_relation.plan),
1502 recursive_term: Box::new(recursive_relation.plan),
1503 union_all: operation.all,
1504 schema: schema.clone(),
1505 limits: RecursiveCteLimits::default(),
1506 },
1507 schema: schema.clone(),
1508 scope: vec![ScopedTable::new(TableMetadata::new(&cte.name, schema), 0)],
1509 };
1510 let mut plans = enclosing_ctes.clone();
1511 plans.insert(cte.name.clone(), relation);
1512 Ok(plans)
1513 }
1514
1515 fn plan_select_relation(
1516 &self,
1517 stmt: &Select,
1518 outer_scope: &[ScopedTable],
1519 enclosing_ctes: &CtePlans,
1520 ) -> Result<PlannedRelation, PlannerError> {
1521 let ctes = self.plan_ctes(stmt, enclosing_ctes)?;
1522 if !stmt.set_operations.is_empty() {
1523 let mut left_select = stmt.clone();
1524 left_select.with = None;
1525 left_select.set_operations.clear();
1526 left_select.order_by.clear();
1527 left_select.limit = None;
1528 left_select.offset = None;
1529 let mut relation = self.plan_select_relation(&left_select, outer_scope, &ctes)?;
1530
1531 for operation in &stmt.set_operations {
1532 let right = self.plan_select_relation(&operation.right, outer_scope, &ctes)?;
1533 if relation.schema.len() != right.schema.len() {
1534 return Err(PlannerError::set_operation_column_count_mismatch(
1535 relation.schema.len(),
1536 right.schema.len(),
1537 operation.span,
1538 ));
1539 }
1540 for (left_column, right_column) in relation.schema.iter().zip(&right.schema) {
1541 if left_column.data_type != right_column.data_type {
1542 return Err(PlannerError::type_mismatch(
1543 left_column.data_type.type_name(),
1544 right_column.data_type.type_name(),
1545 operation.span,
1546 ));
1547 }
1548 }
1549
1550 relation.plan = LogicalPlan::SetOperation {
1551 left: Box::new(relation.plan),
1552 right: Box::new(right.plan),
1553 operator: match operation.operator {
1554 AstSetOperator::Union => SetOperator::Union,
1555 AstSetOperator::Intersect => SetOperator::Intersect,
1556 AstSetOperator::Except => SetOperator::Except,
1557 },
1558 all: operation.all,
1559 };
1560 }
1561
1562 relation.scope = vec![ScopedTable::new(
1563 TableMetadata::new(LITERAL_TABLE, relation.schema.clone()),
1564 0,
1565 )];
1566 if !stmt.order_by.is_empty() {
1567 let order_by = self.build_sort_exprs_with_scope(
1571 &stmt.order_by,
1572 &relation.scope,
1573 &HashMap::new(),
1574 &ctes,
1575 )?;
1576 relation.plan = LogicalPlan::Sort {
1577 input: Box::new(relation.plan),
1578 order_by,
1579 };
1580 }
1581 if stmt.limit.is_some() || stmt.offset.is_some() {
1582 relation.plan = LogicalPlan::Limit {
1583 input: Box::new(relation.plan),
1584 limit: self.extract_limit_value(&stmt.limit, stmt.span)?,
1585 offset: self.extract_limit_value(&stmt.offset, stmt.span)?,
1586 };
1587 }
1588 return Ok(relation);
1589 }
1590
1591 let mut relation = self.plan_from_items(&stmt.from, stmt.span, outer_scope, &ctes)?;
1592 let expr_scope = relation
1593 .scope
1594 .iter()
1595 .cloned()
1596 .chain(offset_scope(outer_scope, relation.schema.len()))
1597 .collect::<Vec<_>>();
1598
1599 let has_group_by = stmt
1600 .group_by
1601 .as_ref()
1602 .is_some_and(|items| !items.is_empty());
1603 let has_aggregate = self.select_contains_aggregate(stmt);
1604 let has_window = select_contains_window(stmt);
1605 let distinct_only =
1606 stmt.distinct && !has_group_by && !has_aggregate && stmt.having.is_none();
1607
1608 if stmt.having.as_ref().is_some_and(expr_contains_window) {
1609 return Err(PlannerError::invalid_expression(
1610 "HAVING cannot contain window functions".to_string(),
1611 ));
1612 }
1613 if stmt
1614 .group_by
1615 .as_ref()
1616 .is_some_and(|items| items.iter().any(expr_contains_window))
1617 {
1618 return Err(PlannerError::invalid_expression(
1619 "GROUP BY cannot contain window functions".to_string(),
1620 ));
1621 }
1622
1623 let projection_aliases = collect_projection_aliases(&stmt.projection);
1627
1628 let final_projection = self.build_projection_with_scope(
1629 &stmt.projection,
1630 &relation.schema,
1631 &expr_scope,
1632 &ctes,
1633 )?;
1634 if !has_window {
1635 install_base_projection(&mut relation.plan, &final_projection);
1636 }
1637 let needs_project_boundary = !matches!(relation.plan, LogicalPlan::Scan { .. });
1638 let base_schema = relation.schema.clone();
1639 let mut plan = relation.plan;
1640
1641 if let Some(ref selection) = stmt.selection {
1643 if expr_contains_window(selection) {
1644 return Err(PlannerError::invalid_expression(
1645 "WHERE cannot contain window functions".to_string(),
1646 ));
1647 }
1648 let predicate = self.infer_expr_with_scope(selection, &expr_scope, &ctes)?;
1649
1650 if predicate.resolved_type != ResolvedType::Boolean {
1652 return Err(PlannerError::type_mismatch(
1653 "Boolean",
1654 predicate.resolved_type.to_string(),
1655 selection.span,
1656 ));
1657 }
1658
1659 plan = LogicalPlan::Filter {
1660 input: Box::new(plan),
1661 predicate,
1662 };
1663 }
1664
1665 if has_window && (has_group_by || has_aggregate || stmt.having.is_some()) {
1666 return self.plan_grouped_window_select(
1667 stmt,
1668 &ctes,
1669 &expr_scope,
1670 &projection_aliases,
1671 plan,
1672 );
1673 }
1674
1675 if has_window {
1676 let mut windows = Vec::new();
1677 let mut window_map = HashMap::new();
1678 if let Projection::Columns(columns) = &final_projection {
1679 for column in columns {
1680 self.collect_windows_from_typed_expr(
1681 &column.expr,
1682 &mut windows,
1683 &mut window_map,
1684 )?;
1685 }
1686 }
1687
1688 let mut outer_order_by = Vec::new();
1689 for order_expr in &stmt.order_by {
1690 let sort_source =
1691 substitute_projection_aliases(&order_expr.expr, &projection_aliases);
1692 let typed = self.infer_expr_with_scope(&sort_source, &expr_scope, &ctes)?;
1693 self.collect_windows_from_typed_expr(&typed, &mut windows, &mut window_map)?;
1694 outer_order_by.push(SortExpr::new(
1695 typed,
1696 order_expr.asc.unwrap_or(true),
1697 order_expr.nulls_first.unwrap_or(false),
1698 ));
1699 }
1700
1701 let window_names = (0..windows.len())
1702 .map(|idx| format!("__window_{idx}"))
1703 .collect::<Vec<_>>();
1704 let mut window_schema = base_schema;
1705 window_schema.extend(windows.iter().enumerate().map(|(idx, window)| {
1706 ColumnMetadata::new(window_names[idx].clone(), window.result_type.clone())
1707 }));
1708
1709 let projection = rewrite_projection_for_windows(
1710 &final_projection,
1711 &window_map,
1712 relation.schema.len(),
1713 &window_names,
1714 )?;
1715 let order_by = outer_order_by
1716 .into_iter()
1717 .map(|sort| {
1718 Ok(SortExpr::new(
1719 rewrite_expr_for_windows(
1720 &sort.expr,
1721 &window_map,
1722 relation.schema.len(),
1723 &window_names,
1724 )?,
1725 sort.asc,
1726 sort.nulls_first,
1727 ))
1728 })
1729 .collect::<Result<Vec<_>, PlannerError>>()?;
1730
1731 return self.finish_window_select(
1732 stmt,
1733 plan,
1734 windows,
1735 projection,
1736 order_by,
1737 window_schema,
1738 );
1739 }
1740
1741 if has_group_by || has_aggregate || stmt.having.is_some() || stmt.distinct {
1742 if !has_group_by && !has_aggregate && stmt.having.is_some() {
1743 return Err(PlannerError::invalid_expression(
1744 "HAVING requires GROUP BY or aggregate functions".to_string(),
1745 ));
1746 }
1747
1748 let (group_keys, projected) = if distinct_only {
1749 let projected = self.build_projected_columns_for_distinct_with_scope(
1750 &stmt.projection,
1751 &relation.schema,
1752 &expr_scope,
1753 &ctes,
1754 )?;
1755 let group_keys = projected.iter().map(|col| col.expr.clone()).collect();
1756 (group_keys, projected)
1757 } else {
1758 let group_keys = self.build_group_keys_with_scope(stmt, &expr_scope, &ctes)?;
1759 let projected = self.build_projected_columns_for_aggregate_with_scope(
1760 &stmt.projection,
1761 &expr_scope,
1762 &ctes,
1763 )?;
1764 (group_keys, projected)
1765 };
1766 let mut aggregates = Vec::new();
1767 let mut agg_map = HashMap::new();
1768
1769 for col in &projected {
1770 self.collect_aggregates_from_typed_expr(&col.expr, &mut aggregates, &mut agg_map)?;
1771 }
1772
1773 let having_typed = if let Some(having) = &stmt.having {
1774 let having = substitute_projection_aliases(having, &projection_aliases);
1775 let typed = self.infer_expr_with_scope(&having, &expr_scope, &ctes)?;
1776 if typed.resolved_type != ResolvedType::Boolean {
1777 return Err(PlannerError::type_mismatch(
1778 "Boolean",
1779 typed.resolved_type.type_name().to_string(),
1780 typed.span,
1781 ));
1782 }
1783 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut agg_map)?;
1784 Some(typed)
1785 } else {
1786 None
1787 };
1788
1789 let mut order_by = Vec::new();
1790 if !stmt.order_by.is_empty() {
1791 for order_expr in &stmt.order_by {
1792 let sort_source =
1793 substitute_projection_aliases(&order_expr.expr, &projection_aliases);
1794 let typed = self.infer_expr_with_scope(&sort_source, &expr_scope, &ctes)?;
1795 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut agg_map)?;
1796 let asc = order_expr.asc.unwrap_or(true);
1797 let nulls_first = order_expr.nulls_first.unwrap_or(false);
1798 order_by.push(SortExpr::new(typed, asc, nulls_first));
1799 }
1800 }
1801
1802 if let Some(ref having) = having_typed {
1803 self.type_checker
1804 .validate_having_expr(having, &group_keys, &aggregates)?;
1805 }
1806
1807 let output_schema = build_aggregate_schema(&group_keys, &aggregates);
1808 let output_names: Vec<String> = output_schema.iter().map(|c| c.name.clone()).collect();
1809
1810 let projection = self.build_aggregate_projection(
1811 projected,
1812 &group_keys,
1813 &aggregates,
1814 &output_names,
1815 )?;
1816
1817 let having = if let Some(having) = having_typed {
1818 Some(self.rewrite_expr_for_aggregate(
1819 &having,
1820 &group_keys,
1821 &aggregates,
1822 &output_names,
1823 )?)
1824 } else {
1825 None
1826 };
1827
1828 let order_by = order_by
1829 .into_iter()
1830 .map(|expr| {
1831 let rewritten = self.rewrite_expr_for_aggregate(
1832 &expr.expr,
1833 &group_keys,
1834 &aggregates,
1835 &output_names,
1836 )?;
1837 Ok(SortExpr::new(rewritten, expr.asc, expr.nulls_first))
1838 })
1839 .collect::<Result<Vec<_>, PlannerError>>()?;
1840
1841 let schema = projection_schema(&projection, &output_schema);
1842 plan = LogicalPlan::Aggregate {
1843 input: Box::new(plan),
1844 group_keys,
1845 aggregates,
1846 having,
1847 projection,
1848 };
1849
1850 if !order_by.is_empty() {
1851 plan = LogicalPlan::Sort {
1852 input: Box::new(plan),
1853 order_by,
1854 };
1855 }
1856
1857 if stmt.limit.is_some() || stmt.offset.is_some() {
1858 let limit = self.extract_limit_value(&stmt.limit, stmt.span)?;
1859 let offset = self.extract_limit_value(&stmt.offset, stmt.span)?;
1860 plan = LogicalPlan::Limit {
1861 input: Box::new(plan),
1862 limit,
1863 offset,
1864 };
1865 }
1866
1867 return Ok(PlannedRelation {
1868 plan,
1869 schema: schema.clone(),
1870 scope: vec![ScopedTable::new(
1871 TableMetadata::new(LITERAL_TABLE, schema),
1872 0,
1873 )],
1874 });
1875 }
1876
1877 if !stmt.order_by.is_empty() {
1879 let order_by = self.build_sort_exprs_with_scope(
1880 &stmt.order_by,
1881 &expr_scope,
1882 &projection_aliases,
1883 &ctes,
1884 )?;
1885 plan = LogicalPlan::Sort {
1886 input: Box::new(plan),
1887 order_by,
1888 };
1889 }
1890
1891 if stmt.limit.is_some() || stmt.offset.is_some() {
1892 let limit = self.extract_limit_value(&stmt.limit, stmt.span)?;
1893 let offset = self.extract_limit_value(&stmt.offset, stmt.span)?;
1894 plan = LogicalPlan::Limit {
1895 input: Box::new(plan),
1896 limit,
1897 offset,
1898 };
1899 }
1900
1901 let output_schema = projection_schema(&final_projection, &relation.schema);
1902 if needs_project_boundary {
1903 plan = LogicalPlan::Project {
1904 input: Box::new(plan),
1905 projection: final_projection,
1906 };
1907 }
1908 Ok(PlannedRelation {
1909 plan,
1910 schema: output_schema.clone(),
1911 scope: vec![ScopedTable::new(
1912 TableMetadata::new(LITERAL_TABLE, output_schema),
1913 0,
1914 )],
1915 })
1916 }
1917
1918 fn plan_grouped_window_select(
1925 &self,
1926 stmt: &Select,
1927 ctes: &CtePlans,
1928 expr_scope: &[ScopedTable],
1929 projection_aliases: &HashMap<String, crate::ast::expr::Expr>,
1930 mut plan: LogicalPlan,
1931 ) -> Result<PlannedRelation, PlannerError> {
1932 let group_keys = self.build_group_keys_with_scope(stmt, expr_scope, ctes)?;
1933 let projected = self.build_projected_columns_for_aggregate_with_scope(
1934 &stmt.projection,
1935 expr_scope,
1936 ctes,
1937 )?;
1938 let mut aggregates = Vec::new();
1939 let mut aggregate_map = HashMap::new();
1940 for column in &projected {
1941 self.collect_aggregates_from_typed_expr(
1942 &column.expr,
1943 &mut aggregates,
1944 &mut aggregate_map,
1945 )?;
1946 }
1947
1948 let having_typed = if let Some(having) = &stmt.having {
1949 let having = substitute_projection_aliases(having, projection_aliases);
1950 if expr_contains_window(&having) {
1951 return Err(PlannerError::invalid_expression(
1952 "HAVING cannot contain window functions".to_string(),
1953 ));
1954 }
1955 let typed = self.infer_expr_with_scope(&having, expr_scope, ctes)?;
1956 if typed.resolved_type != ResolvedType::Boolean {
1957 return Err(PlannerError::type_mismatch(
1958 "Boolean",
1959 typed.resolved_type.type_name().to_string(),
1960 typed.span,
1961 ));
1962 }
1963 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut aggregate_map)?;
1964 Some(typed)
1965 } else {
1966 None
1967 };
1968
1969 let mut outer_order_by = Vec::new();
1970 for order_expr in &stmt.order_by {
1971 let source = substitute_projection_aliases(&order_expr.expr, projection_aliases);
1972 let typed = self.infer_expr_with_scope(&source, expr_scope, ctes)?;
1973 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut aggregate_map)?;
1974 outer_order_by.push(SortExpr::new(
1975 typed,
1976 order_expr.asc.unwrap_or(true),
1977 order_expr.nulls_first.unwrap_or(false),
1978 ));
1979 }
1980
1981 if let Some(having) = &having_typed {
1982 self.type_checker
1983 .validate_having_expr(having, &group_keys, &aggregates)?;
1984 }
1985
1986 let aggregate_schema = build_aggregate_schema(&group_keys, &aggregates);
1987 let aggregate_names = aggregate_schema
1988 .iter()
1989 .map(|column| column.name.clone())
1990 .collect::<Vec<_>>();
1991 let projection =
1992 self.build_aggregate_projection(projected, &group_keys, &aggregates, &aggregate_names)?;
1993 let having = having_typed
1994 .as_ref()
1995 .map(|expr| {
1996 self.rewrite_expr_for_aggregate(expr, &group_keys, &aggregates, &aggregate_names)
1997 })
1998 .transpose()?;
1999 let outer_order_by = outer_order_by
2000 .into_iter()
2001 .map(|sort| {
2002 Ok(SortExpr::new(
2003 self.rewrite_expr_for_aggregate(
2004 &sort.expr,
2005 &group_keys,
2006 &aggregates,
2007 &aggregate_names,
2008 )?,
2009 sort.asc,
2010 sort.nulls_first,
2011 ))
2012 })
2013 .collect::<Result<Vec<_>, PlannerError>>()?;
2014
2015 plan = LogicalPlan::Aggregate {
2016 input: Box::new(plan),
2017 group_keys,
2018 aggregates,
2019 having,
2020 projection: Projection::All(aggregate_names),
2021 };
2022
2023 let mut windows = Vec::new();
2024 let mut window_map = HashMap::new();
2025 if let Projection::Columns(columns) = &projection {
2026 for column in columns {
2027 self.collect_windows_from_typed_expr(&column.expr, &mut windows, &mut window_map)?;
2028 }
2029 }
2030 for sort in &outer_order_by {
2031 self.collect_windows_from_typed_expr(&sort.expr, &mut windows, &mut window_map)?;
2032 }
2033
2034 let window_names = (0..windows.len())
2035 .map(|index| format!("__window_{index}"))
2036 .collect::<Vec<_>>();
2037 let mut window_schema = aggregate_schema;
2038 window_schema.extend(windows.iter().enumerate().map(|(index, window)| {
2039 ColumnMetadata::new(window_names[index].clone(), window.result_type.clone())
2040 }));
2041 let projection = rewrite_projection_for_windows(
2042 &projection,
2043 &window_map,
2044 window_schema.len() - windows.len(),
2045 &window_names,
2046 )?;
2047 let outer_order_by = outer_order_by
2048 .into_iter()
2049 .map(|sort| {
2050 Ok(SortExpr::new(
2051 rewrite_expr_for_windows(
2052 &sort.expr,
2053 &window_map,
2054 window_schema.len() - windows.len(),
2055 &window_names,
2056 )?,
2057 sort.asc,
2058 sort.nulls_first,
2059 ))
2060 })
2061 .collect::<Result<Vec<_>, PlannerError>>()?;
2062
2063 self.finish_window_select(
2064 stmt,
2065 plan,
2066 windows,
2067 projection,
2068 outer_order_by,
2069 window_schema,
2070 )
2071 }
2072
2073 fn finish_window_select(
2075 &self,
2076 stmt: &Select,
2077 mut plan: LogicalPlan,
2078 windows: Vec<WindowExpr>,
2079 projection: Projection,
2080 order_by: Vec<SortExpr>,
2081 window_schema: Vec<ColumnMetadata>,
2082 ) -> Result<PlannedRelation, PlannerError> {
2083 plan = LogicalPlan::Window {
2084 input: Box::new(plan),
2085 windows,
2086 };
2087
2088 let output_schema = projection_schema(&projection, &window_schema);
2089 let mut hidden_order_keys = Vec::new();
2090 let mut projected_order_by = Vec::with_capacity(order_by.len());
2091 for sort in order_by {
2092 let expr =
2093 match rewrite_expr_for_projected_output(&sort.expr, &projection, &output_schema) {
2094 Ok(expr) => expr,
2095 Err(_) if !stmt.distinct => {
2096 let hidden_index = hidden_order_keys.len();
2097 let name = format!("__order_{hidden_index}");
2098 let output_index = output_schema.len() + hidden_index;
2099 hidden_order_keys.push(ProjectedColumn {
2100 expr: sort.expr.clone(),
2101 alias: Some(name.clone()),
2102 });
2103 TypedExpr::column_ref(
2104 "__project__".to_string(),
2105 name,
2106 output_index,
2107 sort.expr.resolved_type.clone(),
2108 sort.expr.span,
2109 )
2110 }
2111 Err(error) => return Err(error),
2112 };
2113 projected_order_by.push(SortExpr::new(expr, sort.asc, sort.nulls_first));
2114 }
2115 let has_hidden_order_keys = !hidden_order_keys.is_empty();
2116 let projection = if has_hidden_order_keys {
2117 let mut columns = match projection {
2118 Projection::Columns(columns) => columns,
2119 Projection::All(_) => output_schema
2120 .iter()
2121 .enumerate()
2122 .map(|(index, column)| ProjectedColumn {
2123 expr: TypedExpr::column_ref(
2124 "__window__".to_string(),
2125 column.name.clone(),
2126 index,
2127 column.data_type.clone(),
2128 stmt.span,
2129 ),
2130 alias: None,
2131 })
2132 .collect(),
2133 };
2134 columns.extend(hidden_order_keys);
2135 Projection::Columns(columns)
2136 } else {
2137 projection
2138 };
2139 plan = LogicalPlan::Project {
2140 input: Box::new(plan),
2141 projection,
2142 };
2143
2144 if stmt.distinct {
2145 let group_keys = output_schema
2146 .iter()
2147 .enumerate()
2148 .map(|(index, column)| {
2149 TypedExpr::column_ref(
2150 LITERAL_TABLE.to_string(),
2151 column.name.clone(),
2152 index,
2153 column.data_type.clone(),
2154 stmt.span,
2155 )
2156 })
2157 .collect();
2158 plan = LogicalPlan::Aggregate {
2159 input: Box::new(plan),
2160 group_keys,
2161 aggregates: Vec::new(),
2162 having: None,
2163 projection: Projection::All(
2164 output_schema
2165 .iter()
2166 .map(|column| column.name.clone())
2167 .collect(),
2168 ),
2169 };
2170 }
2171 if !projected_order_by.is_empty() {
2172 plan = LogicalPlan::Sort {
2173 input: Box::new(plan),
2174 order_by: projected_order_by,
2175 };
2176 }
2177 if stmt.limit.is_some() || stmt.offset.is_some() {
2178 plan = LogicalPlan::Limit {
2179 input: Box::new(plan),
2180 limit: self.extract_limit_value(&stmt.limit, stmt.span)?,
2181 offset: self.extract_limit_value(&stmt.offset, stmt.span)?,
2182 };
2183 }
2184 if has_hidden_order_keys {
2185 plan = LogicalPlan::Project {
2186 input: Box::new(plan),
2187 projection: Projection::Columns(
2188 output_schema
2189 .iter()
2190 .enumerate()
2191 .map(|(index, column)| ProjectedColumn {
2192 expr: TypedExpr::column_ref(
2193 "__ordered__".to_string(),
2194 column.name.clone(),
2195 index,
2196 column.data_type.clone(),
2197 stmt.span,
2198 ),
2199 alias: Some(column.name.clone()),
2200 })
2201 .collect(),
2202 ),
2203 };
2204 }
2205
2206 Ok(PlannedRelation {
2207 plan,
2208 schema: output_schema.clone(),
2209 scope: vec![ScopedTable::new(
2210 TableMetadata::new(LITERAL_TABLE, output_schema),
2211 0,
2212 )],
2213 })
2214 }
2215
2216 fn plan_from_items(
2220 &self,
2221 items: &[FromItem],
2222 select_span: crate::ast::Span,
2223 outer_scope: &[ScopedTable],
2224 ctes: &CtePlans,
2225 ) -> Result<PlannedRelation, PlannerError> {
2226 match items {
2227 [] => {
2228 let schema = Vec::new();
2229 Ok(PlannedRelation {
2230 plan: LogicalPlan::Scan {
2231 table: LITERAL_TABLE.to_string(),
2232 projection: Projection::All(Vec::new()),
2233 },
2234 schema: schema.clone(),
2235 scope: vec![ScopedTable::new(
2236 TableMetadata::new(LITERAL_TABLE, schema),
2237 0,
2238 )],
2239 })
2240 }
2241 [single] => self.plan_from_item(single, 0, outer_scope, ctes),
2242 [first, rest @ ..] => {
2243 let mut relation = self.plan_from_item(first, 0, outer_scope, ctes)?;
2244 for item in rest {
2245 let right =
2246 self.plan_from_item(item, relation.schema.len(), outer_scope, ctes)?;
2247 relation = self.combine_join_relation(
2248 relation,
2249 right,
2250 JoinType::Cross,
2251 None,
2252 None,
2253 select_span,
2254 )?;
2255 }
2256 Ok(relation)
2257 }
2258 }
2259 }
2260
2261 fn plan_from_item(
2262 &self,
2263 item: &FromItem,
2264 start_index: usize,
2265 outer_scope: &[ScopedTable],
2266 ctes: &CtePlans,
2267 ) -> Result<PlannedRelation, PlannerError> {
2268 match item {
2269 FromItem::Table { name, alias, span } => {
2270 if let Some(cte) = ctes.get(name) {
2271 let mut relation = cte.clone();
2272 relation.plan = LogicalPlan::Project {
2273 input: Box::new(relation.plan),
2274 projection: Projection::All(
2275 relation.schema.iter().map(|col| col.name.clone()).collect(),
2276 ),
2277 };
2278 relation.scope = vec![ScopedTable::new(
2279 TableMetadata::new(
2280 alias.clone().unwrap_or_else(|| name.clone()),
2281 relation.schema.clone(),
2282 ),
2283 start_index,
2284 )];
2285 return Ok(relation);
2286 }
2287 let table = self.name_resolver.resolve_table(name, *span)?.clone();
2288 let mut scope_table = table.clone();
2289 if let Some(alias) = alias {
2290 scope_table.name = alias.clone();
2291 }
2292 let schema = table.columns.clone();
2293 Ok(PlannedRelation {
2294 plan: LogicalPlan::Scan {
2295 table: name.clone(),
2296 projection: Projection::All(
2297 schema.iter().map(|col| col.name.clone()).collect(),
2298 ),
2299 },
2300 schema,
2301 scope: vec![ScopedTable::new(scope_table, start_index)],
2302 })
2303 }
2304 FromItem::Join {
2305 left,
2306 right,
2307 join_type,
2308 condition,
2309 using,
2310 natural,
2311 span,
2312 } => {
2313 let left_relation = self.plan_from_item(left, start_index, outer_scope, ctes)?;
2314 let right_relation = self.plan_from_item(
2315 right,
2316 start_index + left_relation.schema.len(),
2317 outer_scope,
2318 ctes,
2319 )?;
2320 let expr_scope = left_relation
2321 .scope
2322 .iter()
2323 .cloned()
2324 .chain(right_relation.scope.iter().cloned())
2325 .chain(offset_scope(
2326 outer_scope,
2327 left_relation.schema.len() + right_relation.schema.len(),
2328 ))
2329 .collect::<Vec<_>>();
2330 let using = if *natural {
2331 Some(natural_join_columns(
2332 &left_relation.schema,
2333 &right_relation.schema,
2334 ))
2335 } else {
2336 using.clone()
2337 };
2338 let typed_condition = if let Some(expr) = condition {
2339 let typed = self.infer_expr_with_scope(expr, &expr_scope, ctes)?;
2340 if typed.resolved_type != ResolvedType::Boolean {
2341 return Err(PlannerError::type_mismatch(
2342 "Boolean",
2343 typed.resolved_type.to_string(),
2344 expr.span,
2345 ));
2346 }
2347 Some(typed)
2348 } else {
2349 self.build_using_condition(
2350 using.as_deref(),
2351 &left_relation,
2352 &right_relation,
2353 *span,
2354 )?
2355 };
2356 self.combine_join_relation(
2357 left_relation,
2358 right_relation,
2359 map_join_type(*join_type),
2360 typed_condition,
2361 using,
2362 *span,
2363 )
2364 }
2365 FromItem::Derived {
2366 subquery,
2367 alias,
2368 span,
2369 } => {
2370 let crate::ast::StatementKind::Select(select) = &subquery.kind else {
2371 return Err(PlannerError::unsupported_feature(
2372 "non-SELECT derived table",
2373 "v0.6.0-subquery Phase 6",
2374 *span,
2375 ));
2376 };
2377 let mut relation = self.plan_select_relation(select, &[], ctes)?;
2384 let alias = alias.clone().ok_or_else(|| {
2385 PlannerError::invalid_expression("derived table requires an alias".to_string())
2386 })?;
2387 relation.plan = LogicalPlan::Project {
2388 input: Box::new(relation.plan),
2389 projection: Projection::All(
2390 relation.schema.iter().map(|col| col.name.clone()).collect(),
2391 ),
2392 };
2393 relation.scope = vec![ScopedTable::new(
2394 TableMetadata::new(alias, relation.schema.clone()),
2395 start_index,
2396 )];
2397 Ok(relation)
2398 }
2399 }
2400 }
2401
2402 fn combine_join_relation(
2403 &self,
2404 left: PlannedRelation,
2405 right: PlannedRelation,
2406 join_type: JoinType,
2407 condition: Option<TypedExpr>,
2408 using: Option<Vec<String>>,
2409 _span: crate::ast::Span,
2410 ) -> Result<PlannedRelation, PlannerError> {
2411 let mut schema = left.schema.clone();
2412 schema.extend(right.schema.clone());
2413 let mut scope = left.scope.clone();
2414 let mut right_scope = right.scope.clone();
2415 if let Some(columns) = &using {
2416 for column in columns {
2420 let right_index = right_scope.iter().find_map(|table| {
2421 table
2422 .table
2423 .get_column_index(column)
2424 .map(|index| table.start_index + index)
2425 });
2426 let Some(right_index) = right_index else {
2427 continue;
2428 };
2429 for table in &mut scope {
2430 if table.table.get_column_index(column).is_some() {
2431 table.merge_column_with(column, right_index);
2432 }
2433 }
2434 }
2435 for table in &mut right_scope {
2436 table.hide_unqualified_columns(columns);
2437 }
2438 }
2439 scope.extend(right_scope);
2440 Ok(PlannedRelation {
2441 plan: LogicalPlan::Join {
2442 left: Box::new(left.plan),
2443 right: Box::new(right.plan),
2444 join_type,
2445 condition,
2446 using,
2447 },
2448 schema,
2449 scope,
2450 })
2451 }
2452
2453 fn build_using_condition(
2454 &self,
2455 using: Option<&[String]>,
2456 left: &PlannedRelation,
2457 right: &PlannedRelation,
2458 span: crate::ast::Span,
2459 ) -> Result<Option<TypedExpr>, PlannerError> {
2460 let Some(columns) = using else {
2461 return Ok(None);
2462 };
2463 let mut condition = None;
2464 for column in columns {
2465 let left_col = find_scoped_column(&left.scope, column, span)?;
2466 let right_col = find_scoped_column(&right.scope, column, span)?;
2467 let left_expr = merged_scoped_column_expr(&left_col, column, span);
2468 let right_expr = merged_scoped_column_expr(&right_col, column, span);
2469 self.type_checker
2470 .check_comparison_op(&left_col.ty, &right_col.ty, span)?;
2471 let eq = TypedExpr::binary_op(
2472 left_expr,
2473 crate::ast::expr::BinaryOp::Eq,
2474 right_expr,
2475 ResolvedType::Boolean,
2476 span,
2477 );
2478 condition = Some(match condition {
2479 Some(prev) => TypedExpr::binary_op(
2480 prev,
2481 crate::ast::expr::BinaryOp::And,
2482 eq,
2483 ResolvedType::Boolean,
2484 span,
2485 ),
2486 None => eq,
2487 });
2488 }
2489 Ok(condition)
2490 }
2491
2492 fn infer_expr_with_scope(
2493 &self,
2494 expr: &crate::ast::expr::Expr,
2495 scope: &[ScopedTable],
2496 ctes: &CtePlans,
2497 ) -> Result<TypedExpr, PlannerError> {
2498 self.type_checker
2499 .infer_type_with_scope(expr, scope, &|stmt, outer_scope| {
2500 let crate::ast::StatementKind::Select(select) = &stmt.kind else {
2501 return Err(PlannerError::unsupported_feature(
2502 "non-SELECT subquery",
2503 "v0.6.0-subquery Phase 6",
2504 stmt.span(),
2505 ));
2506 };
2507 let relation = self.plan_select_relation(select, outer_scope, ctes)?;
2508 Ok((relation.plan, relation.schema))
2509 })
2510 }
2511
2512 #[allow(dead_code)]
2513 fn build_projection(
2514 &self,
2515 items: &[SelectItem],
2516 table: &TableMetadata,
2517 ) -> Result<Projection, PlannerError> {
2518 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
2520 let columns = self.name_resolver.expand_wildcard(table);
2521 return Ok(Projection::All(columns));
2522 }
2523
2524 let mut projected_columns = Vec::new();
2526 for item in items {
2527 match item {
2528 SelectItem::Wildcard { span } => {
2529 for col in &table.columns {
2531 let column_index = table.get_column_index(&col.name).unwrap();
2532 let typed_expr = TypedExpr::column_ref(
2533 table.name.clone(),
2534 col.name.clone(),
2535 column_index,
2536 col.data_type.clone(),
2537 *span,
2538 );
2539 projected_columns.push(ProjectedColumn::new(typed_expr));
2540 }
2541 }
2542 SelectItem::QualifiedWildcard {
2543 table: qualifier,
2544 span,
2545 } => {
2546 if qualifier != &table.name {
2547 return Err(PlannerError::invalid_expression(format!(
2548 "table '{qualifier}' is not available for wildcard projection"
2549 )));
2550 }
2551 for col in &table.columns {
2552 let column_index = table.get_column_index(&col.name).unwrap();
2553 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
2554 table.name.clone(),
2555 col.name.clone(),
2556 column_index,
2557 col.data_type.clone(),
2558 *span,
2559 )));
2560 }
2561 }
2562 SelectItem::Expr { expr, alias, .. } => {
2563 let typed_expr = self.type_checker.infer_type(expr, table)?;
2564 let projected = if let Some(alias) = alias {
2565 ProjectedColumn::with_alias(typed_expr, alias.clone())
2566 } else {
2567 ProjectedColumn::new(typed_expr)
2568 };
2569 projected_columns.push(projected);
2570 }
2571 }
2572 }
2573
2574 Ok(Projection::Columns(projected_columns))
2575 }
2576
2577 fn build_projection_with_scope(
2578 &self,
2579 items: &[SelectItem],
2580 schema: &[ColumnMetadata],
2581 scope: &[ScopedTable],
2582 ctes: &CtePlans,
2583 ) -> Result<Projection, PlannerError> {
2584 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
2585 return Ok(Projection::All(visible_wildcard_columns(schema, scope)));
2586 }
2587
2588 let mut projected_columns = Vec::new();
2589 for item in items {
2590 match item {
2591 SelectItem::Wildcard { span } => {
2592 for scoped in scope {
2593 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
2594 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
2595 scoped.table.name.clone(),
2596 col.name.clone(),
2597 scoped.start_index + local_idx,
2598 col.data_type.clone(),
2599 *span,
2600 )));
2601 }
2602 }
2603 }
2604 SelectItem::QualifiedWildcard { table, span } => {
2605 let scoped = scope
2606 .iter()
2607 .filter(|scoped| scoped.table.name == *table)
2608 .collect::<Vec<_>>();
2609 match scoped.as_slice() {
2610 [] => {
2611 return Err(PlannerError::invalid_expression(format!(
2612 "table '{table}' is not available for wildcard projection"
2613 )));
2614 }
2615 [scoped] => {
2616 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
2617 projected_columns.push(ProjectedColumn::new(
2618 TypedExpr::column_ref(
2619 scoped.table.name.clone(),
2620 col.name.clone(),
2621 scoped.start_index + local_idx,
2622 col.data_type.clone(),
2623 *span,
2624 ),
2625 ));
2626 }
2627 }
2628 _ => {
2629 return Err(PlannerError::ambiguous_column(
2630 table,
2631 scoped
2632 .iter()
2633 .map(|scoped| scoped.table.name.clone())
2634 .collect(),
2635 *span,
2636 ));
2637 }
2638 }
2639 }
2640 SelectItem::Expr { expr, alias, .. } => {
2641 let typed_expr = self.infer_expr_with_scope(expr, scope, ctes)?;
2642 let projected = if let Some(alias) = alias {
2643 ProjectedColumn::with_alias(typed_expr, alias.clone())
2644 } else {
2645 ProjectedColumn::new(typed_expr)
2646 };
2647 projected_columns.push(projected);
2648 }
2649 }
2650 }
2651
2652 Ok(Projection::Columns(projected_columns))
2653 }
2654
2655 #[allow(dead_code)]
2657 fn build_sort_exprs(
2658 &self,
2659 order_by: &[OrderByExpr],
2660 table: &TableMetadata,
2661 ) -> Result<Vec<SortExpr>, PlannerError> {
2662 let mut sort_exprs = Vec::new();
2663
2664 for order_expr in order_by {
2665 let typed_expr = self.type_checker.infer_type(&order_expr.expr, table)?;
2666
2667 let asc = order_expr.asc.unwrap_or(true);
2669
2670 let nulls_first = order_expr.nulls_first.unwrap_or(false);
2672
2673 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
2674 }
2675
2676 Ok(sort_exprs)
2677 }
2678
2679 fn build_sort_exprs_with_scope(
2680 &self,
2681 order_by: &[OrderByExpr],
2682 scope: &[ScopedTable],
2683 projection_aliases: &HashMap<String, crate::ast::expr::Expr>,
2684 ctes: &CtePlans,
2685 ) -> Result<Vec<SortExpr>, PlannerError> {
2686 let mut sort_exprs = Vec::new();
2687 for order_expr in order_by {
2688 let sort_source = substitute_projection_aliases(&order_expr.expr, projection_aliases);
2689 let typed_expr = self.infer_expr_with_scope(&sort_source, scope, ctes)?;
2690 let asc = order_expr.asc.unwrap_or(true);
2691 let nulls_first = order_expr.nulls_first.unwrap_or(false);
2692 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
2693 }
2694 Ok(sort_exprs)
2695 }
2696
2697 fn select_contains_aggregate(&self, stmt: &Select) -> bool {
2698 stmt.projection.iter().any(|item| match item {
2699 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
2700 SelectItem::Expr { expr, .. } => expr_contains_aggregate(expr),
2701 }) || stmt
2702 .group_by
2703 .as_ref()
2704 .map(|items| items.iter().any(expr_contains_aggregate))
2705 .unwrap_or(false)
2706 || stmt
2707 .having
2708 .as_ref()
2709 .map(expr_contains_aggregate)
2710 .unwrap_or(false)
2711 || stmt
2712 .order_by
2713 .iter()
2714 .any(|order| expr_contains_aggregate(&order.expr))
2715 }
2716
2717 #[allow(dead_code)]
2718 fn build_group_keys(
2719 &self,
2720 stmt: &Select,
2721 table: &TableMetadata,
2722 ) -> Result<Vec<TypedExpr>, PlannerError> {
2723 let mut keys = Vec::new();
2724 if let Some(items) = &stmt.group_by {
2725 for expr in items {
2726 let typed = self.type_checker.infer_type(expr, table)?;
2727 if typed_expr_contains_aggregate(&typed) {
2728 return Err(PlannerError::invalid_expression(
2729 "GROUP BY cannot contain aggregate functions".to_string(),
2730 ));
2731 }
2732 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
2733 return Err(PlannerError::invalid_expression(
2734 "GROUP BY expressions must be column references".to_string(),
2735 ));
2736 }
2737 keys.push(typed);
2738 }
2739 }
2740 Ok(keys)
2741 }
2742
2743 fn build_group_keys_with_scope(
2744 &self,
2745 stmt: &Select,
2746 scope: &[ScopedTable],
2747 ctes: &CtePlans,
2748 ) -> Result<Vec<TypedExpr>, PlannerError> {
2749 let mut keys = Vec::new();
2750 if let Some(items) = &stmt.group_by {
2751 for expr in items {
2752 let typed = self.infer_expr_with_scope(expr, scope, ctes)?;
2753 if typed_expr_contains_aggregate(&typed) {
2754 return Err(PlannerError::invalid_expression(
2755 "GROUP BY cannot contain aggregate functions".to_string(),
2756 ));
2757 }
2758 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
2759 return Err(PlannerError::invalid_expression(
2760 "GROUP BY expressions must be column references".to_string(),
2761 ));
2762 }
2763 keys.push(typed);
2764 }
2765 }
2766 Ok(keys)
2767 }
2768
2769 #[allow(dead_code)]
2770 fn build_projected_columns_for_aggregate(
2771 &self,
2772 items: &[SelectItem],
2773 table: &TableMetadata,
2774 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
2775 let mut projected = Vec::new();
2776 for item in items {
2777 match item {
2778 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
2779 return Err(PlannerError::invalid_expression(
2780 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
2781 ));
2782 }
2783 SelectItem::Expr { expr, alias, .. } => {
2784 let typed = self.type_checker.infer_type(expr, table)?;
2785 projected.push(ProjectedColumn {
2786 expr: typed,
2787 alias: alias.clone(),
2788 });
2789 }
2790 }
2791 }
2792 Ok(projected)
2793 }
2794
2795 fn build_projected_columns_for_aggregate_with_scope(
2796 &self,
2797 items: &[SelectItem],
2798 scope: &[ScopedTable],
2799 ctes: &CtePlans,
2800 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
2801 let mut projected = Vec::new();
2802 for item in items {
2803 match item {
2804 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
2805 return Err(PlannerError::invalid_expression(
2806 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
2807 ));
2808 }
2809 SelectItem::Expr { expr, alias, .. } => {
2810 let typed = self.infer_expr_with_scope(expr, scope, ctes)?;
2811 projected.push(ProjectedColumn {
2812 expr: typed,
2813 alias: alias.clone(),
2814 });
2815 }
2816 }
2817 }
2818 Ok(projected)
2819 }
2820
2821 #[allow(dead_code)]
2822 fn build_projected_columns_for_distinct(
2823 &self,
2824 items: &[SelectItem],
2825 table: &TableMetadata,
2826 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
2827 let projection = self.build_projection(items, table)?;
2828 match projection {
2829 Projection::All(columns) => {
2830 let mut projected = Vec::with_capacity(columns.len());
2831 for column in columns {
2832 let column_index = table.get_column_index(&column).ok_or_else(|| {
2833 PlannerError::invalid_expression(format!(
2834 "column '{column}' not found for DISTINCT projection"
2835 ))
2836 })?;
2837 let column_meta = table.get_column(&column).ok_or_else(|| {
2838 PlannerError::invalid_expression(format!(
2839 "column '{column}' not found for DISTINCT projection"
2840 ))
2841 })?;
2842 let typed_expr = TypedExpr::column_ref(
2843 table.name.clone(),
2844 column.clone(),
2845 column_index,
2846 column_meta.data_type.clone(),
2847 crate::ast::Span::default(),
2848 );
2849 projected.push(ProjectedColumn::new(typed_expr));
2850 }
2851 Ok(projected)
2852 }
2853 Projection::Columns(columns) => Ok(columns),
2854 }
2855 }
2856
2857 fn build_projected_columns_for_distinct_with_scope(
2858 &self,
2859 items: &[SelectItem],
2860 schema: &[ColumnMetadata],
2861 scope: &[ScopedTable],
2862 ctes: &CtePlans,
2863 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
2864 let projection = self.build_projection_with_scope(items, schema, scope, ctes)?;
2865 match projection {
2866 Projection::All(columns) => {
2867 let mut projected = Vec::with_capacity(columns.len());
2868 for (idx, column) in columns.into_iter().enumerate() {
2869 let column_meta = schema.get(idx).ok_or_else(|| {
2870 PlannerError::invalid_expression(format!(
2871 "column '{column}' not found for DISTINCT projection"
2872 ))
2873 })?;
2874 projected.push(ProjectedColumn::new(TypedExpr::column_ref(
2875 LITERAL_TABLE.to_string(),
2876 column,
2877 idx,
2878 column_meta.data_type.clone(),
2879 crate::ast::Span::default(),
2880 )));
2881 }
2882 Ok(projected)
2883 }
2884 Projection::Columns(columns) => Ok(columns),
2885 }
2886 }
2887
2888 fn collect_aggregates_from_typed_expr(
2889 &self,
2890 expr: &TypedExpr,
2891 aggregates: &mut Vec<AggregateExpr>,
2892 aggregate_map: &mut HashMap<AggregateSignature, usize>,
2893 ) -> Result<(), PlannerError> {
2894 match &expr.kind {
2895 TypedExprKind::FunctionCall {
2896 name,
2897 args,
2898 distinct,
2899 star,
2900 over: None,
2901 } if is_aggregate_function(name) => {
2902 if args.iter().any(typed_expr_contains_window) {
2903 return Err(PlannerError::invalid_expression(
2904 "aggregate functions cannot contain window functions".to_string(),
2905 ));
2906 }
2907 for arg in args {
2908 if typed_expr_contains_aggregate(arg) {
2909 return Err(PlannerError::invalid_expression(
2910 "nested aggregate functions are not supported".to_string(),
2911 ));
2912 }
2913 }
2914 let (agg, signature) =
2915 self.build_aggregate_expr_from_typed(expr, name, args, *distinct, *star)?;
2916 aggregate_map.entry(signature).or_insert_with(|| {
2917 aggregates.push(agg);
2918 aggregates.len() - 1
2919 });
2920 Ok(())
2921 }
2922 TypedExprKind::BinaryOp { left, right, .. } => {
2923 self.collect_aggregates_from_typed_expr(left, aggregates, aggregate_map)?;
2924 self.collect_aggregates_from_typed_expr(right, aggregates, aggregate_map)?;
2925 Ok(())
2926 }
2927 TypedExprKind::UnaryOp { operand, .. } => {
2928 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)
2929 }
2930 TypedExprKind::Cast { expr, .. } => {
2931 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
2932 }
2933 TypedExprKind::Case {
2934 operand,
2935 branches,
2936 else_expr,
2937 } => {
2938 if let Some(operand) = operand {
2939 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)?;
2940 }
2941 for branch in branches {
2942 self.collect_aggregates_from_typed_expr(
2943 &branch.when,
2944 aggregates,
2945 aggregate_map,
2946 )?;
2947 self.collect_aggregates_from_typed_expr(
2948 &branch.then,
2949 aggregates,
2950 aggregate_map,
2951 )?;
2952 }
2953 if let Some(else_expr) = else_expr {
2954 self.collect_aggregates_from_typed_expr(else_expr, aggregates, aggregate_map)?;
2955 }
2956 Ok(())
2957 }
2958 TypedExprKind::FunctionCall { args, over, .. } => {
2959 for arg in args {
2960 self.collect_aggregates_from_typed_expr(arg, aggregates, aggregate_map)?;
2961 }
2962 if let Some(window) = over {
2963 for expr in &window.partition_by {
2964 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
2965 }
2966 for sort in &window.order_by {
2967 self.collect_aggregates_from_typed_expr(
2968 &sort.expr,
2969 aggregates,
2970 aggregate_map,
2971 )?;
2972 }
2973 }
2974 Ok(())
2975 }
2976 TypedExprKind::Between {
2977 expr, low, high, ..
2978 } => {
2979 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
2980 self.collect_aggregates_from_typed_expr(low, aggregates, aggregate_map)?;
2981 self.collect_aggregates_from_typed_expr(high, aggregates, aggregate_map)?;
2982 Ok(())
2983 }
2984 TypedExprKind::Like {
2985 expr,
2986 pattern,
2987 escape,
2988 ..
2989 } => {
2990 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
2991 self.collect_aggregates_from_typed_expr(pattern, aggregates, aggregate_map)?;
2992 if let Some(esc) = escape {
2993 self.collect_aggregates_from_typed_expr(esc, aggregates, aggregate_map)?;
2994 }
2995 Ok(())
2996 }
2997 TypedExprKind::InList { expr, list, .. } => {
2998 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
2999 for item in list {
3000 self.collect_aggregates_from_typed_expr(item, aggregates, aggregate_map)?;
3001 }
3002 Ok(())
3003 }
3004 TypedExprKind::IsNull { expr, .. } => {
3005 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
3006 }
3007 _ => Ok(()),
3008 }
3009 }
3010
3011 fn collect_windows_from_typed_expr(
3012 &self,
3013 expr: &TypedExpr,
3014 windows: &mut Vec<WindowExpr>,
3015 window_map: &mut HashMap<String, usize>,
3016 ) -> Result<(), PlannerError> {
3017 match &expr.kind {
3018 TypedExprKind::FunctionCall {
3019 name,
3020 args,
3021 distinct,
3022 star,
3023 over: Some(over),
3024 } => {
3025 if args.iter().any(typed_expr_contains_window)
3026 || over.partition_by.iter().any(typed_expr_contains_window)
3027 || over
3028 .order_by
3029 .iter()
3030 .any(|sort| typed_expr_contains_window(&sort.expr))
3031 {
3032 return Err(PlannerError::invalid_expression(
3033 "nested window functions are not supported".to_string(),
3034 ));
3035 }
3036
3037 let key = expr_key(expr);
3038 if window_map.contains_key(&key) {
3039 return Ok(());
3040 }
3041 let function = match name.to_ascii_lowercase().as_str() {
3042 "row_number" => WindowFunction::RowNumber,
3043 "rank" => WindowFunction::Rank,
3044 "dense_rank" => WindowFunction::DenseRank,
3045 "sum" | "count" | "avg" | "min" | "max" => {
3046 let (aggregate, _) = self
3047 .build_aggregate_expr_from_typed(expr, name, args, *distinct, *star)?;
3048 WindowFunction::Aggregate(aggregate)
3049 }
3050 "lag" | "lead" => {
3051 let positional = build_offset_window_function(name, args)?;
3052 if name.eq_ignore_ascii_case("lag") {
3053 WindowFunction::Lag(positional)
3054 } else {
3055 WindowFunction::Lead(positional)
3056 }
3057 }
3058 _ => {
3059 return Err(PlannerError::unsupported_feature(
3060 format!("function '{}' with OVER", name),
3061 "future",
3062 expr.span,
3063 ));
3064 }
3065 };
3066 let index = windows.len();
3067 windows.push(WindowExpr {
3068 function,
3069 partition_by: over.partition_by.clone(),
3070 order_by: over.order_by.clone(),
3071 frame: over.frame.clone(),
3072 result_type: expr.resolved_type.clone(),
3073 });
3074 window_map.insert(key, index);
3075 Ok(())
3076 }
3077 TypedExprKind::FunctionCall { args, .. } => {
3078 for arg in args {
3079 self.collect_windows_from_typed_expr(arg, windows, window_map)?;
3080 }
3081 Ok(())
3082 }
3083 TypedExprKind::BinaryOp { left, right, .. } => {
3084 self.collect_windows_from_typed_expr(left, windows, window_map)?;
3085 self.collect_windows_from_typed_expr(right, windows, window_map)
3086 }
3087 TypedExprKind::UnaryOp { operand, .. } => {
3088 self.collect_windows_from_typed_expr(operand, windows, window_map)
3089 }
3090 TypedExprKind::Cast { expr, .. } | TypedExprKind::IsNull { expr, .. } => {
3091 self.collect_windows_from_typed_expr(expr, windows, window_map)
3092 }
3093 TypedExprKind::Between {
3094 expr, low, high, ..
3095 } => {
3096 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
3097 self.collect_windows_from_typed_expr(low, windows, window_map)?;
3098 self.collect_windows_from_typed_expr(high, windows, window_map)
3099 }
3100 TypedExprKind::Like {
3101 expr,
3102 pattern,
3103 escape,
3104 ..
3105 } => {
3106 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
3107 self.collect_windows_from_typed_expr(pattern, windows, window_map)?;
3108 if let Some(escape) = escape {
3109 self.collect_windows_from_typed_expr(escape, windows, window_map)?;
3110 }
3111 Ok(())
3112 }
3113 TypedExprKind::InList { expr, list, .. } => {
3114 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
3115 for item in list {
3116 self.collect_windows_from_typed_expr(item, windows, window_map)?;
3117 }
3118 Ok(())
3119 }
3120 _ => Ok(()),
3121 }
3122 }
3123
3124 fn build_aggregate_expr_from_typed(
3125 &self,
3126 expr: &TypedExpr,
3127 name: &str,
3128 args: &[TypedExpr],
3129 distinct: bool,
3130 star: bool,
3131 ) -> Result<(AggregateExpr, AggregateSignature), PlannerError> {
3132 let lower = name.to_lowercase();
3133 match lower.as_str() {
3134 "count" => {
3135 if star {
3136 let agg = AggregateExpr::count_star();
3137 let signature = aggregate_signature(name, distinct, star, None, None, expr);
3138 return Ok((agg, signature));
3139 }
3140 if args.len() != 1 {
3141 return Err(PlannerError::type_mismatch(
3142 "1 argument",
3143 format!("{} arguments", args.len()),
3144 expr.span,
3145 ));
3146 }
3147 let agg = AggregateExpr {
3148 function: AggregateFunction::Count,
3149 arg: Some(args[0].clone()),
3150 distinct,
3151 result_type: ResolvedType::BigInt,
3152 };
3153 let signature =
3154 aggregate_signature(name, distinct, star, Some(&args[0]), None, expr);
3155 Ok((agg, signature))
3156 }
3157 "sum" => {
3158 let arg = self.require_single_aggregate_arg(args, expr.span)?;
3159 let agg = AggregateExpr {
3160 function: AggregateFunction::Sum,
3161 arg: Some(arg.clone()),
3162 distinct,
3163 result_type: crate::planner::aggregate_expr::sum_result_type(
3164 &arg.resolved_type,
3165 ),
3166 };
3167 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
3168 Ok((agg, signature))
3169 }
3170 "total" => {
3171 let arg = self.require_single_aggregate_arg(args, expr.span)?;
3172 let agg = AggregateExpr {
3173 function: AggregateFunction::Total,
3174 arg: Some(arg.clone()),
3175 distinct: false,
3176 result_type: ResolvedType::Double,
3177 };
3178 let signature = aggregate_signature(name, false, star, Some(arg), None, expr);
3179 Ok((agg, signature))
3180 }
3181 "avg" => {
3182 let arg = self.require_single_aggregate_arg(args, expr.span)?;
3183 let agg = AggregateExpr {
3184 function: AggregateFunction::Avg,
3185 arg: Some(arg.clone()),
3186 distinct,
3187 result_type: ResolvedType::Double,
3188 };
3189 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
3190 Ok((agg, signature))
3191 }
3192 "min" => {
3193 let arg = self.require_single_aggregate_arg(args, expr.span)?;
3194 let agg = AggregateExpr {
3195 function: AggregateFunction::Min,
3196 arg: Some(arg.clone()),
3197 distinct,
3198 result_type: arg.resolved_type.clone(),
3199 };
3200 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
3201 Ok((agg, signature))
3202 }
3203 "max" => {
3204 let arg = self.require_single_aggregate_arg(args, expr.span)?;
3205 let agg = AggregateExpr {
3206 function: AggregateFunction::Max,
3207 arg: Some(arg.clone()),
3208 distinct,
3209 result_type: arg.resolved_type.clone(),
3210 };
3211 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
3212 Ok((agg, signature))
3213 }
3214 "group_concat" => {
3215 if args.is_empty() || args.len() > 2 {
3216 return Err(PlannerError::type_mismatch(
3217 "1 or 2 arguments",
3218 format!("{} arguments", args.len()),
3219 expr.span,
3220 ));
3221 }
3222 let arg = &args[0];
3223 let mut separator = None;
3224 if args.len() == 2 {
3225 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
3226 separator = Some(value.clone());
3227 } else {
3228 return Err(PlannerError::invalid_expression(
3229 "GROUP_CONCAT separator must be a string literal".to_string(),
3230 ));
3231 }
3232 }
3233 let agg = AggregateExpr {
3234 function: AggregateFunction::GroupConcat { separator },
3235 arg: Some(arg.clone()),
3236 distinct,
3237 result_type: ResolvedType::Text,
3238 };
3239 let signature = aggregate_signature(
3240 name,
3241 distinct,
3242 star,
3243 Some(arg),
3244 match &agg.function {
3245 AggregateFunction::GroupConcat { separator } => separator.as_ref(),
3246 _ => None,
3247 },
3248 expr,
3249 );
3250 Ok((agg, signature))
3251 }
3252 "string_agg" => {
3253 if args.len() != 2 {
3254 return Err(PlannerError::type_mismatch(
3255 "2 arguments",
3256 format!("{} arguments", args.len()),
3257 expr.span,
3258 ));
3259 }
3260 let arg = &args[0];
3261 let separator =
3262 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
3263 Some(value.clone())
3264 } else {
3265 return Err(PlannerError::invalid_expression(
3266 "STRING_AGG separator must be a string literal".to_string(),
3267 ));
3268 };
3269 let agg = AggregateExpr {
3270 function: AggregateFunction::StringAgg { separator },
3271 arg: Some(arg.clone()),
3272 distinct,
3273 result_type: ResolvedType::Text,
3274 };
3275 let signature = aggregate_signature(
3276 name,
3277 distinct,
3278 star,
3279 Some(arg),
3280 match &agg.function {
3281 AggregateFunction::StringAgg { separator } => separator.as_ref(),
3282 _ => None,
3283 },
3284 expr,
3285 );
3286 Ok((agg, signature))
3287 }
3288 _ => Err(PlannerError::unsupported_feature(
3289 format!("function '{}'", name),
3290 "future",
3291 expr.span,
3292 )),
3293 }
3294 }
3295
3296 fn require_single_aggregate_arg<'b>(
3297 &self,
3298 args: &'b [TypedExpr],
3299 span: crate::ast::Span,
3300 ) -> Result<&'b TypedExpr, PlannerError> {
3301 if args.len() != 1 {
3302 return Err(PlannerError::type_mismatch(
3303 "1 argument",
3304 format!("{} arguments", args.len()),
3305 span,
3306 ));
3307 }
3308 Ok(&args[0])
3309 }
3310
3311 fn build_aggregate_projection(
3312 &self,
3313 projected: Vec<ProjectedColumn>,
3314 group_keys: &[TypedExpr],
3315 aggregates: &[AggregateExpr],
3316 output_names: &[String],
3317 ) -> Result<Projection, PlannerError> {
3318 let mut columns = Vec::new();
3319 for col in projected {
3320 let rewritten =
3321 self.rewrite_expr_for_aggregate(&col.expr, group_keys, aggregates, output_names)?;
3322 columns.push(ProjectedColumn {
3323 expr: rewritten,
3324 alias: col.alias,
3325 });
3326 }
3327 Ok(Projection::Columns(columns))
3328 }
3329
3330 fn rewrite_expr_for_aggregate(
3331 &self,
3332 expr: &TypedExpr,
3333 group_keys: &[TypedExpr],
3334 aggregates: &[AggregateExpr],
3335 output_names: &[String],
3336 ) -> Result<TypedExpr, PlannerError> {
3337 let group_key_map = build_group_key_map(group_keys);
3338 let aggregate_map = build_aggregate_map(aggregates);
3339
3340 rewrite_expr_with_maps(expr, &group_key_map, &aggregate_map, output_names)
3341 }
3342
3343 fn extract_limit_value(
3347 &self,
3348 expr: &Option<crate::ast::expr::Expr>,
3349 stmt_span: crate::ast::Span,
3350 ) -> Result<Option<u64>, PlannerError> {
3351 match expr {
3352 None => Ok(None),
3353 Some(e) => {
3354 if let crate::ast::expr::ExprKind::Literal {
3356 literal: Literal::Number(s),
3357 } = &e.kind
3358 {
3359 s.parse::<u64>().map(Some).map_err(|_| {
3360 PlannerError::type_mismatch("unsigned integer", s.clone(), e.span)
3361 })
3362 } else {
3363 Err(PlannerError::unsupported_feature(
3364 "non-literal LIMIT/OFFSET",
3365 "v0.3.0+",
3366 stmt_span,
3367 ))
3368 }
3369 }
3370 }
3371 }
3372
3373 fn plan_insert(&self, stmt: &Insert) -> Result<LogicalPlan, PlannerError> {
3378 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
3380
3381 let columns: Vec<String> = if let Some(ref cols) = stmt.columns {
3383 for col in cols {
3385 self.name_resolver.resolve_column(table, col, stmt.span)?;
3386 }
3387 cols.clone()
3388 } else {
3389 table.column_names().into_iter().map(String::from).collect()
3391 };
3392
3393 match &stmt.source {
3394 InsertSource::Values { values } => {
3395 let mut typed_values: Vec<Vec<TypedExpr>> = Vec::new();
3396
3397 for row in values {
3398 if row.len() != columns.len() {
3399 return Err(PlannerError::column_value_count_mismatch(
3400 columns.len(),
3401 row.len(),
3402 stmt.span,
3403 ));
3404 }
3405
3406 typed_values.push(self.type_check_insert_values(row, &columns, table)?);
3407 }
3408
3409 Ok(LogicalPlan::Insert {
3410 table: table.name.clone(),
3411 columns,
3412 values: typed_values,
3413 })
3414 }
3415 InsertSource::Select { select } => {
3416 let source = self.plan_select_relation(select, &[], &CtePlans::new())?;
3417 if source.schema.len() != columns.len() {
3418 return Err(PlannerError::column_value_count_mismatch(
3419 columns.len(),
3420 source.schema.len(),
3421 stmt.span,
3422 ));
3423 }
3424
3425 for (source_column, target_column) in source.schema.iter().zip(&columns) {
3426 let target = table
3427 .get_column(target_column)
3428 .expect("validated target column");
3429 if target.not_null && source_column.data_type == ResolvedType::Null {
3430 return Err(PlannerError::null_constraint_violation(
3431 target_column,
3432 stmt.span,
3433 ));
3434 }
3435 self.validate_resolved_type_assignment(
3436 &source_column.data_type,
3437 &target.data_type,
3438 stmt.span,
3439 )?;
3440 }
3441
3442 Ok(LogicalPlan::InsertSelect {
3443 table: table.name.clone(),
3444 columns,
3445 source: Box::new(source.plan),
3446 })
3447 }
3448 }
3449 }
3450
3451 fn type_check_insert_values(
3453 &self,
3454 values: &[crate::ast::expr::Expr],
3455 columns: &[String],
3456 table: &TableMetadata,
3457 ) -> Result<Vec<TypedExpr>, PlannerError> {
3458 let mut typed_values = Vec::new();
3459
3460 for (i, value) in values.iter().enumerate() {
3461 let column_name = &columns[i];
3462 let column_meta = table.get_column(column_name).ok_or_else(|| {
3463 PlannerError::column_not_found(column_name, &table.name, value.span)
3464 })?;
3465
3466 let typed_value = self.type_checker.infer_type(value, table)?;
3468
3469 if column_meta.not_null
3471 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
3472 {
3473 return Err(PlannerError::null_constraint_violation(
3474 column_name,
3475 value.span,
3476 ));
3477 }
3478
3479 self.validate_type_assignment(&typed_value, &column_meta.data_type, value.span)?;
3481
3482 let typed_value =
3483 self.coerce_assignment_value(typed_value, &column_meta.data_type, value.span);
3484
3485 typed_values.push(typed_value);
3486 }
3487
3488 Ok(typed_values)
3489 }
3490
3491 fn validate_type_assignment(
3493 &self,
3494 value: &TypedExpr,
3495 target_type: &ResolvedType,
3496 span: crate::ast::Span,
3497 ) -> Result<(), PlannerError> {
3498 self.validate_resolved_type_assignment(&value.resolved_type, target_type, span)
3499 }
3500
3501 fn validate_resolved_type_assignment(
3502 &self,
3503 source_type: &ResolvedType,
3504 target_type: &ResolvedType,
3505 span: crate::ast::Span,
3506 ) -> Result<(), PlannerError> {
3507 if *source_type == ResolvedType::Null {
3509 return Ok(());
3510 }
3511
3512 if self.types_compatible(source_type, target_type) {
3514 return Ok(());
3515 }
3516
3517 Err(PlannerError::type_mismatch(
3518 target_type.to_string(),
3519 source_type.to_string(),
3520 span,
3521 ))
3522 }
3523
3524 fn types_compatible(&self, source: &ResolvedType, target: &ResolvedType) -> bool {
3526 use ResolvedType::*;
3527
3528 if source == target {
3530 return true;
3531 }
3532
3533 match (source, target) {
3535 (Integer, BigInt) | (Integer, Float) | (Integer, Double) => true,
3537 (BigInt, Float) | (BigInt, Double) => true,
3539 (Float, Double) => true,
3541 (Double, Float) => true,
3544 (Text | Integer | BigInt | Float | Double, Timestamp) => true,
3547 (Vector { dimension: d1, .. }, Vector { dimension: d2, .. }) => d1 == d2,
3549 _ => false,
3550 }
3551 }
3552
3553 fn coerce_assignment_value(
3554 &self,
3555 value: TypedExpr,
3556 target_type: &ResolvedType,
3557 span: crate::ast::Span,
3558 ) -> TypedExpr {
3559 if value.resolved_type != *target_type
3560 && value.resolved_type != ResolvedType::Null
3561 && matches!(
3562 target_type,
3563 ResolvedType::Integer
3564 | ResolvedType::BigInt
3565 | ResolvedType::Float
3566 | ResolvedType::Double
3567 | ResolvedType::Timestamp
3568 )
3569 {
3570 TypedExpr::cast(value, target_type.clone(), span)
3571 } else {
3572 value
3573 }
3574 }
3575
3576 fn plan_update(&self, stmt: &Update) -> Result<LogicalPlan, PlannerError> {
3580 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
3582
3583 let mut typed_assignments = Vec::new();
3585
3586 for assignment in &stmt.assignments {
3587 let column_meta =
3589 self.name_resolver
3590 .resolve_column(table, &assignment.column, assignment.span)?;
3591 let column_index = table.get_column_index(&assignment.column).unwrap();
3592
3593 let typed_value = self.type_checker.infer_type(&assignment.value, table)?;
3595
3596 if column_meta.not_null
3598 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
3599 {
3600 return Err(PlannerError::null_constraint_violation(
3601 &assignment.column,
3602 assignment.value.span,
3603 ));
3604 }
3605
3606 self.validate_type_assignment(
3608 &typed_value,
3609 &column_meta.data_type,
3610 assignment.value.span,
3611 )?;
3612
3613 let typed_value = self.coerce_assignment_value(
3614 typed_value,
3615 &column_meta.data_type,
3616 assignment.value.span,
3617 );
3618
3619 typed_assignments.push(TypedAssignment::new(
3620 assignment.column.clone(),
3621 column_index,
3622 typed_value,
3623 ));
3624 }
3625
3626 let filter = if let Some(ref selection) = stmt.selection {
3628 let predicate = self.type_checker.infer_type(selection, table)?;
3629
3630 if predicate.resolved_type != ResolvedType::Boolean {
3632 return Err(PlannerError::type_mismatch(
3633 "Boolean",
3634 predicate.resolved_type.to_string(),
3635 selection.span,
3636 ));
3637 }
3638
3639 Some(predicate)
3640 } else {
3641 None
3642 };
3643
3644 Ok(LogicalPlan::Update {
3645 table: table.name.clone(),
3646 assignments: typed_assignments,
3647 filter,
3648 })
3649 }
3650
3651 fn plan_delete(&self, stmt: &Delete) -> Result<LogicalPlan, PlannerError> {
3655 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
3657
3658 let filter = if let Some(ref selection) = stmt.selection {
3660 let predicate = self.type_checker.infer_type(selection, table)?;
3661
3662 if predicate.resolved_type != ResolvedType::Boolean {
3664 return Err(PlannerError::type_mismatch(
3665 "Boolean",
3666 predicate.resolved_type.to_string(),
3667 selection.span,
3668 ));
3669 }
3670
3671 Some(predicate)
3672 } else {
3673 None
3674 };
3675
3676 Ok(LogicalPlan::Delete {
3677 table: table.name.clone(),
3678 filter,
3679 })
3680 }
3681}
3682
3683#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3684struct AggregateSignature {
3685 name: String,
3686 distinct: bool,
3687 star: bool,
3688 arg_key: Option<String>,
3689 separator: Option<String>,
3690}
3691
3692fn collect_projection_aliases(items: &[SelectItem]) -> HashMap<String, crate::ast::expr::Expr> {
3702 let mut aliases = HashMap::new();
3703 for item in items {
3704 if let SelectItem::Expr {
3705 expr,
3706 alias: Some(alias),
3707 ..
3708 } = item
3709 {
3710 aliases.entry(alias.clone()).or_insert_with(|| expr.clone());
3711 }
3712 }
3713 aliases
3714}
3715
3716fn substitute_projection_aliases(
3734 expr: &crate::ast::expr::Expr,
3735 aliases: &HashMap<String, crate::ast::expr::Expr>,
3736) -> crate::ast::expr::Expr {
3737 use crate::ast::expr::ExprKind;
3738
3739 if aliases.is_empty() {
3740 return expr.clone();
3741 }
3742
3743 let recurse = |e: &crate::ast::expr::Expr| substitute_projection_aliases(e, aliases);
3744
3745 let kind = match &expr.kind {
3746 ExprKind::ColumnRef {
3747 table: None,
3748 column,
3749 } => match aliases.get(column) {
3750 Some(source) => {
3751 let mut replacement = source.clone();
3752 replacement.span = expr.span;
3753 return replacement;
3754 }
3755 None => return expr.clone(),
3756 },
3757 ExprKind::BinaryOp { left, op, right } => ExprKind::BinaryOp {
3758 left: Box::new(recurse(left)),
3759 op: *op,
3760 right: Box::new(recurse(right)),
3761 },
3762 ExprKind::UnaryOp { op, operand } => ExprKind::UnaryOp {
3763 op: *op,
3764 operand: Box::new(recurse(operand)),
3765 },
3766 ExprKind::FunctionCall {
3767 name,
3768 args,
3769 distinct,
3770 star,
3771 over,
3772 } => ExprKind::FunctionCall {
3773 name: name.clone(),
3774 args: args.iter().map(recurse).collect(),
3775 distinct: *distinct,
3776 star: *star,
3777 over: over.as_ref().map(|window| crate::ast::expr::WindowSpec {
3778 partition_by: window.partition_by.iter().map(recurse).collect(),
3779 order_by: window
3780 .order_by
3781 .iter()
3782 .map(|order| OrderByExpr {
3783 expr: recurse(&order.expr),
3784 asc: order.asc,
3785 nulls_first: order.nulls_first,
3786 span: order.span,
3787 })
3788 .collect(),
3789 frame: window.frame.clone(),
3790 }),
3791 },
3792 ExprKind::Case {
3793 operand,
3794 branches,
3795 else_expr,
3796 } => ExprKind::Case {
3797 operand: operand.as_deref().map(|e| Box::new(recurse(e))),
3798 branches: branches
3799 .iter()
3800 .map(|branch| crate::ast::expr::CaseWhen {
3801 when: recurse(&branch.when),
3802 then: recurse(&branch.then),
3803 })
3804 .collect(),
3805 else_expr: else_expr.as_deref().map(|e| Box::new(recurse(e))),
3806 },
3807 ExprKind::Cast { expr, target_type } => ExprKind::Cast {
3808 expr: Box::new(recurse(expr)),
3809 target_type: target_type.clone(),
3810 },
3811 ExprKind::Between {
3812 expr,
3813 low,
3814 high,
3815 negated,
3816 } => ExprKind::Between {
3817 expr: Box::new(recurse(expr)),
3818 low: Box::new(recurse(low)),
3819 high: Box::new(recurse(high)),
3820 negated: *negated,
3821 },
3822 ExprKind::Like {
3823 expr,
3824 pattern,
3825 escape,
3826 negated,
3827 kind,
3828 } => ExprKind::Like {
3829 expr: Box::new(recurse(expr)),
3830 pattern: Box::new(recurse(pattern)),
3831 escape: escape.as_deref().map(|e| Box::new(recurse(e))),
3832 negated: *negated,
3833 kind: *kind,
3834 },
3835 ExprKind::InList {
3836 expr,
3837 list,
3838 negated,
3839 } => ExprKind::InList {
3840 expr: Box::new(recurse(expr)),
3841 list: list.iter().map(recurse).collect(),
3842 negated: *negated,
3843 },
3844 ExprKind::IsNull { expr, negated } => ExprKind::IsNull {
3845 expr: Box::new(recurse(expr)),
3846 negated: *negated,
3847 },
3848 ExprKind::ColumnRef { .. }
3851 | ExprKind::Literal { .. }
3852 | ExprKind::VectorLiteral { .. }
3853 | ExprKind::ScalarSubquery { .. }
3854 | ExprKind::InSubquery { .. }
3855 | ExprKind::Exists { .. }
3856 | ExprKind::Quantified { .. } => return expr.clone(),
3857 };
3858
3859 crate::ast::expr::Expr {
3860 kind,
3861 span: expr.span,
3862 }
3863}
3864
3865fn build_offset_window_function(
3866 name: &str,
3867 args: &[TypedExpr],
3868) -> Result<OffsetWindowFunction, PlannerError> {
3869 let value = args.first().cloned().ok_or_else(|| {
3870 PlannerError::invalid_expression(format!(
3871 "{}() window function expects 1 to 3 arguments",
3872 name.to_ascii_uppercase()
3873 ))
3874 })?;
3875 if args.len() > 3 {
3876 return Err(PlannerError::invalid_expression(format!(
3877 "{}() window function expects 1 to 3 arguments",
3878 name.to_ascii_uppercase()
3879 )));
3880 }
3881 Ok(OffsetWindowFunction {
3882 value,
3883 offset: args.get(1).cloned(),
3884 default: args.get(2).cloned(),
3885 })
3886}
3887
3888fn expr_contains_aggregate(expr: &crate::ast::expr::Expr) -> bool {
3889 use crate::ast::expr::ExprKind;
3890
3891 match &expr.kind {
3892 ExprKind::FunctionCall {
3893 name, args, over, ..
3894 } => {
3895 if over.is_none() && is_aggregate_function(name) {
3896 return true;
3897 }
3898 args.iter().any(expr_contains_aggregate)
3899 || over.as_ref().is_some_and(|window| {
3900 window.partition_by.iter().any(expr_contains_aggregate)
3901 || window
3902 .order_by
3903 .iter()
3904 .any(|sort| expr_contains_aggregate(&sort.expr))
3905 })
3906 }
3907 ExprKind::BinaryOp { left, right, .. } => {
3908 expr_contains_aggregate(left) || expr_contains_aggregate(right)
3909 }
3910 ExprKind::UnaryOp { operand, .. } => expr_contains_aggregate(operand),
3911 ExprKind::Case {
3912 operand,
3913 branches,
3914 else_expr,
3915 } => {
3916 operand.as_deref().is_some_and(expr_contains_aggregate)
3917 || branches.iter().any(|branch| {
3918 expr_contains_aggregate(&branch.when) || expr_contains_aggregate(&branch.then)
3919 })
3920 || else_expr.as_deref().is_some_and(expr_contains_aggregate)
3921 }
3922 ExprKind::Cast { expr, .. } => expr_contains_aggregate(expr),
3923 ExprKind::Between {
3924 expr, low, high, ..
3925 } => {
3926 expr_contains_aggregate(expr)
3927 || expr_contains_aggregate(low)
3928 || expr_contains_aggregate(high)
3929 }
3930 ExprKind::Like {
3931 expr,
3932 pattern,
3933 escape,
3934 ..
3935 } => {
3936 expr_contains_aggregate(expr)
3937 || expr_contains_aggregate(pattern)
3938 || escape.as_deref().is_some_and(expr_contains_aggregate)
3939 }
3940 ExprKind::InList { expr, list, .. } => {
3941 expr_contains_aggregate(expr) || list.iter().any(expr_contains_aggregate)
3942 }
3943 ExprKind::IsNull { expr, .. } => expr_contains_aggregate(expr),
3944 ExprKind::ScalarSubquery { .. }
3945 | ExprKind::InSubquery { .. }
3946 | ExprKind::Exists { .. }
3947 | ExprKind::Quantified { .. }
3948 | ExprKind::Literal { .. }
3949 | ExprKind::VectorLiteral { .. }
3950 | ExprKind::ColumnRef { .. } => false,
3951 }
3952}
3953
3954fn typed_expr_contains_aggregate(expr: &TypedExpr) -> bool {
3955 match &expr.kind {
3956 TypedExprKind::FunctionCall {
3957 name, args, over, ..
3958 } => {
3959 if over.is_none() && is_aggregate_function(name) {
3960 return true;
3961 }
3962 args.iter().any(typed_expr_contains_aggregate)
3963 || over.as_ref().is_some_and(|window| {
3964 window
3965 .partition_by
3966 .iter()
3967 .any(typed_expr_contains_aggregate)
3968 || window
3969 .order_by
3970 .iter()
3971 .any(|sort| typed_expr_contains_aggregate(&sort.expr))
3972 })
3973 }
3974 TypedExprKind::BinaryOp { left, right, .. } => {
3975 typed_expr_contains_aggregate(left) || typed_expr_contains_aggregate(right)
3976 }
3977 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_aggregate(operand),
3978 TypedExprKind::Cast { expr, .. } => typed_expr_contains_aggregate(expr),
3979 TypedExprKind::Case {
3980 operand,
3981 branches,
3982 else_expr,
3983 } => {
3984 operand
3985 .as_deref()
3986 .is_some_and(typed_expr_contains_aggregate)
3987 || branches.iter().any(|branch| {
3988 typed_expr_contains_aggregate(&branch.when)
3989 || typed_expr_contains_aggregate(&branch.then)
3990 })
3991 || else_expr
3992 .as_deref()
3993 .is_some_and(typed_expr_contains_aggregate)
3994 }
3995 TypedExprKind::Between {
3996 expr, low, high, ..
3997 } => {
3998 typed_expr_contains_aggregate(expr)
3999 || typed_expr_contains_aggregate(low)
4000 || typed_expr_contains_aggregate(high)
4001 }
4002 TypedExprKind::Like {
4003 expr,
4004 pattern,
4005 escape,
4006 ..
4007 } => {
4008 typed_expr_contains_aggregate(expr)
4009 || typed_expr_contains_aggregate(pattern)
4010 || escape
4011 .as_ref()
4012 .is_some_and(|inner| typed_expr_contains_aggregate(inner))
4013 }
4014 TypedExprKind::InList { expr, list, .. } => {
4015 typed_expr_contains_aggregate(expr) || list.iter().any(typed_expr_contains_aggregate)
4016 }
4017 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_aggregate(expr),
4018 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_aggregate(expr),
4019 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_aggregate(expr),
4020 TypedExprKind::ScalarSubquery(_) | TypedExprKind::Exists { .. } => false,
4021 _ => false,
4022 }
4023}
4024
4025fn select_contains_window(stmt: &Select) -> bool {
4026 stmt.projection.iter().any(|item| match item {
4027 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
4028 SelectItem::Expr { expr, .. } => expr_contains_window(expr),
4029 }) || stmt
4030 .order_by
4031 .iter()
4032 .any(|order| expr_contains_window(&order.expr))
4033}
4034
4035fn expr_contains_window(expr: &crate::ast::expr::Expr) -> bool {
4036 match &expr.kind {
4037 crate::ast::expr::ExprKind::FunctionCall { args, over, .. } => {
4038 over.is_some() || args.iter().any(expr_contains_window)
4039 }
4040 crate::ast::expr::ExprKind::BinaryOp { left, right, .. } => {
4041 expr_contains_window(left) || expr_contains_window(right)
4042 }
4043 crate::ast::expr::ExprKind::UnaryOp { operand, .. }
4044 | crate::ast::expr::ExprKind::Cast { expr: operand, .. }
4045 | crate::ast::expr::ExprKind::IsNull { expr: operand, .. } => expr_contains_window(operand),
4046 crate::ast::expr::ExprKind::Between {
4047 expr, low, high, ..
4048 } => expr_contains_window(expr) || expr_contains_window(low) || expr_contains_window(high),
4049 crate::ast::expr::ExprKind::Like {
4050 expr,
4051 pattern,
4052 escape,
4053 ..
4054 } => {
4055 expr_contains_window(expr)
4056 || expr_contains_window(pattern)
4057 || escape.as_deref().is_some_and(expr_contains_window)
4058 }
4059 crate::ast::expr::ExprKind::InList { expr, list, .. } => {
4060 expr_contains_window(expr) || list.iter().any(expr_contains_window)
4061 }
4062 _ => false,
4063 }
4064}
4065
4066fn typed_expr_contains_window(expr: &TypedExpr) -> bool {
4067 match &expr.kind {
4068 TypedExprKind::FunctionCall { args, over, .. } => {
4069 over.is_some() || args.iter().any(typed_expr_contains_window)
4070 }
4071 TypedExprKind::BinaryOp { left, right, .. } => {
4072 typed_expr_contains_window(left) || typed_expr_contains_window(right)
4073 }
4074 TypedExprKind::UnaryOp { operand, .. }
4075 | TypedExprKind::Cast { expr: operand, .. }
4076 | TypedExprKind::IsNull { expr: operand, .. } => typed_expr_contains_window(operand),
4077 TypedExprKind::Between {
4078 expr, low, high, ..
4079 } => {
4080 typed_expr_contains_window(expr)
4081 || typed_expr_contains_window(low)
4082 || typed_expr_contains_window(high)
4083 }
4084 TypedExprKind::Like {
4085 expr,
4086 pattern,
4087 escape,
4088 ..
4089 } => {
4090 typed_expr_contains_window(expr)
4091 || typed_expr_contains_window(pattern)
4092 || escape.as_deref().is_some_and(typed_expr_contains_window)
4093 }
4094 TypedExprKind::InList { expr, list, .. } => {
4095 typed_expr_contains_window(expr) || list.iter().any(typed_expr_contains_window)
4096 }
4097 _ => false,
4098 }
4099}
4100
4101fn rewrite_projection_for_windows(
4102 projection: &Projection,
4103 window_map: &HashMap<String, usize>,
4104 base_width: usize,
4105 window_names: &[String],
4106) -> Result<Projection, PlannerError> {
4107 match projection {
4108 Projection::All(names) => Ok(Projection::All(names.clone())),
4109 Projection::Columns(columns) => Ok(Projection::Columns(
4110 columns
4111 .iter()
4112 .map(|column| {
4113 Ok(ProjectedColumn {
4114 expr: rewrite_expr_for_windows(
4115 &column.expr,
4116 window_map,
4117 base_width,
4118 window_names,
4119 )?,
4120 alias: column.alias.clone(),
4121 })
4122 })
4123 .collect::<Result<Vec<_>, PlannerError>>()?,
4124 )),
4125 }
4126}
4127
4128fn rewrite_expr_for_windows(
4129 expr: &TypedExpr,
4130 window_map: &HashMap<String, usize>,
4131 base_width: usize,
4132 window_names: &[String],
4133) -> Result<TypedExpr, PlannerError> {
4134 if let Some(index) = window_map.get(&expr_key(expr)) {
4135 return Ok(TypedExpr::column_ref(
4136 "__window__".to_string(),
4137 window_names
4138 .get(*index)
4139 .cloned()
4140 .unwrap_or_else(|| format!("__window_{index}")),
4141 base_width + index,
4142 expr.resolved_type.clone(),
4143 expr.span,
4144 ));
4145 }
4146
4147 let rewrite =
4148 |inner: &TypedExpr| rewrite_expr_for_windows(inner, window_map, base_width, window_names);
4149 let kind = match &expr.kind {
4150 TypedExprKind::FunctionCall {
4151 name,
4152 args,
4153 distinct,
4154 star,
4155 over,
4156 } => {
4157 if over.is_some() {
4158 return Err(PlannerError::invalid_expression(
4159 "window expression is not part of the window plan".to_string(),
4160 ));
4161 }
4162 TypedExprKind::FunctionCall {
4163 name: name.clone(),
4164 args: args.iter().map(rewrite).collect::<Result<Vec<_>, _>>()?,
4165 distinct: *distinct,
4166 star: *star,
4167 over: None,
4168 }
4169 }
4170 TypedExprKind::BinaryOp { left, op, right } => TypedExprKind::BinaryOp {
4171 left: Box::new(rewrite(left)?),
4172 op: *op,
4173 right: Box::new(rewrite(right)?),
4174 },
4175 TypedExprKind::UnaryOp { op, operand } => TypedExprKind::UnaryOp {
4176 op: *op,
4177 operand: Box::new(rewrite(operand)?),
4178 },
4179 TypedExprKind::Cast {
4180 expr: inner,
4181 target_type,
4182 } => TypedExprKind::Cast {
4183 expr: Box::new(rewrite(inner)?),
4184 target_type: target_type.clone(),
4185 },
4186 TypedExprKind::Between {
4187 expr: inner,
4188 low,
4189 high,
4190 negated,
4191 } => TypedExprKind::Between {
4192 expr: Box::new(rewrite(inner)?),
4193 low: Box::new(rewrite(low)?),
4194 high: Box::new(rewrite(high)?),
4195 negated: *negated,
4196 },
4197 TypedExprKind::Like {
4198 expr: inner,
4199 pattern,
4200 escape,
4201 negated,
4202 kind,
4203 } => TypedExprKind::Like {
4204 expr: Box::new(rewrite(inner)?),
4205 pattern: Box::new(rewrite(pattern)?),
4206 escape: escape.as_deref().map(rewrite).transpose()?.map(Box::new),
4207 negated: *negated,
4208 kind: *kind,
4209 },
4210 TypedExprKind::InList {
4211 expr: inner,
4212 list,
4213 negated,
4214 } => TypedExprKind::InList {
4215 expr: Box::new(rewrite(inner)?),
4216 list: list.iter().map(rewrite).collect::<Result<Vec<_>, _>>()?,
4217 negated: *negated,
4218 },
4219 TypedExprKind::IsNull {
4220 expr: inner,
4221 negated,
4222 } => TypedExprKind::IsNull {
4223 expr: Box::new(rewrite(inner)?),
4224 negated: *negated,
4225 },
4226 _ => return Ok(expr.clone()),
4227 };
4228 Ok(TypedExpr {
4229 kind,
4230 resolved_type: expr.resolved_type.clone(),
4231 span: expr.span,
4232 })
4233}
4234
4235fn rewrite_expr_for_projected_output(
4241 expr: &TypedExpr,
4242 projection: &Projection,
4243 output_schema: &[ColumnMetadata],
4244) -> Result<TypedExpr, PlannerError> {
4245 let index = match projection {
4246 Projection::Columns(columns) => columns
4247 .iter()
4248 .position(|column| expr_key(&column.expr) == expr_key(expr)),
4249 Projection::All(_) => match &expr.kind {
4250 TypedExprKind::ColumnRef { column_index, .. }
4251 if *column_index < output_schema.len() =>
4252 {
4253 Some(*column_index)
4254 }
4255 _ => None,
4256 },
4257 };
4258 let Some(index) = index else {
4259 return Err(PlannerError::invalid_expression(
4260 "ORDER BY expression must appear in the SELECT projection for window queries"
4261 .to_string(),
4262 ));
4263 };
4264 let column = output_schema.get(index).ok_or_else(|| {
4265 PlannerError::invalid_expression(
4266 "ORDER BY projection index is outside the output schema".to_string(),
4267 )
4268 })?;
4269 Ok(TypedExpr::column_ref(
4270 "__project__".to_string(),
4271 column.name.clone(),
4272 index,
4273 column.data_type.clone(),
4274 expr.span,
4275 ))
4276}
4277
4278fn map_join_type(join_type: crate::ast::dml::JoinType) -> JoinType {
4279 match join_type {
4280 crate::ast::dml::JoinType::Inner => JoinType::Inner,
4281 crate::ast::dml::JoinType::Left => JoinType::Left,
4282 crate::ast::dml::JoinType::Right => JoinType::Right,
4283 crate::ast::dml::JoinType::Full => JoinType::Full,
4284 crate::ast::dml::JoinType::Cross => JoinType::Cross,
4285 }
4286}
4287
4288struct FoundScopedColumn {
4289 table: String,
4290 index: usize,
4291 ty: ResolvedType,
4292 partner_indices: Vec<usize>,
4293}
4294
4295fn find_scoped_column(
4296 scope: &[ScopedTable],
4297 column: &str,
4298 span: crate::ast::Span,
4299) -> Result<FoundScopedColumn, PlannerError> {
4300 let mut matches = Vec::new();
4301 for table in scope {
4302 if table.hidden_unqualified_columns.contains(column) {
4303 continue;
4304 }
4305 if let Some(local_idx) = table.table.get_column_index(column) {
4306 let meta = &table.table.columns[local_idx];
4307 matches.push(FoundScopedColumn {
4308 table: table.table.name.clone(),
4309 index: table.start_index + local_idx,
4310 ty: meta.data_type.clone(),
4311 partner_indices: table
4312 .merged_column_partners
4313 .get(column)
4314 .cloned()
4315 .unwrap_or_default(),
4316 });
4317 }
4318 }
4319 match matches.len() {
4320 0 => Err(PlannerError::column_not_found(column, "JOIN input", span)),
4321 1 => Ok(matches.remove(0)),
4322 _ => Err(PlannerError::ambiguous_column(
4323 column,
4324 scope.iter().map(|s| s.table.name.clone()).collect(),
4325 span,
4326 )),
4327 }
4328}
4329
4330fn merged_scoped_column_expr(
4331 found: &FoundScopedColumn,
4332 column: &str,
4333 span: crate::ast::Span,
4334) -> TypedExpr {
4335 let own = TypedExpr::column_ref(
4336 found.table.clone(),
4337 column.to_string(),
4338 found.index,
4339 found.ty.clone(),
4340 span,
4341 );
4342 if found.partner_indices.is_empty() {
4343 return own;
4344 }
4345
4346 let mut args = Vec::with_capacity(found.partner_indices.len() + 1);
4347 args.push(own);
4348 args.extend(found.partner_indices.iter().map(|&index| {
4349 TypedExpr::column_ref(
4350 found.table.clone(),
4351 column.to_string(),
4352 index,
4353 found.ty.clone(),
4354 span,
4355 )
4356 }));
4357 TypedExpr {
4358 kind: TypedExprKind::FunctionCall {
4359 name: "coalesce".to_string(),
4360 args,
4361 distinct: false,
4362 star: false,
4363 over: None,
4364 },
4365 resolved_type: found.ty.clone(),
4366 span,
4367 }
4368}
4369
4370fn projection_schema(
4371 projection: &Projection,
4372 input_schema: &[ColumnMetadata],
4373) -> Vec<ColumnMetadata> {
4374 match projection {
4375 Projection::All(names) => names
4376 .iter()
4377 .enumerate()
4378 .map(|(idx, name)| {
4379 let ty = (names.len() == input_schema.len())
4380 .then(|| input_schema.get(idx))
4381 .flatten()
4382 .or_else(|| input_schema.iter().find(|col| &col.name == name))
4383 .map(|col| col.data_type.clone())
4384 .unwrap_or(ResolvedType::Null);
4385 ColumnMetadata::new(name.clone(), ty)
4386 })
4387 .collect(),
4388 Projection::Columns(columns) => columns
4389 .iter()
4390 .enumerate()
4391 .map(|(idx, col)| {
4392 let name = col
4393 .alias
4394 .clone()
4395 .or_else(|| match &col.expr.kind {
4396 TypedExprKind::ColumnRef { column, .. } => Some(column.clone()),
4397 TypedExprKind::FunctionCall { name, args, .. }
4401 if name == "coalesce" && !args.is_empty() =>
4402 {
4403 let first_column = match &args[0].kind {
4404 TypedExprKind::ColumnRef { column, .. } => Some(column),
4405 _ => None,
4406 };
4407 first_column
4408 .filter(|column| {
4409 args.iter().all(|arg| {
4410 matches!(
4411 &arg.kind,
4412 TypedExprKind::ColumnRef { column: other, .. }
4413 if other == *column
4414 )
4415 })
4416 })
4417 .cloned()
4418 }
4419 _ => None,
4420 })
4421 .unwrap_or_else(|| format!("col_{idx}"));
4422 ColumnMetadata::new(name, col.expr.resolved_type.clone())
4423 })
4424 .collect(),
4425 }
4426}
4427
4428fn visible_wildcard_columns(schema: &[ColumnMetadata], scope: &[ScopedTable]) -> Vec<String> {
4429 schema
4430 .iter()
4431 .enumerate()
4432 .filter(|(index, column)| {
4433 !scope.iter().any(|table| {
4434 *index >= table.start_index
4435 && *index < table.start_index + table.table.columns.len()
4436 && table.hidden_unqualified_columns.contains(&column.name)
4437 })
4438 })
4439 .map(|(_, column)| column.name.clone())
4440 .collect()
4441}
4442
4443fn offset_scope(scope: &[ScopedTable], offset: usize) -> Vec<ScopedTable> {
4444 scope
4445 .iter()
4446 .cloned()
4447 .map(|mut table| {
4448 table.start_index += offset;
4449 table.scope_level += 1;
4450 table
4451 })
4452 .collect()
4453}
4454
4455fn natural_join_columns(
4456 left_schema: &[ColumnMetadata],
4457 right_schema: &[ColumnMetadata],
4458) -> Vec<String> {
4459 let right_names = right_schema
4463 .iter()
4464 .map(|column| column.name.as_str())
4465 .collect::<HashSet<_>>();
4466 left_schema
4467 .iter()
4468 .filter(|left| right_names.contains(left.name.as_str()))
4469 .map(|column| column.name.clone())
4470 .collect()
4471}
4472
4473fn install_base_projection(plan: &mut LogicalPlan, projection: &Projection) {
4474 match plan {
4475 LogicalPlan::Scan {
4476 projection: scan_projection,
4477 ..
4478 } => *scan_projection = projection.clone(),
4479 LogicalPlan::Filter { input, .. } => install_base_projection(input, projection),
4480 _ => {}
4481 }
4482}
4483
4484fn is_aggregate_function(name: &str) -> bool {
4485 matches!(
4486 name.to_ascii_lowercase().as_str(),
4487 "count" | "sum" | "total" | "avg" | "min" | "max" | "group_concat" | "string_agg"
4488 )
4489}
4490
4491fn expr_key(expr: &TypedExpr) -> String {
4492 format!("{:?}", expr.kind)
4493}
4494
4495fn aggregate_signature(
4496 name: &str,
4497 distinct: bool,
4498 star: bool,
4499 arg: Option<&TypedExpr>,
4500 separator: Option<&String>,
4501 _expr: &TypedExpr,
4502) -> AggregateSignature {
4503 AggregateSignature {
4504 name: name.to_ascii_lowercase(),
4505 distinct,
4506 star,
4507 arg_key: arg.map(expr_key),
4508 separator: separator.cloned(),
4509 }
4510}
4511
4512fn build_group_key_map(group_keys: &[TypedExpr]) -> HashMap<String, usize> {
4513 let mut map = HashMap::new();
4514 for (idx, key) in group_keys.iter().enumerate() {
4515 map.insert(expr_key(key), idx);
4516 }
4517 map
4518}
4519
4520fn build_aggregate_map(aggregates: &[AggregateExpr]) -> HashMap<AggregateSignature, usize> {
4521 let mut map = HashMap::new();
4522 for (idx, agg) in aggregates.iter().enumerate() {
4523 let (name, separator, star, arg) = match &agg.function {
4524 AggregateFunction::Count => (
4525 "count".to_string(),
4526 None,
4527 agg.arg.is_none(),
4528 agg.arg.as_ref(),
4529 ),
4530 AggregateFunction::Sum => ("sum".to_string(), None, false, agg.arg.as_ref()),
4531 AggregateFunction::Total => ("total".to_string(), None, false, agg.arg.as_ref()),
4532 AggregateFunction::Avg => ("avg".to_string(), None, false, agg.arg.as_ref()),
4533 AggregateFunction::Min => ("min".to_string(), None, false, agg.arg.as_ref()),
4534 AggregateFunction::Max => ("max".to_string(), None, false, agg.arg.as_ref()),
4535 AggregateFunction::GroupConcat { separator } => (
4536 "group_concat".to_string(),
4537 separator.clone(),
4538 false,
4539 agg.arg.as_ref(),
4540 ),
4541 AggregateFunction::StringAgg { separator } => (
4542 "string_agg".to_string(),
4543 separator.clone(),
4544 false,
4545 agg.arg.as_ref(),
4546 ),
4547 };
4548 let signature = AggregateSignature {
4549 name,
4550 distinct: agg.distinct,
4551 star,
4552 arg_key: arg.map(expr_key),
4553 separator,
4554 };
4555 map.insert(signature, idx);
4556 }
4557 map
4558}
4559
4560fn build_aggregate_schema(
4561 group_keys: &[TypedExpr],
4562 aggregates: &[AggregateExpr],
4563) -> Vec<ColumnMetadata> {
4564 let mut schema = Vec::new();
4565 for (idx, key) in group_keys.iter().enumerate() {
4566 let name = match &key.kind {
4567 TypedExprKind::ColumnRef { column, .. } => column.clone(),
4568 _ => format!("group_{idx}"),
4569 };
4570 schema.push(ColumnMetadata::new(name, key.resolved_type.clone()));
4571 }
4572 for (idx, agg) in aggregates.iter().enumerate() {
4573 let name = match &agg.function {
4574 AggregateFunction::Count => format!("count_{idx}"),
4575 AggregateFunction::Sum => format!("sum_{idx}"),
4576 AggregateFunction::Total => format!("total_{idx}"),
4577 AggregateFunction::Avg => format!("avg_{idx}"),
4578 AggregateFunction::Min => format!("min_{idx}"),
4579 AggregateFunction::Max => format!("max_{idx}"),
4580 AggregateFunction::GroupConcat { .. } => format!("group_concat_{idx}"),
4581 AggregateFunction::StringAgg { .. } => format!("string_agg_{idx}"),
4582 };
4583 schema.push(ColumnMetadata::new(name, agg.result_type.clone()));
4584 }
4585 schema
4586}
4587
4588fn rewrite_expr_with_maps(
4589 expr: &TypedExpr,
4590 group_key_map: &HashMap<String, usize>,
4591 aggregate_map: &HashMap<AggregateSignature, usize>,
4592 output_names: &[String],
4593) -> Result<TypedExpr, PlannerError> {
4594 let group_key_count = output_names.len().saturating_sub(aggregate_map.len());
4595 let key = expr_key(expr);
4596 if let Some(idx) = group_key_map.get(&key) {
4597 return Ok(make_output_column_ref(
4598 *idx,
4599 output_names,
4600 expr.resolved_type.clone(),
4601 expr.span,
4602 ));
4603 }
4604
4605 match &expr.kind {
4606 TypedExprKind::FunctionCall {
4607 name,
4608 args,
4609 distinct,
4610 star,
4611 over: None,
4612 } if is_aggregate_function(name) => {
4613 let separator = if name.eq_ignore_ascii_case("group_concat") && args.len() == 2 {
4614 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
4615 Some(value.clone())
4616 } else {
4617 return Err(PlannerError::invalid_expression(
4618 "GROUP_CONCAT separator must be a string literal".to_string(),
4619 ));
4620 }
4621 } else if name.eq_ignore_ascii_case("string_agg") && args.len() == 2 {
4622 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
4623 Some(value.clone())
4624 } else {
4625 return Err(PlannerError::invalid_expression(
4626 "STRING_AGG separator must be a string literal".to_string(),
4627 ));
4628 }
4629 } else {
4630 None
4631 };
4632 let signature = AggregateSignature {
4633 name: name.to_ascii_lowercase(),
4634 distinct: *distinct,
4635 star: *star,
4636 arg_key: args.first().map(expr_key),
4637 separator,
4638 };
4639 let idx = aggregate_map.get(&signature).ok_or_else(|| {
4640 PlannerError::invalid_expression(
4641 "aggregate in expression is not part of plan".to_string(),
4642 )
4643 })?;
4644 let output_index = group_key_count + idx;
4645 Ok(make_output_column_ref(
4646 output_index,
4647 output_names,
4648 expr.resolved_type.clone(),
4649 expr.span,
4650 ))
4651 }
4652 TypedExprKind::FunctionCall {
4653 name,
4654 args,
4655 distinct,
4656 star,
4657 over,
4658 } => {
4659 if over.is_none() && (*distinct || *star) {
4660 return Err(PlannerError::invalid_expression(
4661 "DISTINCT/STAR modifiers are only supported for aggregates".to_string(),
4662 ));
4663 }
4664 let mut rewritten_args = Vec::with_capacity(args.len());
4665 for arg in args {
4666 rewritten_args.push(rewrite_expr_with_maps(
4667 arg,
4668 group_key_map,
4669 aggregate_map,
4670 output_names,
4671 )?);
4672 }
4673 let over = over
4674 .as_ref()
4675 .map(|window| {
4676 let partition_by = window
4677 .partition_by
4678 .iter()
4679 .map(|expr| {
4680 rewrite_expr_with_maps(expr, group_key_map, aggregate_map, output_names)
4681 })
4682 .collect::<Result<Vec<_>, PlannerError>>()?;
4683 let order_by = window
4684 .order_by
4685 .iter()
4686 .map(|sort| {
4687 Ok(SortExpr::new(
4688 rewrite_expr_with_maps(
4689 &sort.expr,
4690 group_key_map,
4691 aggregate_map,
4692 output_names,
4693 )?,
4694 sort.asc,
4695 sort.nulls_first,
4696 ))
4697 })
4698 .collect::<Result<Vec<_>, PlannerError>>()?;
4699 Ok(crate::planner::typed_expr::TypedWindowSpec {
4700 partition_by,
4701 order_by,
4702 frame: window.frame.clone(),
4703 })
4704 })
4705 .transpose()?;
4706 Ok(TypedExpr {
4707 kind: TypedExprKind::FunctionCall {
4708 name: name.clone(),
4709 args: rewritten_args,
4710 distinct: *distinct,
4711 star: *star,
4712 over,
4713 },
4714 resolved_type: expr.resolved_type.clone(),
4715 span: expr.span,
4716 })
4717 }
4718 TypedExprKind::BinaryOp { left, op, right } => {
4719 let left = rewrite_expr_with_maps(left, group_key_map, aggregate_map, output_names)?;
4720 let right = rewrite_expr_with_maps(right, group_key_map, aggregate_map, output_names)?;
4721 Ok(TypedExpr {
4722 kind: TypedExprKind::BinaryOp {
4723 left: Box::new(left),
4724 op: *op,
4725 right: Box::new(right),
4726 },
4727 resolved_type: expr.resolved_type.clone(),
4728 span: expr.span,
4729 })
4730 }
4731 TypedExprKind::UnaryOp { op, operand } => {
4732 let operand =
4733 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)?;
4734 Ok(TypedExpr {
4735 kind: TypedExprKind::UnaryOp {
4736 op: *op,
4737 operand: Box::new(operand),
4738 },
4739 resolved_type: expr.resolved_type.clone(),
4740 span: expr.span,
4741 })
4742 }
4743 TypedExprKind::Case {
4744 operand,
4745 branches,
4746 else_expr,
4747 } => {
4748 let operand = operand
4749 .as_deref()
4750 .map(|operand| {
4751 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)
4752 .map(Box::new)
4753 })
4754 .transpose()?;
4755 let mut rewritten_branches = Vec::with_capacity(branches.len());
4756 for branch in branches {
4757 rewritten_branches.push(TypedCaseWhen {
4758 when: rewrite_expr_with_maps(
4759 &branch.when,
4760 group_key_map,
4761 aggregate_map,
4762 output_names,
4763 )?,
4764 then: rewrite_expr_with_maps(
4765 &branch.then,
4766 group_key_map,
4767 aggregate_map,
4768 output_names,
4769 )?,
4770 });
4771 }
4772 let else_expr = else_expr
4773 .as_deref()
4774 .map(|else_expr| {
4775 rewrite_expr_with_maps(else_expr, group_key_map, aggregate_map, output_names)
4776 .map(Box::new)
4777 })
4778 .transpose()?;
4779 Ok(TypedExpr {
4780 kind: TypedExprKind::Case {
4781 operand,
4782 branches: rewritten_branches,
4783 else_expr,
4784 },
4785 resolved_type: expr.resolved_type.clone(),
4786 span: expr.span,
4787 })
4788 }
4789 TypedExprKind::Between {
4790 expr: inner,
4791 low,
4792 high,
4793 negated,
4794 } => {
4795 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
4796 let low = rewrite_expr_with_maps(low, group_key_map, aggregate_map, output_names)?;
4797 let high = rewrite_expr_with_maps(high, group_key_map, aggregate_map, output_names)?;
4798 Ok(TypedExpr {
4799 kind: TypedExprKind::Between {
4800 expr: Box::new(inner),
4801 low: Box::new(low),
4802 high: Box::new(high),
4803 negated: *negated,
4804 },
4805 resolved_type: expr.resolved_type.clone(),
4806 span: expr.span,
4807 })
4808 }
4809 TypedExprKind::Like {
4810 expr: inner,
4811 pattern,
4812 escape,
4813 negated,
4814 kind,
4815 } => {
4816 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
4817 let pattern =
4818 rewrite_expr_with_maps(pattern, group_key_map, aggregate_map, output_names)?;
4819 let escape = if let Some(esc) = escape {
4820 Some(Box::new(rewrite_expr_with_maps(
4821 esc,
4822 group_key_map,
4823 aggregate_map,
4824 output_names,
4825 )?))
4826 } else {
4827 None
4828 };
4829 Ok(TypedExpr {
4830 kind: TypedExprKind::Like {
4831 expr: Box::new(inner),
4832 pattern: Box::new(pattern),
4833 escape,
4834 negated: *negated,
4835 kind: *kind,
4836 },
4837 resolved_type: expr.resolved_type.clone(),
4838 span: expr.span,
4839 })
4840 }
4841 TypedExprKind::InList {
4842 expr: inner,
4843 list,
4844 negated,
4845 } => {
4846 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
4847 let mut rewritten_list = Vec::with_capacity(list.len());
4848 for item in list {
4849 rewritten_list.push(rewrite_expr_with_maps(
4850 item,
4851 group_key_map,
4852 aggregate_map,
4853 output_names,
4854 )?);
4855 }
4856 Ok(TypedExpr {
4857 kind: TypedExprKind::InList {
4858 expr: Box::new(inner),
4859 list: rewritten_list,
4860 negated: *negated,
4861 },
4862 resolved_type: expr.resolved_type.clone(),
4863 span: expr.span,
4864 })
4865 }
4866 TypedExprKind::IsNull {
4867 expr: inner,
4868 negated,
4869 } => {
4870 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
4871 Ok(TypedExpr {
4872 kind: TypedExprKind::IsNull {
4873 expr: Box::new(inner),
4874 negated: *negated,
4875 },
4876 resolved_type: expr.resolved_type.clone(),
4877 span: expr.span,
4878 })
4879 }
4880 TypedExprKind::Literal(_) | TypedExprKind::VectorLiteral(_) => Ok(expr.clone()),
4881 TypedExprKind::ColumnRef { .. } => Err(PlannerError::invalid_expression(
4882 "column reference must appear in GROUP BY or be aggregated".to_string(),
4883 )),
4884 TypedExprKind::Cast {
4885 expr: inner,
4886 target_type,
4887 } => {
4888 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
4889 Ok(TypedExpr {
4890 kind: TypedExprKind::Cast {
4891 expr: Box::new(inner),
4892 target_type: target_type.clone(),
4893 },
4894 resolved_type: expr.resolved_type.clone(),
4895 span: expr.span,
4896 })
4897 }
4898 TypedExprKind::ScalarSubquery(_)
4899 | TypedExprKind::InSubquery { .. }
4900 | TypedExprKind::Exists { .. }
4901 | TypedExprKind::Quantified { .. } => Ok(expr.clone()),
4902 }
4903}
4904
4905fn make_output_column_ref(
4906 index: usize,
4907 output_names: &[String],
4908 resolved_type: ResolvedType,
4909 span: crate::ast::Span,
4910) -> TypedExpr {
4911 let name = output_names
4912 .get(index)
4913 .cloned()
4914 .unwrap_or_else(|| format!("col_{index}"));
4915 TypedExpr::column_ref("__agg__".to_string(), name, index, resolved_type, span)
4916}