1pub mod aggregate_expr;
13mod error;
14pub mod knn_optimizer;
15pub mod logical_plan;
16pub mod name_resolver;
17mod named_window;
18pub mod type_checker;
19pub mod typed_expr;
20pub mod types;
21
22#[cfg(test)]
23mod planner_tests;
24
25pub use aggregate_expr::{AggregateExpr, AggregateFunction};
26pub use error::PlannerError;
27pub use knn_optimizer::{KnnPattern, SortDirection, detect_knn_pattern};
28pub use logical_plan::{
29 JoinType, LogicalPlan, OffsetWindowFunction, RecursiveCteLimits, SetOperator,
30 TableFunctionKind, ValueWindowFunction, WindowExpr, WindowFunction,
31};
32pub use name_resolver::{NameResolver, ResolvedColumn};
33pub use type_checker::{ScopedTable, TypeChecker};
34pub use typed_expr::{
35 ProjectedColumn, Projection, SortExpr, TypedAssignment, TypedCaseWhen, TypedExpr, TypedExprKind,
36};
37pub use types::ResolvedType;
38
39use crate::ast::ddl::{
40 ColumnConstraint, ColumnDef, CreateIndex, CreateTable, DropIndex, DropTable,
41};
42use crate::ast::dml::{
43 Delete, FromItem, GroupByItem, Insert, InsertSource, LITERAL_TABLE, OrderByExpr, QueryBody,
44 Select, SelectItem, SetOperation as AstSetOperation, SetOperator as AstSetOperator, Update,
45 Values,
46};
47use crate::ast::expr::{Expr, ExprKind, Literal};
48use crate::ast::{PragmaValue, Spanned, Statement, StatementKind};
49use crate::catalog::{Catalog, ColumnMetadata, IndexMetadata, TableMetadata};
50use crate::{AlopexDialect, DataSourceFormat, Parser, SqlError, TableType};
51use named_window::resolve_named_windows;
52use std::collections::{HashMap, HashSet};
53
54#[derive(Clone)]
55struct PlannedRelation {
56 plan: LogicalPlan,
57 schema: Vec<ColumnMetadata>,
58 scope: Vec<ScopedTable>,
59}
60
61struct WindowSelectStages {
62 plan: LogicalPlan,
63 windows: Vec<WindowExpr>,
64 qualify: Option<TypedExpr>,
65 projection: Projection,
66 order_by: Vec<SortExpr>,
67 window_schema: Vec<ColumnMetadata>,
68}
69
70type CtePlans = HashMap<String, PlannedRelation>;
71
72fn direct_from_reference_count(items: &[FromItem], name: &str) -> usize {
73 items
74 .iter()
75 .map(|item| match item {
76 FromItem::Table {
77 name: table_name, ..
78 } => usize::from(table_name == name),
79 FromItem::Join { left, right, .. } => {
80 direct_from_item_reference_count(left, name)
81 + direct_from_item_reference_count(right, name)
82 }
83 FromItem::Derived { .. } | FromItem::Function { .. } => 0,
84 })
85 .sum()
86}
87
88fn direct_from_item_reference_count(item: &FromItem, name: &str) -> usize {
89 match item {
90 FromItem::Table {
91 name: table_name, ..
92 } => usize::from(table_name == name),
93 FromItem::Join { left, right, .. } => {
94 direct_from_item_reference_count(left, name)
95 + direct_from_item_reference_count(right, name)
96 }
97 FromItem::Derived { .. } | FromItem::Function { .. } => 0,
98 }
99}
100
101fn select_table_reference_count(select: &Select, name: &str) -> usize {
102 fn from_item_count(item: &FromItem, name: &str) -> usize {
103 match item {
104 FromItem::Table {
105 name: table_name, ..
106 } => usize::from(table_name == name),
107 FromItem::Join { left, right, .. } => {
108 from_item_count(left, name) + from_item_count(right, name)
109 }
110 FromItem::Derived { subquery, .. } => query_body_table_reference_count(subquery, name),
111 FromItem::Function { .. } => 0,
115 }
116 }
117
118 let from_count = select
119 .from
120 .iter()
121 .map(|item| from_item_count(item, name))
122 .sum::<usize>();
123 let set_count = select
124 .set_operations
125 .iter()
126 .map(|operation| query_body_table_reference_count(&operation.right, name))
127 .sum::<usize>();
128 let with_count = select.with.as_ref().map_or(0, |with| {
129 with.ctes
130 .iter()
131 .map(|cte| query_body_table_reference_count(&cte.query, name))
132 .sum()
133 });
134 from_count + set_count + with_count
135}
136
137fn values_table_reference_count(values: &Values, name: &str) -> usize {
138 let set_count = values
139 .set_operations
140 .iter()
141 .map(|operation| query_body_table_reference_count(&operation.right, name))
142 .sum::<usize>();
143 let with_count = values.with.as_ref().map_or(0, |with| {
144 with.ctes
145 .iter()
146 .map(|cte| query_body_table_reference_count(&cte.query, name))
147 .sum()
148 });
149 set_count + with_count
150}
151
152fn query_body_table_reference_count(body: &QueryBody, name: &str) -> usize {
153 match body {
154 QueryBody::Select(select) => select_table_reference_count(select, name),
155 QueryBody::Values(values) => values_table_reference_count(values, name),
156 }
157}
158
159fn cte_dependency_cycle(with: &crate::ast::WithClause) -> bool {
160 fn visit(node: usize, dependencies: &[Vec<usize>], state: &mut [u8]) -> bool {
161 state[node] = 1;
162 for &dependency in &dependencies[node] {
163 if state[dependency] == 1
164 || (state[dependency] == 0 && visit(dependency, dependencies, state))
165 {
166 return true;
167 }
168 }
169 state[node] = 2;
170 false
171 }
172
173 let dependencies = with
174 .ctes
175 .iter()
176 .map(|cte| {
177 with.ctes
178 .iter()
179 .enumerate()
180 .filter_map(|(dependency, candidate)| {
181 (query_body_table_reference_count(&cte.query, &candidate.name) > 0)
182 .then_some(dependency)
183 })
184 .collect::<Vec<_>>()
185 })
186 .collect::<Vec<_>>();
187 let mut state = vec![0; dependencies.len()];
188 (0..dependencies.len()).any(|node| state[node] == 0 && visit(node, &dependencies, &mut state))
189}
190
191fn expr_contains_subquery(expr: &Expr) -> bool {
192 match &expr.kind {
193 ExprKind::Literal { .. } | ExprKind::ColumnRef { .. } | ExprKind::VectorLiteral { .. } => {
194 false
195 }
196 ExprKind::BinaryOp { left, right, .. } => {
197 expr_contains_subquery(left) || expr_contains_subquery(right)
198 }
199 ExprKind::UnaryOp { operand, .. } => expr_contains_subquery(operand),
200 ExprKind::TruthPredicate { expr, .. } => expr_contains_subquery(expr),
201 ExprKind::IsDistinctFrom { left, right, .. } => {
202 expr_contains_subquery(left) || expr_contains_subquery(right)
203 }
204 ExprKind::Row { items } => items.iter().any(expr_contains_subquery),
205 ExprKind::Case {
206 operand,
207 branches,
208 else_expr,
209 } => {
210 operand.as_deref().is_some_and(expr_contains_subquery)
211 || branches.iter().any(|branch| {
212 expr_contains_subquery(&branch.when) || expr_contains_subquery(&branch.then)
213 })
214 || else_expr.as_deref().is_some_and(expr_contains_subquery)
215 }
216 ExprKind::FunctionCall {
217 args,
218 order_by,
219 within_group,
220 filter,
221 over,
222 ..
223 } => {
224 args.iter().any(expr_contains_subquery)
225 || order_by
226 .iter()
227 .any(|order| expr_contains_subquery(&order.expr))
228 || within_group
229 .iter()
230 .any(|order| expr_contains_subquery(&order.expr))
231 || filter.as_deref().is_some_and(expr_contains_subquery)
232 || over.as_ref().is_some_and(|window| {
233 window.partition_by.iter().any(expr_contains_subquery)
234 || window
235 .order_by
236 .iter()
237 .any(|order| expr_contains_subquery(&order.expr))
238 })
239 }
240 ExprKind::Cast { expr, .. }
241 | ExprKind::TryCast { expr, .. }
242 | ExprKind::IsNull { expr, .. } => expr_contains_subquery(expr),
243 ExprKind::Between {
244 expr, low, high, ..
245 } => {
246 expr_contains_subquery(expr)
247 || expr_contains_subquery(low)
248 || expr_contains_subquery(high)
249 }
250 ExprKind::Like {
251 expr,
252 pattern,
253 escape,
254 ..
255 } => {
256 expr_contains_subquery(expr)
257 || expr_contains_subquery(pattern)
258 || escape.as_deref().is_some_and(expr_contains_subquery)
259 }
260 ExprKind::InList { expr, list, .. } => {
261 expr_contains_subquery(expr) || list.iter().any(expr_contains_subquery)
262 }
263 ExprKind::ScalarSubquery { .. }
264 | ExprKind::InSubquery { .. }
265 | ExprKind::Exists { .. }
266 | ExprKind::Quantified { .. } => true,
267 }
268}
269
270fn from_item_contains_subquery(item: &FromItem) -> bool {
271 match item {
272 FromItem::Table { .. } => false,
273 FromItem::Derived { .. } | FromItem::Function { .. } => true,
276 FromItem::Join {
277 left,
278 right,
279 condition,
280 ..
281 } => {
282 from_item_contains_subquery(left)
283 || from_item_contains_subquery(right)
284 || condition.as_ref().is_some_and(expr_contains_subquery)
285 }
286 }
287}
288
289fn select_contains_subquery(select: &Select) -> bool {
290 select.projection.iter().any(|item| match item {
291 SelectItem::Expr { expr, .. } => expr_contains_subquery(expr),
292 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
293 }) || select.from.iter().any(from_item_contains_subquery)
294 || select
295 .selection
296 .as_ref()
297 .is_some_and(expr_contains_subquery)
298 || select.group_by.as_ref().is_some_and(|group| {
299 group
300 .iter()
301 .flat_map(GroupByItem::exprs)
302 .any(expr_contains_subquery)
303 })
304 || select.having.as_ref().is_some_and(expr_contains_subquery)
305 || select
306 .set_operations
307 .iter()
308 .any(|operation| query_body_contains_subquery(&operation.right))
309 || select
310 .order_by
311 .iter()
312 .any(|order| expr_contains_subquery(&order.expr))
313 || select.limit.as_ref().is_some_and(expr_contains_subquery)
314 || select.offset.as_ref().is_some_and(expr_contains_subquery)
315 || select.with.as_ref().is_some_and(|with| {
316 with.ctes
317 .iter()
318 .any(|cte| query_body_contains_subquery(&cte.query))
319 })
320}
321
322fn values_contains_subquery(values: &Values) -> bool {
323 values.rows.iter().flatten().any(expr_contains_subquery)
324 || values
325 .set_operations
326 .iter()
327 .any(|operation| query_body_contains_subquery(&operation.right))
328 || values
329 .order_by
330 .iter()
331 .any(|order| expr_contains_subquery(&order.expr))
332 || values.limit.as_ref().is_some_and(expr_contains_subquery)
333 || values.offset.as_ref().is_some_and(expr_contains_subquery)
334 || values.with.as_ref().is_some_and(|with| {
335 with.ctes
336 .iter()
337 .any(|cte| query_body_contains_subquery(&cte.query))
338 })
339}
340
341fn query_body_contains_subquery(body: &QueryBody) -> bool {
342 match body {
343 QueryBody::Select(select) => select_contains_subquery(select),
344 QueryBody::Values(values) => values_contains_subquery(values),
345 }
346}
347
348fn common_values_type(
349 current: &ResolvedType,
350 next: &ResolvedType,
351 span: crate::ast::Span,
352) -> Result<ResolvedType, PlannerError> {
353 use ResolvedType::{BigInt, Double, Float, Integer, Null};
354
355 if *current == Null {
356 return Ok(next.clone());
357 }
358 if *next == Null || current == next {
359 return Ok(current.clone());
360 }
361 match (current, next) {
362 (Integer, BigInt) | (BigInt, Integer) => Ok(BigInt),
363 (Integer | BigInt | Float | Double, Integer | BigInt | Float | Double) => Ok(Double),
364 _ => Err(PlannerError::type_mismatch(
365 current.type_name(),
366 next.type_name(),
367 span,
368 )),
369 }
370}
371
372#[derive(Debug, Clone)]
378pub struct PlannedStatement {
379 pub plan: LogicalPlan,
381 pub routing_input: RoutingInput,
383}
384
385impl PlannedStatement {
386 pub fn statement_kind(&self) -> &StatementKind {
388 &self.routing_input.statement_kind
389 }
390
391 pub fn table_references(&self) -> &[TableReference] {
393 &self.routing_input.table_references
394 }
395
396 pub fn diagnostics(&self) -> &[PlanningDiagnostic] {
399 &self.routing_input.diagnostics
400 }
401}
402
403#[derive(Debug, Clone)]
405pub struct RoutingInput {
406 pub statement_kind: StatementKind,
409 pub table_references: Vec<TableReference>,
411 pub diagnostics: Vec<PlanningDiagnostic>,
413}
414
415#[derive(Debug, Clone, PartialEq, Eq)]
417pub struct TableReference {
418 pub table_name: String,
420 pub access: TableReferenceAccess,
422 pub source: TableReferenceSource,
424}
425
426impl TableReference {
427 pub fn new(
428 table_name: impl Into<String>,
429 access: TableReferenceAccess,
430 source: TableReferenceSource,
431 ) -> Self {
432 Self {
433 table_name: table_name.into(),
434 access,
435 source,
436 }
437 }
438}
439
440#[derive(Debug, Clone, Copy, PartialEq, Eq)]
442pub enum TableReferenceAccess {
443 Read,
445 Write,
447 Create,
449 Drop,
451 Metadata,
453}
454
455#[derive(Debug, Clone, Copy, PartialEq, Eq)]
457pub enum TableReferenceSource {
458 TopLevelPlanTableName,
460 LogicalPlanScan,
462 LogicalPlanMutationTarget,
464 LogicalPlanDdlTarget,
466 LogicalPlanIndexTarget,
468 TypedExprSubquery,
470}
471
472#[derive(Debug, Clone, Copy, PartialEq, Eq)]
474pub enum PlanningDiagnosticSeverity {
475 Info,
476 Warning,
477}
478
479#[derive(Debug, Clone, PartialEq, Eq)]
481pub struct PlanningDiagnostic {
482 pub code: &'static str,
484 pub severity: PlanningDiagnosticSeverity,
486 pub message: String,
488}
489
490impl PlanningDiagnostic {
491 pub fn info(code: &'static str, message: impl Into<String>) -> Self {
492 Self {
493 code,
494 severity: PlanningDiagnosticSeverity::Info,
495 message: message.into(),
496 }
497 }
498
499 pub fn warning(code: &'static str, message: impl Into<String>) -> Self {
500 Self {
501 code,
502 severity: PlanningDiagnosticSeverity::Warning,
503 message: message.into(),
504 }
505 }
506}
507
508pub fn plan_sql_for_routing<C: Catalog + ?Sized>(
510 catalog: &C,
511 sql: &str,
512) -> Result<Vec<PlannedStatement>, SqlError> {
513 let statements = Parser::parse_sql(&AlopexDialect, sql).map_err(SqlError::from)?;
514 statements
515 .iter()
516 .map(|statement| plan_statement_for_routing(catalog, statement).map_err(SqlError::from))
517 .collect()
518}
519
520pub fn plan_statement_for_routing<C: Catalog + ?Sized>(
522 catalog: &C,
523 statement: &Statement,
524) -> Result<PlannedStatement, PlannerError> {
525 let planner = Planner::new(catalog);
526 let plan = planner.plan(statement)?;
527 let routing_input = routing_input_for_plan(statement, &plan)?;
528 Ok(PlannedStatement {
529 plan,
530 routing_input,
531 })
532}
533
534fn routing_input_for_plan(
535 statement: &Statement,
536 plan: &LogicalPlan,
537) -> Result<RoutingInput, PlannerError> {
538 let mut diagnostics = Vec::new();
539 let extractor = TableReferenceExtractor::new();
540 let table_references = extractor.extract_from_logical_plan(
541 plan,
542 table_reference_access(statement)?,
543 &mut diagnostics,
544 );
545
546 Ok(RoutingInput {
547 statement_kind: statement.kind.clone(),
548 table_references,
549 diagnostics,
550 })
551}
552
553#[derive(Debug, Default, Clone, Copy)]
555pub struct TableReferenceExtractor;
556
557impl TableReferenceExtractor {
558 pub fn new() -> Self {
559 Self
560 }
561
562 pub fn extract_from_logical_plan(
565 &self,
566 plan: &LogicalPlan,
567 root_access: TableReferenceAccess,
568 diagnostics: &mut Vec<PlanningDiagnostic>,
569 ) -> Vec<TableReference> {
570 let mut references = Vec::new();
571 self.extract_plan(
572 plan,
573 root_access,
574 TableReferenceSource::LogicalPlanScan,
575 diagnostics,
576 &mut references,
577 );
578 if references.is_empty() {
579 diagnostics.push(PlanningDiagnostic::info(
580 "ALOPEX-PLAN-ROUTE-001",
581 "statement has no physical table reference",
582 ));
583 }
584 references
585 }
586
587 pub fn extract_from_subquery_context(
589 &self,
590 plan: &LogicalPlan,
591 diagnostics: &mut Vec<PlanningDiagnostic>,
592 ) -> Vec<TableReference> {
593 let mut references = Vec::new();
594 self.extract_plan(
595 plan,
596 TableReferenceAccess::Read,
597 TableReferenceSource::TypedExprSubquery,
598 diagnostics,
599 &mut references,
600 );
601 references
602 }
603
604 fn extract_plan(
605 &self,
606 plan: &LogicalPlan,
607 root_access: TableReferenceAccess,
608 scan_source: TableReferenceSource,
609 diagnostics: &mut Vec<PlanningDiagnostic>,
610 references: &mut Vec<TableReference>,
611 ) {
612 match plan {
613 LogicalPlan::Scan { table, projection } => {
614 if table != LITERAL_TABLE {
615 push_table_reference(
616 references,
617 table,
618 TableReferenceAccess::Read,
619 scan_source,
620 );
621 }
622 self.extract_projection(projection, diagnostics, references);
623 }
624 LogicalPlan::Values { rows, .. } => {
625 for row in rows {
626 for value in row {
627 self.extract_typed_expr(value, diagnostics, references);
628 }
629 }
630 }
631 LogicalPlan::Filter { input, predicate } => {
632 self.extract_plan(input, root_access, scan_source, diagnostics, references);
633 self.extract_typed_expr(predicate, diagnostics, references);
634 }
635 LogicalPlan::Project { input, projection } => {
636 self.extract_plan(input, root_access, scan_source, diagnostics, references);
637 self.extract_projection(projection, diagnostics, references);
638 }
639 LogicalPlan::Join {
640 left,
641 right,
642 condition,
643 ..
644 }
645 | LogicalPlan::LateralJoin {
646 left,
647 right,
648 condition,
649 ..
650 } => {
651 self.extract_plan(
652 left,
653 TableReferenceAccess::Read,
654 scan_source,
655 diagnostics,
656 references,
657 );
658 self.extract_plan(
659 right,
660 TableReferenceAccess::Read,
661 scan_source,
662 diagnostics,
663 references,
664 );
665 if let Some(condition) = condition {
666 self.extract_typed_expr(condition, diagnostics, references);
667 }
668 }
669 LogicalPlan::TableFunction { args, .. } => {
670 for arg in args {
671 self.extract_typed_expr(arg, diagnostics, references);
672 }
673 }
674 LogicalPlan::Aggregate {
675 input,
676 group_keys,
677 aggregates,
678 having,
679 projection,
680 grouping_sets: _,
681 } => {
682 self.extract_plan(input, root_access, scan_source, diagnostics, references);
683 for expr in group_keys {
684 self.extract_typed_expr(expr, diagnostics, references);
685 }
686 for aggregate in aggregates {
687 if let Some(arg) = &aggregate.arg {
688 self.extract_typed_expr(arg, diagnostics, references);
689 }
690 }
691 if let Some(having) = having {
692 self.extract_typed_expr(having, diagnostics, references);
693 }
694 self.extract_projection(projection, diagnostics, references);
695 }
696 LogicalPlan::Window { input, windows } => {
697 self.extract_plan(input, root_access, scan_source, diagnostics, references);
698 for window in windows {
699 for expr in &window.partition_by {
700 self.extract_typed_expr(expr, diagnostics, references);
701 }
702 for sort_expr in &window.order_by {
703 self.extract_typed_expr(&sort_expr.expr, diagnostics, references);
704 }
705 match &window.function {
706 WindowFunction::Aggregate(aggregate) => {
707 if let Some(arg) = &aggregate.arg {
708 self.extract_typed_expr(arg, diagnostics, references);
709 }
710 }
711 WindowFunction::Lag(function) | WindowFunction::Lead(function) => {
712 self.extract_typed_expr(&function.value, diagnostics, references);
713 if let Some(offset) = &function.offset {
714 self.extract_typed_expr(offset, diagnostics, references);
715 }
716 if let Some(default) = &function.default {
717 self.extract_typed_expr(default, diagnostics, references);
718 }
719 }
720 WindowFunction::Ntile(argument) => {
721 self.extract_typed_expr(argument, diagnostics, references);
722 }
723 WindowFunction::Value(function) => match function {
724 ValueWindowFunction::FirstValue(value)
725 | ValueWindowFunction::LastValue(value) => {
726 self.extract_typed_expr(value, diagnostics, references);
727 }
728 ValueWindowFunction::NthValue { value, nth } => {
729 self.extract_typed_expr(value, diagnostics, references);
730 self.extract_typed_expr(nth, diagnostics, references);
731 }
732 },
733 WindowFunction::RowNumber
734 | WindowFunction::Rank
735 | WindowFunction::DenseRank
736 | WindowFunction::PercentRank
737 | WindowFunction::CumeDist => {}
738 }
739 }
740 }
741 LogicalPlan::SetOperation { left, right, .. } => {
742 self.extract_plan(left, root_access, scan_source, diagnostics, references);
743 self.extract_plan(right, root_access, scan_source, diagnostics, references);
744 }
745 LogicalPlan::RecursiveCte {
746 anchor,
747 recursive_term,
748 ..
749 } => {
750 self.extract_plan(anchor, root_access, scan_source, diagnostics, references);
751 self.extract_plan(
752 recursive_term,
753 root_access,
754 scan_source,
755 diagnostics,
756 references,
757 );
758 }
759 LogicalPlan::RecursiveReference { .. } => {}
760 LogicalPlan::Sort { input, order_by } => {
761 self.extract_plan(input, root_access, scan_source, diagnostics, references);
762 for sort_expr in order_by {
763 self.extract_typed_expr(&sort_expr.expr, diagnostics, references);
764 }
765 }
766 LogicalPlan::DistinctOn {
767 input, order_by, ..
768 } => {
769 self.extract_plan(input, root_access, scan_source, diagnostics, references);
770 for sort_expr in order_by {
771 self.extract_typed_expr(&sort_expr.expr, diagnostics, references);
772 }
773 }
774 LogicalPlan::Limit { input, .. } => {
775 self.extract_plan(input, root_access, scan_source, diagnostics, references);
776 }
777 LogicalPlan::Insert { table, values, .. } => {
778 push_table_reference(
779 references,
780 table,
781 root_access,
782 TableReferenceSource::LogicalPlanMutationTarget,
783 );
784 for row in values {
785 for value in row {
786 self.extract_typed_expr(value, diagnostics, references);
787 }
788 }
789 }
790 LogicalPlan::InsertSelect { table, source, .. } => {
791 push_table_reference(
792 references,
793 table,
794 root_access,
795 TableReferenceSource::LogicalPlanMutationTarget,
796 );
797 self.extract_plan(
798 source,
799 TableReferenceAccess::Read,
800 scan_source,
801 diagnostics,
802 references,
803 );
804 }
805 LogicalPlan::Update {
806 table,
807 assignments,
808 filter,
809 } => {
810 push_table_reference(
811 references,
812 table,
813 root_access,
814 TableReferenceSource::LogicalPlanMutationTarget,
815 );
816 for assignment in assignments {
817 self.extract_typed_expr(&assignment.value, diagnostics, references);
818 }
819 if let Some(filter) = filter {
820 self.extract_typed_expr(filter, diagnostics, references);
821 }
822 }
823 LogicalPlan::Delete { table, filter } => {
824 push_table_reference(
825 references,
826 table,
827 root_access,
828 TableReferenceSource::LogicalPlanMutationTarget,
829 );
830 if let Some(filter) = filter {
831 self.extract_typed_expr(filter, diagnostics, references);
832 }
833 }
834 LogicalPlan::CreateTable { table, .. } => push_table_reference(
835 references,
836 &table.name,
837 root_access,
838 TableReferenceSource::LogicalPlanDdlTarget,
839 ),
840 LogicalPlan::DropTable { name, .. } => push_table_reference(
841 references,
842 name,
843 root_access,
844 TableReferenceSource::LogicalPlanDdlTarget,
845 ),
846 LogicalPlan::CreateIndex { index, .. } => push_table_reference(
847 references,
848 &index.table,
849 root_access,
850 TableReferenceSource::LogicalPlanIndexTarget,
851 ),
852 LogicalPlan::DropIndex { name, .. } => diagnostics.push(PlanningDiagnostic::warning(
853 "ALOPEX-PLAN-ROUTE-003",
854 format!(
855 "DROP INDEX {name} does not expose a target table in the current logical plan"
856 ),
857 )),
858 LogicalPlan::Pragma { .. } => {}
859 }
860 }
861
862 fn extract_projection(
863 &self,
864 projection: &Projection,
865 diagnostics: &mut Vec<PlanningDiagnostic>,
866 references: &mut Vec<TableReference>,
867 ) {
868 if let Projection::Columns(columns) = projection {
869 for column in columns {
870 self.extract_typed_expr(&column.expr, diagnostics, references);
871 }
872 }
873 }
874
875 fn extract_typed_expr(
876 &self,
877 expr: &TypedExpr,
878 diagnostics: &mut Vec<PlanningDiagnostic>,
879 references: &mut Vec<TableReference>,
880 ) {
881 match &expr.kind {
882 TypedExprKind::Literal(_)
883 | TypedExprKind::ColumnRef { .. }
884 | TypedExprKind::VectorLiteral(_) => {}
885 TypedExprKind::BinaryOp { left, right, .. } => {
886 self.extract_typed_expr(left, diagnostics, references);
887 self.extract_typed_expr(right, diagnostics, references);
888 }
889 TypedExprKind::UnaryOp { operand, .. }
890 | TypedExprKind::Cast { expr: operand, .. }
891 | TypedExprKind::TryCast { expr: operand, .. }
892 | TypedExprKind::IsNull { expr: operand, .. } => {
893 self.extract_typed_expr(operand, diagnostics, references);
894 }
895 TypedExprKind::Case {
896 operand,
897 branches,
898 else_expr,
899 } => {
900 if let Some(operand) = operand {
901 self.extract_typed_expr(operand, diagnostics, references);
902 }
903 for branch in branches {
904 self.extract_typed_expr(&branch.when, diagnostics, references);
905 self.extract_typed_expr(&branch.then, diagnostics, references);
906 }
907 if let Some(else_expr) = else_expr {
908 self.extract_typed_expr(else_expr, diagnostics, references);
909 }
910 }
911 TypedExprKind::FunctionCall {
912 args,
913 filter,
914 order_by,
915 ..
916 } => {
917 for arg in args {
918 self.extract_typed_expr(arg, diagnostics, references);
919 }
920 if let Some(filter) = filter {
921 self.extract_typed_expr(filter, diagnostics, references);
922 }
923 for sort in order_by {
924 self.extract_typed_expr(&sort.expr, diagnostics, references);
925 }
926 }
927 TypedExprKind::Between {
928 expr, low, high, ..
929 } => {
930 self.extract_typed_expr(expr, diagnostics, references);
931 self.extract_typed_expr(low, diagnostics, references);
932 self.extract_typed_expr(high, diagnostics, references);
933 }
934 TypedExprKind::Like {
935 expr,
936 pattern,
937 escape,
938 ..
939 } => {
940 self.extract_typed_expr(expr, diagnostics, references);
941 self.extract_typed_expr(pattern, diagnostics, references);
942 if let Some(escape) = escape {
943 self.extract_typed_expr(escape, diagnostics, references);
944 }
945 }
946 TypedExprKind::InList { expr, list, .. } => {
947 self.extract_typed_expr(expr, diagnostics, references);
948 for item in list {
949 self.extract_typed_expr(item, diagnostics, references);
950 }
951 }
952 TypedExprKind::ScalarSubquery(subquery) => self.extract_plan(
953 subquery,
954 TableReferenceAccess::Read,
955 TableReferenceSource::TypedExprSubquery,
956 diagnostics,
957 references,
958 ),
959 TypedExprKind::InSubquery { expr, subquery, .. } => {
960 self.extract_typed_expr(expr, diagnostics, references);
961 self.extract_plan(
962 subquery,
963 TableReferenceAccess::Read,
964 TableReferenceSource::TypedExprSubquery,
965 diagnostics,
966 references,
967 );
968 }
969 TypedExprKind::Exists { subquery, .. } => self.extract_plan(
970 subquery,
971 TableReferenceAccess::Read,
972 TableReferenceSource::TypedExprSubquery,
973 diagnostics,
974 references,
975 ),
976 TypedExprKind::Quantified { expr, subquery, .. } => {
977 self.extract_typed_expr(expr, diagnostics, references);
978 self.extract_plan(
979 subquery,
980 TableReferenceAccess::Read,
981 TableReferenceSource::TypedExprSubquery,
982 diagnostics,
983 references,
984 );
985 }
986 }
987 }
988}
989
990fn push_table_reference(
991 references: &mut Vec<TableReference>,
992 table_name: &str,
993 access: TableReferenceAccess,
994 source: TableReferenceSource,
995) {
996 if !references.iter().any(|reference| {
997 reference.table_name == table_name
998 && reference.access == access
999 && reference.source == source
1000 }) {
1001 references.push(TableReference::new(table_name, access, source));
1002 }
1003}
1004
1005#[derive(Debug)]
1006enum GenericHostStatement<'a> {
1007 CreateTable(&'a CreateTable),
1008 DropTable(&'a DropTable),
1009 CreateIndex(&'a CreateIndex),
1010 DropIndex(&'a DropIndex),
1011 Pragma {
1012 name: &'a str,
1013 value: &'a Option<PragmaValue>,
1014 },
1015 Select(&'a Select),
1016 Values(&'a Values),
1017 Insert(&'a Insert),
1018 Update(&'a Update),
1019 Delete(&'a Delete),
1020 Unsupported,
1021}
1022
1023fn classify_generic_host_statement(statement_kind: &StatementKind) -> GenericHostStatement<'_> {
1024 #[allow(unreachable_patterns)]
1027 match statement_kind {
1028 StatementKind::CreateTable(statement) => GenericHostStatement::CreateTable(statement),
1029 StatementKind::DropTable(statement) => GenericHostStatement::DropTable(statement),
1030 StatementKind::CreateIndex(statement) => GenericHostStatement::CreateIndex(statement),
1031 StatementKind::DropIndex(statement) => GenericHostStatement::DropIndex(statement),
1032 StatementKind::Pragma { name, value } => GenericHostStatement::Pragma { name, value },
1033 StatementKind::Select(statement) => GenericHostStatement::Select(statement),
1034 StatementKind::Values(statement) => GenericHostStatement::Values(statement),
1035 StatementKind::Insert(statement) => GenericHostStatement::Insert(statement),
1036 StatementKind::Update(statement) => GenericHostStatement::Update(statement),
1037 StatementKind::Delete(statement) => GenericHostStatement::Delete(statement),
1038 _ => GenericHostStatement::Unsupported,
1039 }
1040}
1041
1042fn unsupported_generic_statement(statement: &Statement) -> PlannerError {
1043 PlannerError::unsupported_feature(
1044 "statement kind for the generic SQL planner",
1045 "a statement-specific planner",
1046 statement.span,
1047 )
1048}
1049
1050fn table_reference_access(statement: &Statement) -> Result<TableReferenceAccess, PlannerError> {
1051 table_reference_access_for_classified(
1052 statement,
1053 classify_generic_host_statement(&statement.kind),
1054 )
1055}
1056
1057fn table_reference_access_for_classified(
1058 statement: &Statement,
1059 classified: GenericHostStatement<'_>,
1060) -> Result<TableReferenceAccess, PlannerError> {
1061 match classified {
1062 GenericHostStatement::Select(_) | GenericHostStatement::Values(_) => {
1063 Ok(TableReferenceAccess::Read)
1064 }
1065 GenericHostStatement::Insert(_)
1066 | GenericHostStatement::Update(_)
1067 | GenericHostStatement::Delete(_) => Ok(TableReferenceAccess::Write),
1068 GenericHostStatement::CreateTable(_) => Ok(TableReferenceAccess::Create),
1069 GenericHostStatement::DropTable(_) => Ok(TableReferenceAccess::Drop),
1070 GenericHostStatement::CreateIndex(_)
1071 | GenericHostStatement::DropIndex(_)
1072 | GenericHostStatement::Pragma { .. } => Ok(TableReferenceAccess::Metadata),
1073 GenericHostStatement::Unsupported => Err(unsupported_generic_statement(statement)),
1074 }
1075}
1076
1077pub struct Planner<'a, C: Catalog + ?Sized> {
1104 catalog: &'a C,
1105 name_resolver: NameResolver<'a, C>,
1106 type_checker: TypeChecker<'a, C>,
1107}
1108
1109impl<'a, C: Catalog + ?Sized> Planner<'a, C> {
1110 pub fn new(catalog: &'a C) -> Self {
1112 Self {
1113 catalog,
1114 name_resolver: NameResolver::new(catalog),
1115 type_checker: TypeChecker::new(catalog),
1116 }
1117 }
1118
1119 pub fn plan(&self, stmt: &Statement) -> Result<LogicalPlan, PlannerError> {
1130 self.plan_classified_statement(stmt, classify_generic_host_statement(&stmt.kind))
1131 }
1132
1133 fn plan_classified_statement(
1134 &self,
1135 stmt: &Statement,
1136 classified: GenericHostStatement<'_>,
1137 ) -> Result<LogicalPlan, PlannerError> {
1138 match classified {
1139 GenericHostStatement::CreateTable(statement) => self.plan_create_table(statement),
1141 GenericHostStatement::DropTable(statement) => self.plan_drop_table(statement),
1142 GenericHostStatement::CreateIndex(statement) => self.plan_create_index(statement),
1143 GenericHostStatement::DropIndex(statement) => self.plan_drop_index(statement),
1144 GenericHostStatement::Pragma { name, value } => self.plan_pragma(name, value),
1145
1146 GenericHostStatement::Select(statement) => self.plan_select(statement),
1148 GenericHostStatement::Values(statement) => self.plan_values(statement),
1149 GenericHostStatement::Insert(statement) => self.plan_insert(statement),
1150 GenericHostStatement::Update(statement) => self.plan_update(statement),
1151 GenericHostStatement::Delete(statement) => self.plan_delete(statement),
1152 GenericHostStatement::Unsupported => Err(unsupported_generic_statement(stmt)),
1153 }
1154 }
1155
1156 fn plan_pragma(
1157 &self,
1158 raw_name: &str,
1159 value: &Option<PragmaValue>,
1160 ) -> Result<LogicalPlan, PlannerError> {
1161 let name = raw_name.to_ascii_lowercase();
1162 if !matches!(name.as_str(), "cache_size" | "memory_limit" | "io_stats") {
1163 return Err(PlannerError::InvalidPragma {
1164 name,
1165 reason: "supported names are cache_size, memory_limit, and io_stats".to_string(),
1166 });
1167 }
1168 match name.as_str() {
1169 "cache_size" => match value {
1170 Some(PragmaValue::Int(v)) if *v > 0 => {}
1171 Some(PragmaValue::Int(_)) => {
1172 return Err(PlannerError::InvalidPragma {
1173 name,
1174 reason: "cache_size must be a positive page count".to_string(),
1175 });
1176 }
1177 Some(PragmaValue::Text(_)) => {
1178 return Err(PlannerError::InvalidPragma {
1179 name,
1180 reason: "cache_size requires an integer page count".to_string(),
1181 });
1182 }
1183 None => {}
1184 },
1185 "memory_limit" => {
1186 if let Some(PragmaValue::Int(v)) = value
1187 && *v < 0
1188 {
1189 return Err(PlannerError::InvalidPragma {
1190 name,
1191 reason: "memory_limit cannot be negative".to_string(),
1192 });
1193 }
1194 }
1195 "io_stats" => {
1196 if value.is_some() {
1197 return Err(PlannerError::InvalidPragma {
1198 name,
1199 reason: "io_stats does not accept a value".to_string(),
1200 });
1201 }
1202 }
1203 _ => unreachable!(),
1204 }
1205 Ok(LogicalPlan::Pragma {
1206 name,
1207 value: value.clone(),
1208 })
1209 }
1210
1211 fn plan_create_table(&self, stmt: &CreateTable) -> Result<LogicalPlan, PlannerError> {
1220 if !stmt.if_not_exists && self.catalog.table_exists(&stmt.name) {
1222 return Err(PlannerError::table_already_exists(&stmt.name));
1223 }
1224
1225 let columns: Vec<ColumnMetadata> = stmt
1227 .columns
1228 .iter()
1229 .map(|col| self.convert_column_def(col))
1230 .collect();
1231
1232 let primary_key = Self::extract_primary_key(stmt);
1234
1235 let mut table = TableMetadata::new(stmt.name.clone(), columns);
1238 if let Some(pk) = primary_key {
1239 table = table.with_primary_key(pk);
1240 }
1241 table.catalog_name = "default".to_string();
1242 table.namespace_name = "default".to_string();
1243 table.table_type = TableType::Managed;
1244 table.data_source_format = DataSourceFormat::Alopex;
1245 table.properties = HashMap::new();
1246
1247 Ok(LogicalPlan::CreateTable {
1248 table,
1249 if_not_exists: stmt.if_not_exists,
1250 with_options: stmt
1251 .with_options
1252 .iter()
1253 .map(|opt| (opt.key.clone(), opt.value.clone()))
1254 .collect(),
1255 })
1256 }
1257
1258 fn convert_column_def(&self, col: &ColumnDef) -> ColumnMetadata {
1260 let data_type = ResolvedType::from_ast(&col.data_type);
1261 let mut meta = ColumnMetadata::new(col.name.clone(), data_type);
1262
1263 for constraint in &col.constraints {
1265 meta = Self::apply_column_constraint(meta, constraint);
1266 }
1267
1268 meta
1269 }
1270
1271 fn apply_column_constraint(
1273 mut meta: ColumnMetadata,
1274 constraint: &ColumnConstraint,
1275 ) -> ColumnMetadata {
1276 match constraint {
1277 ColumnConstraint::NotNull { .. } => {
1278 meta.not_null = true;
1279 }
1280 ColumnConstraint::PrimaryKey { .. } => {
1281 meta.primary_key = true;
1282 meta.not_null = true; }
1284 ColumnConstraint::Unique { .. } => {
1285 meta.unique = true;
1286 }
1287 ColumnConstraint::Default { value: expr, .. } => {
1288 meta.default = Some(expr.clone());
1289 }
1290 }
1291 meta
1292 }
1293
1294 fn extract_primary_key(stmt: &CreateTable) -> Option<Vec<String>> {
1296 use crate::ast::ddl::TableConstraint;
1297
1298 if let Some(TableConstraint::PrimaryKey { columns, .. }) = stmt.constraints.first() {
1302 return Some(columns.clone());
1303 }
1304
1305 let pk_columns: Vec<String> = stmt
1307 .columns
1308 .iter()
1309 .filter(|col| col.constraints.iter().any(Self::is_primary_key_constraint))
1310 .map(|col| col.name.clone())
1311 .collect();
1312
1313 if pk_columns.is_empty() {
1314 None
1315 } else {
1316 Some(pk_columns)
1317 }
1318 }
1319
1320 fn is_primary_key_constraint(constraint: &ColumnConstraint) -> bool {
1322 matches!(constraint, ColumnConstraint::PrimaryKey { .. })
1323 }
1324
1325 fn plan_drop_table(&self, stmt: &DropTable) -> Result<LogicalPlan, PlannerError> {
1329 if !stmt.if_exists && !self.table_exists_in_default(&stmt.name) {
1331 return Err(PlannerError::TableNotFound {
1332 name: stmt.name.clone(),
1333 line: stmt.span.start.line,
1334 column: stmt.span.start.column,
1335 });
1336 }
1337
1338 Ok(LogicalPlan::DropTable {
1339 name: stmt.name.clone(),
1340 if_exists: stmt.if_exists,
1341 })
1342 }
1343
1344 fn table_exists_in_default(&self, name: &str) -> bool {
1345 match self.catalog.get_table(name) {
1346 Some(table) => table.catalog_name == "default" && table.namespace_name == "default",
1347 None => false,
1348 }
1349 }
1350
1351 fn plan_create_index(&self, stmt: &CreateIndex) -> Result<LogicalPlan, PlannerError> {
1358 if !stmt.if_not_exists && self.catalog.index_exists(&stmt.name) {
1360 return Err(PlannerError::index_already_exists(&stmt.name));
1361 }
1362
1363 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
1365
1366 self.name_resolver
1368 .resolve_column(table, &stmt.column, stmt.span)?;
1369
1370 let mut index = IndexMetadata::new(
1374 0,
1375 stmt.name.clone(),
1376 stmt.table.clone(),
1377 vec![stmt.column.clone()],
1378 );
1379
1380 if let Some(method) = stmt.method {
1381 index = index.with_method(method);
1382 }
1383
1384 let options: Vec<(String, String)> = stmt
1385 .options
1386 .iter()
1387 .map(|opt| (opt.key.clone(), opt.value.clone()))
1388 .collect();
1389 if !options.is_empty() {
1390 index = index.with_options(options);
1391 }
1392
1393 Ok(LogicalPlan::CreateIndex {
1394 index,
1395 if_not_exists: stmt.if_not_exists,
1396 })
1397 }
1398
1399 fn plan_drop_index(&self, stmt: &DropIndex) -> Result<LogicalPlan, PlannerError> {
1403 if !stmt.if_exists && !self.index_exists_in_default(&stmt.name) {
1405 return Err(PlannerError::index_not_found(&stmt.name));
1406 }
1407
1408 Ok(LogicalPlan::DropIndex {
1409 name: stmt.name.clone(),
1410 if_exists: stmt.if_exists,
1411 })
1412 }
1413
1414 fn index_exists_in_default(&self, name: &str) -> bool {
1415 match self.catalog.get_index(name) {
1416 Some(index) => index.catalog_name == "default" && index.namespace_name == "default",
1417 None => false,
1418 }
1419 }
1420
1421 fn plan_select(&self, stmt: &Select) -> Result<LogicalPlan, PlannerError> {
1430 self.plan_select_relation(stmt, &[], &CtePlans::new())
1431 .map(|relation| relation.plan)
1432 }
1433
1434 fn plan_values(&self, stmt: &Values) -> Result<LogicalPlan, PlannerError> {
1435 self.plan_values_relation(stmt, &[], &CtePlans::new())
1436 .map(|relation| relation.plan)
1437 }
1438
1439 fn plan_query_body_relation(
1440 &self,
1441 body: &QueryBody,
1442 outer_scope: &[ScopedTable],
1443 enclosing_ctes: &CtePlans,
1444 ) -> Result<PlannedRelation, PlannerError> {
1445 match body {
1446 QueryBody::Select(select) => {
1447 self.plan_select_relation(select, outer_scope, enclosing_ctes)
1448 }
1449 QueryBody::Values(values) => {
1450 self.plan_values_relation(values, outer_scope, enclosing_ctes)
1451 }
1452 }
1453 }
1454
1455 fn plan_ctes(
1456 &self,
1457 stmt: &Select,
1458 enclosing_ctes: &CtePlans,
1459 ) -> Result<CtePlans, PlannerError> {
1460 self.plan_with_clause(stmt.with.as_ref(), enclosing_ctes)
1461 }
1462
1463 fn plan_with_clause(
1464 &self,
1465 with: Option<&crate::ast::WithClause>,
1466 enclosing_ctes: &CtePlans,
1467 ) -> Result<CtePlans, PlannerError> {
1468 let Some(with) = with else {
1469 return Ok(enclosing_ctes.clone());
1470 };
1471 let mut declared_names = HashSet::new();
1472 for cte in &with.ctes {
1473 if !declared_names.insert(&cte.name) {
1474 return Err(PlannerError::invalid_expression(format!(
1475 "common table expression '{}' is defined more than once",
1476 cte.name
1477 )));
1478 }
1479 }
1480 if with.recursive && cte_dependency_cycle(with) {
1481 return self.plan_recursive_cte(with, enclosing_ctes);
1482 }
1483
1484 let mut plans = enclosing_ctes.clone();
1485 let mut local_names = HashSet::new();
1486 for cte in &with.ctes {
1487 if !local_names.insert(cte.name.clone()) {
1488 return Err(PlannerError::invalid_expression(format!(
1489 "common table expression '{}' is defined more than once",
1490 cte.name
1491 )));
1492 }
1493 let mut relation = self.plan_query_body_relation(&cte.query, &[], &plans)?;
1494 if !cte.columns.is_empty() {
1495 if cte.columns.len() != relation.schema.len() {
1496 return Err(PlannerError::cte_column_count_mismatch(
1497 &cte.name,
1498 cte.columns.len(),
1499 relation.schema.len(),
1500 cte.span,
1501 ));
1502 }
1503
1504 let mut column_names = HashSet::new();
1505 for column_name in &cte.columns {
1506 if !column_names.insert(column_name) {
1507 return Err(PlannerError::duplicate_cte_column(
1508 &cte.name,
1509 column_name,
1510 cte.span,
1511 ));
1512 }
1513 }
1514
1515 for (column, column_name) in relation.schema.iter_mut().zip(&cte.columns) {
1516 column.name.clone_from(column_name);
1517 }
1518 }
1519 plans.insert(cte.name.clone(), relation);
1520 }
1521 Ok(plans)
1522 }
1523
1524 fn plan_recursive_cte(
1525 &self,
1526 with: &crate::ast::WithClause,
1527 enclosing_ctes: &CtePlans,
1528 ) -> Result<CtePlans, PlannerError> {
1529 if with.ctes.len() != 1 {
1530 return Err(PlannerError::unsupported_feature(
1531 "recursive WITH containing anything other than exactly one common table expression",
1532 "a future version",
1533 with.span,
1534 ));
1535 }
1536
1537 let cte = &with.ctes[0];
1538 let mut column_names = HashSet::new();
1539 for column_name in &cte.columns {
1540 if !column_names.insert(column_name) {
1541 return Err(PlannerError::duplicate_cte_column(
1542 &cte.name,
1543 column_name,
1544 cte.span,
1545 ));
1546 }
1547 }
1548
1549 let QueryBody::Select(body) = cte.query.as_ref() else {
1550 return Err(PlannerError::unsupported_feature(
1551 "recursive common table expression whose outer body is not SELECT",
1552 "a future version",
1553 cte.span,
1554 ));
1555 };
1556 if body.set_operations.len() != 1 {
1557 return Err(PlannerError::unsupported_feature(
1558 "recursive common table expression without exactly one UNION or UNION ALL",
1559 "a future version",
1560 cte.span,
1561 ));
1562 }
1563 if !body.order_by.is_empty()
1564 || body.limit.is_some()
1565 || body.offset.is_some()
1566 || body.limit_with_ties
1567 {
1568 return Err(PlannerError::unsupported_feature(
1569 "ORDER BY, LIMIT, OFFSET, or FETCH inside a recursive common table expression",
1570 "a future version",
1571 body.span,
1572 ));
1573 }
1574
1575 let operation = &body.set_operations[0];
1576 if operation.operator != AstSetOperator::Union {
1577 return Err(PlannerError::unsupported_feature(
1578 "recursive common table expression using an operator other than UNION or UNION ALL",
1579 "a future version",
1580 operation.span,
1581 ));
1582 }
1583
1584 let mut anchor = body.clone();
1585 anchor.with = None;
1586 anchor.set_operations.clear();
1587 if select_table_reference_count(&anchor, &cte.name) != 0 {
1588 return Err(PlannerError::unsupported_feature(
1589 "recursive common table expression without an anchor term that does not reference itself",
1590 "a future version",
1591 anchor.span,
1592 ));
1593 }
1594
1595 let QueryBody::Select(recursive_term) = operation.right.as_ref() else {
1596 return Err(PlannerError::unsupported_feature(
1597 "recursive common table expression whose recursive term is not SELECT",
1598 "a future version",
1599 operation.span,
1600 ));
1601 };
1602 if select_contains_subquery(recursive_term) {
1603 return Err(PlannerError::unsupported_feature(
1604 "subquery in a recursive term",
1605 "a future version",
1606 recursive_term.span,
1607 ));
1608 }
1609 let total_references = select_table_reference_count(recursive_term, &cte.name);
1610 let direct_references = direct_from_reference_count(&recursive_term.from, &cte.name);
1611 if total_references != 1 || direct_references != 1 {
1612 return Err(PlannerError::unsupported_feature(
1613 "recursive term without exactly one direct self-reference",
1614 "a future version",
1615 recursive_term.span,
1616 ));
1617 }
1618 if recursive_term.with.is_some() || !recursive_term.set_operations.is_empty() {
1619 return Err(PlannerError::unsupported_feature(
1620 "nested WITH or set operation in a recursive term",
1621 "a future version",
1622 recursive_term.span,
1623 ));
1624 }
1625
1626 let mut anchor_relation = self.plan_select_relation(&anchor, &[], enclosing_ctes)?;
1627 if cte.columns.is_empty() {
1628 let mut anchor_names = HashSet::new();
1629 for column in &anchor_relation.schema {
1630 if !anchor_names.insert(&column.name) {
1631 return Err(PlannerError::duplicate_cte_column(
1632 &cte.name,
1633 &column.name,
1634 cte.span,
1635 ));
1636 }
1637 }
1638 } else {
1639 if cte.columns.len() != anchor_relation.schema.len() {
1640 return Err(PlannerError::cte_column_count_mismatch(
1641 &cte.name,
1642 cte.columns.len(),
1643 anchor_relation.schema.len(),
1644 cte.span,
1645 ));
1646 }
1647 for (column, name) in anchor_relation.schema.iter_mut().zip(&cte.columns) {
1648 column.name.clone_from(name);
1649 }
1650 }
1651
1652 let mut recursive_scope = enclosing_ctes.clone();
1653 recursive_scope.insert(
1654 cte.name.clone(),
1655 PlannedRelation {
1656 plan: LogicalPlan::RecursiveReference {
1657 name: cte.name.clone(),
1658 schema: anchor_relation.schema.clone(),
1659 },
1660 schema: anchor_relation.schema.clone(),
1661 scope: vec![ScopedTable::new(
1662 TableMetadata::new(&cte.name, anchor_relation.schema.clone()),
1663 0,
1664 )],
1665 },
1666 );
1667 let recursive_relation =
1668 self.plan_select_relation(recursive_term, &[], &recursive_scope)?;
1669 if recursive_relation.schema.len() != anchor_relation.schema.len() {
1670 return Err(PlannerError::set_operation_column_count_mismatch(
1671 anchor_relation.schema.len(),
1672 recursive_relation.schema.len(),
1673 operation.span,
1674 ));
1675 }
1676 for (anchor_column, recursive_column) in anchor_relation
1677 .schema
1678 .iter()
1679 .zip(&recursive_relation.schema)
1680 {
1681 if anchor_column.data_type != recursive_column.data_type {
1682 return Err(PlannerError::type_mismatch(
1683 anchor_column.data_type.type_name(),
1684 recursive_column.data_type.type_name(),
1685 operation.span,
1686 ));
1687 }
1688 }
1689
1690 let schema = anchor_relation.schema.clone();
1691 let relation = PlannedRelation {
1692 plan: LogicalPlan::RecursiveCte {
1693 name: cte.name.clone(),
1694 anchor: Box::new(anchor_relation.plan),
1695 recursive_term: Box::new(recursive_relation.plan),
1696 union_all: operation.all,
1697 schema: schema.clone(),
1698 limits: RecursiveCteLimits::default(),
1699 },
1700 schema: schema.clone(),
1701 scope: vec![ScopedTable::new(TableMetadata::new(&cte.name, schema), 0)],
1702 };
1703 let mut plans = enclosing_ctes.clone();
1704 plans.insert(cte.name.clone(), relation);
1705 Ok(plans)
1706 }
1707
1708 fn plan_values_relation(
1709 &self,
1710 stmt: &Values,
1711 outer_scope: &[ScopedTable],
1712 enclosing_ctes: &CtePlans,
1713 ) -> Result<PlannedRelation, PlannerError> {
1714 let ctes = self.plan_with_clause(stmt.with.as_ref(), enclosing_ctes)?;
1715 let relation = self.plan_values_core(stmt, outer_scope, &ctes)?;
1716 self.apply_set_operations_and_tail(
1717 relation,
1718 &stmt.set_operations,
1719 &stmt.order_by,
1720 &stmt.limit,
1721 &stmt.offset,
1722 stmt.limit_with_ties,
1723 outer_scope,
1724 &ctes,
1725 )
1726 }
1727
1728 fn plan_values_core(
1729 &self,
1730 stmt: &Values,
1731 outer_scope: &[ScopedTable],
1732 ctes: &CtePlans,
1733 ) -> Result<PlannedRelation, PlannerError> {
1734 if stmt.rows.is_empty() {
1735 let schema = Vec::new();
1736 return Ok(PlannedRelation {
1737 plan: LogicalPlan::Values {
1738 rows: Vec::new(),
1739 schema: schema.clone(),
1740 },
1741 schema: schema.clone(),
1742 scope: vec![ScopedTable::new(
1743 TableMetadata::new(LITERAL_TABLE, schema),
1744 0,
1745 )],
1746 });
1747 }
1748
1749 let width = stmt.rows[0].len();
1750 if width == 0 {
1751 return Err(PlannerError::values_column_count_mismatch(
1752 1, 1, 0, stmt.span,
1753 ));
1754 }
1755
1756 let mut typed_rows = Vec::with_capacity(stmt.rows.len());
1757 for (row_index, row) in stmt.rows.iter().enumerate() {
1758 if row.len() != width {
1759 return Err(PlannerError::values_column_count_mismatch(
1760 row_index + 1,
1761 width,
1762 row.len(),
1763 stmt.span,
1764 ));
1765 }
1766 let mut typed_row = Vec::with_capacity(width);
1767 for expr in row {
1768 if expr_contains_subquery(expr) {
1769 return Err(PlannerError::invalid_expression(
1770 "VALUES expressions cannot contain subqueries".to_string(),
1771 ));
1772 }
1773 let typed = self.infer_expr_with_scope(expr, outer_scope, ctes)?;
1774 if typed_expr_contains_aggregate(&typed) || typed_expr_contains_window(&typed) {
1775 return Err(PlannerError::invalid_expression(
1776 "VALUES expressions must be scalar".to_string(),
1777 ));
1778 }
1779 typed_row.push(typed);
1780 }
1781 typed_rows.push(typed_row);
1782 }
1783
1784 let mut common_types = vec![ResolvedType::Null; width];
1785 for row in &typed_rows {
1786 for (column_index, value) in row.iter().enumerate() {
1787 common_types[column_index] = common_values_type(
1788 &common_types[column_index],
1789 &value.resolved_type,
1790 value.span,
1791 )?;
1792 }
1793 }
1794 for row in &mut typed_rows {
1795 for (value, target) in row.iter_mut().zip(&common_types) {
1796 if value.resolved_type != *target && value.resolved_type != ResolvedType::Null {
1797 *value = TypedExpr::cast(value.clone(), target.clone(), value.span);
1798 }
1799 }
1800 }
1801
1802 let schema = common_types
1803 .into_iter()
1804 .enumerate()
1805 .map(|(index, data_type)| {
1806 ColumnMetadata::new(format!("column{}", index + 1), data_type)
1807 })
1808 .collect::<Vec<_>>();
1809 Ok(PlannedRelation {
1810 plan: LogicalPlan::Values {
1811 rows: typed_rows,
1812 schema: schema.clone(),
1813 },
1814 schema: schema.clone(),
1815 scope: vec![ScopedTable::new(
1816 TableMetadata::new(LITERAL_TABLE, schema),
1817 0,
1818 )],
1819 })
1820 }
1821
1822 #[allow(clippy::too_many_arguments)]
1823 fn apply_set_operations_and_tail(
1824 &self,
1825 mut relation: PlannedRelation,
1826 operations: &[AstSetOperation],
1827 order_by: &[OrderByExpr],
1828 limit: &Option<Expr>,
1829 offset: &Option<Expr>,
1830 limit_with_ties: bool,
1831 outer_scope: &[ScopedTable],
1832 ctes: &CtePlans,
1833 ) -> Result<PlannedRelation, PlannerError> {
1834 for operation in operations {
1835 let right = self.plan_query_body_relation(&operation.right, outer_scope, ctes)?;
1836 if relation.schema.len() != right.schema.len() {
1837 return Err(PlannerError::set_operation_column_count_mismatch(
1838 relation.schema.len(),
1839 right.schema.len(),
1840 operation.span,
1841 ));
1842 }
1843 for (left_column, right_column) in relation.schema.iter().zip(&right.schema) {
1844 if left_column.data_type != right_column.data_type {
1845 return Err(PlannerError::type_mismatch(
1846 left_column.data_type.type_name(),
1847 right_column.data_type.type_name(),
1848 operation.span,
1849 ));
1850 }
1851 }
1852
1853 relation.plan = LogicalPlan::SetOperation {
1854 left: Box::new(relation.plan),
1855 right: Box::new(right.plan),
1856 operator: match operation.operator {
1857 AstSetOperator::Union => SetOperator::Union,
1858 AstSetOperator::Intersect => SetOperator::Intersect,
1859 AstSetOperator::Except => SetOperator::Except,
1860 },
1861 all: operation.all,
1862 };
1863 }
1864
1865 relation.scope = vec![ScopedTable::new(
1866 TableMetadata::new(LITERAL_TABLE, relation.schema.clone()),
1867 0,
1868 )];
1869 if !order_by.is_empty() {
1870 let order_by =
1871 self.build_sort_exprs_with_scope(order_by, &relation.scope, &HashMap::new(), ctes)?;
1872 relation.plan = LogicalPlan::Sort {
1873 input: Box::new(relation.plan),
1874 order_by,
1875 };
1876 }
1877 relation.plan = self.apply_pagination(relation.plan, limit, offset, limit_with_ties)?;
1878 Ok(relation)
1879 }
1880
1881 fn plan_select_relation(
1882 &self,
1883 stmt: &Select,
1884 outer_scope: &[ScopedTable],
1885 enclosing_ctes: &CtePlans,
1886 ) -> Result<PlannedRelation, PlannerError> {
1887 let resolved_stmt = resolve_named_windows(stmt)?;
1888 let stmt = &resolved_stmt;
1889 let ctes = self.plan_ctes(stmt, enclosing_ctes)?;
1890 if !stmt.set_operations.is_empty() {
1891 if !stmt.distinct_on.is_empty() {
1895 return Err(PlannerError::unsupported_feature(
1896 "DISTINCT ON combined with UNION, INTERSECT, or EXCEPT",
1897 "a future version",
1898 stmt.span,
1899 ));
1900 }
1901 let mut left_select = stmt.clone();
1902 left_select.with = None;
1903 left_select.set_operations.clear();
1904 left_select.order_by.clear();
1905 left_select.limit = None;
1906 left_select.offset = None;
1907 left_select.limit_with_ties = false;
1908 let relation = self.plan_select_relation(&left_select, outer_scope, &ctes)?;
1909
1910 return self.apply_set_operations_and_tail(
1911 relation,
1912 &stmt.set_operations,
1913 &stmt.order_by,
1914 &stmt.limit,
1915 &stmt.offset,
1916 stmt.limit_with_ties,
1917 outer_scope,
1918 &ctes,
1919 );
1920 }
1921
1922 let mut relation = self.plan_from_items(&stmt.from, stmt.span, outer_scope, &ctes)?;
1923 let expr_scope = relation
1924 .scope
1925 .iter()
1926 .cloned()
1927 .chain(offset_scope(outer_scope, relation.schema.len()))
1928 .collect::<Vec<_>>();
1929
1930 let has_group_by = stmt
1931 .group_by
1932 .as_ref()
1933 .is_some_and(|items| !items.is_empty());
1934 let has_aggregate = self.select_contains_aggregate(stmt);
1935 let has_window = select_contains_window(stmt);
1936 if stmt.qualify.is_some() && !has_window {
1937 return Err(PlannerError::invalid_expression(
1938 "QUALIFY requires at least one window function in the query block".to_string(),
1939 ));
1940 }
1941 let distinct_only =
1942 stmt.distinct && !has_group_by && !has_aggregate && stmt.having.is_none();
1943
1944 let has_distinct_on = !stmt.distinct_on.is_empty();
1945 if has_distinct_on {
1946 if stmt.distinct {
1949 return Err(PlannerError::invalid_expression(
1950 "DISTINCT and DISTINCT ON cannot be combined".to_string(),
1951 ));
1952 }
1953 for key in &stmt.distinct_on {
1955 if expr_contains_aggregate(key) || expr_contains_window(key) {
1956 return Err(PlannerError::invalid_expression(
1957 "DISTINCT ON expressions cannot contain aggregate or window functions"
1958 .to_string(),
1959 ));
1960 }
1961 if expr_contains_subquery(key) {
1962 return Err(PlannerError::invalid_expression(
1963 "DISTINCT ON expressions cannot contain subqueries".to_string(),
1964 ));
1965 }
1966 }
1967 if has_group_by || has_aggregate || stmt.having.is_some() {
1970 return Err(PlannerError::unsupported_feature(
1971 "DISTINCT ON with GROUP BY, aggregate functions, or HAVING",
1972 "a future version",
1973 stmt.span,
1974 ));
1975 }
1976 if has_window || stmt.qualify.is_some() {
1977 return Err(PlannerError::unsupported_feature(
1978 "DISTINCT ON with window functions or QUALIFY",
1979 "a future version",
1980 stmt.span,
1981 ));
1982 }
1983 }
1984
1985 if stmt.having.as_ref().is_some_and(expr_contains_window) {
1986 return Err(PlannerError::invalid_expression(
1987 "HAVING cannot contain window functions".to_string(),
1988 ));
1989 }
1990 if stmt.group_by.as_ref().is_some_and(|items| {
1991 items
1992 .iter()
1993 .flat_map(GroupByItem::exprs)
1994 .any(expr_contains_window)
1995 }) {
1996 return Err(PlannerError::invalid_expression(
1997 "GROUP BY cannot contain window functions".to_string(),
1998 ));
1999 }
2000 if stmt.group_by.as_ref().is_some_and(|items| {
2002 items
2003 .iter()
2004 .flat_map(GroupByItem::exprs)
2005 .any(expr_contains_grouping)
2006 }) {
2007 return Err(PlannerError::invalid_expression(
2008 "GROUPING is not allowed in GROUP BY".to_string(),
2009 ));
2010 }
2011 if stmt.selection.as_ref().is_some_and(expr_contains_grouping) {
2012 return Err(PlannerError::invalid_expression(
2013 "GROUPING is not allowed in WHERE".to_string(),
2014 ));
2015 }
2016 if !(has_group_by || has_aggregate || stmt.having.is_some()) {
2017 let grouping_present = stmt.projection.iter().any(|item| match item {
2018 SelectItem::Expr { expr, .. } => expr_contains_grouping(expr),
2019 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
2020 }) || stmt
2021 .order_by
2022 .iter()
2023 .any(|order| expr_contains_grouping(&order.expr))
2024 || stmt.qualify.as_ref().is_some_and(expr_contains_grouping);
2025 if grouping_present {
2026 return Err(PlannerError::invalid_expression(
2027 "GROUPING is only allowed in grouped queries".to_string(),
2028 ));
2029 }
2030 }
2031
2032 let projection_aliases = collect_projection_aliases(&stmt.projection);
2036
2037 let final_projection = self.build_projection_with_scope(
2038 &stmt.projection,
2039 &relation.schema,
2040 &expr_scope,
2041 &ctes,
2042 )?;
2043 if !has_window {
2044 install_base_projection(&mut relation.plan, &final_projection);
2045 }
2046 let needs_project_boundary = !matches!(relation.plan, LogicalPlan::Scan { .. });
2047 let base_schema = relation.schema.clone();
2048 let mut plan = relation.plan;
2049
2050 if let Some(ref selection) = stmt.selection {
2052 if expr_contains_window(selection) {
2053 return Err(PlannerError::invalid_expression(
2054 "WHERE cannot contain window functions".to_string(),
2055 ));
2056 }
2057 let predicate = self.infer_expr_with_scope(selection, &expr_scope, &ctes)?;
2058
2059 if predicate.resolved_type != ResolvedType::Boolean {
2061 return Err(PlannerError::type_mismatch(
2062 "Boolean",
2063 predicate.resolved_type.to_string(),
2064 selection.span,
2065 ));
2066 }
2067
2068 plan = LogicalPlan::Filter {
2069 input: Box::new(plan),
2070 predicate,
2071 };
2072 }
2073
2074 if has_window && (has_group_by || has_aggregate || stmt.having.is_some()) {
2075 return self.plan_grouped_window_select(
2076 stmt,
2077 &ctes,
2078 &expr_scope,
2079 &projection_aliases,
2080 plan,
2081 );
2082 }
2083
2084 if has_window {
2085 let mut windows = Vec::new();
2086 let mut window_map = HashMap::new();
2087 if let Projection::Columns(columns) = &final_projection {
2088 for column in columns {
2089 self.collect_windows_from_typed_expr(
2090 &column.expr,
2091 &mut windows,
2092 &mut window_map,
2093 )?;
2094 }
2095 }
2096
2097 let mut outer_order_by = Vec::new();
2098 for order_expr in &stmt.order_by {
2099 let sort_source =
2100 substitute_projection_aliases(&order_expr.expr, &projection_aliases);
2101 let typed = self.infer_expr_with_scope(&sort_source, &expr_scope, &ctes)?;
2102 self.collect_windows_from_typed_expr(&typed, &mut windows, &mut window_map)?;
2103 outer_order_by.push(SortExpr::new(
2104 typed,
2105 order_expr.asc.unwrap_or(true),
2106 order_expr.nulls_first.unwrap_or(false),
2107 ));
2108 }
2109
2110 let qualify = if let Some(qualify) = &stmt.qualify {
2111 let source = substitute_projection_aliases(qualify, &projection_aliases);
2112 let typed = self.infer_expr_with_scope(&source, &expr_scope, &ctes)?;
2113 if typed.resolved_type != ResolvedType::Boolean {
2114 return Err(PlannerError::type_mismatch(
2115 "Boolean",
2116 typed.resolved_type.type_name().to_string(),
2117 typed.span,
2118 ));
2119 }
2120 self.collect_windows_from_typed_expr(&typed, &mut windows, &mut window_map)?;
2121 Some(typed)
2122 } else {
2123 None
2124 };
2125
2126 let window_names = (0..windows.len())
2127 .map(|idx| format!("__window_{idx}"))
2128 .collect::<Vec<_>>();
2129 let mut window_schema = base_schema;
2130 window_schema.extend(windows.iter().enumerate().map(|(idx, window)| {
2131 ColumnMetadata::new(window_names[idx].clone(), window.result_type.clone())
2132 }));
2133
2134 let projection = rewrite_projection_for_windows(
2135 &final_projection,
2136 &window_map,
2137 relation.schema.len(),
2138 &window_names,
2139 )?;
2140 let order_by = outer_order_by
2141 .into_iter()
2142 .map(|sort| {
2143 Ok(SortExpr::new(
2144 rewrite_expr_for_windows(
2145 &sort.expr,
2146 &window_map,
2147 relation.schema.len(),
2148 &window_names,
2149 )?,
2150 sort.asc,
2151 sort.nulls_first,
2152 ))
2153 })
2154 .collect::<Result<Vec<_>, PlannerError>>()?;
2155 let qualify = qualify
2156 .as_ref()
2157 .map(|expr| {
2158 rewrite_expr_for_windows(
2159 expr,
2160 &window_map,
2161 relation.schema.len(),
2162 &window_names,
2163 )
2164 })
2165 .transpose()?;
2166
2167 return self.finish_window_select(
2168 stmt,
2169 WindowSelectStages {
2170 plan,
2171 windows,
2172 qualify,
2173 projection,
2174 order_by,
2175 window_schema,
2176 },
2177 );
2178 }
2179
2180 if has_group_by || has_aggregate || stmt.having.is_some() || stmt.distinct {
2181 if !has_group_by && !has_aggregate && stmt.having.is_some() {
2182 return Err(PlannerError::invalid_expression(
2183 "HAVING requires GROUP BY or aggregate functions".to_string(),
2184 ));
2185 }
2186
2187 let (group_keys, grouping_sets, projected) = if distinct_only {
2188 let projected = self.build_projected_columns_for_distinct_with_scope(
2189 &stmt.projection,
2190 &relation.schema,
2191 &expr_scope,
2192 &ctes,
2193 )?;
2194 let group_keys = projected.iter().map(|col| col.expr.clone()).collect();
2195 (group_keys, None, projected)
2196 } else {
2197 let expanded = self.expand_group_by_items(stmt, &expr_scope, &ctes)?;
2198 let projected = self.build_projected_columns_for_aggregate_with_scope(
2199 &stmt.projection,
2200 &expr_scope,
2201 &ctes,
2202 )?;
2203 (expanded.group_keys, expanded.grouping_sets, projected)
2204 };
2205 let mut aggregates = Vec::new();
2206 let mut agg_map = HashMap::new();
2207
2208 for col in &projected {
2209 self.collect_aggregates_from_typed_expr(&col.expr, &mut aggregates, &mut agg_map)?;
2210 }
2211
2212 let having_typed = if let Some(having) = &stmt.having {
2213 let having = substitute_projection_aliases(having, &projection_aliases);
2214 let typed = self.infer_expr_with_scope(&having, &expr_scope, &ctes)?;
2215 if typed.resolved_type != ResolvedType::Boolean {
2216 return Err(PlannerError::type_mismatch(
2217 "Boolean",
2218 typed.resolved_type.type_name().to_string(),
2219 typed.span,
2220 ));
2221 }
2222 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut agg_map)?;
2223 Some(typed)
2224 } else {
2225 None
2226 };
2227
2228 let mut order_by = Vec::new();
2229 if !stmt.order_by.is_empty() {
2230 for order_expr in &stmt.order_by {
2231 let sort_source =
2232 substitute_projection_aliases(&order_expr.expr, &projection_aliases);
2233 let typed = self.infer_expr_with_scope(&sort_source, &expr_scope, &ctes)?;
2234 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut agg_map)?;
2235 let asc = order_expr.asc.unwrap_or(true);
2236 let nulls_first = order_expr.nulls_first.unwrap_or(false);
2237 order_by.push(SortExpr::new(typed, asc, nulls_first));
2238 }
2239 }
2240
2241 if let Some(ref having) = having_typed {
2242 self.type_checker
2243 .validate_having_expr(having, &group_keys, &aggregates)?;
2244 }
2245
2246 let mut output_schema = build_aggregate_schema(&group_keys, &aggregates);
2247 let output_names: Vec<String> = output_schema.iter().map(|c| c.name.clone()).collect();
2252 let grouping_rewrite = GroupingRewrite::new(&group_keys, &aggregates, &grouping_sets);
2253 if grouping_sets.is_some() {
2254 output_schema.push(ColumnMetadata::new(
2255 GROUPING_ID_COLUMN,
2256 ResolvedType::BigInt,
2257 ));
2258 }
2259
2260 let projection = self.build_aggregate_projection(
2261 projected,
2262 &group_keys,
2263 &aggregates,
2264 &output_names,
2265 Some(&grouping_rewrite),
2266 )?;
2267
2268 let having = if let Some(having) = having_typed {
2269 Some(self.rewrite_expr_for_aggregate(
2270 &having,
2271 &group_keys,
2272 &aggregates,
2273 &output_names,
2274 Some(&grouping_rewrite),
2275 )?)
2276 } else {
2277 None
2278 };
2279
2280 let order_by = order_by
2281 .into_iter()
2282 .map(|expr| {
2283 let rewritten = self.rewrite_expr_for_aggregate(
2284 &expr.expr,
2285 &group_keys,
2286 &aggregates,
2287 &output_names,
2288 Some(&grouping_rewrite),
2289 )?;
2290 Ok(SortExpr::new(rewritten, expr.asc, expr.nulls_first))
2291 })
2292 .collect::<Result<Vec<_>, PlannerError>>()?;
2293
2294 let schema = projection_schema(&projection, &output_schema);
2295 plan = LogicalPlan::Aggregate {
2296 input: Box::new(plan),
2297 group_keys,
2298 aggregates,
2299 having,
2300 projection,
2301 grouping_sets,
2302 };
2303
2304 if !order_by.is_empty() {
2305 plan = LogicalPlan::Sort {
2306 input: Box::new(plan),
2307 order_by,
2308 };
2309 }
2310
2311 plan = self.apply_pagination(plan, &stmt.limit, &stmt.offset, stmt.limit_with_ties)?;
2312
2313 return Ok(PlannedRelation {
2314 plan,
2315 schema: schema.clone(),
2316 scope: vec![ScopedTable::new(
2317 TableMetadata::new(LITERAL_TABLE, schema),
2318 0,
2319 )],
2320 });
2321 }
2322
2323 let mut distinct_on_tie_keys: Option<Vec<SortExpr>> = None;
2325 if has_distinct_on {
2326 let mut key_exprs: Vec<TypedExpr> = Vec::new();
2333 let mut key_signatures = HashSet::new();
2334 for key in &stmt.distinct_on {
2335 let source = substitute_projection_aliases(key, &projection_aliases);
2336 let typed = self.infer_expr_with_scope(&source, &expr_scope, &ctes)?;
2337 if key_signatures.insert(distinct_on_expr_signature(&typed)) {
2338 key_exprs.push(typed);
2339 }
2340 }
2341 let user_order_by = self.build_sort_exprs_with_scope(
2342 &stmt.order_by,
2343 &expr_scope,
2344 &projection_aliases,
2345 &ctes,
2346 )?;
2347 let user_order_by_len = user_order_by.len();
2355 let (key_count, order_by) =
2356 build_distinct_on_sort_spec(key_exprs, user_order_by, &base_schema, stmt.span)?;
2357 distinct_on_tie_keys = Some(order_by[..user_order_by_len.min(order_by.len())].to_vec());
2358 plan = LogicalPlan::DistinctOn {
2359 input: Box::new(plan),
2360 key_count,
2361 order_by,
2362 };
2363 } else if !stmt.order_by.is_empty() {
2364 let order_by = self.build_sort_exprs_with_scope(
2365 &stmt.order_by,
2366 &expr_scope,
2367 &projection_aliases,
2368 &ctes,
2369 )?;
2370 plan = LogicalPlan::Sort {
2371 input: Box::new(plan),
2372 order_by,
2373 };
2374 }
2375
2376 plan = self.apply_pagination_with_tie_keys(
2377 plan,
2378 &stmt.limit,
2379 &stmt.offset,
2380 stmt.limit_with_ties,
2381 distinct_on_tie_keys.as_deref(),
2382 )?;
2383
2384 let output_schema = projection_schema(&final_projection, &relation.schema);
2385 if needs_project_boundary {
2386 plan = LogicalPlan::Project {
2387 input: Box::new(plan),
2388 projection: final_projection,
2389 };
2390 }
2391 Ok(PlannedRelation {
2392 plan,
2393 schema: output_schema.clone(),
2394 scope: vec![ScopedTable::new(
2395 TableMetadata::new(LITERAL_TABLE, output_schema),
2396 0,
2397 )],
2398 })
2399 }
2400
2401 fn plan_grouped_window_select(
2408 &self,
2409 stmt: &Select,
2410 ctes: &CtePlans,
2411 expr_scope: &[ScopedTable],
2412 projection_aliases: &HashMap<String, crate::ast::expr::Expr>,
2413 mut plan: LogicalPlan,
2414 ) -> Result<PlannedRelation, PlannerError> {
2415 let expanded = self.expand_group_by_items(stmt, expr_scope, ctes)?;
2416 let group_keys = expanded.group_keys;
2417 let grouping_sets = expanded.grouping_sets;
2418 let projected = self.build_projected_columns_for_aggregate_with_scope(
2419 &stmt.projection,
2420 expr_scope,
2421 ctes,
2422 )?;
2423 let mut aggregates = Vec::new();
2424 let mut aggregate_map = HashMap::new();
2425 for column in &projected {
2426 self.collect_aggregates_from_typed_expr(
2427 &column.expr,
2428 &mut aggregates,
2429 &mut aggregate_map,
2430 )?;
2431 }
2432
2433 let having_typed = if let Some(having) = &stmt.having {
2434 let having = substitute_projection_aliases(having, projection_aliases);
2435 if expr_contains_window(&having) {
2436 return Err(PlannerError::invalid_expression(
2437 "HAVING cannot contain window functions".to_string(),
2438 ));
2439 }
2440 let typed = self.infer_expr_with_scope(&having, expr_scope, ctes)?;
2441 if typed.resolved_type != ResolvedType::Boolean {
2442 return Err(PlannerError::type_mismatch(
2443 "Boolean",
2444 typed.resolved_type.type_name().to_string(),
2445 typed.span,
2446 ));
2447 }
2448 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut aggregate_map)?;
2449 Some(typed)
2450 } else {
2451 None
2452 };
2453
2454 let qualify_typed = if let Some(qualify) = &stmt.qualify {
2455 let qualify = substitute_projection_aliases(qualify, projection_aliases);
2456 let typed = self.infer_expr_with_scope(&qualify, expr_scope, ctes)?;
2457 if typed.resolved_type != ResolvedType::Boolean {
2458 return Err(PlannerError::type_mismatch(
2459 "Boolean",
2460 typed.resolved_type.type_name().to_string(),
2461 typed.span,
2462 ));
2463 }
2464 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut aggregate_map)?;
2465 Some(typed)
2466 } else {
2467 None
2468 };
2469
2470 let mut outer_order_by = Vec::new();
2471 for order_expr in &stmt.order_by {
2472 let source = substitute_projection_aliases(&order_expr.expr, projection_aliases);
2473 let typed = self.infer_expr_with_scope(&source, expr_scope, ctes)?;
2474 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut aggregate_map)?;
2475 outer_order_by.push(SortExpr::new(
2476 typed,
2477 order_expr.asc.unwrap_or(true),
2478 order_expr.nulls_first.unwrap_or(false),
2479 ));
2480 }
2481
2482 if let Some(having) = &having_typed {
2483 self.type_checker
2484 .validate_having_expr(having, &group_keys, &aggregates)?;
2485 }
2486
2487 let mut aggregate_schema = build_aggregate_schema(&group_keys, &aggregates);
2488 let rewrite_names = aggregate_schema
2491 .iter()
2492 .map(|column| column.name.clone())
2493 .collect::<Vec<_>>();
2494 let grouping_rewrite = GroupingRewrite::new(&group_keys, &aggregates, &grouping_sets);
2495 if grouping_sets.is_some() {
2496 aggregate_schema.push(ColumnMetadata::new(
2497 GROUPING_ID_COLUMN,
2498 ResolvedType::BigInt,
2499 ));
2500 }
2501 let aggregate_names = aggregate_schema
2502 .iter()
2503 .map(|column| column.name.clone())
2504 .collect::<Vec<_>>();
2505 let projection = self.build_aggregate_projection(
2506 projected,
2507 &group_keys,
2508 &aggregates,
2509 &rewrite_names,
2510 Some(&grouping_rewrite),
2511 )?;
2512 let having = having_typed
2513 .as_ref()
2514 .map(|expr| {
2515 self.rewrite_expr_for_aggregate(
2516 expr,
2517 &group_keys,
2518 &aggregates,
2519 &rewrite_names,
2520 Some(&grouping_rewrite),
2521 )
2522 })
2523 .transpose()?;
2524 let outer_order_by = outer_order_by
2525 .into_iter()
2526 .map(|sort| {
2527 Ok(SortExpr::new(
2528 self.rewrite_expr_for_aggregate(
2529 &sort.expr,
2530 &group_keys,
2531 &aggregates,
2532 &rewrite_names,
2533 Some(&grouping_rewrite),
2534 )?,
2535 sort.asc,
2536 sort.nulls_first,
2537 ))
2538 })
2539 .collect::<Result<Vec<_>, PlannerError>>()?;
2540 let qualify = qualify_typed
2541 .as_ref()
2542 .map(|expr| {
2543 self.rewrite_expr_for_aggregate(
2544 expr,
2545 &group_keys,
2546 &aggregates,
2547 &rewrite_names,
2548 Some(&grouping_rewrite),
2549 )
2550 })
2551 .transpose()?;
2552
2553 plan = LogicalPlan::Aggregate {
2554 input: Box::new(plan),
2555 group_keys,
2556 aggregates,
2557 having,
2558 projection: Projection::All(aggregate_names),
2559 grouping_sets,
2560 };
2561
2562 let mut windows = Vec::new();
2563 let mut window_map = HashMap::new();
2564 if let Projection::Columns(columns) = &projection {
2565 for column in columns {
2566 self.collect_windows_from_typed_expr(&column.expr, &mut windows, &mut window_map)?;
2567 }
2568 }
2569 for sort in &outer_order_by {
2570 self.collect_windows_from_typed_expr(&sort.expr, &mut windows, &mut window_map)?;
2571 }
2572 if let Some(qualify) = &qualify {
2573 self.collect_windows_from_typed_expr(qualify, &mut windows, &mut window_map)?;
2574 }
2575
2576 let window_names = (0..windows.len())
2577 .map(|index| format!("__window_{index}"))
2578 .collect::<Vec<_>>();
2579 let mut window_schema = aggregate_schema;
2580 window_schema.extend(windows.iter().enumerate().map(|(index, window)| {
2581 ColumnMetadata::new(window_names[index].clone(), window.result_type.clone())
2582 }));
2583 let projection = rewrite_projection_for_windows(
2584 &projection,
2585 &window_map,
2586 window_schema.len() - windows.len(),
2587 &window_names,
2588 )?;
2589 let outer_order_by = outer_order_by
2590 .into_iter()
2591 .map(|sort| {
2592 Ok(SortExpr::new(
2593 rewrite_expr_for_windows(
2594 &sort.expr,
2595 &window_map,
2596 window_schema.len() - windows.len(),
2597 &window_names,
2598 )?,
2599 sort.asc,
2600 sort.nulls_first,
2601 ))
2602 })
2603 .collect::<Result<Vec<_>, PlannerError>>()?;
2604 let qualify = qualify
2605 .as_ref()
2606 .map(|expr| {
2607 rewrite_expr_for_windows(
2608 expr,
2609 &window_map,
2610 window_schema.len() - windows.len(),
2611 &window_names,
2612 )
2613 })
2614 .transpose()?;
2615
2616 self.finish_window_select(
2617 stmt,
2618 WindowSelectStages {
2619 plan,
2620 windows,
2621 qualify,
2622 projection,
2623 order_by: outer_order_by,
2624 window_schema,
2625 },
2626 )
2627 }
2628
2629 fn finish_window_select(
2631 &self,
2632 stmt: &Select,
2633 stages: WindowSelectStages,
2634 ) -> Result<PlannedRelation, PlannerError> {
2635 let WindowSelectStages {
2636 mut plan,
2637 windows,
2638 qualify,
2639 projection,
2640 order_by,
2641 window_schema,
2642 } = stages;
2643 plan = LogicalPlan::Window {
2644 input: Box::new(plan),
2645 windows,
2646 };
2647 if let Some(predicate) = qualify {
2648 plan = LogicalPlan::Filter {
2649 input: Box::new(plan),
2650 predicate,
2651 };
2652 }
2653
2654 let output_schema = projection_schema(&projection, &window_schema);
2655 let mut hidden_order_keys = Vec::new();
2656 let mut projected_order_by = Vec::with_capacity(order_by.len());
2657 for sort in order_by {
2658 let expr =
2659 match rewrite_expr_for_projected_output(&sort.expr, &projection, &output_schema) {
2660 Ok(expr) => expr,
2661 Err(_) if !stmt.distinct => {
2662 let hidden_index = hidden_order_keys.len();
2663 let name = format!("__order_{hidden_index}");
2664 let output_index = output_schema.len() + hidden_index;
2665 hidden_order_keys.push(ProjectedColumn {
2666 expr: sort.expr.clone(),
2667 alias: Some(name.clone()),
2668 });
2669 TypedExpr::column_ref(
2670 "__project__".to_string(),
2671 name,
2672 output_index,
2673 sort.expr.resolved_type.clone(),
2674 sort.expr.span,
2675 )
2676 }
2677 Err(error) => return Err(error),
2678 };
2679 projected_order_by.push(SortExpr::new(expr, sort.asc, sort.nulls_first));
2680 }
2681 let has_hidden_order_keys = !hidden_order_keys.is_empty();
2682 let projection = if has_hidden_order_keys {
2683 let mut columns = match projection {
2684 Projection::Columns(columns) => columns,
2685 Projection::All(_) => output_schema
2686 .iter()
2687 .enumerate()
2688 .map(|(index, column)| ProjectedColumn {
2689 expr: TypedExpr::column_ref(
2690 "__window__".to_string(),
2691 column.name.clone(),
2692 index,
2693 column.data_type.clone(),
2694 stmt.span,
2695 ),
2696 alias: None,
2697 })
2698 .collect(),
2699 };
2700 columns.extend(hidden_order_keys);
2701 Projection::Columns(columns)
2702 } else {
2703 projection
2704 };
2705 plan = LogicalPlan::Project {
2706 input: Box::new(plan),
2707 projection,
2708 };
2709
2710 if stmt.distinct {
2711 let group_keys = output_schema
2712 .iter()
2713 .enumerate()
2714 .map(|(index, column)| {
2715 TypedExpr::column_ref(
2716 LITERAL_TABLE.to_string(),
2717 column.name.clone(),
2718 index,
2719 column.data_type.clone(),
2720 stmt.span,
2721 )
2722 })
2723 .collect();
2724 plan = LogicalPlan::Aggregate {
2725 input: Box::new(plan),
2726 group_keys,
2727 aggregates: Vec::new(),
2728 having: None,
2729 projection: Projection::All(
2730 output_schema
2731 .iter()
2732 .map(|column| column.name.clone())
2733 .collect(),
2734 ),
2735 grouping_sets: None,
2736 };
2737 }
2738 if !projected_order_by.is_empty() {
2739 plan = LogicalPlan::Sort {
2740 input: Box::new(plan),
2741 order_by: projected_order_by,
2742 };
2743 }
2744 plan = self.apply_pagination(plan, &stmt.limit, &stmt.offset, stmt.limit_with_ties)?;
2745 if has_hidden_order_keys {
2746 plan = LogicalPlan::Project {
2747 input: Box::new(plan),
2748 projection: Projection::Columns(
2749 output_schema
2750 .iter()
2751 .enumerate()
2752 .map(|(index, column)| ProjectedColumn {
2753 expr: TypedExpr::column_ref(
2754 "__ordered__".to_string(),
2755 column.name.clone(),
2756 index,
2757 column.data_type.clone(),
2758 stmt.span,
2759 ),
2760 alias: Some(column.name.clone()),
2761 })
2762 .collect(),
2763 ),
2764 };
2765 }
2766
2767 Ok(PlannedRelation {
2768 plan,
2769 schema: output_schema.clone(),
2770 scope: vec![ScopedTable::new(
2771 TableMetadata::new(LITERAL_TABLE, output_schema),
2772 0,
2773 )],
2774 })
2775 }
2776
2777 fn plan_from_items(
2781 &self,
2782 items: &[FromItem],
2783 select_span: crate::ast::Span,
2784 outer_scope: &[ScopedTable],
2785 ctes: &CtePlans,
2786 ) -> Result<PlannedRelation, PlannerError> {
2787 match items {
2788 [] => {
2789 let schema = Vec::new();
2790 Ok(PlannedRelation {
2791 plan: LogicalPlan::Scan {
2792 table: LITERAL_TABLE.to_string(),
2793 projection: Projection::All(Vec::new()),
2794 },
2795 schema: schema.clone(),
2796 scope: vec![ScopedTable::new(
2797 TableMetadata::new(LITERAL_TABLE, schema),
2798 0,
2799 )],
2800 })
2801 }
2802 [single] => self.plan_from_item(single, 0, outer_scope, outer_scope, ctes),
2803 [first, rest @ ..] => {
2804 let mut relation = self.plan_from_item(first, 0, outer_scope, outer_scope, ctes)?;
2805 for item in rest {
2806 let left_width = relation.schema.len();
2809 let lateral_scope =
2810 lateral_outer_scope(&relation.scope, 0, left_width, outer_scope);
2811 let right =
2812 self.plan_from_item(item, left_width, outer_scope, &lateral_scope, ctes)?;
2813 relation = if from_item_is_lateral(item) {
2814 self.combine_lateral_join_relation(
2815 relation,
2816 right,
2817 JoinType::Cross,
2818 None,
2819 None,
2820 select_span,
2821 )?
2822 } else {
2823 self.combine_join_relation(
2824 relation,
2825 right,
2826 JoinType::Cross,
2827 None,
2828 None,
2829 select_span,
2830 )?
2831 };
2832 }
2833 Ok(relation)
2834 }
2835 }
2836 }
2837
2838 fn plan_from_item(
2839 &self,
2840 item: &FromItem,
2841 start_index: usize,
2842 outer_scope: &[ScopedTable],
2843 lateral_scope: &[ScopedTable],
2844 ctes: &CtePlans,
2845 ) -> Result<PlannedRelation, PlannerError> {
2846 match item {
2847 FromItem::Table {
2848 name,
2849 alias,
2850 columns,
2851 span,
2852 } => {
2853 if let Some(cte) = ctes.get(name) {
2854 let mut relation = cte.clone();
2855 relation.plan = LogicalPlan::Project {
2856 input: Box::new(relation.plan),
2857 projection: Projection::All(
2858 relation.schema.iter().map(|col| col.name.clone()).collect(),
2859 ),
2860 };
2861 let relation_name = alias.clone().unwrap_or_else(|| name.clone());
2862 apply_alias_columns(&relation_name, columns, &mut relation.schema, *span)?;
2863 relation.scope = vec![ScopedTable::new(
2864 TableMetadata::new(relation_name, relation.schema.clone()),
2865 start_index,
2866 )];
2867 return Ok(relation);
2868 }
2869 let table = self.name_resolver.resolve_table(name, *span)?.clone();
2870 let mut scope_table = table.clone();
2871 if let Some(alias) = alias {
2872 scope_table.name = alias.clone();
2873 }
2874 let mut schema = table.columns.clone();
2875 let projection =
2878 Projection::All(schema.iter().map(|col| col.name.clone()).collect());
2879 apply_alias_columns(&scope_table.name, columns, &mut schema, *span)?;
2880 scope_table.columns = schema.clone();
2881 Ok(PlannedRelation {
2882 plan: LogicalPlan::Scan {
2883 table: name.clone(),
2884 projection,
2885 },
2886 schema,
2887 scope: vec![ScopedTable::new(scope_table, start_index)],
2888 })
2889 }
2890 FromItem::Function {
2891 name,
2892 args,
2893 alias,
2894 columns,
2895 span,
2896 ..
2897 } => self.plan_table_function(
2898 name,
2899 args,
2900 alias.as_deref(),
2901 columns,
2902 *span,
2903 start_index,
2904 lateral_scope,
2905 ctes,
2906 ),
2907 FromItem::Join {
2908 left,
2909 right,
2910 join_type,
2911 condition,
2912 using,
2913 natural,
2914 span,
2915 } => {
2916 let right_lateral = from_item_is_lateral(right);
2917 if right_lateral {
2918 let unsupported = match join_type {
2921 crate::ast::dml::JoinType::Right => Some("RIGHT"),
2922 crate::ast::dml::JoinType::Full => Some("FULL"),
2923 _ => None,
2924 };
2925 if let Some(kind) = unsupported {
2926 return Err(PlannerError::lateral_join_type_unsupported(kind, *span));
2927 }
2928 }
2929 let left_relation =
2930 self.plan_from_item(left, start_index, outer_scope, lateral_scope, ctes)?;
2931 let right_lateral_scope = right_lateral.then(|| {
2936 lateral_outer_scope(
2937 &left_relation.scope,
2938 start_index,
2939 left_relation.schema.len(),
2940 outer_scope,
2941 )
2942 });
2943 let right_relation = self.plan_from_item(
2944 right,
2945 start_index + left_relation.schema.len(),
2946 outer_scope,
2947 right_lateral_scope.as_deref().unwrap_or(lateral_scope),
2948 ctes,
2949 )?;
2950 let expr_scope = left_relation
2951 .scope
2952 .iter()
2953 .cloned()
2954 .chain(right_relation.scope.iter().cloned())
2955 .chain(offset_scope(
2956 outer_scope,
2957 left_relation.schema.len() + right_relation.schema.len(),
2958 ))
2959 .collect::<Vec<_>>();
2960 let using = if *natural {
2961 Some(natural_join_columns(
2962 &left_relation.schema,
2963 &right_relation.schema,
2964 ))
2965 } else {
2966 using.clone()
2967 };
2968 let typed_condition = if let Some(expr) = condition {
2969 let typed = self.infer_expr_with_scope(expr, &expr_scope, ctes)?;
2970 if typed.resolved_type != ResolvedType::Boolean {
2971 return Err(PlannerError::type_mismatch(
2972 "Boolean",
2973 typed.resolved_type.to_string(),
2974 expr.span,
2975 ));
2976 }
2977 Some(typed)
2978 } else {
2979 self.build_using_condition(
2980 using.as_deref(),
2981 &left_relation,
2982 &right_relation,
2983 *span,
2984 )?
2985 };
2986 if right_lateral {
2987 self.combine_lateral_join_relation(
2988 left_relation,
2989 right_relation,
2990 map_join_type(*join_type),
2991 typed_condition,
2992 using.as_deref(),
2993 *span,
2994 )
2995 } else {
2996 self.combine_join_relation(
2997 left_relation,
2998 right_relation,
2999 map_join_type(*join_type),
3000 typed_condition,
3001 using,
3002 *span,
3003 )
3004 }
3005 }
3006 FromItem::Derived {
3007 subquery,
3008 alias,
3009 columns,
3010 lateral,
3011 span,
3012 } => {
3013 let visible: &[ScopedTable] = if *lateral { lateral_scope } else { &[] };
3019 let mut relation = self.plan_query_body_relation(subquery, visible, ctes)?;
3020 let alias = alias.clone().ok_or_else(|| {
3021 PlannerError::invalid_expression("derived table requires an alias".to_string())
3022 })?;
3023 let source_names = relation
3024 .schema
3025 .iter()
3026 .map(|column| column.name.clone())
3027 .collect();
3028 relation.plan = LogicalPlan::Project {
3029 input: Box::new(relation.plan),
3030 projection: Projection::All(source_names),
3031 };
3032 apply_alias_columns(&alias, columns, &mut relation.schema, *span)?;
3033 relation.scope = vec![ScopedTable::new(
3034 TableMetadata::new(alias, relation.schema.clone()),
3035 start_index,
3036 )];
3037 Ok(relation)
3038 }
3039 }
3040 }
3041
3042 #[allow(clippy::too_many_arguments)]
3050 fn plan_table_function(
3051 &self,
3052 name: &str,
3053 args: &[Expr],
3054 alias: Option<&str>,
3055 columns: &[String],
3056 span: crate::ast::Span,
3057 start_index: usize,
3058 lateral_scope: &[ScopedTable],
3059 ctes: &CtePlans,
3060 ) -> Result<PlannedRelation, PlannerError> {
3061 let Some(function) = TableFunctionKind::from_name(name) else {
3062 return Err(PlannerError::unknown_table_function(name, span));
3063 };
3064 if function == TableFunctionKind::GenerateSeries {
3065 return Err(PlannerError::unsupported_feature(
3066 "table function GENERATE_SERIES",
3067 "a future version (issue #157)",
3068 span,
3069 ));
3070 }
3071
3072 if args.len() != 1 {
3073 return Err(PlannerError::invalid_expression(format!(
3074 "table function UNNEST takes exactly 1 argument, found {}",
3075 args.len()
3076 )));
3077 }
3078 let typed_arg = self.infer_expr_with_scope(&args[0], lateral_scope, ctes)?;
3079 if !matches!(
3080 typed_arg.resolved_type,
3081 ResolvedType::Vector { .. } | ResolvedType::Null
3082 ) {
3083 return Err(PlannerError::type_mismatch(
3084 "VECTOR",
3085 typed_arg.resolved_type.to_string(),
3086 args[0].span,
3087 ));
3088 }
3089
3090 let mut schema = vec![ColumnMetadata::new(
3093 function.default_relation_name(),
3094 ResolvedType::Float,
3095 )];
3096 let relation_name = alias
3097 .map(str::to_string)
3098 .unwrap_or_else(|| function.default_relation_name().to_string());
3099 apply_alias_columns(&relation_name, columns, &mut schema, span)?;
3100
3101 Ok(PlannedRelation {
3102 plan: LogicalPlan::TableFunction {
3103 function,
3104 args: vec![typed_arg],
3105 schema: schema.clone(),
3106 },
3107 schema: schema.clone(),
3108 scope: vec![ScopedTable::new(
3109 TableMetadata::new(relation_name, schema),
3110 start_index,
3111 )],
3112 })
3113 }
3114
3115 fn combine_lateral_join_relation(
3116 &self,
3117 left: PlannedRelation,
3118 right: PlannedRelation,
3119 join_type: JoinType,
3120 condition: Option<TypedExpr>,
3121 using: Option<&[String]>,
3122 _span: crate::ast::Span,
3123 ) -> Result<PlannedRelation, PlannerError> {
3124 let (schema, scope) = combine_join_shape(&left, &right, using);
3129 Ok(PlannedRelation {
3130 plan: LogicalPlan::LateralJoin {
3131 left: Box::new(left.plan),
3132 right: Box::new(right.plan),
3133 join_type,
3134 condition,
3135 right_schema: right.schema,
3136 },
3137 schema,
3138 scope,
3139 })
3140 }
3141
3142 fn combine_join_relation(
3143 &self,
3144 left: PlannedRelation,
3145 right: PlannedRelation,
3146 join_type: JoinType,
3147 condition: Option<TypedExpr>,
3148 using: Option<Vec<String>>,
3149 _span: crate::ast::Span,
3150 ) -> Result<PlannedRelation, PlannerError> {
3151 let (schema, scope) = combine_join_shape(&left, &right, using.as_deref());
3152 Ok(PlannedRelation {
3153 plan: LogicalPlan::Join {
3154 left: Box::new(left.plan),
3155 right: Box::new(right.plan),
3156 join_type,
3157 condition,
3158 using,
3159 },
3160 schema,
3161 scope,
3162 })
3163 }
3164
3165 fn build_using_condition(
3166 &self,
3167 using: Option<&[String]>,
3168 left: &PlannedRelation,
3169 right: &PlannedRelation,
3170 span: crate::ast::Span,
3171 ) -> Result<Option<TypedExpr>, PlannerError> {
3172 let Some(columns) = using else {
3173 return Ok(None);
3174 };
3175 let mut condition = None;
3176 for column in columns {
3177 let left_col = find_scoped_column(&left.scope, column, span)?;
3178 let right_col = find_scoped_column(&right.scope, column, span)?;
3179 let left_expr = merged_scoped_column_expr(&left_col, column, span);
3180 let right_expr = merged_scoped_column_expr(&right_col, column, span);
3181 self.type_checker
3182 .check_comparison_op(&left_col.ty, &right_col.ty, span)?;
3183 let eq = TypedExpr::binary_op(
3184 left_expr,
3185 crate::ast::expr::BinaryOp::Eq,
3186 right_expr,
3187 ResolvedType::Boolean,
3188 span,
3189 );
3190 condition = Some(match condition {
3191 Some(prev) => TypedExpr::binary_op(
3192 prev,
3193 crate::ast::expr::BinaryOp::And,
3194 eq,
3195 ResolvedType::Boolean,
3196 span,
3197 ),
3198 None => eq,
3199 });
3200 }
3201 Ok(condition)
3202 }
3203
3204 fn infer_expr_with_scope(
3205 &self,
3206 expr: &crate::ast::expr::Expr,
3207 scope: &[ScopedTable],
3208 ctes: &CtePlans,
3209 ) -> Result<TypedExpr, PlannerError> {
3210 self.type_checker
3211 .infer_type_with_scope(expr, scope, &|stmt, outer_scope| {
3212 let relation = match &stmt.kind {
3213 StatementKind::Select(select) => {
3214 self.plan_select_relation(select, outer_scope, ctes)?
3215 }
3216 StatementKind::Values(values) => {
3217 self.plan_values_relation(values, outer_scope, ctes)?
3218 }
3219 _ => {
3220 return Err(PlannerError::unsupported_feature(
3221 "non-query subquery",
3222 "a future version",
3223 stmt.span(),
3224 ));
3225 }
3226 };
3227 Ok((relation.plan, relation.schema))
3228 })
3229 }
3230
3231 #[allow(dead_code)]
3232 fn build_projection(
3233 &self,
3234 items: &[SelectItem],
3235 table: &TableMetadata,
3236 ) -> Result<Projection, PlannerError> {
3237 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
3239 let columns = self.name_resolver.expand_wildcard(table);
3240 return Ok(Projection::All(columns));
3241 }
3242
3243 let mut projected_columns = Vec::new();
3245 for item in items {
3246 match item {
3247 SelectItem::Wildcard { span } => {
3248 for col in &table.columns {
3250 let column_index = table.get_column_index(&col.name).unwrap();
3251 let typed_expr = TypedExpr::column_ref(
3252 table.name.clone(),
3253 col.name.clone(),
3254 column_index,
3255 col.data_type.clone(),
3256 *span,
3257 );
3258 projected_columns.push(ProjectedColumn::new(typed_expr));
3259 }
3260 }
3261 SelectItem::QualifiedWildcard {
3262 table: qualifier,
3263 span,
3264 } => {
3265 if qualifier != &table.name {
3266 return Err(PlannerError::invalid_expression(format!(
3267 "table '{qualifier}' is not available for wildcard projection"
3268 )));
3269 }
3270 for col in &table.columns {
3271 let column_index = table.get_column_index(&col.name).unwrap();
3272 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
3273 table.name.clone(),
3274 col.name.clone(),
3275 column_index,
3276 col.data_type.clone(),
3277 *span,
3278 )));
3279 }
3280 }
3281 SelectItem::Expr { expr, alias, .. } => {
3282 let typed_expr = self.type_checker.infer_type(expr, table)?;
3283 let projected = if let Some(alias) = alias {
3284 ProjectedColumn::with_alias(typed_expr, alias.clone())
3285 } else {
3286 ProjectedColumn::new(typed_expr)
3287 };
3288 projected_columns.push(projected);
3289 }
3290 }
3291 }
3292
3293 Ok(Projection::Columns(projected_columns))
3294 }
3295
3296 fn build_projection_with_scope(
3297 &self,
3298 items: &[SelectItem],
3299 schema: &[ColumnMetadata],
3300 scope: &[ScopedTable],
3301 ctes: &CtePlans,
3302 ) -> Result<Projection, PlannerError> {
3303 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
3304 return Ok(Projection::All(visible_wildcard_columns(schema, scope)));
3305 }
3306
3307 let mut projected_columns = Vec::new();
3308 for item in items {
3309 match item {
3310 SelectItem::Wildcard { span } => {
3311 for scoped in scope {
3312 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
3313 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
3314 scoped.table.name.clone(),
3315 col.name.clone(),
3316 scoped.start_index + local_idx,
3317 col.data_type.clone(),
3318 *span,
3319 )));
3320 }
3321 }
3322 }
3323 SelectItem::QualifiedWildcard { table, span } => {
3324 let scoped = scope
3325 .iter()
3326 .filter(|scoped| scoped.table.name == *table)
3327 .collect::<Vec<_>>();
3328 match scoped.as_slice() {
3329 [] => {
3330 return Err(PlannerError::invalid_expression(format!(
3331 "table '{table}' is not available for wildcard projection"
3332 )));
3333 }
3334 [scoped] => {
3335 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
3336 projected_columns.push(ProjectedColumn::new(
3337 TypedExpr::column_ref(
3338 scoped.table.name.clone(),
3339 col.name.clone(),
3340 scoped.start_index + local_idx,
3341 col.data_type.clone(),
3342 *span,
3343 ),
3344 ));
3345 }
3346 }
3347 _ => {
3348 return Err(PlannerError::ambiguous_column(
3349 table,
3350 scoped
3351 .iter()
3352 .map(|scoped| scoped.table.name.clone())
3353 .collect(),
3354 *span,
3355 ));
3356 }
3357 }
3358 }
3359 SelectItem::Expr { expr, alias, .. } => {
3360 let typed_expr = self.infer_expr_with_scope(expr, scope, ctes)?;
3361 let projected = if let Some(alias) = alias {
3362 ProjectedColumn::with_alias(typed_expr, alias.clone())
3363 } else {
3364 ProjectedColumn::new(typed_expr)
3365 };
3366 projected_columns.push(projected);
3367 }
3368 }
3369 }
3370
3371 Ok(Projection::Columns(projected_columns))
3372 }
3373
3374 #[allow(dead_code)]
3376 fn build_sort_exprs(
3377 &self,
3378 order_by: &[OrderByExpr],
3379 table: &TableMetadata,
3380 ) -> Result<Vec<SortExpr>, PlannerError> {
3381 let mut sort_exprs = Vec::new();
3382
3383 for order_expr in order_by {
3384 let typed_expr = self.type_checker.infer_type(&order_expr.expr, table)?;
3385
3386 let asc = order_expr.asc.unwrap_or(true);
3388
3389 let nulls_first = order_expr.nulls_first.unwrap_or(false);
3391
3392 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
3393 }
3394
3395 Ok(sort_exprs)
3396 }
3397
3398 fn build_sort_exprs_with_scope(
3399 &self,
3400 order_by: &[OrderByExpr],
3401 scope: &[ScopedTable],
3402 projection_aliases: &HashMap<String, crate::ast::expr::Expr>,
3403 ctes: &CtePlans,
3404 ) -> Result<Vec<SortExpr>, PlannerError> {
3405 let mut sort_exprs = Vec::new();
3406 for order_expr in order_by {
3407 let sort_source = substitute_projection_aliases(&order_expr.expr, projection_aliases);
3408 let typed_expr = self.infer_expr_with_scope(&sort_source, scope, ctes)?;
3409 let asc = order_expr.asc.unwrap_or(true);
3410 let nulls_first = order_expr.nulls_first.unwrap_or(false);
3411 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
3412 }
3413 Ok(sort_exprs)
3414 }
3415
3416 fn select_contains_aggregate(&self, stmt: &Select) -> bool {
3417 stmt.projection.iter().any(|item| match item {
3418 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
3419 SelectItem::Expr { expr, .. } => expr_contains_aggregate(expr),
3420 }) || stmt
3421 .group_by
3422 .as_ref()
3423 .map(|items| {
3424 items
3425 .iter()
3426 .flat_map(GroupByItem::exprs)
3427 .any(expr_contains_aggregate)
3428 })
3429 .unwrap_or(false)
3430 || stmt
3431 .having
3432 .as_ref()
3433 .map(expr_contains_aggregate)
3434 .unwrap_or(false)
3435 || stmt
3436 .qualify
3437 .as_ref()
3438 .map(expr_contains_aggregate)
3439 .unwrap_or(false)
3440 || stmt
3441 .order_by
3442 .iter()
3443 .any(|order| expr_contains_aggregate(&order.expr))
3444 }
3445
3446 #[allow(dead_code)]
3447 fn build_group_keys(
3448 &self,
3449 stmt: &Select,
3450 table: &TableMetadata,
3451 ) -> Result<Vec<TypedExpr>, PlannerError> {
3452 let mut keys = Vec::new();
3453 if let Some(items) = &stmt.group_by {
3454 for expr in items.iter().flat_map(GroupByItem::exprs) {
3455 let typed = self.type_checker.infer_type(expr, table)?;
3456 if typed_expr_contains_aggregate(&typed) {
3457 return Err(PlannerError::invalid_expression(
3458 "GROUP BY cannot contain aggregate functions".to_string(),
3459 ));
3460 }
3461 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
3462 return Err(PlannerError::invalid_expression(
3463 "GROUP BY expressions must be column references".to_string(),
3464 ));
3465 }
3466 keys.push(typed);
3467 }
3468 }
3469 Ok(keys)
3470 }
3471
3472 fn type_group_key_with_scope(
3474 &self,
3475 expr: &Expr,
3476 scope: &[ScopedTable],
3477 ctes: &CtePlans,
3478 ) -> Result<TypedExpr, PlannerError> {
3479 let typed = self.infer_expr_with_scope(expr, scope, ctes)?;
3480 if typed_expr_contains_aggregate(&typed) {
3481 return Err(PlannerError::invalid_expression(
3482 "GROUP BY cannot contain aggregate functions".to_string(),
3483 ));
3484 }
3485 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
3486 return Err(PlannerError::invalid_expression(
3487 "GROUP BY expressions must be column references".to_string(),
3488 ));
3489 }
3490 Ok(typed)
3491 }
3492
3493 fn expand_group_by_items(
3501 &self,
3502 stmt: &Select,
3503 scope: &[ScopedTable],
3504 ctes: &CtePlans,
3505 ) -> Result<ExpandedGroupBy, PlannerError> {
3506 let Some(items) = &stmt.group_by else {
3507 return Ok(ExpandedGroupBy {
3508 group_keys: Vec::new(),
3509 grouping_sets: None,
3510 });
3511 };
3512
3513 if items
3514 .iter()
3515 .all(|item| matches!(item, GroupByItem::Expr { .. }))
3516 {
3517 let mut keys = Vec::new();
3519 for item in items {
3520 if let GroupByItem::Expr { expr } = item {
3521 keys.push(self.type_group_key_with_scope(expr, scope, ctes)?);
3522 }
3523 }
3524 return Ok(ExpandedGroupBy {
3525 group_keys: keys,
3526 grouping_sets: None,
3527 });
3528 }
3529
3530 let mut keys: Vec<TypedExpr> = Vec::new();
3531 let mut key_index: HashMap<String, usize> = HashMap::new();
3532 let mut add_key = |planner: &Self, expr: &Expr| -> Result<usize, PlannerError> {
3533 let typed = planner.type_group_key_with_scope(expr, scope, ctes)?;
3534 let signature = expr_key(&typed);
3535 if let Some(&index) = key_index.get(&signature) {
3536 return Ok(index);
3537 }
3538 let index = keys.len();
3539 keys.push(typed);
3540 key_index.insert(signature, index);
3541 Ok(index)
3542 };
3543
3544 let mut sets: Vec<Vec<usize>> = vec![Vec::new()];
3547 for item in items {
3548 let item_sets: Vec<Vec<usize>> = match item {
3549 GroupByItem::Expr { expr } => vec![vec![add_key(self, expr)?]],
3550 GroupByItem::Rollup { exprs } => {
3551 if exprs.is_empty() {
3552 return Err(PlannerError::invalid_expression(
3553 "ROLLUP requires at least one expression".to_string(),
3554 ));
3555 }
3556 let indexes = exprs
3557 .iter()
3558 .map(|expr| add_key(self, expr))
3559 .collect::<Result<Vec<_>, _>>()?;
3560 (0..=indexes.len())
3561 .rev()
3562 .map(|len| indexes[..len].to_vec())
3563 .collect()
3564 }
3565 GroupByItem::Cube { exprs } => {
3566 if exprs.is_empty() {
3567 return Err(PlannerError::invalid_expression(
3568 "CUBE requires at least one expression".to_string(),
3569 ));
3570 }
3571 if exprs.len() > MAX_CUBE_COLUMNS {
3572 return Err(PlannerError::invalid_expression(format!(
3573 "too many grouping sets (max {MAX_GROUPING_SETS})"
3574 )));
3575 }
3576 let indexes = exprs
3577 .iter()
3578 .map(|expr| add_key(self, expr))
3579 .collect::<Result<Vec<_>, _>>()?;
3580 let n = indexes.len();
3581 (0..(1usize << n))
3582 .rev()
3583 .map(|included| {
3584 indexes
3585 .iter()
3586 .enumerate()
3587 .filter(|(position, _)| (included >> (n - 1 - position)) & 1 == 1)
3588 .map(|(_, &index)| index)
3589 .collect()
3590 })
3591 .collect()
3592 }
3593 GroupByItem::GroupingSets { sets: listed } => {
3594 if listed.is_empty() {
3595 return Err(PlannerError::invalid_expression(
3596 "GROUPING SETS requires at least one grouping set".to_string(),
3597 ));
3598 }
3599 listed
3600 .iter()
3601 .map(|set| {
3602 set.iter()
3603 .map(|expr| add_key(self, expr))
3604 .collect::<Result<Vec<_>, _>>()
3605 })
3606 .collect::<Result<Vec<_>, _>>()?
3607 }
3608 };
3609
3610 let mut combined = Vec::with_capacity(sets.len().saturating_mul(item_sets.len()));
3611 for base in &sets {
3612 for item_set in &item_sets {
3613 if combined.len() >= MAX_GROUPING_SETS {
3614 return Err(PlannerError::invalid_expression(format!(
3615 "too many grouping sets (max {MAX_GROUPING_SETS})"
3616 )));
3617 }
3618 let mut set = base.clone();
3619 set.extend(item_set.iter().copied());
3620 combined.push(set);
3621 }
3622 }
3623 sets = combined;
3624 }
3625
3626 if keys.len() > MAX_GROUPING_KEYS {
3627 return Err(PlannerError::invalid_expression(format!(
3628 "too many grouping columns (max {MAX_GROUPING_KEYS})"
3629 )));
3630 }
3631
3632 let key_count = keys.len();
3633 let full_mask = grouping_full_mask(key_count);
3634 let masks = sets
3635 .iter()
3636 .map(|set| {
3637 let mut mask = full_mask;
3638 for &index in set {
3639 mask &= !(1u64 << (key_count - 1 - index));
3640 }
3641 mask
3642 })
3643 .collect();
3644
3645 Ok(ExpandedGroupBy {
3646 group_keys: keys,
3647 grouping_sets: Some(masks),
3648 })
3649 }
3650
3651 #[allow(dead_code)]
3652 fn build_projected_columns_for_aggregate(
3653 &self,
3654 items: &[SelectItem],
3655 table: &TableMetadata,
3656 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3657 let mut projected = Vec::new();
3658 for item in items {
3659 match item {
3660 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
3661 return Err(PlannerError::invalid_expression(
3662 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
3663 ));
3664 }
3665 SelectItem::Expr { expr, alias, .. } => {
3666 let typed = self.type_checker.infer_type(expr, table)?;
3667 projected.push(ProjectedColumn {
3668 expr: typed,
3669 alias: alias.clone(),
3670 });
3671 }
3672 }
3673 }
3674 Ok(projected)
3675 }
3676
3677 fn build_projected_columns_for_aggregate_with_scope(
3678 &self,
3679 items: &[SelectItem],
3680 scope: &[ScopedTable],
3681 ctes: &CtePlans,
3682 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3683 let mut projected = Vec::new();
3684 for item in items {
3685 match item {
3686 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
3687 return Err(PlannerError::invalid_expression(
3688 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
3689 ));
3690 }
3691 SelectItem::Expr { expr, alias, .. } => {
3692 let typed = self.infer_expr_with_scope(expr, scope, ctes)?;
3693 projected.push(ProjectedColumn {
3694 expr: typed,
3695 alias: alias.clone(),
3696 });
3697 }
3698 }
3699 }
3700 Ok(projected)
3701 }
3702
3703 #[allow(dead_code)]
3704 fn build_projected_columns_for_distinct(
3705 &self,
3706 items: &[SelectItem],
3707 table: &TableMetadata,
3708 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3709 let projection = self.build_projection(items, table)?;
3710 match projection {
3711 Projection::All(columns) => {
3712 let mut projected = Vec::with_capacity(columns.len());
3713 for column in columns {
3714 let column_index = table.get_column_index(&column).ok_or_else(|| {
3715 PlannerError::invalid_expression(format!(
3716 "column '{column}' not found for DISTINCT projection"
3717 ))
3718 })?;
3719 let column_meta = table.get_column(&column).ok_or_else(|| {
3720 PlannerError::invalid_expression(format!(
3721 "column '{column}' not found for DISTINCT projection"
3722 ))
3723 })?;
3724 let typed_expr = TypedExpr::column_ref(
3725 table.name.clone(),
3726 column.clone(),
3727 column_index,
3728 column_meta.data_type.clone(),
3729 crate::ast::Span::default(),
3730 );
3731 projected.push(ProjectedColumn::new(typed_expr));
3732 }
3733 Ok(projected)
3734 }
3735 Projection::Columns(columns) => Ok(columns),
3736 }
3737 }
3738
3739 fn build_projected_columns_for_distinct_with_scope(
3740 &self,
3741 items: &[SelectItem],
3742 schema: &[ColumnMetadata],
3743 scope: &[ScopedTable],
3744 ctes: &CtePlans,
3745 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3746 let projection = self.build_projection_with_scope(items, schema, scope, ctes)?;
3747 match projection {
3748 Projection::All(columns) => {
3749 let mut projected = Vec::with_capacity(columns.len());
3750 for (idx, column) in columns.into_iter().enumerate() {
3751 let column_meta = schema.get(idx).ok_or_else(|| {
3752 PlannerError::invalid_expression(format!(
3753 "column '{column}' not found for DISTINCT projection"
3754 ))
3755 })?;
3756 projected.push(ProjectedColumn::new(TypedExpr::column_ref(
3757 LITERAL_TABLE.to_string(),
3758 column,
3759 idx,
3760 column_meta.data_type.clone(),
3761 crate::ast::Span::default(),
3762 )));
3763 }
3764 Ok(projected)
3765 }
3766 Projection::Columns(columns) => Ok(columns),
3767 }
3768 }
3769
3770 fn collect_aggregates_from_typed_expr(
3771 &self,
3772 expr: &TypedExpr,
3773 aggregates: &mut Vec<AggregateExpr>,
3774 aggregate_map: &mut HashMap<AggregateSignature, usize>,
3775 ) -> Result<(), PlannerError> {
3776 match &expr.kind {
3777 TypedExprKind::FunctionCall {
3778 name,
3779 args,
3780 distinct,
3781 star,
3782 filter,
3783 order_by,
3784 over: None,
3785 } if is_aggregate_function(name) => {
3786 if args.iter().any(typed_expr_contains_window) {
3787 return Err(PlannerError::invalid_expression(
3788 "aggregate functions cannot contain window functions".to_string(),
3789 ));
3790 }
3791 for arg in args {
3792 if typed_expr_contains_aggregate(arg) {
3793 return Err(PlannerError::invalid_expression(
3794 "nested aggregate functions are not supported".to_string(),
3795 ));
3796 }
3797 }
3798 if let Some(filter) = filter {
3802 if typed_expr_contains_aggregate(filter) {
3803 return Err(PlannerError::invalid_expression(
3804 "aggregate functions are not allowed in FILTER".to_string(),
3805 ));
3806 }
3807 if typed_expr_contains_window(filter) {
3808 return Err(PlannerError::invalid_expression(
3809 "window functions are not allowed in FILTER".to_string(),
3810 ));
3811 }
3812 }
3813 for sort in order_by {
3814 if typed_expr_contains_aggregate(&sort.expr) {
3815 return Err(PlannerError::invalid_expression(
3816 "aggregate functions are not allowed in aggregate ORDER BY".to_string(),
3817 ));
3818 }
3819 if typed_expr_contains_window(&sort.expr) {
3820 return Err(PlannerError::invalid_expression(
3821 "window functions are not allowed in aggregate ORDER BY".to_string(),
3822 ));
3823 }
3824 }
3825 let (agg, signature) = self.build_aggregate_expr_from_typed(
3826 expr,
3827 name,
3828 args,
3829 *distinct,
3830 *star,
3831 filter.as_deref(),
3832 order_by,
3833 )?;
3834 aggregate_map.entry(signature).or_insert_with(|| {
3835 aggregates.push(agg);
3836 aggregates.len() - 1
3837 });
3838 Ok(())
3839 }
3840 TypedExprKind::BinaryOp { left, right, .. } => {
3841 self.collect_aggregates_from_typed_expr(left, aggregates, aggregate_map)?;
3842 self.collect_aggregates_from_typed_expr(right, aggregates, aggregate_map)?;
3843 Ok(())
3844 }
3845 TypedExprKind::UnaryOp { operand, .. } => {
3846 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)
3847 }
3848 TypedExprKind::Cast { expr, .. } | TypedExprKind::TryCast { expr, .. } => {
3849 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
3850 }
3851 TypedExprKind::Case {
3852 operand,
3853 branches,
3854 else_expr,
3855 } => {
3856 if let Some(operand) = operand {
3857 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)?;
3858 }
3859 for branch in branches {
3860 self.collect_aggregates_from_typed_expr(
3861 &branch.when,
3862 aggregates,
3863 aggregate_map,
3864 )?;
3865 self.collect_aggregates_from_typed_expr(
3866 &branch.then,
3867 aggregates,
3868 aggregate_map,
3869 )?;
3870 }
3871 if let Some(else_expr) = else_expr {
3872 self.collect_aggregates_from_typed_expr(else_expr, aggregates, aggregate_map)?;
3873 }
3874 Ok(())
3875 }
3876 TypedExprKind::FunctionCall { args, over, .. } => {
3877 for arg in args {
3878 self.collect_aggregates_from_typed_expr(arg, aggregates, aggregate_map)?;
3879 }
3880 if let Some(window) = over {
3881 for expr in &window.partition_by {
3882 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3883 }
3884 for sort in &window.order_by {
3885 self.collect_aggregates_from_typed_expr(
3886 &sort.expr,
3887 aggregates,
3888 aggregate_map,
3889 )?;
3890 }
3891 }
3892 Ok(())
3893 }
3894 TypedExprKind::Between {
3895 expr, low, high, ..
3896 } => {
3897 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3898 self.collect_aggregates_from_typed_expr(low, aggregates, aggregate_map)?;
3899 self.collect_aggregates_from_typed_expr(high, aggregates, aggregate_map)?;
3900 Ok(())
3901 }
3902 TypedExprKind::Like {
3903 expr,
3904 pattern,
3905 escape,
3906 ..
3907 } => {
3908 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3909 self.collect_aggregates_from_typed_expr(pattern, aggregates, aggregate_map)?;
3910 if let Some(esc) = escape {
3911 self.collect_aggregates_from_typed_expr(esc, aggregates, aggregate_map)?;
3912 }
3913 Ok(())
3914 }
3915 TypedExprKind::InList { expr, list, .. } => {
3916 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3917 for item in list {
3918 self.collect_aggregates_from_typed_expr(item, aggregates, aggregate_map)?;
3919 }
3920 Ok(())
3921 }
3922 TypedExprKind::IsNull { expr, .. } => {
3923 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
3924 }
3925 _ => Ok(()),
3926 }
3927 }
3928
3929 fn collect_windows_from_typed_expr(
3930 &self,
3931 expr: &TypedExpr,
3932 windows: &mut Vec<WindowExpr>,
3933 window_map: &mut HashMap<String, usize>,
3934 ) -> Result<(), PlannerError> {
3935 match &expr.kind {
3936 TypedExprKind::FunctionCall {
3937 name,
3938 args,
3939 distinct,
3940 star,
3941 filter,
3942 order_by,
3943 over: Some(over),
3944 } => {
3945 if filter.is_some() || !order_by.is_empty() {
3946 return Err(PlannerError::invalid_expression(
3950 "FILTER and aggregate ORDER BY cannot be combined with OVER".to_string(),
3951 ));
3952 }
3953 if args.iter().any(typed_expr_contains_window)
3954 || over.partition_by.iter().any(typed_expr_contains_window)
3955 || over
3956 .order_by
3957 .iter()
3958 .any(|sort| typed_expr_contains_window(&sort.expr))
3959 {
3960 return Err(PlannerError::invalid_expression(
3961 "nested window functions are not supported".to_string(),
3962 ));
3963 }
3964
3965 let key = expr_key(expr);
3966 if window_map.contains_key(&key) {
3967 return Ok(());
3968 }
3969 let function = match name.to_ascii_lowercase().as_str() {
3970 "row_number" => WindowFunction::RowNumber,
3971 "rank" => WindowFunction::Rank,
3972 "dense_rank" => WindowFunction::DenseRank,
3973 "percent_rank" => WindowFunction::PercentRank,
3974 "cume_dist" => WindowFunction::CumeDist,
3975 "ntile" => WindowFunction::Ntile(args[0].clone()),
3976 "first_value" => {
3977 WindowFunction::Value(ValueWindowFunction::FirstValue(args[0].clone()))
3978 }
3979 "last_value" => {
3980 WindowFunction::Value(ValueWindowFunction::LastValue(args[0].clone()))
3981 }
3982 "nth_value" => WindowFunction::Value(ValueWindowFunction::NthValue {
3983 value: args[0].clone(),
3984 nth: args[1].clone(),
3985 }),
3986 "sum" | "count" | "avg" | "min" | "max" => {
3987 let (aggregate, _) = self.build_aggregate_expr_from_typed(
3988 expr,
3989 name,
3990 args,
3991 *distinct,
3992 *star,
3993 None,
3994 &[],
3995 )?;
3996 WindowFunction::Aggregate(aggregate)
3997 }
3998 "lag" | "lead" => {
3999 let positional = build_offset_window_function(name, args)?;
4000 if name.eq_ignore_ascii_case("lag") {
4001 WindowFunction::Lag(positional)
4002 } else {
4003 WindowFunction::Lead(positional)
4004 }
4005 }
4006 _ => {
4007 return Err(PlannerError::unsupported_feature(
4008 format!("function '{}' with OVER", name),
4009 "future",
4010 expr.span,
4011 ));
4012 }
4013 };
4014 let index = windows.len();
4015 windows.push(WindowExpr {
4016 function,
4017 partition_by: over.partition_by.clone(),
4018 order_by: over.order_by.clone(),
4019 frame: over.frame.clone(),
4020 result_type: expr.resolved_type.clone(),
4021 });
4022 window_map.insert(key, index);
4023 Ok(())
4024 }
4025 TypedExprKind::FunctionCall { args, .. } => {
4026 for arg in args {
4027 self.collect_windows_from_typed_expr(arg, windows, window_map)?;
4028 }
4029 Ok(())
4030 }
4031 TypedExprKind::BinaryOp { left, right, .. } => {
4032 self.collect_windows_from_typed_expr(left, windows, window_map)?;
4033 self.collect_windows_from_typed_expr(right, windows, window_map)
4034 }
4035 TypedExprKind::UnaryOp { operand, .. } => {
4036 self.collect_windows_from_typed_expr(operand, windows, window_map)
4037 }
4038 TypedExprKind::Cast { expr, .. }
4039 | TypedExprKind::TryCast { expr, .. }
4040 | TypedExprKind::IsNull { expr, .. } => {
4041 self.collect_windows_from_typed_expr(expr, windows, window_map)
4042 }
4043 TypedExprKind::Between {
4044 expr, low, high, ..
4045 } => {
4046 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
4047 self.collect_windows_from_typed_expr(low, windows, window_map)?;
4048 self.collect_windows_from_typed_expr(high, windows, window_map)
4049 }
4050 TypedExprKind::Like {
4051 expr,
4052 pattern,
4053 escape,
4054 ..
4055 } => {
4056 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
4057 self.collect_windows_from_typed_expr(pattern, windows, window_map)?;
4058 if let Some(escape) = escape {
4059 self.collect_windows_from_typed_expr(escape, windows, window_map)?;
4060 }
4061 Ok(())
4062 }
4063 TypedExprKind::InList { expr, list, .. } => {
4064 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
4065 for item in list {
4066 self.collect_windows_from_typed_expr(item, windows, window_map)?;
4067 }
4068 Ok(())
4069 }
4070 _ => Ok(()),
4071 }
4072 }
4073
4074 #[allow(clippy::too_many_arguments)]
4075 fn build_aggregate_expr_from_typed(
4076 &self,
4077 expr: &TypedExpr,
4078 name: &str,
4079 args: &[TypedExpr],
4080 distinct: bool,
4081 star: bool,
4082 filter: Option<&TypedExpr>,
4083 order_by: &[SortExpr],
4084 ) -> Result<(AggregateExpr, AggregateSignature), PlannerError> {
4085 let lower = name.to_lowercase();
4086 let filter_owned = filter.cloned();
4087 let retained_order_by: Vec<SortExpr> = if is_order_sensitive_aggregate(&lower) {
4092 order_by.to_vec()
4093 } else {
4094 Vec::new()
4095 };
4096 match lower.as_str() {
4097 "count" => {
4098 if star {
4099 let mut agg = AggregateExpr::count_star();
4100 agg.filter = filter_owned;
4101 let signature = aggregate_signature(
4102 name,
4103 distinct,
4104 star,
4105 None,
4106 None,
4107 expr,
4108 filter,
4109 &retained_order_by,
4110 );
4111 return Ok((agg, signature));
4112 }
4113 if args.len() != 1 {
4114 return Err(PlannerError::type_mismatch(
4115 "1 argument",
4116 format!("{} arguments", args.len()),
4117 expr.span,
4118 ));
4119 }
4120 let agg = AggregateExpr {
4121 function: AggregateFunction::Count,
4122 arg: Some(args[0].clone()),
4123 distinct,
4124 result_type: ResolvedType::BigInt,
4125 filter: filter_owned,
4126 order_by: retained_order_by.clone(),
4127 };
4128 let signature = aggregate_signature(
4129 name,
4130 distinct,
4131 star,
4132 Some(&args[0]),
4133 None,
4134 expr,
4135 filter,
4136 &retained_order_by,
4137 );
4138 Ok((agg, signature))
4139 }
4140 "sum" => {
4141 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4142 let agg = AggregateExpr {
4143 function: AggregateFunction::Sum,
4144 arg: Some(arg.clone()),
4145 distinct,
4146 result_type: crate::planner::aggregate_expr::sum_result_type(
4147 &arg.resolved_type,
4148 ),
4149 filter: filter_owned,
4150 order_by: retained_order_by.clone(),
4151 };
4152 let signature = aggregate_signature(
4153 name,
4154 distinct,
4155 star,
4156 Some(arg),
4157 None,
4158 expr,
4159 filter,
4160 &retained_order_by,
4161 );
4162 Ok((agg, signature))
4163 }
4164 "total" => {
4165 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4166 let agg = AggregateExpr {
4167 function: AggregateFunction::Total,
4168 arg: Some(arg.clone()),
4169 distinct: false,
4170 result_type: ResolvedType::Double,
4171 filter: filter_owned,
4172 order_by: retained_order_by.clone(),
4173 };
4174 let signature = aggregate_signature(
4175 name,
4176 false,
4177 star,
4178 Some(arg),
4179 None,
4180 expr,
4181 filter,
4182 &retained_order_by,
4183 );
4184 Ok((agg, signature))
4185 }
4186 "avg" => {
4187 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4188 let agg = AggregateExpr {
4189 function: AggregateFunction::Avg,
4190 arg: Some(arg.clone()),
4191 distinct,
4192 result_type: ResolvedType::Double,
4193 filter: filter_owned,
4194 order_by: retained_order_by.clone(),
4195 };
4196 let signature = aggregate_signature(
4197 name,
4198 distinct,
4199 star,
4200 Some(arg),
4201 None,
4202 expr,
4203 filter,
4204 &retained_order_by,
4205 );
4206 Ok((agg, signature))
4207 }
4208 "min" => {
4209 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4210 let agg = AggregateExpr {
4211 function: AggregateFunction::Min,
4212 arg: Some(arg.clone()),
4213 distinct,
4214 result_type: arg.resolved_type.clone(),
4215 filter: filter_owned,
4216 order_by: retained_order_by.clone(),
4217 };
4218 let signature = aggregate_signature(
4219 name,
4220 distinct,
4221 star,
4222 Some(arg),
4223 None,
4224 expr,
4225 filter,
4226 &retained_order_by,
4227 );
4228 Ok((agg, signature))
4229 }
4230 "max" => {
4231 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4232 let agg = AggregateExpr {
4233 function: AggregateFunction::Max,
4234 arg: Some(arg.clone()),
4235 distinct,
4236 result_type: arg.resolved_type.clone(),
4237 filter: filter_owned,
4238 order_by: retained_order_by.clone(),
4239 };
4240 let signature = aggregate_signature(
4241 name,
4242 distinct,
4243 star,
4244 Some(arg),
4245 None,
4246 expr,
4247 filter,
4248 &retained_order_by,
4249 );
4250 Ok((agg, signature))
4251 }
4252 "group_concat" => {
4253 if args.is_empty() || args.len() > 2 {
4254 return Err(PlannerError::type_mismatch(
4255 "1 or 2 arguments",
4256 format!("{} arguments", args.len()),
4257 expr.span,
4258 ));
4259 }
4260 let arg = &args[0];
4261 let mut separator = None;
4262 if args.len() == 2 {
4263 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
4264 separator = Some(value.clone());
4265 } else {
4266 return Err(PlannerError::invalid_expression(
4267 "GROUP_CONCAT separator must be a string literal".to_string(),
4268 ));
4269 }
4270 }
4271 let agg = AggregateExpr {
4272 function: AggregateFunction::GroupConcat { separator },
4273 arg: Some(arg.clone()),
4274 distinct,
4275 result_type: ResolvedType::Text,
4276 filter: filter_owned,
4277 order_by: retained_order_by.clone(),
4278 };
4279 let signature = aggregate_signature(
4280 name,
4281 distinct,
4282 star,
4283 Some(arg),
4284 match &agg.function {
4285 AggregateFunction::GroupConcat { separator } => separator.as_ref(),
4286 _ => None,
4287 },
4288 expr,
4289 filter,
4290 &retained_order_by,
4291 );
4292 Ok((agg, signature))
4293 }
4294 "string_agg" => {
4295 if args.len() != 2 {
4296 return Err(PlannerError::type_mismatch(
4297 "2 arguments",
4298 format!("{} arguments", args.len()),
4299 expr.span,
4300 ));
4301 }
4302 let arg = &args[0];
4303 let separator =
4304 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
4305 Some(value.clone())
4306 } else {
4307 return Err(PlannerError::invalid_expression(
4308 "STRING_AGG separator must be a string literal".to_string(),
4309 ));
4310 };
4311 let agg = AggregateExpr {
4312 function: AggregateFunction::StringAgg { separator },
4313 arg: Some(arg.clone()),
4314 distinct,
4315 result_type: ResolvedType::Text,
4316 filter: filter_owned,
4317 order_by: retained_order_by.clone(),
4318 };
4319 let signature = aggregate_signature(
4320 name,
4321 distinct,
4322 star,
4323 Some(arg),
4324 match &agg.function {
4325 AggregateFunction::StringAgg { separator } => separator.as_ref(),
4326 _ => None,
4327 },
4328 expr,
4329 filter,
4330 &retained_order_by,
4331 );
4332 Ok((agg, signature))
4333 }
4334 "percentile_disc" => {
4335 if args.len() != 1 {
4336 return Err(PlannerError::type_mismatch(
4337 "1 argument",
4338 format!("{} arguments", args.len()),
4339 expr.span,
4340 ));
4341 }
4342 let fraction = type_checker::percentile_fraction(&args[0])?;
4343 if retained_order_by.len() != 1 {
4344 return Err(PlannerError::invalid_expression(
4345 "PERCENTILE_DISC requires WITHIN GROUP (ORDER BY ...) with exactly \
4346 one sort expression"
4347 .to_string(),
4348 ));
4349 }
4350 let sort = &retained_order_by[0];
4351 let agg = AggregateExpr {
4352 function: AggregateFunction::PercentileDisc { fraction },
4353 arg: Some(sort.expr.clone()),
4354 distinct: false,
4355 result_type: sort.expr.resolved_type.clone(),
4356 filter: filter_owned,
4357 order_by: retained_order_by.clone(),
4358 };
4359 let fraction_key = format!("{fraction:?}");
4362 let signature = aggregate_signature(
4363 name,
4364 false,
4365 star,
4366 None,
4367 Some(&fraction_key),
4368 expr,
4369 filter,
4370 &retained_order_by,
4371 );
4372 Ok((agg, signature))
4373 }
4374 _ => Err(PlannerError::unsupported_feature(
4375 format!("function '{}'", name),
4376 "future",
4377 expr.span,
4378 )),
4379 }
4380 }
4381 fn require_single_aggregate_arg<'b>(
4382 &self,
4383 args: &'b [TypedExpr],
4384 span: crate::ast::Span,
4385 ) -> Result<&'b TypedExpr, PlannerError> {
4386 if args.len() != 1 {
4387 return Err(PlannerError::type_mismatch(
4388 "1 argument",
4389 format!("{} arguments", args.len()),
4390 span,
4391 ));
4392 }
4393 Ok(&args[0])
4394 }
4395
4396 fn build_aggregate_projection(
4397 &self,
4398 projected: Vec<ProjectedColumn>,
4399 group_keys: &[TypedExpr],
4400 aggregates: &[AggregateExpr],
4401 output_names: &[String],
4402 grouping: Option<&GroupingRewrite>,
4403 ) -> Result<Projection, PlannerError> {
4404 let mut columns = Vec::new();
4405 for col in projected {
4406 let rewritten = self.rewrite_expr_for_aggregate(
4407 &col.expr,
4408 group_keys,
4409 aggregates,
4410 output_names,
4411 grouping,
4412 )?;
4413 columns.push(ProjectedColumn {
4414 expr: rewritten,
4415 alias: col.alias,
4416 });
4417 }
4418 Ok(Projection::Columns(columns))
4419 }
4420
4421 fn rewrite_expr_for_aggregate(
4429 &self,
4430 expr: &TypedExpr,
4431 group_keys: &[TypedExpr],
4432 aggregates: &[AggregateExpr],
4433 output_names: &[String],
4434 grouping: Option<&GroupingRewrite>,
4435 ) -> Result<TypedExpr, PlannerError> {
4436 let group_key_map = build_group_key_map(group_keys);
4437 let aggregate_map = build_aggregate_map(aggregates);
4438
4439 let expr = match grouping {
4440 Some(context) => rewrite_grouping_calls(expr, context)?,
4441 None => expr.clone(),
4442 };
4443 rewrite_expr_with_maps(&expr, &group_key_map, &aggregate_map, output_names)
4444 }
4445
4446 fn resolve_pagination_count(
4457 &self,
4458 expr: &Option<crate::ast::expr::Expr>,
4459 clause: &str,
4460 ) -> Result<Option<u64>, PlannerError> {
4461 let Some(expr) = expr else {
4462 return Ok(None);
4463 };
4464 if expr_contains_subquery(expr) {
4465 return Err(PlannerError::unsupported_feature(
4466 format!("subquery in {clause}"),
4467 "a future version",
4468 expr.span,
4469 ));
4470 }
4471 let typed = self
4473 .type_checker
4474 .infer_type_with_scope(expr, &[], &|stmt, _outer| {
4475 Err(PlannerError::unsupported_feature(
4476 format!("subquery in {clause}"),
4477 "a future version",
4478 stmt.span(),
4479 ))
4480 })?;
4481 if typed_expr_contains_aggregate(&typed) {
4482 return Err(PlannerError::invalid_expression(format!(
4483 "aggregate functions are not allowed in {clause}"
4484 )));
4485 }
4486 if typed_expr_contains_window(&typed) {
4487 return Err(PlannerError::invalid_expression(format!(
4488 "window functions are not allowed in {clause}"
4489 )));
4490 }
4491 match typed.resolved_type {
4492 ResolvedType::Integer | ResolvedType::BigInt | ResolvedType::Null => {}
4493 ref other => {
4494 return Err(PlannerError::type_mismatch(
4495 "BIGINT",
4496 other.type_name().to_string(),
4497 expr.span,
4498 ));
4499 }
4500 }
4501 let context = crate::executor::evaluator::EvalContext::new(&[]);
4504 let value = crate::executor::evaluator::evaluate(&typed, &context).map_err(|error| {
4505 PlannerError::invalid_expression(format!("{clause} expression is invalid: {error}"))
4506 })?;
4507 let count = match value {
4508 crate::storage::SqlValue::Null => return Ok(None),
4509 crate::storage::SqlValue::Integer(value) => i64::from(value),
4510 crate::storage::SqlValue::BigInt(value) => value,
4511 other => {
4512 return Err(PlannerError::type_mismatch(
4513 "BIGINT",
4514 format!("{other:?}"),
4515 expr.span,
4516 ));
4517 }
4518 };
4519 if count < 0 {
4520 return Err(PlannerError::invalid_expression(format!(
4521 "{clause} must not be negative"
4522 )));
4523 }
4524 Ok(Some(count as u64))
4525 }
4526
4527 fn apply_pagination(
4532 &self,
4533 plan: LogicalPlan,
4534 limit: &Option<Expr>,
4535 offset: &Option<Expr>,
4536 with_ties: bool,
4537 ) -> Result<LogicalPlan, PlannerError> {
4538 self.apply_pagination_with_tie_keys(plan, limit, offset, with_ties, None)
4539 }
4540
4541 fn apply_pagination_with_tie_keys(
4549 &self,
4550 plan: LogicalPlan,
4551 limit: &Option<Expr>,
4552 offset: &Option<Expr>,
4553 with_ties: bool,
4554 tie_keys: Option<&[SortExpr]>,
4555 ) -> Result<LogicalPlan, PlannerError> {
4556 if limit.is_none() && offset.is_none() && !with_ties {
4557 return Ok(plan);
4558 }
4559 let ties = if with_ties {
4560 let keys = match tie_keys {
4561 Some(keys) => keys.to_vec(),
4562 None => match &plan {
4563 LogicalPlan::Sort { order_by, .. } => order_by.clone(),
4564 _ => Vec::new(),
4565 },
4566 };
4567 if keys.is_empty() {
4568 return Err(PlannerError::invalid_expression(
4569 "FETCH ... WITH TIES requires ORDER BY".to_string(),
4570 ));
4571 }
4572 Some(keys)
4573 } else {
4574 None
4575 };
4576 Ok(LogicalPlan::Limit {
4577 input: Box::new(plan),
4578 limit: self.resolve_pagination_count(limit, "LIMIT")?,
4579 offset: self.resolve_pagination_count(offset, "OFFSET")?,
4580 ties,
4581 })
4582 }
4583
4584 fn plan_insert(&self, stmt: &Insert) -> Result<LogicalPlan, PlannerError> {
4589 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
4591
4592 let columns: Vec<String> = if let Some(ref cols) = stmt.columns {
4594 for col in cols {
4596 self.name_resolver.resolve_column(table, col, stmt.span)?;
4597 }
4598 cols.clone()
4599 } else {
4600 table.column_names().into_iter().map(String::from).collect()
4602 };
4603
4604 match &stmt.source {
4605 InsertSource::Values { values } => {
4606 let mut typed_values: Vec<Vec<TypedExpr>> = Vec::new();
4607
4608 for row in values {
4609 if row.len() != columns.len() {
4610 return Err(PlannerError::column_value_count_mismatch(
4611 columns.len(),
4612 row.len(),
4613 stmt.span,
4614 ));
4615 }
4616
4617 typed_values.push(self.type_check_insert_values(row, &columns, table)?);
4618 }
4619
4620 Ok(LogicalPlan::Insert {
4621 table: table.name.clone(),
4622 columns,
4623 values: typed_values,
4624 })
4625 }
4626 InsertSource::Select { select } => {
4627 let source = self.plan_select_relation(select, &[], &CtePlans::new())?;
4628 self.finish_insert_query(stmt, table, columns, source)
4629 }
4630 InsertSource::Query { query } => {
4631 let source = self.plan_query_body_relation(query, &[], &CtePlans::new())?;
4632 self.finish_insert_query(stmt, table, columns, source)
4633 }
4634 }
4635 }
4636
4637 fn finish_insert_query(
4638 &self,
4639 stmt: &Insert,
4640 table: &TableMetadata,
4641 columns: Vec<String>,
4642 source: PlannedRelation,
4643 ) -> Result<LogicalPlan, PlannerError> {
4644 if source.schema.len() != columns.len() {
4645 return Err(PlannerError::column_value_count_mismatch(
4646 columns.len(),
4647 source.schema.len(),
4648 stmt.span,
4649 ));
4650 }
4651
4652 for (source_column, target_column) in source.schema.iter().zip(&columns) {
4653 let target = table
4654 .get_column(target_column)
4655 .expect("validated target column");
4656 if target.not_null && source_column.data_type == ResolvedType::Null {
4657 return Err(PlannerError::null_constraint_violation(
4658 target_column,
4659 stmt.span,
4660 ));
4661 }
4662 self.validate_resolved_type_assignment(
4663 &source_column.data_type,
4664 &target.data_type,
4665 stmt.span,
4666 )?;
4667 }
4668
4669 Ok(LogicalPlan::InsertSelect {
4670 table: table.name.clone(),
4671 columns,
4672 source: Box::new(source.plan),
4673 })
4674 }
4675
4676 fn type_check_insert_values(
4678 &self,
4679 values: &[crate::ast::expr::Expr],
4680 columns: &[String],
4681 table: &TableMetadata,
4682 ) -> Result<Vec<TypedExpr>, PlannerError> {
4683 let mut typed_values = Vec::new();
4684
4685 for (i, value) in values.iter().enumerate() {
4686 let column_name = &columns[i];
4687 let column_meta = table.get_column(column_name).ok_or_else(|| {
4688 PlannerError::column_not_found(column_name, &table.name, value.span)
4689 })?;
4690
4691 let typed_value = self.type_checker.infer_type(value, table)?;
4693
4694 if column_meta.not_null
4696 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
4697 {
4698 return Err(PlannerError::null_constraint_violation(
4699 column_name,
4700 value.span,
4701 ));
4702 }
4703
4704 self.validate_type_assignment(&typed_value, &column_meta.data_type, value.span)?;
4706
4707 let typed_value =
4708 self.coerce_assignment_value(typed_value, &column_meta.data_type, value.span);
4709
4710 typed_values.push(typed_value);
4711 }
4712
4713 Ok(typed_values)
4714 }
4715
4716 fn validate_type_assignment(
4718 &self,
4719 value: &TypedExpr,
4720 target_type: &ResolvedType,
4721 span: crate::ast::Span,
4722 ) -> Result<(), PlannerError> {
4723 self.validate_resolved_type_assignment(&value.resolved_type, target_type, span)
4724 }
4725
4726 fn validate_resolved_type_assignment(
4727 &self,
4728 source_type: &ResolvedType,
4729 target_type: &ResolvedType,
4730 span: crate::ast::Span,
4731 ) -> Result<(), PlannerError> {
4732 if *source_type == ResolvedType::Null {
4734 return Ok(());
4735 }
4736
4737 if self.types_compatible(source_type, target_type) {
4739 return Ok(());
4740 }
4741
4742 Err(PlannerError::type_mismatch(
4743 target_type.to_string(),
4744 source_type.to_string(),
4745 span,
4746 ))
4747 }
4748
4749 fn types_compatible(&self, source: &ResolvedType, target: &ResolvedType) -> bool {
4751 use ResolvedType::*;
4752
4753 if source == target {
4755 return true;
4756 }
4757
4758 match (source, target) {
4760 (Integer, BigInt) | (Integer, Float) | (Integer, Double) => true,
4762 (BigInt, Float) | (BigInt, Double) => true,
4764 (Float, Double) => true,
4766 (Double, Float) => true,
4769 (Text | Integer | BigInt | Float | Double, Timestamp) => true,
4772 (Vector { dimension: d1, .. }, Vector { dimension: d2, .. }) => d1 == d2,
4774 _ => false,
4775 }
4776 }
4777
4778 fn coerce_assignment_value(
4779 &self,
4780 value: TypedExpr,
4781 target_type: &ResolvedType,
4782 span: crate::ast::Span,
4783 ) -> TypedExpr {
4784 if value.resolved_type != *target_type
4785 && value.resolved_type != ResolvedType::Null
4786 && matches!(
4787 target_type,
4788 ResolvedType::Integer
4789 | ResolvedType::BigInt
4790 | ResolvedType::Float
4791 | ResolvedType::Double
4792 | ResolvedType::Timestamp
4793 )
4794 {
4795 TypedExpr::cast(value, target_type.clone(), span)
4796 } else {
4797 value
4798 }
4799 }
4800
4801 fn plan_update(&self, stmt: &Update) -> Result<LogicalPlan, PlannerError> {
4805 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
4807
4808 let mut typed_assignments = Vec::new();
4810
4811 for assignment in &stmt.assignments {
4812 let column_meta =
4814 self.name_resolver
4815 .resolve_column(table, &assignment.column, assignment.span)?;
4816 let column_index = table.get_column_index(&assignment.column).unwrap();
4817
4818 let typed_value = self.type_checker.infer_type(&assignment.value, table)?;
4820
4821 if column_meta.not_null
4823 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
4824 {
4825 return Err(PlannerError::null_constraint_violation(
4826 &assignment.column,
4827 assignment.value.span,
4828 ));
4829 }
4830
4831 self.validate_type_assignment(
4833 &typed_value,
4834 &column_meta.data_type,
4835 assignment.value.span,
4836 )?;
4837
4838 let typed_value = self.coerce_assignment_value(
4839 typed_value,
4840 &column_meta.data_type,
4841 assignment.value.span,
4842 );
4843
4844 typed_assignments.push(TypedAssignment::new(
4845 assignment.column.clone(),
4846 column_index,
4847 typed_value,
4848 ));
4849 }
4850
4851 let filter = if let Some(ref selection) = stmt.selection {
4853 let predicate = self.type_checker.infer_type(selection, table)?;
4854
4855 if predicate.resolved_type != ResolvedType::Boolean {
4857 return Err(PlannerError::type_mismatch(
4858 "Boolean",
4859 predicate.resolved_type.to_string(),
4860 selection.span,
4861 ));
4862 }
4863
4864 Some(predicate)
4865 } else {
4866 None
4867 };
4868
4869 Ok(LogicalPlan::Update {
4870 table: table.name.clone(),
4871 assignments: typed_assignments,
4872 filter,
4873 })
4874 }
4875
4876 fn plan_delete(&self, stmt: &Delete) -> Result<LogicalPlan, PlannerError> {
4880 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
4882
4883 let filter = if let Some(ref selection) = stmt.selection {
4885 let predicate = self.type_checker.infer_type(selection, table)?;
4886
4887 if predicate.resolved_type != ResolvedType::Boolean {
4889 return Err(PlannerError::type_mismatch(
4890 "Boolean",
4891 predicate.resolved_type.to_string(),
4892 selection.span,
4893 ));
4894 }
4895
4896 Some(predicate)
4897 } else {
4898 None
4899 };
4900
4901 Ok(LogicalPlan::Delete {
4902 table: table.name.clone(),
4903 filter,
4904 })
4905 }
4906}
4907
4908#[derive(Debug, Clone, PartialEq, Eq, Hash)]
4909struct AggregateSignature {
4910 name: String,
4911 distinct: bool,
4912 star: bool,
4913 arg_key: Option<String>,
4914 separator: Option<String>,
4915 filter_key: Option<String>,
4918 order_key: Option<String>,
4922}
4923
4924fn collect_projection_aliases(items: &[SelectItem]) -> HashMap<String, crate::ast::expr::Expr> {
4934 let mut aliases = HashMap::new();
4935 for item in items {
4936 if let SelectItem::Expr {
4937 expr,
4938 alias: Some(alias),
4939 ..
4940 } = item
4941 {
4942 aliases.entry(alias.clone()).or_insert_with(|| expr.clone());
4943 }
4944 }
4945 aliases
4946}
4947
4948fn substitute_projection_aliases(
4966 expr: &crate::ast::expr::Expr,
4967 aliases: &HashMap<String, crate::ast::expr::Expr>,
4968) -> crate::ast::expr::Expr {
4969 use crate::ast::expr::ExprKind;
4970
4971 if aliases.is_empty() {
4972 return expr.clone();
4973 }
4974
4975 let recurse = |e: &crate::ast::expr::Expr| substitute_projection_aliases(e, aliases);
4976
4977 let kind = match &expr.kind {
4978 ExprKind::ColumnRef {
4979 table: None,
4980 column,
4981 } => match aliases.get(column) {
4982 Some(source) => {
4983 let mut replacement = source.clone();
4984 replacement.span = expr.span;
4985 return replacement;
4986 }
4987 None => return expr.clone(),
4988 },
4989 ExprKind::BinaryOp { left, op, right } => ExprKind::BinaryOp {
4990 left: Box::new(recurse(left)),
4991 op: *op,
4992 right: Box::new(recurse(right)),
4993 },
4994 ExprKind::UnaryOp { op, operand } => ExprKind::UnaryOp {
4995 op: *op,
4996 operand: Box::new(recurse(operand)),
4997 },
4998 ExprKind::FunctionCall {
4999 name,
5000 args,
5001 distinct,
5002 star,
5003 order_by,
5004 within_group,
5005 filter,
5006 over,
5007 } => ExprKind::FunctionCall {
5008 name: name.clone(),
5009 args: args.iter().map(recurse).collect(),
5010 distinct: *distinct,
5011 star: *star,
5012 order_by: order_by
5013 .iter()
5014 .map(|order| OrderByExpr {
5015 expr: recurse(&order.expr),
5016 asc: order.asc,
5017 nulls_first: order.nulls_first,
5018 span: order.span,
5019 })
5020 .collect(),
5021 within_group: within_group
5022 .iter()
5023 .map(|order| OrderByExpr {
5024 expr: recurse(&order.expr),
5025 asc: order.asc,
5026 nulls_first: order.nulls_first,
5027 span: order.span,
5028 })
5029 .collect(),
5030 filter: filter
5031 .as_deref()
5032 .map(|predicate| Box::new(recurse(predicate))),
5033 over: over.as_ref().map(|window| crate::ast::expr::WindowSpec {
5034 base: window.base.clone(),
5035 partition_by: window.partition_by.iter().map(recurse).collect(),
5036 order_by: window
5037 .order_by
5038 .iter()
5039 .map(|order| OrderByExpr {
5040 expr: recurse(&order.expr),
5041 asc: order.asc,
5042 nulls_first: order.nulls_first,
5043 span: order.span,
5044 })
5045 .collect(),
5046 frame: window.frame.clone(),
5047 }),
5048 },
5049 ExprKind::Case {
5050 operand,
5051 branches,
5052 else_expr,
5053 } => ExprKind::Case {
5054 operand: operand.as_deref().map(|e| Box::new(recurse(e))),
5055 branches: branches
5056 .iter()
5057 .map(|branch| crate::ast::expr::CaseWhen {
5058 when: recurse(&branch.when),
5059 then: recurse(&branch.then),
5060 })
5061 .collect(),
5062 else_expr: else_expr.as_deref().map(|e| Box::new(recurse(e))),
5063 },
5064 ExprKind::Cast { expr, target_type } => ExprKind::Cast {
5065 expr: Box::new(recurse(expr)),
5066 target_type: target_type.clone(),
5067 },
5068 ExprKind::TryCast { expr, target_type } => ExprKind::TryCast {
5069 expr: Box::new(recurse(expr)),
5070 target_type: target_type.clone(),
5071 },
5072 ExprKind::Between {
5073 expr,
5074 low,
5075 high,
5076 negated,
5077 } => ExprKind::Between {
5078 expr: Box::new(recurse(expr)),
5079 low: Box::new(recurse(low)),
5080 high: Box::new(recurse(high)),
5081 negated: *negated,
5082 },
5083 ExprKind::Like {
5084 expr,
5085 pattern,
5086 escape,
5087 negated,
5088 kind,
5089 } => ExprKind::Like {
5090 expr: Box::new(recurse(expr)),
5091 pattern: Box::new(recurse(pattern)),
5092 escape: escape.as_deref().map(|e| Box::new(recurse(e))),
5093 negated: *negated,
5094 kind: *kind,
5095 },
5096 ExprKind::InList {
5097 expr,
5098 list,
5099 negated,
5100 } => ExprKind::InList {
5101 expr: Box::new(recurse(expr)),
5102 list: list.iter().map(recurse).collect(),
5103 negated: *negated,
5104 },
5105 ExprKind::IsNull { expr, negated } => ExprKind::IsNull {
5106 expr: Box::new(recurse(expr)),
5107 negated: *negated,
5108 },
5109 ExprKind::TruthPredicate {
5110 expr,
5111 value,
5112 negated,
5113 } => ExprKind::TruthPredicate {
5114 expr: Box::new(recurse(expr)),
5115 value: *value,
5116 negated: *negated,
5117 },
5118 ExprKind::IsDistinctFrom {
5119 left,
5120 right,
5121 negated,
5122 } => ExprKind::IsDistinctFrom {
5123 left: Box::new(recurse(left)),
5124 right: Box::new(recurse(right)),
5125 negated: *negated,
5126 },
5127 ExprKind::Row { items } => ExprKind::Row {
5128 items: items.iter().map(recurse).collect(),
5129 },
5130 ExprKind::ColumnRef { .. }
5133 | ExprKind::Literal { .. }
5134 | ExprKind::VectorLiteral { .. }
5135 | ExprKind::ScalarSubquery { .. }
5136 | ExprKind::InSubquery { .. }
5137 | ExprKind::Exists { .. }
5138 | ExprKind::Quantified { .. } => return expr.clone(),
5139 };
5140
5141 crate::ast::expr::Expr {
5142 kind,
5143 span: expr.span,
5144 }
5145}
5146
5147fn build_offset_window_function(
5148 name: &str,
5149 args: &[TypedExpr],
5150) -> Result<OffsetWindowFunction, PlannerError> {
5151 let value = args.first().cloned().ok_or_else(|| {
5152 PlannerError::invalid_expression(format!(
5153 "{}() window function expects 1 to 3 arguments",
5154 name.to_ascii_uppercase()
5155 ))
5156 })?;
5157 if args.len() > 3 {
5158 return Err(PlannerError::invalid_expression(format!(
5159 "{}() window function expects 1 to 3 arguments",
5160 name.to_ascii_uppercase()
5161 )));
5162 }
5163 Ok(OffsetWindowFunction {
5164 value,
5165 offset: args.get(1).cloned(),
5166 default: args.get(2).cloned(),
5167 })
5168}
5169
5170fn expr_contains_aggregate(expr: &crate::ast::expr::Expr) -> bool {
5171 use crate::ast::expr::ExprKind;
5172
5173 match &expr.kind {
5174 ExprKind::FunctionCall {
5175 name,
5176 args,
5177 order_by,
5178 within_group,
5179 filter,
5180 over,
5181 ..
5182 } => {
5183 if over.is_none() && is_aggregate_function(name) {
5184 return true;
5185 }
5186 args.iter().any(expr_contains_aggregate)
5187 || order_by
5188 .iter()
5189 .any(|sort| expr_contains_aggregate(&sort.expr))
5190 || within_group
5191 .iter()
5192 .any(|sort| expr_contains_aggregate(&sort.expr))
5193 || filter.as_deref().is_some_and(expr_contains_aggregate)
5194 || over.as_ref().is_some_and(|window| {
5195 window.partition_by.iter().any(expr_contains_aggregate)
5196 || window
5197 .order_by
5198 .iter()
5199 .any(|sort| expr_contains_aggregate(&sort.expr))
5200 })
5201 }
5202 ExprKind::BinaryOp { left, right, .. } => {
5203 expr_contains_aggregate(left) || expr_contains_aggregate(right)
5204 }
5205 ExprKind::UnaryOp { operand, .. } => expr_contains_aggregate(operand),
5206 ExprKind::TruthPredicate { expr, .. } => expr_contains_aggregate(expr),
5207 ExprKind::IsDistinctFrom { left, right, .. } => {
5208 expr_contains_aggregate(left) || expr_contains_aggregate(right)
5209 }
5210 ExprKind::Row { items } => items.iter().any(expr_contains_aggregate),
5211 ExprKind::Case {
5212 operand,
5213 branches,
5214 else_expr,
5215 } => {
5216 operand.as_deref().is_some_and(expr_contains_aggregate)
5217 || branches.iter().any(|branch| {
5218 expr_contains_aggregate(&branch.when) || expr_contains_aggregate(&branch.then)
5219 })
5220 || else_expr.as_deref().is_some_and(expr_contains_aggregate)
5221 }
5222 ExprKind::Cast { expr, .. } | ExprKind::TryCast { expr, .. } => {
5223 expr_contains_aggregate(expr)
5224 }
5225 ExprKind::Between {
5226 expr, low, high, ..
5227 } => {
5228 expr_contains_aggregate(expr)
5229 || expr_contains_aggregate(low)
5230 || expr_contains_aggregate(high)
5231 }
5232 ExprKind::Like {
5233 expr,
5234 pattern,
5235 escape,
5236 ..
5237 } => {
5238 expr_contains_aggregate(expr)
5239 || expr_contains_aggregate(pattern)
5240 || escape.as_deref().is_some_and(expr_contains_aggregate)
5241 }
5242 ExprKind::InList { expr, list, .. } => {
5243 expr_contains_aggregate(expr) || list.iter().any(expr_contains_aggregate)
5244 }
5245 ExprKind::IsNull { expr, .. } => expr_contains_aggregate(expr),
5246 ExprKind::ScalarSubquery { .. }
5247 | ExprKind::InSubquery { .. }
5248 | ExprKind::Exists { .. }
5249 | ExprKind::Quantified { .. }
5250 | ExprKind::Literal { .. }
5251 | ExprKind::VectorLiteral { .. }
5252 | ExprKind::ColumnRef { .. } => false,
5253 }
5254}
5255
5256fn typed_expr_contains_aggregate(expr: &TypedExpr) -> bool {
5257 match &expr.kind {
5258 TypedExprKind::FunctionCall {
5259 name,
5260 args,
5261 filter,
5262 order_by,
5263 over,
5264 ..
5265 } => {
5266 if over.is_none() && is_aggregate_function(name) {
5267 return true;
5268 }
5269 args.iter().any(typed_expr_contains_aggregate)
5270 || filter.as_deref().is_some_and(typed_expr_contains_aggregate)
5271 || order_by
5272 .iter()
5273 .any(|sort| typed_expr_contains_aggregate(&sort.expr))
5274 || over.as_ref().is_some_and(|window| {
5275 window
5276 .partition_by
5277 .iter()
5278 .any(typed_expr_contains_aggregate)
5279 || window
5280 .order_by
5281 .iter()
5282 .any(|sort| typed_expr_contains_aggregate(&sort.expr))
5283 })
5284 }
5285 TypedExprKind::BinaryOp { left, right, .. } => {
5286 typed_expr_contains_aggregate(left) || typed_expr_contains_aggregate(right)
5287 }
5288 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_aggregate(operand),
5289 TypedExprKind::Cast { expr, .. } | TypedExprKind::TryCast { expr, .. } => {
5290 typed_expr_contains_aggregate(expr)
5291 }
5292 TypedExprKind::Case {
5293 operand,
5294 branches,
5295 else_expr,
5296 } => {
5297 operand
5298 .as_deref()
5299 .is_some_and(typed_expr_contains_aggregate)
5300 || branches.iter().any(|branch| {
5301 typed_expr_contains_aggregate(&branch.when)
5302 || typed_expr_contains_aggregate(&branch.then)
5303 })
5304 || else_expr
5305 .as_deref()
5306 .is_some_and(typed_expr_contains_aggregate)
5307 }
5308 TypedExprKind::Between {
5309 expr, low, high, ..
5310 } => {
5311 typed_expr_contains_aggregate(expr)
5312 || typed_expr_contains_aggregate(low)
5313 || typed_expr_contains_aggregate(high)
5314 }
5315 TypedExprKind::Like {
5316 expr,
5317 pattern,
5318 escape,
5319 ..
5320 } => {
5321 typed_expr_contains_aggregate(expr)
5322 || typed_expr_contains_aggregate(pattern)
5323 || escape
5324 .as_ref()
5325 .is_some_and(|inner| typed_expr_contains_aggregate(inner))
5326 }
5327 TypedExprKind::InList { expr, list, .. } => {
5328 typed_expr_contains_aggregate(expr) || list.iter().any(typed_expr_contains_aggregate)
5329 }
5330 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_aggregate(expr),
5331 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_aggregate(expr),
5332 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_aggregate(expr),
5333 TypedExprKind::ScalarSubquery(_) | TypedExprKind::Exists { .. } => false,
5334 _ => false,
5335 }
5336}
5337
5338fn select_contains_window(stmt: &Select) -> bool {
5339 stmt.projection.iter().any(|item| match item {
5340 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
5341 SelectItem::Expr { expr, .. } => expr_contains_window(expr),
5342 }) || stmt.qualify.as_ref().is_some_and(expr_contains_window)
5343 || stmt
5344 .order_by
5345 .iter()
5346 .any(|order| expr_contains_window(&order.expr))
5347}
5348
5349fn expr_contains_window(expr: &crate::ast::expr::Expr) -> bool {
5350 match &expr.kind {
5351 crate::ast::expr::ExprKind::FunctionCall {
5352 args,
5353 order_by,
5354 within_group,
5355 filter,
5356 over,
5357 ..
5358 } => {
5359 over.is_some()
5360 || args.iter().any(expr_contains_window)
5361 || order_by.iter().any(|sort| expr_contains_window(&sort.expr))
5362 || within_group
5363 .iter()
5364 .any(|sort| expr_contains_window(&sort.expr))
5365 || filter.as_deref().is_some_and(expr_contains_window)
5366 }
5367 crate::ast::expr::ExprKind::BinaryOp { left, right, .. } => {
5368 expr_contains_window(left) || expr_contains_window(right)
5369 }
5370 crate::ast::expr::ExprKind::UnaryOp { operand, .. }
5371 | crate::ast::expr::ExprKind::Cast { expr: operand, .. }
5372 | crate::ast::expr::ExprKind::TryCast { expr: operand, .. }
5373 | crate::ast::expr::ExprKind::IsNull { expr: operand, .. } => expr_contains_window(operand),
5374 crate::ast::expr::ExprKind::Between {
5375 expr, low, high, ..
5376 } => expr_contains_window(expr) || expr_contains_window(low) || expr_contains_window(high),
5377 crate::ast::expr::ExprKind::Like {
5378 expr,
5379 pattern,
5380 escape,
5381 ..
5382 } => {
5383 expr_contains_window(expr)
5384 || expr_contains_window(pattern)
5385 || escape.as_deref().is_some_and(expr_contains_window)
5386 }
5387 crate::ast::expr::ExprKind::InList { expr, list, .. } => {
5388 expr_contains_window(expr) || list.iter().any(expr_contains_window)
5389 }
5390 _ => false,
5391 }
5392}
5393
5394fn typed_expr_contains_window(expr: &TypedExpr) -> bool {
5395 match &expr.kind {
5396 TypedExprKind::FunctionCall {
5397 args,
5398 filter,
5399 order_by,
5400 over,
5401 ..
5402 } => {
5403 over.is_some()
5404 || args.iter().any(typed_expr_contains_window)
5405 || filter.as_deref().is_some_and(typed_expr_contains_window)
5406 || order_by
5407 .iter()
5408 .any(|sort| typed_expr_contains_window(&sort.expr))
5409 }
5410 TypedExprKind::BinaryOp { left, right, .. } => {
5411 typed_expr_contains_window(left) || typed_expr_contains_window(right)
5412 }
5413 TypedExprKind::UnaryOp { operand, .. }
5414 | TypedExprKind::Cast { expr: operand, .. }
5415 | TypedExprKind::TryCast { expr: operand, .. }
5416 | TypedExprKind::IsNull { expr: operand, .. } => typed_expr_contains_window(operand),
5417 TypedExprKind::Between {
5418 expr, low, high, ..
5419 } => {
5420 typed_expr_contains_window(expr)
5421 || typed_expr_contains_window(low)
5422 || typed_expr_contains_window(high)
5423 }
5424 TypedExprKind::Like {
5425 expr,
5426 pattern,
5427 escape,
5428 ..
5429 } => {
5430 typed_expr_contains_window(expr)
5431 || typed_expr_contains_window(pattern)
5432 || escape.as_deref().is_some_and(typed_expr_contains_window)
5433 }
5434 TypedExprKind::InList { expr, list, .. } => {
5435 typed_expr_contains_window(expr) || list.iter().any(typed_expr_contains_window)
5436 }
5437 _ => false,
5438 }
5439}
5440
5441fn rewrite_projection_for_windows(
5442 projection: &Projection,
5443 window_map: &HashMap<String, usize>,
5444 base_width: usize,
5445 window_names: &[String],
5446) -> Result<Projection, PlannerError> {
5447 match projection {
5448 Projection::All(names) => Ok(Projection::All(names.clone())),
5449 Projection::Columns(columns) => Ok(Projection::Columns(
5450 columns
5451 .iter()
5452 .map(|column| {
5453 Ok(ProjectedColumn {
5454 expr: rewrite_expr_for_windows(
5455 &column.expr,
5456 window_map,
5457 base_width,
5458 window_names,
5459 )?,
5460 alias: column.alias.clone(),
5461 })
5462 })
5463 .collect::<Result<Vec<_>, PlannerError>>()?,
5464 )),
5465 }
5466}
5467
5468fn rewrite_expr_for_windows(
5469 expr: &TypedExpr,
5470 window_map: &HashMap<String, usize>,
5471 base_width: usize,
5472 window_names: &[String],
5473) -> Result<TypedExpr, PlannerError> {
5474 if let Some(index) = window_map.get(&expr_key(expr)) {
5475 return Ok(TypedExpr::column_ref(
5476 "__window__".to_string(),
5477 window_names
5478 .get(*index)
5479 .cloned()
5480 .unwrap_or_else(|| format!("__window_{index}")),
5481 base_width + index,
5482 expr.resolved_type.clone(),
5483 expr.span,
5484 ));
5485 }
5486
5487 let rewrite =
5488 |inner: &TypedExpr| rewrite_expr_for_windows(inner, window_map, base_width, window_names);
5489 let kind = match &expr.kind {
5490 TypedExprKind::FunctionCall {
5491 name,
5492 args,
5493 distinct,
5494 star,
5495 filter,
5496 order_by,
5497 over,
5498 } => {
5499 if over.is_some() {
5500 return Err(PlannerError::invalid_expression(
5501 "window expression is not part of the window plan".to_string(),
5502 ));
5503 }
5504 TypedExprKind::FunctionCall {
5505 name: name.clone(),
5506 args: args.iter().map(rewrite).collect::<Result<Vec<_>, _>>()?,
5507 distinct: *distinct,
5508 star: *star,
5509 filter: filter.as_deref().map(rewrite).transpose()?.map(Box::new),
5510 order_by: order_by
5511 .iter()
5512 .map(|sort| {
5513 Ok(SortExpr::new(
5514 rewrite(&sort.expr)?,
5515 sort.asc,
5516 sort.nulls_first,
5517 ))
5518 })
5519 .collect::<Result<Vec<_>, PlannerError>>()?,
5520 over: None,
5521 }
5522 }
5523 TypedExprKind::BinaryOp { left, op, right } => TypedExprKind::BinaryOp {
5524 left: Box::new(rewrite(left)?),
5525 op: *op,
5526 right: Box::new(rewrite(right)?),
5527 },
5528 TypedExprKind::UnaryOp { op, operand } => TypedExprKind::UnaryOp {
5529 op: *op,
5530 operand: Box::new(rewrite(operand)?),
5531 },
5532 TypedExprKind::Cast {
5533 expr: inner,
5534 target_type,
5535 } => TypedExprKind::Cast {
5536 expr: Box::new(rewrite(inner)?),
5537 target_type: target_type.clone(),
5538 },
5539 TypedExprKind::TryCast {
5540 expr: inner,
5541 target_type,
5542 } => TypedExprKind::TryCast {
5543 expr: Box::new(rewrite(inner)?),
5544 target_type: target_type.clone(),
5545 },
5546 TypedExprKind::Between {
5547 expr: inner,
5548 low,
5549 high,
5550 negated,
5551 } => TypedExprKind::Between {
5552 expr: Box::new(rewrite(inner)?),
5553 low: Box::new(rewrite(low)?),
5554 high: Box::new(rewrite(high)?),
5555 negated: *negated,
5556 },
5557 TypedExprKind::Like {
5558 expr: inner,
5559 pattern,
5560 escape,
5561 negated,
5562 kind,
5563 } => TypedExprKind::Like {
5564 expr: Box::new(rewrite(inner)?),
5565 pattern: Box::new(rewrite(pattern)?),
5566 escape: escape.as_deref().map(rewrite).transpose()?.map(Box::new),
5567 negated: *negated,
5568 kind: *kind,
5569 },
5570 TypedExprKind::InList {
5571 expr: inner,
5572 list,
5573 negated,
5574 } => TypedExprKind::InList {
5575 expr: Box::new(rewrite(inner)?),
5576 list: list.iter().map(rewrite).collect::<Result<Vec<_>, _>>()?,
5577 negated: *negated,
5578 },
5579 TypedExprKind::IsNull {
5580 expr: inner,
5581 negated,
5582 } => TypedExprKind::IsNull {
5583 expr: Box::new(rewrite(inner)?),
5584 negated: *negated,
5585 },
5586 _ => return Ok(expr.clone()),
5587 };
5588 Ok(TypedExpr {
5589 kind,
5590 resolved_type: expr.resolved_type.clone(),
5591 span: expr.span,
5592 })
5593}
5594
5595fn rewrite_expr_for_projected_output(
5601 expr: &TypedExpr,
5602 projection: &Projection,
5603 output_schema: &[ColumnMetadata],
5604) -> Result<TypedExpr, PlannerError> {
5605 let index = match projection {
5606 Projection::Columns(columns) => columns
5607 .iter()
5608 .position(|column| expr_key(&column.expr) == expr_key(expr)),
5609 Projection::All(_) => match &expr.kind {
5610 TypedExprKind::ColumnRef { column_index, .. }
5611 if *column_index < output_schema.len() =>
5612 {
5613 Some(*column_index)
5614 }
5615 _ => None,
5616 },
5617 };
5618 let Some(index) = index else {
5619 return Err(PlannerError::invalid_expression(
5620 "ORDER BY expression must appear in the SELECT projection for window queries"
5621 .to_string(),
5622 ));
5623 };
5624 let column = output_schema.get(index).ok_or_else(|| {
5625 PlannerError::invalid_expression(
5626 "ORDER BY projection index is outside the output schema".to_string(),
5627 )
5628 })?;
5629 Ok(TypedExpr::column_ref(
5630 "__project__".to_string(),
5631 column.name.clone(),
5632 index,
5633 column.data_type.clone(),
5634 expr.span,
5635 ))
5636}
5637
5638fn map_join_type(join_type: crate::ast::dml::JoinType) -> JoinType {
5639 match join_type {
5640 crate::ast::dml::JoinType::Inner => JoinType::Inner,
5641 crate::ast::dml::JoinType::Left => JoinType::Left,
5642 crate::ast::dml::JoinType::Right => JoinType::Right,
5643 crate::ast::dml::JoinType::Full => JoinType::Full,
5644 crate::ast::dml::JoinType::Cross => JoinType::Cross,
5645 }
5646}
5647
5648struct FoundScopedColumn {
5649 table: String,
5650 index: usize,
5651 ty: ResolvedType,
5652 partner_indices: Vec<usize>,
5653}
5654
5655fn find_scoped_column(
5656 scope: &[ScopedTable],
5657 column: &str,
5658 span: crate::ast::Span,
5659) -> Result<FoundScopedColumn, PlannerError> {
5660 let mut matches = Vec::new();
5661 for table in scope {
5662 if table.hidden_unqualified_columns.contains(column) {
5663 continue;
5664 }
5665 if let Some(local_idx) = table.table.get_column_index(column) {
5666 let meta = &table.table.columns[local_idx];
5667 matches.push(FoundScopedColumn {
5668 table: table.table.name.clone(),
5669 index: table.start_index + local_idx,
5670 ty: meta.data_type.clone(),
5671 partner_indices: table
5672 .merged_column_partners
5673 .get(column)
5674 .cloned()
5675 .unwrap_or_default(),
5676 });
5677 }
5678 }
5679 match matches.len() {
5680 0 => Err(PlannerError::column_not_found(column, "JOIN input", span)),
5681 1 => Ok(matches.remove(0)),
5682 _ => Err(PlannerError::ambiguous_column(
5683 column,
5684 scope.iter().map(|s| s.table.name.clone()).collect(),
5685 span,
5686 )),
5687 }
5688}
5689
5690fn merged_scoped_column_expr(
5691 found: &FoundScopedColumn,
5692 column: &str,
5693 span: crate::ast::Span,
5694) -> TypedExpr {
5695 let own = TypedExpr::column_ref(
5696 found.table.clone(),
5697 column.to_string(),
5698 found.index,
5699 found.ty.clone(),
5700 span,
5701 );
5702 if found.partner_indices.is_empty() {
5703 return own;
5704 }
5705
5706 let mut args = Vec::with_capacity(found.partner_indices.len() + 1);
5707 args.push(own);
5708 args.extend(found.partner_indices.iter().map(|&index| {
5709 TypedExpr::column_ref(
5710 found.table.clone(),
5711 column.to_string(),
5712 index,
5713 found.ty.clone(),
5714 span,
5715 )
5716 }));
5717 TypedExpr {
5718 kind: TypedExprKind::FunctionCall {
5719 name: "coalesce".to_string(),
5720 args,
5721 distinct: false,
5722 star: false,
5723 filter: None,
5724 order_by: Vec::new(),
5725 over: None,
5726 },
5727 resolved_type: found.ty.clone(),
5728 span,
5729 }
5730}
5731
5732fn projection_schema(
5733 projection: &Projection,
5734 input_schema: &[ColumnMetadata],
5735) -> Vec<ColumnMetadata> {
5736 match projection {
5737 Projection::All(names) => names
5738 .iter()
5739 .enumerate()
5740 .map(|(idx, name)| {
5741 let ty = (names.len() == input_schema.len())
5742 .then(|| input_schema.get(idx))
5743 .flatten()
5744 .or_else(|| input_schema.iter().find(|col| &col.name == name))
5745 .map(|col| col.data_type.clone())
5746 .unwrap_or(ResolvedType::Null);
5747 ColumnMetadata::new(name.clone(), ty)
5748 })
5749 .collect(),
5750 Projection::Columns(columns) => columns
5751 .iter()
5752 .enumerate()
5753 .map(|(idx, col)| {
5754 let name = col
5755 .alias
5756 .clone()
5757 .or_else(|| match &col.expr.kind {
5758 TypedExprKind::ColumnRef { column, .. } => Some(column.clone()),
5759 TypedExprKind::FunctionCall { name, args, .. }
5763 if name == "coalesce" && !args.is_empty() =>
5764 {
5765 let first_column = match &args[0].kind {
5766 TypedExprKind::ColumnRef { column, .. } => Some(column),
5767 _ => None,
5768 };
5769 first_column
5770 .filter(|column| {
5771 args.iter().all(|arg| {
5772 matches!(
5773 &arg.kind,
5774 TypedExprKind::ColumnRef { column: other, .. }
5775 if other == *column
5776 )
5777 })
5778 })
5779 .cloned()
5780 }
5781 _ => None,
5782 })
5783 .unwrap_or_else(|| format!("col_{idx}"));
5784 ColumnMetadata::new(name, col.expr.resolved_type.clone())
5785 })
5786 .collect(),
5787 }
5788}
5789
5790fn visible_wildcard_columns(schema: &[ColumnMetadata], scope: &[ScopedTable]) -> Vec<String> {
5791 schema
5792 .iter()
5793 .enumerate()
5794 .filter(|(index, column)| {
5795 !scope.iter().any(|table| {
5796 *index >= table.start_index
5797 && *index < table.start_index + table.table.columns.len()
5798 && table.hidden_unqualified_columns.contains(&column.name)
5799 })
5800 })
5801 .map(|(_, column)| column.name.clone())
5802 .collect()
5803}
5804
5805fn combine_join_shape(
5810 left: &PlannedRelation,
5811 right: &PlannedRelation,
5812 using: Option<&[String]>,
5813) -> (Vec<ColumnMetadata>, Vec<ScopedTable>) {
5814 let mut schema = left.schema.clone();
5815 schema.extend(right.schema.clone());
5816 let mut scope = left.scope.clone();
5817 let mut right_scope = right.scope.clone();
5818 if let Some(columns) = using {
5819 for column in columns {
5823 let right_index = right_scope.iter().find_map(|table| {
5824 table
5825 .table
5826 .get_column_index(column)
5827 .map(|index| table.start_index + index)
5828 });
5829 let Some(right_index) = right_index else {
5830 continue;
5831 };
5832 for table in &mut scope {
5833 if table.table.get_column_index(column).is_some() {
5834 table.merge_column_with(column, right_index);
5835 }
5836 }
5837 }
5838 for table in &mut right_scope {
5839 table.hide_unqualified_columns(columns);
5840 }
5841 }
5842 scope.extend(right_scope);
5843 (schema, scope)
5844}
5845
5846fn from_item_is_lateral(item: &FromItem) -> bool {
5851 match item {
5852 FromItem::Derived { lateral, .. } => *lateral,
5853 FromItem::Function { .. } => true,
5854 FromItem::Table { .. } | FromItem::Join { .. } => false,
5855 }
5856}
5857
5858fn lateral_outer_scope(
5866 left_scope: &[ScopedTable],
5867 base: usize,
5868 left_width: usize,
5869 outer_scope: &[ScopedTable],
5870) -> Vec<ScopedTable> {
5871 debug_assert!(
5872 left_scope.iter().all(|table| table.start_index >= base),
5873 "a FROM item's left sibling scope must start at the join's own base"
5874 );
5875 left_scope
5876 .iter()
5877 .cloned()
5878 .map(|mut table| {
5879 table.start_index -= base;
5880 table
5881 })
5882 .chain(outer_scope.iter().cloned().map(|mut table| {
5883 table.start_index += left_width;
5884 table
5885 }))
5886 .collect()
5887}
5888
5889fn apply_alias_columns(
5894 alias: &str,
5895 columns: &[String],
5896 schema: &mut [ColumnMetadata],
5897 span: crate::ast::Span,
5898) -> Result<(), PlannerError> {
5899 if columns.is_empty() {
5900 return Ok(());
5901 }
5902 if columns.len() != schema.len() {
5903 return Err(PlannerError::table_alias_column_count_mismatch(
5904 alias,
5905 columns.len(),
5906 schema.len(),
5907 span,
5908 ));
5909 }
5910 let mut names = HashSet::new();
5911 for name in columns {
5912 if !names.insert(name) {
5913 return Err(PlannerError::invalid_expression(format!(
5914 "relation alias '{alias}' declares column '{name}' more than once"
5915 )));
5916 }
5917 }
5918 for (column, name) in schema.iter_mut().zip(columns) {
5919 column.name.clone_from(name);
5920 }
5921 Ok(())
5922}
5923
5924fn offset_scope(scope: &[ScopedTable], offset: usize) -> Vec<ScopedTable> {
5925 scope
5926 .iter()
5927 .cloned()
5928 .map(|mut table| {
5929 table.start_index += offset;
5930 table.scope_level += 1;
5931 table
5932 })
5933 .collect()
5934}
5935
5936fn natural_join_columns(
5937 left_schema: &[ColumnMetadata],
5938 right_schema: &[ColumnMetadata],
5939) -> Vec<String> {
5940 let right_names = right_schema
5944 .iter()
5945 .map(|column| column.name.as_str())
5946 .collect::<HashSet<_>>();
5947 left_schema
5948 .iter()
5949 .filter(|left| right_names.contains(left.name.as_str()))
5950 .map(|column| column.name.clone())
5951 .collect()
5952}
5953
5954fn install_base_projection(plan: &mut LogicalPlan, projection: &Projection) {
5955 match plan {
5956 LogicalPlan::Scan {
5957 projection: scan_projection,
5958 ..
5959 } => *scan_projection = projection.clone(),
5960 LogicalPlan::Filter { input, .. } => install_base_projection(input, projection),
5961 _ => {}
5962 }
5963}
5964
5965fn is_aggregate_function(name: &str) -> bool {
5966 matches!(
5967 name.to_ascii_lowercase().as_str(),
5968 "count"
5969 | "sum"
5970 | "total"
5971 | "avg"
5972 | "min"
5973 | "max"
5974 | "group_concat"
5975 | "string_agg"
5976 | "percentile_disc"
5977 )
5978}
5979
5980fn expr_key(expr: &TypedExpr) -> String {
5981 format!("{:?}", expr.kind)
5982}
5983
5984fn distinct_on_expr_signature(expr: &TypedExpr) -> String {
5994 const MARKER: &str = "span: Span { start: Location { line: ";
5995 let rendered = format!("{:?}", expr.kind);
5996 let mut result = String::with_capacity(rendered.len());
5997 let mut rest = rendered.as_str();
5998 while let Some(position) = rest.find(MARKER) {
5999 let after = &rest[position + MARKER.len()..];
6000 match debug_span_tail_length(after) {
6001 Some(consumed) => {
6002 result.push_str(&rest[..position]);
6003 result.push_str("span: _");
6004 rest = &after[consumed..];
6005 }
6006 None => {
6007 let keep = position + MARKER.len();
6008 result.push_str(&rest[..keep]);
6009 rest = &rest[keep..];
6010 }
6011 }
6012 }
6013 result.push_str(rest);
6014 result
6015}
6016
6017fn debug_span_tail_length(input: &str) -> Option<usize> {
6021 fn digits(input: &str, offset: &mut usize) -> bool {
6022 let start = *offset;
6023 while input
6024 .as_bytes()
6025 .get(*offset)
6026 .is_some_and(u8::is_ascii_digit)
6027 {
6028 *offset += 1;
6029 }
6030 *offset > start
6031 }
6032 fn literal(input: &str, offset: &mut usize, expected: &str) -> bool {
6033 if input[*offset..].starts_with(expected) {
6034 *offset += expected.len();
6035 true
6036 } else {
6037 false
6038 }
6039 }
6040 let mut offset = 0;
6041 (digits(input, &mut offset)
6042 && literal(input, &mut offset, ", column: ")
6043 && digits(input, &mut offset)
6044 && literal(input, &mut offset, " }, end: Location { line: ")
6045 && digits(input, &mut offset)
6046 && literal(input, &mut offset, ", column: ")
6047 && digits(input, &mut offset)
6048 && literal(input, &mut offset, " } }"))
6049 .then_some(offset)
6050}
6051
6052fn build_distinct_on_sort_spec(
6069 key_exprs: Vec<TypedExpr>,
6070 user_order_by: Vec<SortExpr>,
6071 base_schema: &[ColumnMetadata],
6072 fallback_span: crate::ast::Span,
6073) -> Result<(usize, Vec<SortExpr>), PlannerError> {
6074 let key_signatures: Vec<String> = key_exprs.iter().map(distinct_on_expr_signature).collect();
6075 let mut consumed = vec![false; key_exprs.len()];
6076 let mut prefix: Vec<SortExpr> = Vec::new();
6077 let mut tail: Vec<SortExpr> = Vec::new();
6078 let mut prefix_ended = false;
6079 for sort in user_order_by {
6080 let signature = distinct_on_expr_signature(&sort.expr);
6081 if let Some(index) = key_signatures
6082 .iter()
6083 .position(|candidate| candidate == &signature)
6084 {
6085 if prefix_ended {
6086 return Err(PlannerError::distinct_on_order_by_mismatch(sort.expr.span));
6089 }
6090 consumed[index] = true;
6091 prefix.push(sort);
6092 } else {
6093 prefix_ended = true;
6094 tail.push(sort);
6095 }
6096 }
6097 let mut implicit: Vec<SortExpr> = Vec::new();
6098 for (index, key) in key_exprs.into_iter().enumerate() {
6099 if consumed[index] {
6100 continue;
6101 }
6102 if prefix_ended {
6103 return Err(PlannerError::distinct_on_order_by_mismatch(key.span));
6106 }
6107 implicit.push(SortExpr::new(key, true, false));
6108 }
6109 let key_count = prefix.len() + implicit.len();
6110 let mut order_by = prefix;
6111 order_by.append(&mut implicit);
6112 order_by.append(&mut tail);
6113 let mut seen_columns: HashSet<usize> = order_by
6114 .iter()
6115 .filter_map(|sort| match &sort.expr.kind {
6116 TypedExprKind::ColumnRef { column_index, .. } => Some(*column_index),
6117 _ => None,
6118 })
6119 .collect();
6120 for (index, column) in base_schema.iter().enumerate() {
6121 if seen_columns.insert(index) {
6122 order_by.push(SortExpr::new(
6123 TypedExpr::column_ref(
6124 String::new(),
6125 column.name.clone(),
6126 index,
6127 column.data_type.clone(),
6128 fallback_span,
6129 ),
6130 true,
6131 false,
6132 ));
6133 }
6134 }
6135 Ok((key_count, order_by))
6136}
6137
6138fn is_order_sensitive_aggregate(name: &str) -> bool {
6140 matches!(
6141 name.to_ascii_lowercase().as_str(),
6142 "group_concat" | "string_agg" | "percentile_disc"
6143 )
6144}
6145
6146fn sort_exprs_key(order_by: &[SortExpr]) -> Option<String> {
6147 if order_by.is_empty() {
6148 return None;
6149 }
6150 Some(
6151 order_by
6152 .iter()
6153 .map(|sort| format!("{}|{}|{}", expr_key(&sort.expr), sort.asc, sort.nulls_first))
6154 .collect::<Vec<_>>()
6155 .join(","),
6156 )
6157}
6158
6159#[allow(clippy::too_many_arguments)]
6160fn aggregate_signature(
6161 name: &str,
6162 distinct: bool,
6163 star: bool,
6164 arg: Option<&TypedExpr>,
6165 separator: Option<&String>,
6166 _expr: &TypedExpr,
6167 filter: Option<&TypedExpr>,
6168 order_by: &[SortExpr],
6169) -> AggregateSignature {
6170 AggregateSignature {
6171 name: name.to_ascii_lowercase(),
6172 distinct,
6173 star,
6174 arg_key: arg.map(expr_key),
6175 separator: separator.cloned(),
6176 filter_key: filter.map(expr_key),
6177 order_key: sort_exprs_key(order_by),
6178 }
6179}
6180
6181fn build_group_key_map(group_keys: &[TypedExpr]) -> HashMap<String, usize> {
6182 let mut map = HashMap::new();
6183 for (idx, key) in group_keys.iter().enumerate() {
6184 map.insert(expr_key(key), idx);
6185 }
6186 map
6187}
6188
6189fn build_aggregate_map(aggregates: &[AggregateExpr]) -> HashMap<AggregateSignature, usize> {
6190 let mut map = HashMap::new();
6191 for (idx, agg) in aggregates.iter().enumerate() {
6192 let (name, separator, star, arg) = match &agg.function {
6193 AggregateFunction::Count => (
6194 "count".to_string(),
6195 None,
6196 agg.arg.is_none(),
6197 agg.arg.as_ref(),
6198 ),
6199 AggregateFunction::Sum => ("sum".to_string(), None, false, agg.arg.as_ref()),
6200 AggregateFunction::Total => ("total".to_string(), None, false, agg.arg.as_ref()),
6201 AggregateFunction::Avg => ("avg".to_string(), None, false, agg.arg.as_ref()),
6202 AggregateFunction::Min => ("min".to_string(), None, false, agg.arg.as_ref()),
6203 AggregateFunction::Max => ("max".to_string(), None, false, agg.arg.as_ref()),
6204 AggregateFunction::GroupConcat { separator } => (
6205 "group_concat".to_string(),
6206 separator.clone(),
6207 false,
6208 agg.arg.as_ref(),
6209 ),
6210 AggregateFunction::StringAgg { separator } => (
6211 "string_agg".to_string(),
6212 separator.clone(),
6213 false,
6214 agg.arg.as_ref(),
6215 ),
6216 AggregateFunction::PercentileDisc { fraction } => (
6217 "percentile_disc".to_string(),
6218 Some(format!("{fraction:?}")),
6219 false,
6220 None,
6221 ),
6222 };
6223 let signature = AggregateSignature {
6224 name,
6225 distinct: agg.distinct,
6226 star,
6227 arg_key: arg.map(expr_key),
6228 separator,
6229 filter_key: agg.filter.as_ref().map(expr_key),
6230 order_key: sort_exprs_key(&agg.order_by),
6231 };
6232 map.insert(signature, idx);
6233 }
6234 map
6235}
6236
6237pub(crate) const GROUPING_ID_COLUMN: &str = "__grouping_id";
6239const MAX_GROUPING_SETS: usize = 4096;
6241const MAX_CUBE_COLUMNS: usize = 12;
6243const MAX_GROUPING_KEYS: usize = 63;
6245
6246struct ExpandedGroupBy {
6248 group_keys: Vec<TypedExpr>,
6249 grouping_sets: Option<Vec<u64>>,
6250}
6251
6252fn grouping_full_mask(key_count: usize) -> u64 {
6253 if key_count == 0 {
6254 0
6255 } else {
6256 (1u64 << key_count) - 1
6257 }
6258}
6259
6260fn is_grouping_function(name: &str) -> bool {
6261 name.eq_ignore_ascii_case("grouping") || name.eq_ignore_ascii_case("grouping_id")
6262}
6263
6264struct GroupingRewrite {
6266 key_index: HashMap<String, usize>,
6268 key_count: usize,
6269 gid_index: usize,
6271 sets_present: bool,
6274}
6275
6276impl GroupingRewrite {
6277 fn new(
6278 group_keys: &[TypedExpr],
6279 aggregates: &[AggregateExpr],
6280 grouping_sets: &Option<Vec<u64>>,
6281 ) -> Self {
6282 let key_index = group_keys
6283 .iter()
6284 .enumerate()
6285 .map(|(index, key)| (expr_key(key), index))
6286 .collect();
6287 Self {
6288 key_index,
6289 key_count: group_keys.len(),
6290 gid_index: group_keys.len() + aggregates.len(),
6291 sets_present: grouping_sets.is_some(),
6292 }
6293 }
6294}
6295
6296fn expr_contains_grouping(expr: &crate::ast::expr::Expr) -> bool {
6298 use crate::ast::expr::ExprKind;
6299
6300 match &expr.kind {
6301 ExprKind::FunctionCall {
6302 name,
6303 args,
6304 order_by,
6305 within_group,
6306 filter,
6307 over,
6308 ..
6309 } => {
6310 is_grouping_function(name)
6311 || args.iter().any(expr_contains_grouping)
6312 || order_by
6313 .iter()
6314 .any(|sort| expr_contains_grouping(&sort.expr))
6315 || within_group
6316 .iter()
6317 .any(|sort| expr_contains_grouping(&sort.expr))
6318 || filter.as_deref().is_some_and(expr_contains_grouping)
6319 || over.as_ref().is_some_and(|window| {
6320 window.partition_by.iter().any(expr_contains_grouping)
6321 || window
6322 .order_by
6323 .iter()
6324 .any(|sort| expr_contains_grouping(&sort.expr))
6325 })
6326 }
6327 ExprKind::BinaryOp { left, right, .. } => {
6328 expr_contains_grouping(left) || expr_contains_grouping(right)
6329 }
6330 ExprKind::UnaryOp { operand, .. } => expr_contains_grouping(operand),
6331 ExprKind::TruthPredicate { expr, .. } => expr_contains_grouping(expr),
6332 ExprKind::IsDistinctFrom { left, right, .. } => {
6333 expr_contains_grouping(left) || expr_contains_grouping(right)
6334 }
6335 ExprKind::Row { items } => items.iter().any(expr_contains_grouping),
6336 ExprKind::Case {
6337 operand,
6338 branches,
6339 else_expr,
6340 } => {
6341 operand.as_deref().is_some_and(expr_contains_grouping)
6342 || branches.iter().any(|branch| {
6343 expr_contains_grouping(&branch.when) || expr_contains_grouping(&branch.then)
6344 })
6345 || else_expr.as_deref().is_some_and(expr_contains_grouping)
6346 }
6347 ExprKind::Cast { expr, .. } | ExprKind::TryCast { expr, .. } => {
6348 expr_contains_grouping(expr)
6349 }
6350 ExprKind::Between {
6351 expr, low, high, ..
6352 } => {
6353 expr_contains_grouping(expr)
6354 || expr_contains_grouping(low)
6355 || expr_contains_grouping(high)
6356 }
6357 ExprKind::Like {
6358 expr,
6359 pattern,
6360 escape,
6361 ..
6362 } => {
6363 expr_contains_grouping(expr)
6364 || expr_contains_grouping(pattern)
6365 || escape.as_deref().is_some_and(expr_contains_grouping)
6366 }
6367 ExprKind::InList { expr, list, .. } => {
6368 expr_contains_grouping(expr) || list.iter().any(expr_contains_grouping)
6369 }
6370 ExprKind::IsNull { expr, .. } => expr_contains_grouping(expr),
6371 ExprKind::ScalarSubquery { .. }
6372 | ExprKind::InSubquery { .. }
6373 | ExprKind::Exists { .. }
6374 | ExprKind::Quantified { .. }
6375 | ExprKind::Literal { .. }
6376 | ExprKind::VectorLiteral { .. }
6377 | ExprKind::ColumnRef { .. } => false,
6378 }
6379}
6380
6381fn typed_expr_contains_grouping(expr: &TypedExpr) -> bool {
6382 match &expr.kind {
6383 TypedExprKind::FunctionCall {
6384 name,
6385 args,
6386 filter,
6387 order_by,
6388 over,
6389 ..
6390 } => {
6391 is_grouping_function(name)
6392 || args.iter().any(typed_expr_contains_grouping)
6393 || filter.as_deref().is_some_and(typed_expr_contains_grouping)
6394 || order_by
6395 .iter()
6396 .any(|sort| typed_expr_contains_grouping(&sort.expr))
6397 || over.as_ref().is_some_and(|window| {
6398 window.partition_by.iter().any(typed_expr_contains_grouping)
6399 || window
6400 .order_by
6401 .iter()
6402 .any(|sort| typed_expr_contains_grouping(&sort.expr))
6403 })
6404 }
6405 TypedExprKind::BinaryOp { left, right, .. } => {
6406 typed_expr_contains_grouping(left) || typed_expr_contains_grouping(right)
6407 }
6408 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_grouping(operand),
6409 TypedExprKind::Cast { expr, .. } | TypedExprKind::TryCast { expr, .. } => {
6410 typed_expr_contains_grouping(expr)
6411 }
6412 TypedExprKind::Case {
6413 operand,
6414 branches,
6415 else_expr,
6416 } => {
6417 operand.as_deref().is_some_and(typed_expr_contains_grouping)
6418 || branches.iter().any(|branch| {
6419 typed_expr_contains_grouping(&branch.when)
6420 || typed_expr_contains_grouping(&branch.then)
6421 })
6422 || else_expr
6423 .as_deref()
6424 .is_some_and(typed_expr_contains_grouping)
6425 }
6426 TypedExprKind::Between {
6427 expr, low, high, ..
6428 } => {
6429 typed_expr_contains_grouping(expr)
6430 || typed_expr_contains_grouping(low)
6431 || typed_expr_contains_grouping(high)
6432 }
6433 TypedExprKind::Like {
6434 expr,
6435 pattern,
6436 escape,
6437 ..
6438 } => {
6439 typed_expr_contains_grouping(expr)
6440 || typed_expr_contains_grouping(pattern)
6441 || escape
6442 .as_ref()
6443 .is_some_and(|inner| typed_expr_contains_grouping(inner))
6444 }
6445 TypedExprKind::InList { expr, list, .. } => {
6446 typed_expr_contains_grouping(expr) || list.iter().any(typed_expr_contains_grouping)
6447 }
6448 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_grouping(expr),
6449 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_grouping(expr),
6450 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_grouping(expr),
6451 TypedExprKind::Literal(_)
6452 | TypedExprKind::VectorLiteral(_)
6453 | TypedExprKind::ColumnRef { .. }
6454 | TypedExprKind::ScalarSubquery(_)
6455 | TypedExprKind::Exists { .. } => false,
6456 }
6457}
6458
6459fn bigint_literal(value: u64, span: crate::ast::Span) -> TypedExpr {
6460 TypedExpr {
6461 kind: TypedExprKind::Literal(Literal::Number(value.to_string())),
6462 resolved_type: ResolvedType::BigInt,
6463 span,
6464 }
6465}
6466
6467fn bigint_binary_op(
6468 left: TypedExpr,
6469 op: crate::ast::expr::BinaryOp,
6470 right: TypedExpr,
6471 span: crate::ast::Span,
6472) -> TypedExpr {
6473 TypedExpr {
6474 kind: TypedExprKind::BinaryOp {
6475 left: Box::new(left),
6476 op,
6477 right: Box::new(right),
6478 },
6479 resolved_type: ResolvedType::BigInt,
6480 span,
6481 }
6482}
6483
6484fn lower_grouping_call(
6492 args: &[TypedExpr],
6493 context: &GroupingRewrite,
6494 span: crate::ast::Span,
6495) -> Result<TypedExpr, PlannerError> {
6496 use crate::ast::expr::BinaryOp as AstBinaryOp;
6497
6498 if args.is_empty() {
6499 return Err(PlannerError::invalid_expression(
6500 "GROUPING requires at least one argument".to_string(),
6501 ));
6502 }
6503 if args.len() > MAX_GROUPING_KEYS {
6504 return Err(PlannerError::invalid_expression(format!(
6505 "GROUPING accepts at most {MAX_GROUPING_KEYS} arguments"
6506 )));
6507 }
6508 let mut key_positions = Vec::with_capacity(args.len());
6509 for arg in args {
6510 let Some(&position) = context.key_index.get(&expr_key(arg)) else {
6511 return Err(PlannerError::invalid_expression(
6512 "arguments to GROUPING must be grouping expressions of the query".to_string(),
6513 ));
6514 };
6515 key_positions.push(position);
6516 }
6517
6518 if !context.sets_present {
6519 return Ok(bigint_literal(0, span));
6521 }
6522
6523 let argument_count = key_positions.len();
6524 let mut sum: Option<TypedExpr> = None;
6525 for (argument, key_position) in key_positions.into_iter().enumerate() {
6526 let gid_ref = TypedExpr::column_ref(
6527 "__agg__".to_string(),
6528 GROUPING_ID_COLUMN.to_string(),
6529 context.gid_index,
6530 ResolvedType::BigInt,
6531 span,
6532 );
6533 let excluded_shift = (context.key_count - 1 - key_position) as u32;
6534 let bit = bigint_binary_op(
6535 bigint_binary_op(
6536 gid_ref,
6537 AstBinaryOp::Div,
6538 bigint_literal(1u64 << excluded_shift, span),
6539 span,
6540 ),
6541 AstBinaryOp::Mod,
6542 bigint_literal(2, span),
6543 span,
6544 );
6545 let weight = 1u64 << (argument_count - 1 - argument);
6546 let term = if weight == 1 {
6547 bit
6548 } else {
6549 bigint_binary_op(bit, AstBinaryOp::Mul, bigint_literal(weight, span), span)
6550 };
6551 sum = Some(match sum {
6552 None => term,
6553 Some(current) => bigint_binary_op(current, AstBinaryOp::Add, term, span),
6554 });
6555 }
6556 Ok(sum.expect("GROUPING argument list is non-empty"))
6557}
6558
6559fn rewrite_grouping_calls(
6566 expr: &TypedExpr,
6567 context: &GroupingRewrite,
6568) -> Result<TypedExpr, PlannerError> {
6569 let rebuild = |inner: &TypedExpr| rewrite_grouping_calls(inner, context);
6570 let rebuild_box = |inner: &TypedExpr| -> Result<Box<TypedExpr>, PlannerError> {
6571 Ok(Box::new(rewrite_grouping_calls(inner, context)?))
6572 };
6573 let kind = match &expr.kind {
6574 TypedExprKind::FunctionCall {
6575 name,
6576 args,
6577 distinct,
6578 star,
6579 filter,
6580 order_by,
6581 over,
6582 } => {
6583 if is_grouping_function(name) {
6584 if over.is_some() {
6585 return Err(PlannerError::invalid_expression(
6586 "GROUPING cannot be used as a window function".to_string(),
6587 ));
6588 }
6589 return lower_grouping_call(args, context, expr.span);
6590 }
6591 if over.is_none() && is_aggregate_function(name) {
6592 if args.iter().any(typed_expr_contains_grouping)
6593 || filter.as_deref().is_some_and(typed_expr_contains_grouping)
6594 || order_by
6595 .iter()
6596 .any(|sort| typed_expr_contains_grouping(&sort.expr))
6597 {
6598 return Err(PlannerError::invalid_expression(
6599 "GROUPING cannot appear inside aggregate function arguments".to_string(),
6600 ));
6601 }
6602 return Ok(expr.clone());
6603 }
6604 TypedExprKind::FunctionCall {
6605 name: name.clone(),
6606 args: args.iter().map(rebuild).collect::<Result<Vec<_>, _>>()?,
6607 distinct: *distinct,
6608 star: *star,
6609 filter: filter.as_deref().map(rebuild_box).transpose()?,
6610 order_by: order_by
6611 .iter()
6612 .map(|sort| {
6613 Ok(SortExpr::new(
6614 rebuild(&sort.expr)?,
6615 sort.asc,
6616 sort.nulls_first,
6617 ))
6618 })
6619 .collect::<Result<Vec<_>, PlannerError>>()?,
6620 over: over
6621 .as_ref()
6622 .map(|window| {
6623 Ok(typed_expr::TypedWindowSpec {
6624 partition_by: window
6625 .partition_by
6626 .iter()
6627 .map(rebuild)
6628 .collect::<Result<Vec<_>, _>>()?,
6629 order_by: window
6630 .order_by
6631 .iter()
6632 .map(|sort| {
6633 Ok(SortExpr::new(
6634 rebuild(&sort.expr)?,
6635 sort.asc,
6636 sort.nulls_first,
6637 ))
6638 })
6639 .collect::<Result<Vec<_>, PlannerError>>()?,
6640 frame: window.frame.clone(),
6641 })
6642 })
6643 .transpose()
6644 .map_err(|error: PlannerError| error)?,
6645 }
6646 }
6647 TypedExprKind::BinaryOp { left, op, right } => TypedExprKind::BinaryOp {
6648 left: rebuild_box(left)?,
6649 op: *op,
6650 right: rebuild_box(right)?,
6651 },
6652 TypedExprKind::UnaryOp { op, operand } => TypedExprKind::UnaryOp {
6653 op: *op,
6654 operand: rebuild_box(operand)?,
6655 },
6656 TypedExprKind::Case {
6657 operand,
6658 branches,
6659 else_expr,
6660 } => TypedExprKind::Case {
6661 operand: operand.as_deref().map(rebuild_box).transpose()?,
6662 branches: branches
6663 .iter()
6664 .map(|branch| {
6665 Ok(TypedCaseWhen {
6666 when: rebuild(&branch.when)?,
6667 then: rebuild(&branch.then)?,
6668 })
6669 })
6670 .collect::<Result<Vec<_>, PlannerError>>()?,
6671 else_expr: else_expr.as_deref().map(rebuild_box).transpose()?,
6672 },
6673 TypedExprKind::Cast {
6674 expr: inner,
6675 target_type,
6676 } => TypedExprKind::Cast {
6677 expr: rebuild_box(inner)?,
6678 target_type: target_type.clone(),
6679 },
6680 TypedExprKind::TryCast {
6681 expr: inner,
6682 target_type,
6683 } => TypedExprKind::TryCast {
6684 expr: rebuild_box(inner)?,
6685 target_type: target_type.clone(),
6686 },
6687 TypedExprKind::Between {
6688 expr: inner,
6689 low,
6690 high,
6691 negated,
6692 } => TypedExprKind::Between {
6693 expr: rebuild_box(inner)?,
6694 low: rebuild_box(low)?,
6695 high: rebuild_box(high)?,
6696 negated: *negated,
6697 },
6698 TypedExprKind::Like {
6699 expr: inner,
6700 pattern,
6701 escape,
6702 negated,
6703 kind,
6704 } => TypedExprKind::Like {
6705 expr: rebuild_box(inner)?,
6706 pattern: rebuild_box(pattern)?,
6707 escape: escape.as_deref().map(rebuild_box).transpose()?,
6708 negated: *negated,
6709 kind: *kind,
6710 },
6711 TypedExprKind::InList {
6712 expr: inner,
6713 list,
6714 negated,
6715 } => TypedExprKind::InList {
6716 expr: rebuild_box(inner)?,
6717 list: list.iter().map(rebuild).collect::<Result<Vec<_>, _>>()?,
6718 negated: *negated,
6719 },
6720 TypedExprKind::IsNull {
6721 expr: inner,
6722 negated,
6723 } => TypedExprKind::IsNull {
6724 expr: rebuild_box(inner)?,
6725 negated: *negated,
6726 },
6727 TypedExprKind::InSubquery {
6728 expr: inner,
6729 subquery,
6730 negated,
6731 } => TypedExprKind::InSubquery {
6732 expr: rebuild_box(inner)?,
6733 subquery: subquery.clone(),
6734 negated: *negated,
6735 },
6736 TypedExprKind::Quantified {
6737 expr: inner,
6738 op,
6739 quantifier,
6740 subquery,
6741 } => TypedExprKind::Quantified {
6742 expr: rebuild_box(inner)?,
6743 op: *op,
6744 quantifier: *quantifier,
6745 subquery: subquery.clone(),
6746 },
6747 TypedExprKind::Literal(_)
6748 | TypedExprKind::VectorLiteral(_)
6749 | TypedExprKind::ColumnRef { .. }
6750 | TypedExprKind::ScalarSubquery(_)
6751 | TypedExprKind::Exists { .. } => return Ok(expr.clone()),
6752 };
6753 Ok(TypedExpr {
6754 kind,
6755 resolved_type: expr.resolved_type.clone(),
6756 span: expr.span,
6757 })
6758}
6759
6760fn build_aggregate_schema(
6761 group_keys: &[TypedExpr],
6762 aggregates: &[AggregateExpr],
6763) -> Vec<ColumnMetadata> {
6764 let mut schema = Vec::new();
6765 for (idx, key) in group_keys.iter().enumerate() {
6766 let name = match &key.kind {
6767 TypedExprKind::ColumnRef { column, .. } => column.clone(),
6768 _ => format!("group_{idx}"),
6769 };
6770 schema.push(ColumnMetadata::new(name, key.resolved_type.clone()));
6771 }
6772 for (idx, agg) in aggregates.iter().enumerate() {
6773 let name = match &agg.function {
6774 AggregateFunction::Count => format!("count_{idx}"),
6775 AggregateFunction::Sum => format!("sum_{idx}"),
6776 AggregateFunction::Total => format!("total_{idx}"),
6777 AggregateFunction::Avg => format!("avg_{idx}"),
6778 AggregateFunction::Min => format!("min_{idx}"),
6779 AggregateFunction::Max => format!("max_{idx}"),
6780 AggregateFunction::GroupConcat { .. } => format!("group_concat_{idx}"),
6781 AggregateFunction::StringAgg { .. } => format!("string_agg_{idx}"),
6782 AggregateFunction::PercentileDisc { .. } => format!("percentile_disc_{idx}"),
6783 };
6784 schema.push(ColumnMetadata::new(name, agg.result_type.clone()));
6785 }
6786 schema
6787}
6788
6789fn rewrite_expr_with_maps(
6790 expr: &TypedExpr,
6791 group_key_map: &HashMap<String, usize>,
6792 aggregate_map: &HashMap<AggregateSignature, usize>,
6793 output_names: &[String],
6794) -> Result<TypedExpr, PlannerError> {
6795 let group_key_count = output_names.len().saturating_sub(aggregate_map.len());
6796 let key = expr_key(expr);
6797 if let Some(idx) = group_key_map.get(&key) {
6798 return Ok(make_output_column_ref(
6799 *idx,
6800 output_names,
6801 expr.resolved_type.clone(),
6802 expr.span,
6803 ));
6804 }
6805
6806 match &expr.kind {
6807 TypedExprKind::FunctionCall {
6808 name,
6809 args,
6810 distinct,
6811 star,
6812 filter,
6813 order_by,
6814 over: None,
6815 } if is_aggregate_function(name) => {
6816 let is_percentile = name.eq_ignore_ascii_case("percentile_disc");
6817 let separator = if name.eq_ignore_ascii_case("group_concat") && args.len() == 2 {
6818 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
6819 Some(value.clone())
6820 } else {
6821 return Err(PlannerError::invalid_expression(
6822 "GROUP_CONCAT separator must be a string literal".to_string(),
6823 ));
6824 }
6825 } else if name.eq_ignore_ascii_case("string_agg") && args.len() == 2 {
6826 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
6827 Some(value.clone())
6828 } else {
6829 return Err(PlannerError::invalid_expression(
6830 "STRING_AGG separator must be a string literal".to_string(),
6831 ));
6832 }
6833 } else if is_percentile && args.len() == 1 {
6834 Some(format!(
6835 "{:?}",
6836 type_checker::percentile_fraction(&args[0])?
6837 ))
6838 } else {
6839 None
6840 };
6841 let signature = AggregateSignature {
6842 name: name.to_ascii_lowercase(),
6843 distinct: *distinct,
6844 star: *star,
6845 arg_key: if is_percentile {
6846 None
6847 } else {
6848 args.first().map(expr_key)
6849 },
6850 separator,
6851 filter_key: filter.as_deref().map(expr_key),
6852 order_key: if is_order_sensitive_aggregate(name) {
6853 sort_exprs_key(order_by)
6854 } else {
6855 None
6856 },
6857 };
6858 let idx = aggregate_map.get(&signature).ok_or_else(|| {
6859 PlannerError::invalid_expression(
6860 "aggregate in expression is not part of plan".to_string(),
6861 )
6862 })?;
6863 let output_index = group_key_count + idx;
6864 Ok(make_output_column_ref(
6865 output_index,
6866 output_names,
6867 expr.resolved_type.clone(),
6868 expr.span,
6869 ))
6870 }
6871 TypedExprKind::FunctionCall {
6872 name,
6873 args,
6874 distinct,
6875 star,
6876 filter,
6877 order_by,
6878 over,
6879 } => {
6880 if over.is_none() && (*distinct || *star) {
6881 return Err(PlannerError::invalid_expression(
6882 "DISTINCT/STAR modifiers are only supported for aggregates".to_string(),
6883 ));
6884 }
6885 if filter.is_some() || !order_by.is_empty() {
6886 return Err(PlannerError::invalid_expression(
6890 "aggregate in expression is not part of plan".to_string(),
6891 ));
6892 }
6893 let mut rewritten_args = Vec::with_capacity(args.len());
6894 for arg in args {
6895 rewritten_args.push(rewrite_expr_with_maps(
6896 arg,
6897 group_key_map,
6898 aggregate_map,
6899 output_names,
6900 )?);
6901 }
6902 let over = over
6903 .as_ref()
6904 .map(|window| {
6905 let partition_by = window
6906 .partition_by
6907 .iter()
6908 .map(|expr| {
6909 rewrite_expr_with_maps(expr, group_key_map, aggregate_map, output_names)
6910 })
6911 .collect::<Result<Vec<_>, PlannerError>>()?;
6912 let order_by = window
6913 .order_by
6914 .iter()
6915 .map(|sort| {
6916 Ok(SortExpr::new(
6917 rewrite_expr_with_maps(
6918 &sort.expr,
6919 group_key_map,
6920 aggregate_map,
6921 output_names,
6922 )?,
6923 sort.asc,
6924 sort.nulls_first,
6925 ))
6926 })
6927 .collect::<Result<Vec<_>, PlannerError>>()?;
6928 Ok(crate::planner::typed_expr::TypedWindowSpec {
6929 partition_by,
6930 order_by,
6931 frame: window.frame.clone(),
6932 })
6933 })
6934 .transpose()?;
6935 Ok(TypedExpr {
6936 kind: TypedExprKind::FunctionCall {
6937 name: name.clone(),
6938 args: rewritten_args,
6939 distinct: *distinct,
6940 star: *star,
6941 filter: None,
6942 order_by: Vec::new(),
6943 over,
6944 },
6945 resolved_type: expr.resolved_type.clone(),
6946 span: expr.span,
6947 })
6948 }
6949 TypedExprKind::BinaryOp { left, op, right } => {
6950 let left = rewrite_expr_with_maps(left, group_key_map, aggregate_map, output_names)?;
6951 let right = rewrite_expr_with_maps(right, group_key_map, aggregate_map, output_names)?;
6952 Ok(TypedExpr {
6953 kind: TypedExprKind::BinaryOp {
6954 left: Box::new(left),
6955 op: *op,
6956 right: Box::new(right),
6957 },
6958 resolved_type: expr.resolved_type.clone(),
6959 span: expr.span,
6960 })
6961 }
6962 TypedExprKind::UnaryOp { op, operand } => {
6963 let operand =
6964 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)?;
6965 Ok(TypedExpr {
6966 kind: TypedExprKind::UnaryOp {
6967 op: *op,
6968 operand: Box::new(operand),
6969 },
6970 resolved_type: expr.resolved_type.clone(),
6971 span: expr.span,
6972 })
6973 }
6974 TypedExprKind::Case {
6975 operand,
6976 branches,
6977 else_expr,
6978 } => {
6979 let operand = operand
6980 .as_deref()
6981 .map(|operand| {
6982 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)
6983 .map(Box::new)
6984 })
6985 .transpose()?;
6986 let mut rewritten_branches = Vec::with_capacity(branches.len());
6987 for branch in branches {
6988 rewritten_branches.push(TypedCaseWhen {
6989 when: rewrite_expr_with_maps(
6990 &branch.when,
6991 group_key_map,
6992 aggregate_map,
6993 output_names,
6994 )?,
6995 then: rewrite_expr_with_maps(
6996 &branch.then,
6997 group_key_map,
6998 aggregate_map,
6999 output_names,
7000 )?,
7001 });
7002 }
7003 let else_expr = else_expr
7004 .as_deref()
7005 .map(|else_expr| {
7006 rewrite_expr_with_maps(else_expr, group_key_map, aggregate_map, output_names)
7007 .map(Box::new)
7008 })
7009 .transpose()?;
7010 Ok(TypedExpr {
7011 kind: TypedExprKind::Case {
7012 operand,
7013 branches: rewritten_branches,
7014 else_expr,
7015 },
7016 resolved_type: expr.resolved_type.clone(),
7017 span: expr.span,
7018 })
7019 }
7020 TypedExprKind::Between {
7021 expr: inner,
7022 low,
7023 high,
7024 negated,
7025 } => {
7026 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7027 let low = rewrite_expr_with_maps(low, group_key_map, aggregate_map, output_names)?;
7028 let high = rewrite_expr_with_maps(high, group_key_map, aggregate_map, output_names)?;
7029 Ok(TypedExpr {
7030 kind: TypedExprKind::Between {
7031 expr: Box::new(inner),
7032 low: Box::new(low),
7033 high: Box::new(high),
7034 negated: *negated,
7035 },
7036 resolved_type: expr.resolved_type.clone(),
7037 span: expr.span,
7038 })
7039 }
7040 TypedExprKind::Like {
7041 expr: inner,
7042 pattern,
7043 escape,
7044 negated,
7045 kind,
7046 } => {
7047 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7048 let pattern =
7049 rewrite_expr_with_maps(pattern, group_key_map, aggregate_map, output_names)?;
7050 let escape = if let Some(esc) = escape {
7051 Some(Box::new(rewrite_expr_with_maps(
7052 esc,
7053 group_key_map,
7054 aggregate_map,
7055 output_names,
7056 )?))
7057 } else {
7058 None
7059 };
7060 Ok(TypedExpr {
7061 kind: TypedExprKind::Like {
7062 expr: Box::new(inner),
7063 pattern: Box::new(pattern),
7064 escape,
7065 negated: *negated,
7066 kind: *kind,
7067 },
7068 resolved_type: expr.resolved_type.clone(),
7069 span: expr.span,
7070 })
7071 }
7072 TypedExprKind::InList {
7073 expr: inner,
7074 list,
7075 negated,
7076 } => {
7077 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7078 let mut rewritten_list = Vec::with_capacity(list.len());
7079 for item in list {
7080 rewritten_list.push(rewrite_expr_with_maps(
7081 item,
7082 group_key_map,
7083 aggregate_map,
7084 output_names,
7085 )?);
7086 }
7087 Ok(TypedExpr {
7088 kind: TypedExprKind::InList {
7089 expr: Box::new(inner),
7090 list: rewritten_list,
7091 negated: *negated,
7092 },
7093 resolved_type: expr.resolved_type.clone(),
7094 span: expr.span,
7095 })
7096 }
7097 TypedExprKind::IsNull {
7098 expr: inner,
7099 negated,
7100 } => {
7101 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7102 Ok(TypedExpr {
7103 kind: TypedExprKind::IsNull {
7104 expr: Box::new(inner),
7105 negated: *negated,
7106 },
7107 resolved_type: expr.resolved_type.clone(),
7108 span: expr.span,
7109 })
7110 }
7111 TypedExprKind::Literal(_) | TypedExprKind::VectorLiteral(_) => Ok(expr.clone()),
7112 TypedExprKind::ColumnRef { table, .. } if table == "__agg__" => Ok(expr.clone()),
7115 TypedExprKind::ColumnRef { .. } => Err(PlannerError::invalid_expression(
7116 "column reference must appear in GROUP BY or be aggregated".to_string(),
7117 )),
7118 TypedExprKind::Cast {
7119 expr: inner,
7120 target_type,
7121 } => {
7122 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7123 Ok(TypedExpr {
7124 kind: TypedExprKind::Cast {
7125 expr: Box::new(inner),
7126 target_type: target_type.clone(),
7127 },
7128 resolved_type: expr.resolved_type.clone(),
7129 span: expr.span,
7130 })
7131 }
7132 TypedExprKind::TryCast {
7133 expr: inner,
7134 target_type,
7135 } => {
7136 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7137 Ok(TypedExpr {
7138 kind: TypedExprKind::TryCast {
7139 expr: Box::new(inner),
7140 target_type: target_type.clone(),
7141 },
7142 resolved_type: expr.resolved_type.clone(),
7143 span: expr.span,
7144 })
7145 }
7146 TypedExprKind::ScalarSubquery(_)
7147 | TypedExprKind::InSubquery { .. }
7148 | TypedExprKind::Exists { .. }
7149 | TypedExprKind::Quantified { .. } => Ok(expr.clone()),
7150 }
7151}
7152
7153fn make_output_column_ref(
7154 index: usize,
7155 output_names: &[String],
7156 resolved_type: ResolvedType,
7157 span: crate::ast::Span,
7158) -> TypedExpr {
7159 let name = output_names
7160 .get(index)
7161 .cloned()
7162 .unwrap_or_else(|| format!("col_{index}"));
7163 TypedExpr::column_ref("__agg__".to_string(), name, index, resolved_type, span)
7164}